From cabb604195221ed9da35132a8c5a167a5660a5cb Mon Sep 17 00:00:00 2001 From: Krisztian Litkey Date: Tue, 16 Jul 2024 15:31:49 +0300 Subject: [PATCH 1/4] kubernetes: split out client setup code from agent. Split out kubernetes client setup code from agent, to make it available explicitly and more generally for any kind of plugins, not just resource management ones. Signed-off-by: Krisztian Litkey --- pkg/agent/agent.go | 60 ++++++------- pkg/kubernetes/client/client.go | 149 ++++++++++++++++++++++++++++++++ 2 files changed, 174 insertions(+), 35 deletions(-) create mode 100644 pkg/kubernetes/client/client.go diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 9f63cbae9..7af43f808 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -22,6 +22,7 @@ import ( "os" "sync" + "github.com/containers/nri-plugins/pkg/kubernetes/client" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -33,7 +34,6 @@ import ( "github.com/containers/nri-plugins/pkg/agent/podresapi" "github.com/containers/nri-plugins/pkg/agent/watch" cfgapi "github.com/containers/nri-plugins/pkg/apis/config/v1alpha1" - k8sclient "k8s.io/client-go/kubernetes" logger "github.com/containers/nri-plugins/pkg/log" ) @@ -127,12 +127,11 @@ type Agent struct { kubeConfig string // kubeconfig path configFile string // configuration file to use instead of custom resource - cfgIf ConfigInterface // custom resource access interface - httpCli *http.Client // shared HTTP client - k8sCli *k8sclient.Clientset // kubernetes client - nrtCli *nrtapi.Client // NRT custom resources client - nrtLock sync.Mutex // serialize NRT custom resource updates - podResCli *podresapi.Client // pod resources API client + cfgIf ConfigInterface // custom resource access interface + k8sCli *client.Client // kubernetes client + nrtCli *nrtapi.Client // NRT custom resources client + nrtLock sync.Mutex // serialize NRT custom resource updates + podResCli *podresapi.Client // pod resources API client notifyFn NotifyFn // config resource change notification callback nodeWatch watch.Interface // kubernetes node watch @@ -295,7 +294,8 @@ func (a *Agent) configure(newConfig metav1.Object) { log.Error("failed to setup NRT client: %w", err) break } - cli, err := nrtapi.NewForConfigAndClient(cfg, a.httpCli) + + cli, err := nrtapi.NewForConfigAndClient(cfg, a.k8sCli.HttpClient()) if err != nil { log.Error("failed to setup NRT client: %w", err) break @@ -331,37 +331,28 @@ func (a *Agent) hasLocalConfig() bool { } func (a *Agent) setupClients() error { + var err error + if a.hasLocalConfig() { log.Warn("running with local configuration, skipping cluster access client setup...") return nil } - // Create HTTP/REST client and K8s client on initial startup. Any failure - // to create these is a failure start up. - if a.httpCli == nil { - log.Info("setting up HTTP/REST client...") - restCfg, err := a.getRESTConfig() - if err != nil { - return err - } - - a.httpCli, err = rest.HTTPClientFor(restCfg) - if err != nil { - return fmt.Errorf("failed to setup kubernetes HTTP client: %w", err) - } + a.k8sCli, err = client.New(client.WithKubeOrInClusterConfig(a.kubeConfig)) + if err != nil { + return err + } - log.Info("setting up K8s client...") - a.k8sCli, err = k8sclient.NewForConfigAndClient(restCfg, a.httpCli) - if err != nil { - a.cleanupClients() - return fmt.Errorf("failed to setup kubernetes client: %w", err) - } + a.nrtCli, err = nrtapi.NewForConfigAndClient(a.k8sCli.RestConfig(), a.k8sCli.HttpClient()) + if err != nil { + a.cleanupClients() + return fmt.Errorf("failed to setup NRT client: %w", err) + } - kubeCfg := *restCfg - err = a.cfgIf.SetKubeClient(a.httpCli, &kubeCfg) - if err != nil { - return fmt.Errorf("failed to setup kubernetes config resource client: %w", err) - } + err = a.cfgIf.SetKubeClient(a.k8sCli.HttpClient(), a.k8sCli.RestConfig()) + if err != nil { + a.cleanupClients() + return fmt.Errorf("failed to setup kubernetes config resource client: %w", err) } a.configure(a.currentCfg) @@ -370,10 +361,9 @@ func (a *Agent) setupClients() error { } func (a *Agent) cleanupClients() { - if a.httpCli != nil { - a.httpCli.CloseIdleConnections() + if a.k8sCli != nil { + a.k8sCli.Close() } - a.httpCli = nil a.k8sCli = nil a.nrtCli = nil } diff --git a/pkg/kubernetes/client/client.go b/pkg/kubernetes/client/client.go new file mode 100644 index 000000000..0e50a575f --- /dev/null +++ b/pkg/kubernetes/client/client.go @@ -0,0 +1,149 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package client + +import ( + "net/http" + + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" +) + +// Option is an option that can be applied to a Client. +type Option func(*Client) error + +// Client enacapsulates our Kubernetes client. +type Client struct { + cfg *rest.Config + http *http.Client + *kubernetes.Clientset +} + +// GetConfigForFile returns a REST configuration for the given file. +func GetConfigForFile(kubeConfig string) (*rest.Config, error) { + return clientcmd.BuildConfigFromFlags("", kubeConfig) +} + +// InClusterConfig returns the in-cluster REST configuration. +func InClusterConfig() (*rest.Config, error) { + return rest.InClusterConfig() +} + +// WithKubeConfig returns a Client Option for using the given kubeconfig file. +func WithKubeConfig(file string) Option { + return func(c *Client) error { + cfg, err := GetConfigForFile(file) + if err != nil { + return err + } + return WithRestConfig(cfg)(c) + } +} + +// WithInClusterConfig returns a Client Option for using the in-cluster configuration. +func WithInClusterConfig() Option { + return func(c *Client) error { + cfg, err := rest.InClusterConfig() + if err != nil { + return err + } + return WithRestConfig(cfg)(c) + } +} + +// WithKubeOrInClusterConfig returns a Client Option for using in-cluster configuration +// if a configuration file is not given. +func WithKubeOrInClusterConfig(file string) Option { + if file == "" { + return WithInClusterConfig() + } + return WithKubeConfig(file) +} + +// WithRestConfig returns a Client Option for using the given REST configuration. +func WithRestConfig(cfg *rest.Config) Option { + return func(c *Client) error { + c.cfg = cfg + return nil + } +} + +// WithHttpClient returns a Client Option for using/sharing the given HTTP client. +func WithHttpClient(hc *http.Client) Option { + return func(c *Client) error { + c.http = hc + return nil + } +} + +// New creates a new Client with the given options. +func New(options ...Option) (*Client, error) { + c := &Client{} + + for _, o := range options { + if err := o(c); err != nil { + return nil, err + } + } + + if c.cfg == nil { + if err := WithInClusterConfig()(c); err != nil { + return nil, err + } + } + + if c.http == nil { + hc, err := rest.HTTPClientFor(c.cfg) + if err != nil { + return nil, err + } + c.http = hc + } + + client, err := kubernetes.NewForConfigAndClient(c.cfg, c.http) + if err != nil { + return nil, err + } + c.Clientset = client + + return c, nil +} + +// RestConfig returns a shallow copy of the REST configuration of the Client. +func (c *Client) RestConfig() *rest.Config { + cfg := *c.cfg + return &cfg +} + +// HttpClient returns the HTTP client of the Client. +func (c *Client) HttpClient() *http.Client { + return c.http +} + +// K8sClient returns the K8s Clientset of the Client. +func (c *Client) K8sClient() *kubernetes.Clientset { + return c.Clientset +} + +// Close closes the Client. +func (c *Client) Close() { + if c.http != nil { + c.http.CloseIdleConnections() + } + c.cfg = nil + c.http = nil + c.Clientset = nil +} From 041c16258195facbc2374520fffa81fef1233df0 Mon Sep 17 00:00:00 2001 From: Krisztian Litkey Date: Wed, 4 Jun 2025 09:21:57 +0300 Subject: [PATCH 2/4] kubernetes: allow setting accepted and wire content types. Allow setting the content types a client accepts and the type it uses on the wire. Provide constants for JSON and protobuf content types. Signed-off-by: Krisztian Litkey --- pkg/kubernetes/client/client.go | 49 +++++++++++++++++++++++++++++++-- 1 file changed, 47 insertions(+), 2 deletions(-) diff --git a/pkg/kubernetes/client/client.go b/pkg/kubernetes/client/client.go index 0e50a575f..c60243514 100644 --- a/pkg/kubernetes/client/client.go +++ b/pkg/kubernetes/client/client.go @@ -15,7 +15,9 @@ package client import ( + "errors" "net/http" + "strings" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -76,7 +78,7 @@ func WithKubeOrInClusterConfig(file string) Option { // WithRestConfig returns a Client Option for using the given REST configuration. func WithRestConfig(cfg *rest.Config) Option { return func(c *Client) error { - c.cfg = cfg + c.cfg = rest.CopyConfig(cfg) return nil } } @@ -89,13 +91,50 @@ func WithHttpClient(hc *http.Client) Option { } } +// WithAcceptContentTypes returns a Client Option for setting the accepted content types. +func WithAcceptContentTypes(contentTypes ...string) Option { + return func(c *Client) error { + if c.cfg == nil { + return errRetryWhenConfigSet + } + c.cfg.AcceptContentTypes = strings.Join(contentTypes, ",") + return nil + } +} + +// WithContentType returns a Client Option for setting the wire format content type. +func WithContentType(contentType string) Option { + return func(c *Client) error { + if c.cfg == nil { + return errRetryWhenConfigSet + } + c.cfg.ContentType = contentType + return nil + } +} + +const ( + ContentTypeJSON = "application/json" + ContentTypeProtobuf = "application/vnd.kubernetes.protobuf" +) + +var ( + // returned by options if applied too early, before a configuration is set + errRetryWhenConfigSet = errors.New("retry when client config is set") +) + // New creates a new Client with the given options. func New(options ...Option) (*Client, error) { c := &Client{} + var retry []Option for _, o := range options { if err := o(c); err != nil { - return nil, err + if err == errRetryWhenConfigSet { + retry = append(retry, o) + } else { + return nil, err + } } } @@ -105,6 +144,12 @@ func New(options ...Option) (*Client, error) { } } + for _, o := range retry { + if err := o(c); err != nil { + return nil, err + } + } + if c.http == nil { hc, err := rest.HTTPClientFor(c.cfg) if err != nil { From 66b4d4ae4679657caa933f7c81d59cd88c324ade Mon Sep 17 00:00:00 2001 From: Krisztian Litkey Date: Wed, 17 Jul 2024 19:36:53 +0300 Subject: [PATCH 3/4] kubernetes: split out watch wrapper from agent. Split out kubernetes watch wrapper and setup code from agent, to make it available explicitly and more generally for other kinds of plugins than just resource management ones. Signed-off-by: Krisztian Litkey --- pkg/kubernetes/watch/file.go | 178 +++++++++++++++++++++++++++++++++ pkg/kubernetes/watch/object.go | 160 +++++++++++++++++++++++++++++ pkg/kubernetes/watch/watch.go | 38 +++++++ 3 files changed, 376 insertions(+) create mode 100644 pkg/kubernetes/watch/file.go create mode 100644 pkg/kubernetes/watch/object.go create mode 100644 pkg/kubernetes/watch/watch.go diff --git a/pkg/kubernetes/watch/file.go b/pkg/kubernetes/watch/file.go new file mode 100644 index 000000000..bd28c9689 --- /dev/null +++ b/pkg/kubernetes/watch/file.go @@ -0,0 +1,178 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package watch + +import ( + "errors" + "io/fs" + "os" + "path" + "path/filepath" + "sync" + + "github.com/fsnotify/fsnotify" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + k8swatch "k8s.io/apimachinery/pkg/watch" +) + +// FileClient takes care of unmarshaling file content to a target object. +type FileClient interface { + Unmarshal([]byte, string) (runtime.Object, error) +} + +type fileWatch struct { + dir string + file string + client FileClient + fsw *fsnotify.Watcher + resultC chan Event + stopOnce sync.Once + stopC chan struct{} + doneC chan struct{} +} + +// File creates a k8s-compatible watch for the given file. Contents of the +// file are unmarshalled into the object of choice by the client. The file +// is monitored for changes and corresponding watch events are generated. +func File(client FileClient, file string) (Interface, error) { + absPath, err := filepath.Abs(file) + if err != nil { + return nil, err + } + + fsw, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + if err = fsw.Add(filepath.Dir(absPath)); err != nil { + return nil, err + } + + fw := &fileWatch{ + dir: filepath.Dir(absPath), + file: filepath.Base(absPath), + client: client, + fsw: fsw, + resultC: make(chan Event, k8swatch.DefaultChanSize), + stopC: make(chan struct{}), + doneC: make(chan struct{}), + } + + obj, err := fw.readObject() + if err != nil && !errors.Is(err, fs.ErrNotExist) { + return nil, err + } + + fw.sendEvent(Added, obj) + + go fw.run() + + return fw, nil +} + +// Stop stops the watch. +func (w *fileWatch) Stop() { + w.stopOnce.Do(func() { + close(w.stopC) + <-w.doneC + w.stopC = nil + }) +} + +// ResultChan returns the channel for receiving events from the watch. +func (w *fileWatch) ResultChan() <-chan Event { + return w.resultC +} + +func (w *fileWatch) run() { + for { + select { + case <-w.stopC: + w.fsw.Close() + close(w.resultC) + close(w.doneC) + return + + case e, ok := <-w.fsw.Events: + log.Debug("%s got event %+v", w.name(), e) + + if !ok { + w.sendEvent( + Error, + &metav1.Status{ + Status: metav1.StatusFailure, + Message: "failed to receive fsnotify event", + }, + ) + close(w.resultC) + close(w.doneC) + return + } + + if path.Base(e.Name) != w.file { + continue + } + + switch { + case (e.Op & (fsnotify.Create | fsnotify.Write)) != 0: + obj, err := w.readObject() + if err != nil { + log.Debug("%s failed to read/unmarshal: %v", w.name(), err) + continue + } + + if (e.Op & fsnotify.Create) != 0 { + w.sendEvent(Added, obj) + } else { + w.sendEvent(Added, obj) + } + + case (e.Op & (fsnotify.Remove | fsnotify.Rename)) != 0: + w.sendEvent(Deleted, nil) + } + } + } +} + +func (w *fileWatch) sendEvent(t EventType, obj runtime.Object) { + select { + case w.resultC <- Event{Type: t, Object: obj}: + default: + log.Warn("failed to deliver fileWatch %v event", t) + } +} + +func (w *fileWatch) readObject() (runtime.Object, error) { + file := path.Join(w.dir, w.file) + data, err := os.ReadFile(file) + if err != nil { + return nil, err + } + + obj, err := w.client.Unmarshal(data, file) + if err != nil { + return nil, err + } + + log.Debug("%s read object %+v", w.name(), obj) + + return obj, nil +} + +func (w *fileWatch) name() string { + return path.Join("filewatch/path:", path.Join(w.dir, w.file)) +} diff --git a/pkg/kubernetes/watch/object.go b/pkg/kubernetes/watch/object.go new file mode 100644 index 000000000..eefe3fed4 --- /dev/null +++ b/pkg/kubernetes/watch/object.go @@ -0,0 +1,160 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package watch + +import ( + "path" + "sync" + "time" + + k8swatch "k8s.io/apimachinery/pkg/watch" +) + +const ( + createDelay = 1 * time.Second +) + +type watch struct { + sync.Mutex + subject string + client ObjectClient + w Interface + resultC chan Event + createC <-chan time.Time + + stopOnce sync.Once + stopC chan struct{} + doneC chan struct{} +} + +// ObjectClient takes care of low-level details of creating a wrapped watch. +type ObjectClient interface { + CreateWatch() (Interface, error) +} + +// Object creates a wrapped watch using the given client. The watch +// is transparently recreated upon expiration and any errors. +func Object(client ObjectClient, subject string) (Interface, error) { + w := &watch{ + subject: subject, + client: client, + resultC: make(chan Event, k8swatch.DefaultChanSize), + stopC: make(chan struct{}), + doneC: make(chan struct{}), + } + + if err := w.create(); err != nil { + return nil, err + } + + go w.run() + + return w, nil +} + +// Stop stops the watch. +func (w *watch) Stop() { + w.stopOnce.Do(func() { + close(w.stopC) + <-w.doneC + w.stop() + w.stopC = nil + }) +} + +// ResultChan returns the channel for receiving events from the watch. +func (w *watch) ResultChan() <-chan Event { + return w.resultC +} + +func (w *watch) eventChan() <-chan Event { + if w.w == nil { + return nil + } + + return w.w.ResultChan() +} + +func (w *watch) run() { + for { + select { + case <-w.stopC: + close(w.resultC) + close(w.doneC) + return + + case e, ok := <-w.eventChan(): + if !ok { + log.Debug("%s failed...", w.name()) + w.stop() + w.scheduleCreate() + continue + } + + if e.Type == Error { + log.Debug("%s failed with an error...", w.name()) + w.stop() + w.scheduleCreate() + continue + } + + select { + case w.resultC <- e: + default: + log.Debug("%s failed to deliver %v event", w.name(), e.Type) + w.stop() + w.scheduleCreate() + } + + case <-w.createC: + w.createC = nil + log.Debug("reopening %s...", w.name()) + if err := w.create(); err != nil { + w.scheduleCreate() + } + } + } +} + +func (w *watch) create() error { + w.stop() + + wif, err := w.client.CreateWatch() + if err != nil { + return err + } + w.w = wif + + return nil +} + +func (w *watch) scheduleCreate() { + if w.createC == nil { + w.createC = time.After(createDelay) + } +} + +func (w *watch) stop() { + if w.w == nil { + return + } + + w.w.Stop() + w.w = nil +} + +func (w *watch) name() string { + return path.Join("watch", w.subject) +} diff --git a/pkg/kubernetes/watch/watch.go b/pkg/kubernetes/watch/watch.go new file mode 100644 index 000000000..095215211 --- /dev/null +++ b/pkg/kubernetes/watch/watch.go @@ -0,0 +1,38 @@ +// Copyright The NRI Plugins Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package watch + +import ( + logger "github.com/containers/nri-plugins/pkg/log" + k8swatch "k8s.io/apimachinery/pkg/watch" +) + +type ( + Interface = k8swatch.Interface + EventType = k8swatch.EventType + Event = k8swatch.Event +) + +const ( + Added = k8swatch.Added + Modified = k8swatch.Modified + Deleted = k8swatch.Deleted + Bookmark = k8swatch.Bookmark + Error = k8swatch.Error +) + +var ( + log = logger.Get("watch") +) From 0e0686b38e974d448a9892b0451d27d1c2e7ac4f Mon Sep 17 00:00:00 2001 From: Krisztian Litkey Date: Wed, 4 Jun 2025 09:20:32 +0300 Subject: [PATCH 4/4] agent: expose node name, kube client and config. Add functions for retrieving from the agent the node name, the kubernetes client, and the REST configuration in use. Signed-off-by: Krisztian Litkey --- pkg/agent/agent.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 7af43f808..f97766f94 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -263,6 +263,18 @@ func (a *Agent) Stop() { } } +func (a *Agent) NodeName() string { + return a.nodeName +} + +func (a *Agent) KubeClient() *client.Client { + return a.k8sCli +} + +func (a *Agent) KubeConfig() string { + return a.kubeConfig +} + var ( defaultConfig = &cfgapi.AgentConfig{ NodeResourceTopology: true,