Refactor kubernetes client

pkg/agent/agent.go

@@ -29,13 +29,12 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/clientcmd"
 
 	nrtapi "github.com/containers/nri-plugins/pkg/agent/nrtapi"
 	"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"
+	"github.com/containers/nri-plugins/pkg/kubernetes/client"
+	"github.com/containers/nri-plugins/pkg/kubernetes/watch"
 
 	logger "github.com/containers/nri-plugins/pkg/log"
 )
@@ -129,12 +128,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    // wrapped kubernetes client + REST config + HTTP 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
@@ -300,12 +298,11 @@ func (a *Agent) configure(newConfig metav1.Object) {
 	switch {
 	case cfg.NodeResourceTopology && a.nrtCli == nil:
 		log.Infof("enabling NRT client")
-		cfg, err := a.getRESTConfig()
-		if err != nil {
-			log.Errorf("failed to setup NRT client: %v", err)
+		if a.k8sCli == nil {
+			log.Errorf("failed to setup NRT client: no kubernetes client")
 			break
 		}
-		cli, err := nrtapi.NewForConfigAndClient(cfg, a.httpCli)
+		cli, err := nrtapi.NewForConfigAndClient(a.k8sCli.RestConfig(), a.k8sCli.HttpClient())
 		if err != nil {
 			log.Errorf("failed to setup NRT client: %v", err)
 			break
@@ -346,30 +343,17 @@ func (a *Agent) setupClients() error {
 		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.Infof("setting up HTTP/REST client...")
-		restCfg, err := a.getRESTConfig()
-		if err != nil {
-			return err
-		}
-
-		a.httpCli, err = rest.HTTPClientFor(restCfg)
+	// Create the kubernetes client on initial startup. Any failure is fatal.
+	if a.k8sCli == nil {
+		log.Infof("setting up kubernetes client...")
+		c, err := client.New(client.WithKubeOrInClusterConfig(a.kubeConfig))
 		if err != nil {
-			return fmt.Errorf("failed to setup kubernetes HTTP client: %w", err)
-		}
-
-		log.Infof("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.k8sCli = c
 
-		kubeCfg := *restCfg
-		err = a.cfgIf.SetKubeClient(a.httpCli, &kubeCfg)
-		if err != nil {
+		if err := a.cfgIf.SetKubeClient(a.k8sCli.HttpClient(), a.k8sCli.RestConfig()); err != nil {
+			a.cleanupClients()
 			return fmt.Errorf("failed to setup kubernetes config resource client: %w", err)
 		}
 	}
@@ -380,31 +364,35 @@ func (a *Agent) setupClients() error {
 }
 
 func (a *Agent) cleanupClients() {
-	if a.httpCli != nil {
-		a.httpCli.CloseIdleConnections()
-	}
-	a.httpCli = nil
+	a.k8sCli.Close()
 	a.k8sCli = nil
 	a.nrtCli = nil
 }
 
-func (a *Agent) getRESTConfig() (*rest.Config, error) {
-	var (
-		cfg *rest.Config
-		err error
-	)
+// NodeName returns the kubernetes node name this agent is running on.
+func (a *Agent) NodeName() string {
+	return a.nodeName
+}
 
-	if a.kubeConfig == "" {
-		cfg, err = rest.InClusterConfig()
-	} else {
-		cfg, err = clientcmd.BuildConfigFromFlags("", a.kubeConfig)
-	}
+// KubeClient returns the shared kubernetes client wrapper. Returns nil
+// before setupClients has run successfully.
+func (a *Agent) KubeClient() *client.Client {
+	return a.k8sCli
+}
 
-	if err != nil {
-		return nil, fmt.Errorf("failed to get kubernetes REST client config: %w", err)
-	}
+// KubeConfig returns the kubeconfig file path this agent was configured
+// with. Returns the empty string when running with in-cluster credentials.
+func (a *Agent) KubeConfig() string {
+	return a.kubeConfig
+}
 
-	return cfg, err
+// RestConfig returns a copy of the REST config used by the shared
+// kubernetes client, or nil before setupClients has run successfully.
+func (a *Agent) RestConfig() *rest.Config {
+	if a.k8sCli == nil {
+		return nil
+	}
+	return a.k8sCli.RestConfig()
 }
 
 func (a *Agent) setupNodeWatch() error {

pkg/kubernetes/client/client.go

@@ -0,0 +1,219 @@
+/*
+Copyright The NRI Plugins Authors.
+
+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 builds a *kubernetes.Clientset from a kubeconfig file or
+// in-cluster credentials, and exposes the REST config and HTTP client it
+// was built from so callers can share one client.
+package client
+
+import (
+	"errors"
+	"net/http"
+	"strings"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+// Wire content types accepted by the Kubernetes API server.
+const (
+	ContentTypeJSON     = "application/json"
+	ContentTypeProtobuf = "application/vnd.kubernetes.protobuf"
+)
+
+// Client wraps a Kubernetes clientset together with the REST config and
+// HTTP client it was built from. Use the embedded Clientset directly for
+// API calls, or HttpClient()/RestConfig() to build other clients sharing
+// the same transport.
+type Client struct {
+	cfg  *rest.Config
+	http *http.Client
+	*kubernetes.Clientset
+}
+
+// Option configures a Client during construction via New. Options apply
+// in order; config-dependent options (WithContentType,
+// WithAcceptContentTypes) require a config-source option (WithKubeConfig,
+// WithInClusterConfig, or WithRestConfig) earlier in the list.
+type Option func(*Client) error
+
+// errNoConfigSet is returned by options that require the REST config
+// to be present but are called before any config-source option.
+var errNoConfigSet = errors.New("option requires REST config; pass a config-source option (WithKubeConfig, WithInClusterConfig, or WithRestConfig) before this option")
+
+// GetConfigForFile returns a REST configuration parsed from the given
+// kubeconfig file path. Thin wrapper over clientcmd.BuildConfigFromFlags
+// exposed for callers that need a config but not a full Client.
+func GetConfigForFile(kubeConfig string) (*rest.Config, error) {
+	return clientcmd.BuildConfigFromFlags("", kubeConfig)
+}
+
+// InClusterConfig returns the REST configuration for the pod's service
+// account, if the process is running inside a Kubernetes cluster.
+// Returns rest.ErrNotInCluster (wrapped) when not in a cluster.
+func InClusterConfig() (*rest.Config, error) {
+	return rest.InClusterConfig()
+}
+
+// New constructs a Client by applying the given options in order,
+// defaulting to WithInClusterConfig() if none set a REST config.
+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
+	}
+
+	cs, err := kubernetes.NewForConfigAndClient(c.cfg, c.http)
+	if err != nil {
+		return nil, err
+	}
+	c.Clientset = cs
+
+	return c, nil
+}
+
+// WithKubeConfig returns an Option that resolves the REST config from
+// 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 an Option that resolves the REST config
+// from the pod's service-account credentials.
+func WithInClusterConfig() Option {
+	return func(c *Client) error {
+		cfg, err := InClusterConfig()
+		if err != nil {
+			return err
+		}
+		return WithRestConfig(cfg)(c)
+	}
+}
+
+// WithKubeOrInClusterConfig resolves the REST config from the given
+// kubeconfig file if non-empty, or from in-cluster credentials otherwise.
+func WithKubeOrInClusterConfig(file string) Option {
+	if file == "" {
+		return WithInClusterConfig()
+	}
+	return WithKubeConfig(file)
+}
+
+// WithRestConfig uses a deep copy (via rest.CopyConfig) of the given REST
+// config, so the caller keeps ownership of the original.
+func WithRestConfig(cfg *rest.Config) Option {
+	return func(c *Client) error {
+		if cfg == nil {
+			return errors.New("rest config must not be nil")
+		}
+		c.cfg = rest.CopyConfig(cfg)
+		return nil
+	}
+}
+
+// WithHttpClient returns an Option that uses the given pre-built HTTP
+// client. Useful when multiple components should share one client
+// (and therefore its connection pool).
+func WithHttpClient(hc *http.Client) Option {
+	return func(c *Client) error {
+		c.http = hc
+		return nil
+	}
+}
+
+// WithAcceptContentTypes sets the Accept content types to negotiate with
+// the API server, joined with commas. Requires a config-source option
+// earlier in the list.
+func WithAcceptContentTypes(contentTypes ...string) Option {
+	return func(c *Client) error {
+		if c.cfg == nil {
+			return errNoConfigSet
+		}
+		c.cfg.AcceptContentTypes = strings.Join(contentTypes, ",")
+		return nil
+	}
+}
+
+// WithContentType sets the wire content type used for requests. Requires
+// a config-source option earlier in the list.
+func WithContentType(contentType string) Option {
+	return func(c *Client) error {
+		if c.cfg == nil {
+			return errNoConfigSet
+		}
+		c.cfg.ContentType = contentType
+		return nil
+	}
+}
+
+// RestConfig returns a copy of the Client's REST config. Top-level and
+// value-typed nested fields may be freely overwritten, but nested
+// maps/slices (e.g. TLSClientConfig.CAData) share storage with the
+// Client's internal config and must not be mutated.
+func (c *Client) RestConfig() *rest.Config {
+	return rest.CopyConfig(c.cfg)
+}
+
+// HttpClient returns the Client's underlying HTTP client, e.g. for
+// constructing other clients that share the same transport.
+func (c *Client) HttpClient() *http.Client {
+	return c.http
+}
+
+// K8sClient returns the Client's underlying *kubernetes.Clientset.
+// Callers may alternatively use the embedded Clientset directly on
+// the Client value.
+func (c *Client) K8sClient() *kubernetes.Clientset {
+	return c.Clientset
+}
+
+// Close releases resources held by the Client. Safe to call on a nil or
+// already-closed Client.
+func (c *Client) Close() {
+	if c == nil {
+		return
+	}
+	if c.http != nil {
+		c.http.CloseIdleConnections()
+	}
+	c.cfg = nil
+	c.http = nil
+	c.Clientset = nil
+}