feat(client): add resuable kubernetes client package (#172)

Signed-off-by: AmitKumarDas <[email protected]>

Author: Amit Kumar Das

pkg/kubernetes/retry.go

@@ -37,16 +37,24 @@ func (rt *RetryTimeout) Error() string {
 // Retryable helps executing user provided functions as
 // conditions in a repeated manner till this condition succeeds
 type Retryable struct {
-	Message string
+	//Message string
 
 	WaitTimeout  time.Duration
 	WaitInterval time.Duration
+
+	// RunOnce will run the function only once
+	//
+	// NOTE:
+	// 	In other words, this makes the retry option
+	// as a No Operation i.e. noop
+	RunOnce bool
 }
 
 // RetryConfig helps in creating an instance of Retryable
 type RetryConfig struct {
 	WaitTimeout  *time.Duration
 	WaitInterval *time.Duration
+	RunOnce      bool
 }
 
 // NewRetry returns a new instance of Retryable
@@ -55,34 +63,46 @@ func NewRetry(config RetryConfig) *Retryable {
 	timeout := 60 * time.Second
 	// sleep interval defaults to 1 second
 	interval := 1 * time.Second
+
 	// override timeout with user specified value
 	if config.WaitTimeout != nil {
 		timeout = *config.WaitTimeout
 	}
+
 	// override interval with user specified value
 	if config.WaitInterval != nil {
 		interval = *config.WaitInterval
 	}
+
 	return &Retryable{
 		WaitTimeout:  timeout,
 		WaitInterval: interval,
+		RunOnce:      config.RunOnce,
 	}
 }
 
 // Waitf retries this provided function as a condition till
 // this condition succeeds.
 //
-// Clients invoking this method need to return appropriate
-// values in the function implementation to let this function
-// to be either returned, or exited or retried.
+// NOTE:
+// 	Clients invoking this method need to return appropriate
+// values (i.e. bool & error) within the condition implementation.
+// These return values let the condition to be either returned or
+// retried.
 func (r *Retryable) Waitf(
-	condition func() (bool, error),
-	message string,
-	args ...interface{},
+	condition func() (bool, error), // condition that gets retried
+	msgFormat string,
+	msgArgs ...interface{},
 ) error {
+	if r.RunOnce {
+		// No need to retry if this condition is meant to be run once
+		_, err := condition()
+		return err
+	}
+
 	context := fmt.Sprintf(
-		message,
-		args...,
+		msgFormat,
+		msgArgs...,
 	)
 	// mark the start time
 	start := time.Now()
@@ -130,6 +150,7 @@ func (r *Retryable) Waitf(
 				context,
 			)
 		}
+		// retry after sleeping for specified interval
 		time.Sleep(r.WaitInterval)
 	}
 }

pkg/kubernetes/utility.go

