agent: split out client and watch for other classes of plugins.

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
@@ -264,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,
@@ -295,7 +306,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 +343,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 +373,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
 }

pkg/kubernetes/client/client.go

@@ -0,0 +1,194 @@
+// 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 (
+	"errors"
+	"net/http"
+	"strings"
+
+	"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 = rest.CopyConfig(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
+	}
+}
+
+// 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 {
+			if err == errRetryWhenConfigSet {
+				retry = append(retry, o)
+			} else {
+				return nil, err
+			}
+		}
+	}
+
+	if c.cfg == nil {
+		if err := WithInClusterConfig()(c); err != nil {
+			return nil, err
+		}
+	}
+
+	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 {
+			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
+}