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"
@@ -32,7 +33,6 @@ import (
 	nrtapi "github.com/containers/nri-plugins/pkg/agent/nrtapi"
 	"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"
 )
@@ -126,11 +126,10 @@ 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
+	cfgIf   ConfigInterface // custom resource access interface
+	k8sCli  *client.Client
+	nrtCli  *nrtapi.Client // NRT custom resources client
+	nrtLock sync.Mutex     // serialize NRT custom resource updates
 
 	notifyFn      NotifyFn        // config resource change notification callback
 	nodeWatch     watch.Interface // kubernetes node watch
@@ -267,47 +266,36 @@ func (a *Agent) hasLocalConfig() bool {
 }
 
 func (a *Agent) setupClients() error {
+	var err error
+
 	if a.hasLocalConfig() {
 		return nil
 	}
 
-	cfg, err := a.getRESTConfig()
+	a.k8sCli, err = client.New(client.WithKubeOrInClusterConfig(a.kubeConfig))
 	if err != nil {
 		return err
 	}
 
-	a.httpCli, err = rest.HTTPClientFor(cfg)
-	if err != nil {
-		return fmt.Errorf("failed to setup kubernetes HTTP client: %w", err)
-	}
-
-	a.k8sCli, err = k8sclient.NewForConfigAndClient(cfg, a.httpCli)
-	if err != nil {
-		a.cleanupClients()
-		return fmt.Errorf("failed to setup kubernetes client: %w", err)
-	}
-
-	restCfg := *cfg
-	a.nrtCli, err = nrtapi.NewForConfigAndClient(&restCfg, a.httpCli)
+	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)
 	}
 
-	restCfg = *cfg
-	err = a.cfgIf.SetKubeClient(a.httpCli, &restCfg)
+	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)
 	}
 
 	return nil
 }
 
 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,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
+}