@@ -17,8 +17,12 @@ limitations under the License.
 package kubernetes
 
 import (
+	"sync"
+	"time"
+
 	"github.com/pkg/errors"
 	apiextnv1beta1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1"
+	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
@@ -47,13 +51,14 @@ type UtilityFuncs struct {
 	getAPIForAPIVersionAndResourceFn    func(string, string) *dynamicdiscovery.APIResource
 }
 
-// Utility exposes instances needed to invoke kubernetes
-// api operations
+// Utility provides options to invoke kubernetes APIs
 type Utility struct {
 	*UtilityFuncs
 	Retry *Retryable
 
-	apiDiscovery *dynamicdiscovery.APIResourceDiscovery
+	kubeConfig *rest.Config
+
+	apiResourceDiscovery *dynamicdiscovery.APIResourceDiscovery
 
 	// dynamic client to invoke kubernetes operations
 	// against kubernetes native as well as custom resources
@@ -79,53 +84,62 @@ type Utility struct {
 	err error
 }
 
-func (u *Utility) setTeardownFlag(config UtilityConfig) {
-	u.isTeardown = config.IsTeardown
-}
+var doOnce sync.Once
 
-func (u *Utility) setAPIDiscovery(config UtilityConfig) {
-	u.apiDiscovery = config.APIDiscovery
-}
+// singleton instance of Utility to be used across the
+// project to invoke Kubernetes operations
+var singleton *Utility
 
-func (u *Utility) setCRDClient(config UtilityConfig) {
-	if config.KubeConfig == nil {
-		u.err = errors.Errorf(
-			"Failed to set crd client: Nil kube config provided",
-		)
-		return
+// Singleton returns a new or existing instance of Utility
+func Singleton(config UtilityConfig) (*Utility, error) {
+	if singleton != nil {
+		return singleton, nil
 	}
-	u.crdClient, u.err = apiextnv1beta1.NewForConfig(
-		config.KubeConfig,
-	)
+	var err error
+	doOnce.Do(func() {
+		singleton, err = NewUtility(config)
+	})
+	return singleton, err
 }
 
-func (u *Utility) setDynamicClientset(config UtilityConfig) {
-	u.dynamicClientset, u.err = dynamicclientset.New(
-		config.KubeConfig,
-		config.APIDiscovery,
-	)
-}
-
-func (u *Utility) setKubeClientset(config UtilityConfig) {
-	u.kubeClientset, u.err = kubernetes.NewForConfig(
-		config.KubeConfig,
-	)
-}
-
-// NewUtility returns a new instance of Fixture
+// NewUtility returns a new instance of Kubernetes utility
 func NewUtility(config UtilityConfig) (*Utility, error) {
-	// check retry
+	// initialize retry instance to default behaviour
 	var retry = NewRetry(RetryConfig{})
 	if config.Retry != nil {
+		// override from config if available
 		retry = config.Retry
 	}
+
+	// initialize utility instance
 	u := &Utility{
 		UtilityFuncs: &UtilityFuncs{},
 		Retry:        retry,
+		kubeConfig:   config.KubeConfig,
 	}
+
+	if u.kubeConfig == nil {
+		// Set kube config to in-cluster config
+		u.kubeConfig, u.err = rest.InClusterConfig()
+		if u.err != nil {
+			return nil, errors.Wrapf(
+				u.err,
+				"Failed to create k8s utility",
+			)
+		}
+	}
+
+	// setup options to mutate utility instance
+	// based on the provided config
+	//
+	// NOTE:
+	// 	Following order needs to be maintained
 	var setters = []func(UtilityConfig){
+		// pre settings
 		u.setTeardownFlag,
-		u.setAPIDiscovery,
+		u.setAPIResourceDiscoveryOrDefault,
+
+		// post settings
 		u.setCRDClient,
 		u.setDynamicClientset,
 		u.setKubeClientset,
@@ -139,11 +153,50 @@ func NewUtility(config UtilityConfig) (*Utility, error) {
 	return u, nil
 }
 
-// Teardown deletes resources created through this instance
-func (u *Utility) Teardown() {
-	if !u.isTeardown {
+func (u *Utility) setTeardownFlag(config UtilityConfig) {
+	u.isTeardown = config.IsTeardown
+}
+
+func (u *Utility) setAPIResourceDiscoveryOrDefault(config UtilityConfig) {
+	u.apiResourceDiscovery = config.APIDiscovery
+	if u.apiResourceDiscovery != nil {
 		return
 	}
+	discoveryClient := discovery.NewDiscoveryClientForConfigOrDie(
+		u.kubeConfig,
+	)
+	d := dynamicdiscovery.NewAPIResourceDiscoverer(discoveryClient)
+
+	// This needs to be started with appropriate refresh interval
+	// before being used. We set to 30 seconds discovery interval.
+	// In other words, if new CRDs are applied then it will take
+	// this interval for these new CRDs to be discovered.
+	d.Start(time.Duration(30) * time.Second)
+
+	u.apiResourceDiscovery = d
+}
+
+func (u *Utility) setCRDClient(config UtilityConfig) {
+	u.crdClient, u.err = apiextnv1beta1.NewForConfig(
+		u.kubeConfig,
+	)
+}
+
+func (u *Utility) setDynamicClientset(config UtilityConfig) {
+	u.dynamicClientset, u.err = dynamicclientset.New(
+		u.kubeConfig,
+		u.apiResourceDiscovery, // this must be set previously
+	)
+}
+
+func (u *Utility) setKubeClientset(config UtilityConfig) {
+	u.kubeClientset, u.err = kubernetes.NewForConfig(
+		u.kubeConfig,
+	)
+}
+
+// MustTeardown deletes resources created through this instance
+func (u *Utility) MustTeardown() {
 	// cleanup in descending order
 	for i := len(u.teardownFuncs) - 1; i >= 0; i-- {
 		teardown := u.teardownFuncs[i]
@@ -173,13 +226,31 @@ func (u *Utility) Teardown() {
 	}
 }
 
-// AddToTeardown adds the given teardown function to
+// Teardown optionally deletes resources created through this
+// instance
+func (u *Utility) Teardown() {
+	if !u.isTeardown {
+		return
+	}
+	u.MustTeardown()
+}
+
+// MustAddToTeardown adds the given teardown function to
+// the list of teardown functions
+func (u *Utility) MustAddToTeardown(teardown func() error) {
+	if teardown == nil {
+		return
+	}
+	u.teardownFuncs = append(u.teardownFuncs, teardown)
+}
+
+// AddToTeardown optionally adds the given teardown function to
 // the list of teardown functions
 func (u *Utility) AddToTeardown(teardown func() error) {
 	if !u.isTeardown {
 		return
 	}
-	u.teardownFuncs = append(u.teardownFuncs, teardown)
+	u.MustAddToTeardown(teardown)
 }
 
 // GetClientForAPIVersionAndKind returns the dynamic client for the
@@ -198,32 +269,42 @@ func (u *Utility) GetClientForAPIVersionAndKind(
 }
 
 // GetClientForAPIVersionAndResource returns the dynamic client for the
-// given api version & resource
+// given api version & resource name
 func (u *Utility) GetClientForAPIVersionAndResource(
 	apiversion string,
-	resource string,
+	resourceName string,
 ) (*clientset.ResourceClient, error) {
 	if u.getClientForAPIVersionAndResourceFn != nil {
-		return u.getClientForAPIVersionAndResourceFn(apiversion, resource)
+		return u.getClientForAPIVersionAndResourceFn(apiversion, resourceName)
 	}
 	return u.dynamicClientset.GetClientForAPIVersionAndResource(
 		apiversion,
-		resource,
+		resourceName,
 	)
 }
 
-// GetAPIForAPIVersionAndResource returns the discovered api based
-// on the provided api version & resource
-func (u *Utility) GetAPIForAPIVersionAndResource(
+// GetAPIResourceForAPIVersionAndResourceName returns the
+// discovered API resource based on the provided api version &
+// resource name
+func (u *Utility) GetAPIResourceForAPIVersionAndResourceName(
 	apiversion string,
-	resource string,
+	resourceName string,
 ) *dynamicdiscovery.APIResource {
 	if u.getAPIForAPIVersionAndResourceFn != nil {
-		return u.getAPIForAPIVersionAndResourceFn(apiversion, resource)
+		return u.getAPIForAPIVersionAndResourceFn(apiversion, resourceName)
 	}
-	return u.apiDiscovery.
-		GetAPIForAPIVersionAndResource(
-			apiversion,
-			resource,
-		)
+	return u.apiResourceDiscovery.GetAPIForAPIVersionAndResource(
+		apiversion,
+		resourceName,
+	)
+}
+
+// GetAPIResourceDiscovery returns the api resource discovery instance
+func (u *Utility) GetAPIResourceDiscovery() *dynamicdiscovery.APIResourceDiscovery {
+	return u.apiResourceDiscovery
+}
+
+// GetKubeConfig returns the Kubernetes config instance
+func (u *Utility) GetKubeConfig() *rest.Config {
+	return u.kubeConfig
 }

pkg/recipe/base.go

@@ -16,14 +16,18 @@ limitations under the License.
 
 package recipe
 
-import types "mayadata.io/d-operators/types/recipe"
+import (
+	"mayadata.io/d-operators/pkg/kubernetes"
+	types "mayadata.io/d-operators/types/recipe"
+)
 
 // BaseRunner is the common runner used by all action runners
 type BaseRunner struct {
 	*Fixture
-	TaskIndex    int
-	TaskName     string
-	Retry        *Retryable
+	TaskIndex int
+	TaskName  string
+	//Retry        *Retryable
+	Retry        *kubernetes.Retryable
 	FailFastRule types.FailFastRule
 }