From 2a4321338754a9938d7e3c1660d5ffdacde48e15 Mon Sep 17 00:00:00 2001 From: Sanket Sudake Date: Thu, 21 Apr 2022 16:49:52 +0530 Subject: [PATCH] Use Clientset interface instead of type for Fission/kubernetes clients (#2416) Using interface makes it easy to create a fake client and unit test a specific portion of the code. We should be able to more write unit test and increase coverage of code with this change. Signed-off-by: Sanket Sudake --- cmd/fission-bundle/mqtrigger/mqtrigger.go | 2 +- cmd/preupgradechecks/checks.go | 7 +++--- pkg/buildermgr/buildermgr.go | 2 +- pkg/buildermgr/common.go | 6 ++--- pkg/buildermgr/envwatcher.go | 10 ++++---- pkg/buildermgr/pkgwatcher.go | 10 ++++---- pkg/canaryconfigmgr/canaryConfigMgr.go | 8 +++---- pkg/controller/api.go | 6 ++--- pkg/controller/config.go | 6 ++--- pkg/controller/controller.go | 2 +- pkg/controller/functionApi.go | 2 +- pkg/crd/client.go | 24 ++++++------------- pkg/crd/crd.go | 2 +- pkg/executor/cms/cmhandler.go | 8 +++---- pkg/executor/cms/cmscontroller.go | 8 +++---- pkg/executor/cms/secrethandler.go | 8 +++---- pkg/executor/executor.go | 7 +++--- pkg/executor/executor_test.go | 8 +++---- pkg/executor/executortype/container/common.go | 2 +- .../executortype/container/containermgr.go | 10 ++++---- .../executortype/newdeploy/newdeploy.go | 2 +- .../executortype/newdeploy/newdeploymgr.go | 10 ++++---- pkg/executor/reaper/reaper.go | 4 ++-- pkg/executor/util/util.go | 2 +- pkg/fetcher/fetcher.go | 7 +++--- .../cmd/support/resources/kubernetes.go | 14 +++++------ pkg/fission-cli/util/util.go | 2 +- pkg/healthcheck/healthcheck.go | 4 ++-- pkg/kubewatcher/kubewatcher.go | 10 ++++---- pkg/kubewatcher/main.go | 2 +- pkg/kubewatcher/watchSync.go | 6 ++--- pkg/mqtrigger/mqtmanager.go | 6 ++--- pkg/mqtrigger/scalermanager.go | 16 ++++++------- pkg/router/httpTriggers.go | 10 ++++---- pkg/router/ingress.go | 6 ++--- pkg/router/router.go | 2 +- pkg/storagesvc/archivePruner.go | 3 ++- pkg/timer/main.go | 2 +- pkg/timer/timerSync.go | 6 ++--- 39 files changed, 123 insertions(+), 129 deletions(-) diff --git a/cmd/fission-bundle/mqtrigger/mqtrigger.go b/cmd/fission-bundle/mqtrigger/mqtrigger.go index 92c92d15..50f65b29 100644 --- a/cmd/fission-bundle/mqtrigger/mqtrigger.go +++ b/cmd/fission-bundle/mqtrigger/mqtrigger.go @@ -41,7 +41,7 @@ func Start(ctx context.Context, logger *zap.Logger, routerUrl string) error { return errors.Wrap(err, "failed to get fission or kubernetes client") } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/cmd/preupgradechecks/checks.go b/cmd/preupgradechecks/checks.go index 16eae2c1..36756dc5 100644 --- a/cmd/preupgradechecks/checks.go +++ b/cmd/preupgradechecks/checks.go @@ -32,14 +32,15 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) type ( PreUpgradeTaskClient struct { logger *zap.Logger - fissionClient *crd.FissionClient - k8sClient *kubernetes.Clientset - apiExtClient *apiextensionsclient.Clientset + fissionClient versioned.Interface + k8sClient kubernetes.Interface + apiExtClient apiextensionsclient.Interface fnPodNs string envBuilderNs string } diff --git a/pkg/buildermgr/buildermgr.go b/pkg/buildermgr/buildermgr.go index 33adcb5a..db9081bc 100644 --- a/pkg/buildermgr/buildermgr.go +++ b/pkg/buildermgr/buildermgr.go @@ -38,7 +38,7 @@ func Start(ctx context.Context, logger *zap.Logger, storageSvcUrl string, envBui return errors.Wrap(err, "failed to get fission or kubernetes client") } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/buildermgr/common.go b/pkg/buildermgr/common.go index 137d9c94..0f7d6a97 100644 --- a/pkg/buildermgr/common.go +++ b/pkg/buildermgr/common.go @@ -31,10 +31,10 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/builder" builderClient "github.com/fission/fission/pkg/builder/client" - "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/fetcher" fetcherClient "github.com/fission/fission/pkg/fetcher/client" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) // buildPackage helps to build source package into deployment package. @@ -44,7 +44,7 @@ import ( // 3. Send upload request to fetcher to upload deployment package. // 4. Return upload response and build logs. // *. Return build logs and error if any one of steps above failed. -func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, envBuilderNamespace string, +func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface, envBuilderNamespace string, storageSvcUrl string, pkg *fv1.Package) (uploadResp *fetcher.ArchiveUploadResponse, buildLogs string, err error) { env, err := fissionClient.CoreV1().Environments(pkg.Spec.Environment.Namespace).Get(ctx, pkg.Spec.Environment.Name, metav1.GetOptions{}) @@ -121,7 +121,7 @@ func buildPackage(ctx context.Context, logger *zap.Logger, fissionClient *crd.Fi return uploadResp, buildResp.BuildLogs, nil } -func updatePackage(logger *zap.Logger, fissionClient *crd.FissionClient, +func updatePackage(logger *zap.Logger, fissionClient versioned.Interface, pkg *fv1.Package, status fv1.BuildStatus, buildLogs string, uploadResp *fetcher.ArchiveUploadResponse) (*fv1.Package, error) { diff --git a/pkg/buildermgr/envwatcher.go b/pkg/buildermgr/envwatcher.go index 59fb65d4..7a780976 100644 --- a/pkg/buildermgr/envwatcher.go +++ b/pkg/buildermgr/envwatcher.go @@ -34,9 +34,9 @@ import ( "k8s.io/client-go/kubernetes" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/util" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/utils" ) @@ -82,8 +82,8 @@ type ( cache map[string]*builderInfo requestChan chan envwRequest builderNamespace string - fissionClient *crd.FissionClient - kubernetesClient *kubernetes.Clientset + fissionClient versioned.Interface + kubernetesClient kubernetes.Interface fetcherConfig *fetcherConfig.Config builderImagePullPolicy apiv1.PullPolicy useIstio bool @@ -92,8 +92,8 @@ type ( func makeEnvironmentWatcher( logger *zap.Logger, - fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, + fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, fetcherConfig *fetcherConfig.Config, builderNamespace string) *environmentWatcher { diff --git a/pkg/buildermgr/pkgwatcher.go b/pkg/buildermgr/pkgwatcher.go index 12f3d6b6..c98b5d8e 100644 --- a/pkg/buildermgr/pkgwatcher.go +++ b/pkg/buildermgr/pkgwatcher.go @@ -30,7 +30,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/cache" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/metrics" ) @@ -38,8 +38,8 @@ import ( type ( packageWatcher struct { logger *zap.Logger - fissionClient *crd.FissionClient - k8sClient *kubernetes.Clientset + fissionClient versioned.Interface + k8sClient kubernetes.Interface podInformer *k8sCache.SharedIndexInformer pkgInformer *k8sCache.SharedIndexInformer builderNamespace string @@ -48,7 +48,7 @@ type ( } ) -func makePackageWatcher(logger *zap.Logger, fissionClient *crd.FissionClient, k8sClientSet *kubernetes.Clientset, +func makePackageWatcher(logger *zap.Logger, fissionClient versioned.Interface, k8sClientSet kubernetes.Interface, builderNamespace string, storageSvcUrl string, podInformer *k8sCache.SharedIndexInformer, pkgInformer *k8sCache.SharedIndexInformer) *packageWatcher { pkgw := &packageWatcher{ @@ -333,7 +333,7 @@ func (pkgw *packageWatcher) Run(ctx context.Context) { // setInitialBuildStatus sets initial build status to a package if it is empty. // This normally occurs when the user applies package YAML files that have no status field // through kubectl. -func setInitialBuildStatus(fissionClient *crd.FissionClient, pkg *fv1.Package) (*fv1.Package, error) { +func setInitialBuildStatus(fissionClient versioned.Interface, pkg *fv1.Package) (*fv1.Package, error) { pkg.Status = fv1.PackageStatus{ LastUpdateTimestamp: metav1.Time{Time: time.Now().UTC()}, } diff --git a/pkg/canaryconfigmgr/canaryConfigMgr.go b/pkg/canaryconfigmgr/canaryConfigMgr.go index a285895f..6975c63f 100644 --- a/pkg/canaryconfigmgr/canaryConfigMgr.go +++ b/pkg/canaryconfigmgr/canaryConfigMgr.go @@ -32,7 +32,7 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" ) @@ -42,14 +42,14 @@ const ( type canaryConfigMgr struct { logger *zap.Logger - fissionClient *crd.FissionClient - kubeClient *kubernetes.Clientset + fissionClient versioned.Interface + kubeClient kubernetes.Interface canaryConfigInformer *k8sCache.SharedIndexInformer promClient *PrometheusApiClient canaryCfgCancelFuncMap *canaryConfigCancelFuncMap } -func MakeCanaryConfigMgr(logger *zap.Logger, fissionClient *crd.FissionClient, kubeClient *kubernetes.Clientset, prometheusSvc string) (*canaryConfigMgr, error) { +func MakeCanaryConfigMgr(logger *zap.Logger, fissionClient versioned.Interface, kubeClient kubernetes.Interface, prometheusSvc string) (*canaryConfigMgr, error) { if prometheusSvc == "" { logger.Info("try to retrieve prometheus server information from environment variables") diff --git a/pkg/controller/api.go b/pkg/controller/api.go index b8a44bb2..9792fbfc 100644 --- a/pkg/controller/api.go +++ b/pkg/controller/api.go @@ -31,9 +31,9 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" - "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/fission-cli/logdb" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/info" "github.com/fission/fission/pkg/utils/httpserver" "github.com/fission/fission/pkg/utils/metrics" @@ -52,8 +52,8 @@ func init() { type ( API struct { logger *zap.Logger - fissionClient *crd.FissionClient - kubernetesClient *kubernetes.Clientset + fissionClient versioned.Interface + kubernetesClient kubernetes.Interface storageServiceUrl string builderManagerUrl string workflowApiUrl string diff --git a/pkg/controller/config.go b/pkg/controller/config.go index 346f4fed..04756911 100644 --- a/pkg/controller/config.go +++ b/pkg/controller/config.go @@ -24,11 +24,11 @@ import ( "k8s.io/client-go/kubernetes" "github.com/fission/fission/pkg/canaryconfigmgr" - "github.com/fission/fission/pkg/crd" config "github.com/fission/fission/pkg/featureconfig" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) -func ConfigCanaryFeature(context context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, kubeClient *kubernetes.Clientset, featureConfig *config.FeatureConfig, featureStatus map[string]string) error { +func ConfigCanaryFeature(context context.Context, logger *zap.Logger, fissionClient versioned.Interface, kubeClient kubernetes.Interface, featureConfig *config.FeatureConfig, featureStatus map[string]string) error { // start the appropriate controller if featureConfig.CanaryConfig.IsEnabled { canaryCfgMgr, err := canaryconfigmgr.MakeCanaryConfigMgr(logger, fissionClient, kubeClient, featureConfig.CanaryConfig.PrometheusSvc) @@ -44,7 +44,7 @@ func ConfigCanaryFeature(context context.Context, logger *zap.Logger, fissionCli } // ConfigureFeatures gets the feature config and configures the features that are enabled -func ConfigureFeatures(context context.Context, logger *zap.Logger, unitTestMode bool, fissionClient *crd.FissionClient, kubeClient *kubernetes.Clientset) (map[string]string, error) { +func ConfigureFeatures(context context.Context, logger *zap.Logger, unitTestMode bool, fissionClient versioned.Interface, kubeClient kubernetes.Interface) (map[string]string, error) { // set feature enabled to false if unitTestMode if unitTestMode { return nil, nil diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index a7f44413..653edcd3 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -37,7 +37,7 @@ func Start(ctx context.Context, logger *zap.Logger, port int, unitTestFlag bool, cLogger.Fatal("failed to find fission CRDs", zap.Error(err)) } - err = fc.WaitForCRDs() + err = crd.WaitForCRDs(fc) if err != nil { cLogger.Fatal("error waiting for CRDs", zap.Error(err)) } diff --git a/pkg/controller/functionApi.go b/pkg/controller/functionApi.go index 08e58929..420bf271 100644 --- a/pkg/controller/functionApi.go +++ b/pkg/controller/functionApi.go @@ -339,7 +339,7 @@ func (a *API) FunctionPodLogs(w http.ResponseWriter, r *http.Request) { } } -func getContainerLog(kubernetesClient *kubernetes.Clientset, w http.ResponseWriter, fn *fv1.Function, pod *apiv1.Pod) error { +func getContainerLog(kubernetesClient kubernetes.Interface, w http.ResponseWriter, fn *fv1.Function, pod *apiv1.Pod) error { seq := strings.Repeat("=", 35) for _, container := range pod.Spec.Containers { diff --git a/pkg/crd/client.go b/pkg/crd/client.go index 88735845..49b38ecd 100644 --- a/pkg/crd/client.go +++ b/pkg/crd/client.go @@ -31,20 +31,13 @@ import ( "k8s.io/client-go/tools/clientcmd" metricsclient "k8s.io/metrics/pkg/client/clientset/versioned" - genClientset "github.com/fission/fission/pkg/generated/clientset/versioned" -) - -type ( - // FissionClient exports the client interface to be used - FissionClient struct { - genClientset.Interface - } + "github.com/fission/fission/pkg/generated/clientset/versioned" ) // GetKubernetesClient gets a kubernetes client using the kubeconfig file at the // environment var $KUBECONFIG, or an in-cluster config if that's // undefined. -func GetKubernetesClient() (*rest.Config, *kubernetes.Clientset, *apiextensionsclient.Clientset, *metricsclient.Clientset, error) { +func GetKubernetesClient() (*rest.Config, kubernetes.Interface, apiextensionsclient.Interface, metricsclient.Interface, error) { var config *rest.Config var err error @@ -79,29 +72,26 @@ func GetKubernetesClient() (*rest.Config, *kubernetes.Clientset, *apiextensionsc return config, clientset, apiExtClientset, metricsClient, nil } -func MakeFissionClient() (*FissionClient, *kubernetes.Clientset, *apiextensionsclient.Clientset, *metricsclient.Clientset, error) { +func MakeFissionClient() (versioned.Interface, kubernetes.Interface, apiextensionsclient.Interface, metricsclient.Interface, error) { config, kubeClient, apiExtClient, metricsClient, err := GetKubernetesClient() if err != nil { return nil, nil, nil, nil, err } // make a CRD REST client with the config - crdClient, err := genClientset.NewForConfig(config) + crdClient, err := versioned.NewForConfig(config) if err != nil { return nil, nil, nil, nil, err } - fc := &FissionClient{ - Interface: crdClient, - } - return fc, kubeClient, apiExtClient, metricsClient, nil + return crdClient, kubeClient, apiExtClient, metricsClient, nil } // WaitForCRDs does a timeout to check if CRDs have been installed -func (fc *FissionClient) WaitForCRDs() error { +func WaitForCRDs(fissionClient versioned.Interface) error { start := time.Now() for { - fi := fc.CoreV1().Functions(metav1.NamespaceDefault) + fi := fissionClient.CoreV1().Functions(metav1.NamespaceDefault) _, err := fi.List(context.TODO(), metav1.ListOptions{}) if err != nil { time.Sleep(100 * time.Millisecond) diff --git a/pkg/crd/crd.go b/pkg/crd/crd.go index 729e889d..ac9b1829 100644 --- a/pkg/crd/crd.go +++ b/pkg/crd/crd.go @@ -27,7 +27,7 @@ import ( ) // EnsureFissionCRDs checks if all Fission CRDs are present -func EnsureFissionCRDs(logger *zap.Logger, clientset *apiextensionsclient.Clientset) error { +func EnsureFissionCRDs(logger *zap.Logger, clientset apiextensionsclient.Interface) error { crdsExpected := []string{ "canaryconfigs.fission.io", "environments.fission.io", diff --git a/pkg/executor/cms/cmhandler.go b/pkg/executor/cms/cmhandler.go index a7990eee..45f9cded 100644 --- a/pkg/executor/cms/cmhandler.go +++ b/pkg/executor/cms/cmhandler.go @@ -25,11 +25,11 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) -func getConfigmapRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { +func getConfigmapRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient versioned.Interface) ([]fv1.Function, error) { funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { return nil, err @@ -47,8 +47,8 @@ func getConfigmapRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1 return relatedFunctions, nil } -func ConfigMapEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { +func ConfigMapEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) {}, diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index 876b476e..6ee5e8c7 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -25,8 +25,8 @@ import ( "k8s.io/client-go/kubernetes" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) type ( @@ -34,13 +34,13 @@ type ( ConfigSecretController struct { logger *zap.Logger - fissionClient *crd.FissionClient + fissionClient versioned.Interface } ) // MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions -func MakeConfigSecretController(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType, +func MakeConfigSecretController(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, types map[fv1.ExecutorType]executortype.ExecutorType, configmapInformer informerv1.ConfigMapInformer, secretInformer informerv1.SecretInformer) *ConfigSecretController { logger.Debug("Creating ConfigMap & Secret Controller") diff --git a/pkg/executor/cms/secrethandler.go b/pkg/executor/cms/secrethandler.go index 13a257b4..7c56ebf6 100644 --- a/pkg/executor/cms/secrethandler.go +++ b/pkg/executor/cms/secrethandler.go @@ -25,11 +25,11 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) -func getSecretRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) { +func getSecretRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.ObjectMeta, fissionClient versioned.Interface) ([]fv1.Function, error) { funcList, err := fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) if err != nil { return nil, err @@ -47,8 +47,8 @@ func getSecretRelatedFuncs(ctx context.Context, logger *zap.Logger, m *metav1.Ob return relatedFunctions, nil } -func SecretEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { +func SecretEventHandlers(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, types map[fv1.ExecutorType]executortype.ExecutorType) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) {}, DeleteFunc: func(obj interface{}) {}, diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index c58b438a..1a726752 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -42,6 +42,7 @@ import ( "github.com/fission/fission/pkg/executor/reaper" "github.com/fission/fission/pkg/executor/util" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" + "github.com/fission/fission/pkg/generated/clientset/versioned" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils/metrics" @@ -56,7 +57,7 @@ type ( executorTypes map[fv1.ExecutorType]executortype.ExecutorType cms *cms.ConfigSecretController - fissionClient *crd.FissionClient + fissionClient versioned.Interface requestChan chan *createFuncServiceRequest fsCreateWg sync.Map @@ -76,7 +77,7 @@ type ( // MakeExecutor returns an Executor for given ExecutorType(s). func MakeExecutor(ctx context.Context, logger *zap.Logger, cms *cms.ConfigSecretController, - fissionClient *crd.FissionClient, types map[fv1.ExecutorType]executortype.ExecutorType, + fissionClient versioned.Interface, types map[fv1.ExecutorType]executortype.ExecutorType, informers []k8sCache.SharedIndexInformer) (*Executor, error) { executor := &Executor{ logger: logger.Named("executor"), @@ -257,7 +258,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace st return errors.Wrap(err, "failed to get kubernetes client") } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/executor/executor_test.go b/pkg/executor/executor_test.go index 9c363800..149c3111 100644 --- a/pkg/executor/executor_test.go +++ b/pkg/executor/executor_test.go @@ -51,7 +51,7 @@ func panicIf(err error) { } // return the number of pods in the given namespace matching the given labels -func countPods(kubeClient *kubernetes.Clientset, ns string, labelz map[string]string) int { +func countPods(kubeClient kubernetes.Interface, ns string, labelz map[string]string) int { pods, err := kubeClient.CoreV1().Pods(ns).List(context.TODO(), metav1.ListOptions{ LabelSelector: labels.Set(labelz).AsSelector().String(), }) @@ -61,7 +61,7 @@ func countPods(kubeClient *kubernetes.Clientset, ns string, labelz map[string]st return len(pods.Items) } -func createTestNamespace(kubeClient *kubernetes.Clientset, ns string) { +func createTestNamespace(kubeClient kubernetes.Interface, ns string) { _, err := kubeClient.CoreV1().Namespaces().Create(context.TODO(), &apiv1.Namespace{ ObjectMeta: metav1.ObjectMeta{ Name: ns, @@ -74,7 +74,7 @@ func createTestNamespace(kubeClient *kubernetes.Clientset, ns string) { } // create a nodeport service -func createSvc(kubeClient *kubernetes.Clientset, ns string, name string, targetPort int, nodePort int32, labels map[string]string) *apiv1.Service { +func createSvc(kubeClient kubernetes.Interface, ns string, name string, targetPort int, nodePort int32, labels map[string]string) *apiv1.Service { svc, err := kubeClient.CoreV1().Services(ns).Create(context.TODO(), &apiv1.Service{ ObjectMeta: metav1.ObjectMeta{ Name: name, @@ -148,7 +148,7 @@ func TestExecutor(t *testing.T) { log.Panicf("failed to ensure crds: %v", err) } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { log.Panicf("failed to wait crds: %v", err) } diff --git a/pkg/executor/executortype/container/common.go b/pkg/executor/executortype/container/common.go index 695d5a79..d5dfa0ea 100644 --- a/pkg/executor/executortype/container/common.go +++ b/pkg/executor/executortype/container/common.go @@ -106,7 +106,7 @@ func (cn *Container) cleanupContainer(ctx context.Context, ns string, name strin // identical way to get a value that can reflect resources changed without affecting by the time. // To achieve this goal, the sum of the resource version of all referenced resources is a good fit for our // scenario since the sum of the resource version is always the same as long as no resources changed. -func referencedResourcesRVSum(ctx context.Context, client *kubernetes.Clientset, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { +func referencedResourcesRVSum(ctx context.Context, client kubernetes.Interface, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { rvCount := 0 if len(secrets) > 0 { diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index b5109c71..4c507cb2 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -42,11 +42,11 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" "github.com/fission/fission/pkg/executor/fscache" "github.com/fission/fission/pkg/executor/metrics" "github.com/fission/fission/pkg/executor/reaper" + "github.com/fission/fission/pkg/generated/clientset/versioned" finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" @@ -61,8 +61,8 @@ type ( Container struct { logger *zap.Logger - kubernetesClient *kubernetes.Clientset - fissionClient *crd.FissionClient + kubernetesClient kubernetes.Interface + fissionClient versioned.Interface instanceID string // fetcherConfig *fetcherConfig.Config @@ -88,8 +88,8 @@ type ( func MakeContainer( ctx context.Context, logger *zap.Logger, - fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, + fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, namespace string, instanceID string, funcInformer finformerv1.FunctionInformer, diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index 9e43bb83..2c269831 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -583,7 +583,7 @@ func (deploy *NewDeploy) cleanupNewdeploy(ctx context.Context, ns string, name s // identical way to get a value that can reflect resources changed without affecting by the time. // To achieve this goal, the sum of the resource version of all referenced resources is a good fit for our // scenario since the sum of the resource version is always the same as long as no resources changed. -func referencedResourcesRVSum(ctx context.Context, client *kubernetes.Clientset, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { +func referencedResourcesRVSum(ctx context.Context, client kubernetes.Interface, namespace string, secrets []fv1.SecretReference, cfgmaps []fv1.ConfigMapReference) (int, error) { rvCount := 0 if len(secrets) > 0 { diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index af767ebd..5308b9ca 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -42,12 +42,12 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/executortype" "github.com/fission/fission/pkg/executor/fscache" "github.com/fission/fission/pkg/executor/metrics" "github.com/fission/fission/pkg/executor/reaper" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" + "github.com/fission/fission/pkg/generated/clientset/versioned" finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" @@ -62,8 +62,8 @@ type ( NewDeploy struct { logger *zap.Logger - kubernetesClient *kubernetes.Clientset - fissionClient *crd.FissionClient + kubernetesClient kubernetes.Interface + fissionClient versioned.Interface instanceID string fetcherConfig *fetcherConfig.Config @@ -88,8 +88,8 @@ type ( // MakeNewDeploy initializes and returns an instance of NewDeploy. func MakeNewDeploy( logger *zap.Logger, - fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, + fissionClient versioned.Interface, + kubernetesClient kubernetes.Interface, namespace string, fetcherConfig *fetcherConfig.Config, instanceID string, diff --git a/pkg/executor/reaper/reaper.go b/pkg/executor/reaper/reaper.go index f3551ea1..da828783 100644 --- a/pkg/executor/reaper/reaper.go +++ b/pkg/executor/reaper/reaper.go @@ -27,7 +27,7 @@ import ( "k8s.io/client-go/kubernetes" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/utils" ) @@ -180,7 +180,7 @@ func CleanupHpa(ctx context.Context, logger *zap.Logger, client kubernetes.Inter // CleanupRoleBindings periodically lists rolebindings across all namespaces and removes Service Accounts from them or // deletes the rolebindings completely if there are no Service Accounts in a rolebinding object. -func CleanupRoleBindings(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, fissionClient *crd.FissionClient, functionNs, envBuilderNs string, cleanupRoleBindingInterval time.Duration) { +func CleanupRoleBindings(ctx context.Context, logger *zap.Logger, client kubernetes.Interface, fissionClient versioned.Interface, functionNs, envBuilderNs string, cleanupRoleBindingInterval time.Duration) { for { // some sleep before the next reaper iteration time.Sleep(cleanupRoleBindingInterval) diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index 58df2c13..e1b48880 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -57,7 +57,7 @@ func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) { } // ConvertConfigSecrets returns envFromSource which can be passed directly into the pod spec -func ConvertConfigSecrets(ctx context.Context, fn *fv1.Function, kc *kubernetes.Clientset) ([]apiv1.EnvFromSource, error) { +func ConvertConfigSecrets(ctx context.Context, fn *fv1.Function, kc kubernetes.Interface) ([]apiv1.EnvFromSource, error) { cmList := fn.Spec.ConfigMaps secList := fn.Spec.Secrets diff --git a/pkg/fetcher/fetcher.go b/pkg/fetcher/fetcher.go index aeba19f4..be11487e 100644 --- a/pkg/fetcher/fetcher.go +++ b/pkg/fetcher/fetcher.go @@ -48,6 +48,7 @@ import ( "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/error/network" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/info" storageSvcClient "github.com/fission/fission/pkg/storagesvc/client" "github.com/fission/fission/pkg/utils" @@ -61,8 +62,8 @@ type ( sharedVolumePath string sharedSecretPath string sharedConfigPath string - fissionClient *crd.FissionClient - kubeClient *kubernetes.Clientset + fissionClient versioned.Interface + kubeClient kubernetes.Interface httpClient *http.Client Info PodInfo } @@ -824,7 +825,7 @@ func (fetcher *Fetcher) WsEndHandler(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } -func eventRecorder(kubeClient *kubernetes.Clientset) (record.EventRecorder, error) { +func eventRecorder(kubeClient kubernetes.Interface) (record.EventRecorder, error) { eventBroadcaster := record.NewBroadcaster() eventBroadcaster.StartLogging(zap.S().Infof) eventBroadcaster.StartRecordingToSink( diff --git a/pkg/fission-cli/cmd/support/resources/kubernetes.go b/pkg/fission-cli/cmd/support/resources/kubernetes.go index 9d54057f..ad655fc7 100644 --- a/pkg/fission-cli/cmd/support/resources/kubernetes.go +++ b/pkg/fission-cli/cmd/support/resources/kubernetes.go @@ -43,15 +43,15 @@ const ( // Kubernetes Version type KubernetesVersion struct { - client *kubernetes.Clientset + client kubernetes.Interface } -func NewKubernetesVersion(clientset *kubernetes.Clientset) Resource { +func NewKubernetesVersion(clientset kubernetes.Interface) Resource { return KubernetesVersion{client: clientset} } func (res KubernetesVersion) Dump(dumpDir string) { - serverVer, err := res.client.ServerVersion() + serverVer, err := res.client.Discovery().ServerVersion() if err != nil { console.Error(fmt.Sprintf("Error setting up kubernetes client: %v", err)) return @@ -63,12 +63,12 @@ func (res KubernetesVersion) Dump(dumpDir string) { // Kubernetes Object Dumper type KubernetesObjectDumper struct { - client *kubernetes.Clientset + client kubernetes.Interface objType string selector string } -func NewKubernetesObjectDumper(clientset *kubernetes.Clientset, objType string, selector string) Resource { +func NewKubernetesObjectDumper(clientset kubernetes.Interface, objType string, selector string) Resource { return KubernetesObjectDumper{ client: clientset, objType: objType, @@ -184,11 +184,11 @@ func nodeClean(node corev1.Node) corev1.Node { } type KubernetesPodLogDumper struct { - client *kubernetes.Clientset + client kubernetes.Interface labelSelector string } -func NewKubernetesPodLogDumper(clientset *kubernetes.Clientset, selector string) Resource { +func NewKubernetesPodLogDumper(clientset kubernetes.Interface, selector string) Resource { return KubernetesPodLogDumper{ client: clientset, labelSelector: selector, diff --git a/pkg/fission-cli/util/util.go b/pkg/fission-cli/util/util.go index 0d55793f..4bd3d28a 100644 --- a/pkg/fission-cli/util/util.go +++ b/pkg/fission-cli/util/util.go @@ -105,7 +105,7 @@ func KubifyName(old string) string { // GetKubernetesClient builds a new kubernetes client. If the KUBECONFIG // environment variable is empty or doesn't exist, ~/.kube/config is used for // the kube config path -func GetKubernetesClient(kubeContext string) (*restclient.Config, *kubernetes.Clientset, error) { +func GetKubernetesClient(kubeContext string) (*restclient.Config, kubernetes.Interface, error) { loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() kubeConfigPath := os.Getenv("KUBECONFIG") diff --git a/pkg/healthcheck/healthcheck.go b/pkg/healthcheck/healthcheck.go index e905a617..dd336c10 100644 --- a/pkg/healthcheck/healthcheck.go +++ b/pkg/healthcheck/healthcheck.go @@ -62,7 +62,7 @@ type HealthChecker struct { categories []*Category *Options - kubeAPI *kubernetes.Clientset + kubeAPI kubernetes.Interface fissionNamespace string } @@ -84,7 +84,7 @@ func isCompatibleVersion(minimalRequirementVersion [3]int, actualVersion [3]int) func (hc *HealthChecker) CheckKubeVersion() (err error) { - version, err := hc.kubeAPI.ServerVersion() + version, err := hc.kubeAPI.Discovery().ServerVersion() if err != nil { return err } diff --git a/pkg/kubewatcher/kubewatcher.go b/pkg/kubewatcher/kubewatcher.go index a7ddaacf..eb13acf2 100644 --- a/pkg/kubewatcher/kubewatcher.go +++ b/pkg/kubewatcher/kubewatcher.go @@ -52,7 +52,7 @@ type ( KubeWatcher struct { logger *zap.Logger watches map[types.UID]watchSubscription - kubernetesClient *kubernetes.Clientset + kubernetesClient kubernetes.Interface requestChannel chan *kubeWatcherRequest publisher publisher.Publisher } @@ -63,7 +63,7 @@ type ( kubeWatch watch.Interface lastResourceVersion string stopped *int32 - kubernetesClient *kubernetes.Clientset + kubernetesClient kubernetes.Interface publisher publisher.Publisher } @@ -77,7 +77,7 @@ type ( } ) -func MakeKubeWatcher(ctx context.Context, logger *zap.Logger, kubernetesClient *kubernetes.Clientset, publisher publisher.Publisher) *KubeWatcher { +func MakeKubeWatcher(ctx context.Context, logger *zap.Logger, kubernetesClient kubernetes.Interface, publisher publisher.Publisher) *KubeWatcher { kw := &KubeWatcher{ logger: logger.Named("kube_watcher"), watches: make(map[types.UID]watchSubscription), @@ -149,7 +149,7 @@ func printKubernetesObject(obj runtime.Object, w io.Writer) error { return err } -func createKubernetesWatch(ctx context.Context, kubeClient *kubernetes.Clientset, w *fv1.KubernetesWatchTrigger, resourceVersion string) (watch.Interface, error) { +func createKubernetesWatch(ctx context.Context, kubeClient kubernetes.Interface, w *fv1.KubernetesWatchTrigger, resourceVersion string) (watch.Interface, error) { var wi watch.Interface var err error var watchTimeoutSec int64 = 120 @@ -198,7 +198,7 @@ func (kw *KubeWatcher) removeWatch(w *fv1.KubernetesWatchTrigger) error { return nil } -func MakeWatchSubscription(ctx context.Context, logger *zap.Logger, w *fv1.KubernetesWatchTrigger, kubeClient *kubernetes.Clientset, publisher publisher.Publisher) (*watchSubscription, error) { +func MakeWatchSubscription(ctx context.Context, logger *zap.Logger, w *fv1.KubernetesWatchTrigger, kubeClient kubernetes.Interface, publisher publisher.Publisher) (*watchSubscription, error) { var stopped int32 = 0 ws := &watchSubscription{ logger: logger.Named("watch_subscription"), diff --git a/pkg/kubewatcher/main.go b/pkg/kubewatcher/main.go index 62901482..a76544a7 100644 --- a/pkg/kubewatcher/main.go +++ b/pkg/kubewatcher/main.go @@ -32,7 +32,7 @@ func Start(ctx context.Context, logger *zap.Logger, routerUrl string) error { return errors.Wrap(err, "failed to get fission or kubernetes client") } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/kubewatcher/watchSync.go b/pkg/kubewatcher/watchSync.go index 523bdb65..161ce219 100644 --- a/pkg/kubewatcher/watchSync.go +++ b/pkg/kubewatcher/watchSync.go @@ -23,18 +23,18 @@ import ( "go.uber.org/zap" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) type ( WatchSync struct { logger *zap.Logger - client *crd.FissionClient + client versioned.Interface kubeWatcher *KubeWatcher } ) -func MakeWatchSync(logger *zap.Logger, client *crd.FissionClient, kubeWatcher *KubeWatcher) *WatchSync { +func MakeWatchSync(logger *zap.Logger, client versioned.Interface, kubeWatcher *KubeWatcher) *WatchSync { ws := &WatchSync{ logger: logger.Named("watch_sync"), client: client, diff --git a/pkg/mqtrigger/mqtmanager.go b/pkg/mqtrigger/mqtmanager.go index 4e3a1509..77c9df18 100644 --- a/pkg/mqtrigger/mqtmanager.go +++ b/pkg/mqtrigger/mqtmanager.go @@ -25,7 +25,7 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/mqtrigger/messageQueue" "github.com/fission/fission/pkg/utils/metrics" @@ -44,7 +44,7 @@ type ( logger *zap.Logger reqChan chan request triggers map[string]*triggerSubscription - fissionClient *crd.FissionClient + fissionClient versioned.Interface messageQueueType fv1.MessageQueueType messageQueue messageQueue.MessageQueue } @@ -66,7 +66,7 @@ type ( ) func MakeMessageQueueTriggerManager(logger *zap.Logger, - fissionClient *crd.FissionClient, mqType fv1.MessageQueueType, messageQueue messageQueue.MessageQueue) *MessageQueueTriggerManager { + fissionClient versioned.Interface, mqType fv1.MessageQueueType, messageQueue messageQueue.MessageQueue) *MessageQueueTriggerManager { mqTriggerMgr := MessageQueueTriggerManager{ logger: logger.Named("message_queue_trigger_manager"), reqChan: make(chan request), diff --git a/pkg/mqtrigger/scalermanager.go b/pkg/mqtrigger/scalermanager.go index cfd7e623..38c28981 100644 --- a/pkg/mqtrigger/scalermanager.go +++ b/pkg/mqtrigger/scalermanager.go @@ -66,7 +66,7 @@ func getAuthTriggerClient(namespace string) (dynamic.ResourceInterface, error) { return dynamicClient.Resource(authTriggerGVR).Namespace(namespace), nil } -func mqTriggerEventHandlers(logger *zap.Logger, kubeClient *kubernetes.Clientset, routerURL string) k8sCache.ResourceEventHandlerFuncs { +func mqTriggerEventHandlers(logger *zap.Logger, kubeClient kubernetes.Interface, routerURL string) k8sCache.ResourceEventHandlerFuncs { return k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { go func() { @@ -154,7 +154,7 @@ func StartScalerManager(ctx context.Context, logger *zap.Logger, routerURL strin if err != nil { return err } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } @@ -338,7 +338,7 @@ func getAuthTriggerSpec(mqt *fv1.MessageQueueTrigger, authenticationRef string, return authTriggerObj, nil } -func createAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient *kubernetes.Clientset) error { +func createAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient kubernetes.Interface) error { authTriggerObj, err := getAuthTriggerSpec(mqt, authenticationRef, kubeClient) if err != nil { return err @@ -354,7 +354,7 @@ func createAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, k return nil } -func updateAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient *kubernetes.Clientset) error { +func updateAuthTrigger(mqt *fv1.MessageQueueTrigger, authenticationRef string, kubeClient kubernetes.Interface) error { authTriggerClient, err := getAuthTriggerClient(mqt.ObjectMeta.Namespace) if err != nil { return err @@ -389,7 +389,7 @@ func deleteAuthTrigger(name, namespace string) error { return nil } -func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) (*appsv1.Deployment, error) { +func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient kubernetes.Interface) (*appsv1.Deployment, error) { envVars, err := getEnvVarlist(mqt, routerURL, kubeClient) if err != nil { return nil, err @@ -448,7 +448,7 @@ func getDeploymentSpec(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClien }, nil } -func createDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) error { +func createDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient kubernetes.Interface) error { deployment, err := getDeploymentSpec(mqt, routerURL, kubeClient) if err != nil { return err @@ -460,7 +460,7 @@ func createDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient return nil } -func updateDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient *kubernetes.Clientset) error { +func updateDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient kubernetes.Interface) error { deployment, err := getDeploymentSpec(mqt, routerURL, kubeClient) if err != nil { return err @@ -472,7 +472,7 @@ func updateDeployment(mqt *fv1.MessageQueueTrigger, routerURL string, kubeClient return nil } -func deleteDeployment(name string, namespace string, kubeClient *kubernetes.Clientset) error { +func deleteDeployment(name string, namespace string, kubeClient kubernetes.Interface) error { deletePolicy := metav1.DeletePropagationForeground if err := kubeClient.AppsV1().Deployments(namespace).Delete(context.TODO(), name, metav1.DeleteOptions{ PropagationPolicy: &deletePolicy, diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index 0a0ab146..914d1f2b 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -35,9 +35,9 @@ import ( k8sCache "k8s.io/client-go/tools/cache" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" executorClient "github.com/fission/fission/pkg/executor/client" config "github.com/fission/fission/pkg/featureconfig" + "github.com/fission/fission/pkg/generated/clientset/versioned" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/utils" @@ -54,8 +54,8 @@ type HTTPTriggerSet struct { *mutableRouter logger *zap.Logger - fissionClient *crd.FissionClient - kubeClient *kubernetes.Clientset + fissionClient versioned.Interface + kubeClient kubernetes.Interface executor *executorClient.Client resolver *functionReferenceResolver triggers []fv1.HTTPTrigger @@ -83,8 +83,8 @@ func loadFeatureConfigmap() error { return nil } -func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionClient *crd.FissionClient, - kubeClient *kubernetes.Clientset, executor *executorClient.Client, params *tsRoundTripperParams, isDebugEnv bool, unTapServiceTimeout time.Duration, actionThrottler *throttler.Throttler) *HTTPTriggerSet { +func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionClient versioned.Interface, + kubeClient kubernetes.Interface, executor *executorClient.Client, params *tsRoundTripperParams, isDebugEnv bool, unTapServiceTimeout time.Duration, actionThrottler *throttler.Throttler) *HTTPTriggerSet { httpTriggerSet := &HTTPTriggerSet{ logger: logger.Named("http_trigger_set"), diff --git a/pkg/router/ingress.go b/pkg/router/ingress.go index 59eb5692..cdfd3ebb 100644 --- a/pkg/router/ingress.go +++ b/pkg/router/ingress.go @@ -39,7 +39,7 @@ func init() { } } -func createIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kubernetes.Clientset) { +func createIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient kubernetes.Interface) { if !trigger.Spec.CreateIngress { return } @@ -51,7 +51,7 @@ func createIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kub logger.Debug("created ingress successfully for trigger", zap.String("trigger", trigger.ObjectMeta.Name)) } -func deleteIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kubernetes.Clientset) { +func deleteIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient kubernetes.Interface) { if !trigger.Spec.CreateIngress { return } @@ -71,7 +71,7 @@ func deleteIngress(logger *zap.Logger, trigger *fv1.HTTPTrigger, kubeClient *kub } } -func updateIngress(logger *zap.Logger, oldT *fv1.HTTPTrigger, newT *fv1.HTTPTrigger, kubeClient *kubernetes.Clientset) { +func updateIngress(logger *zap.Logger, oldT *fv1.HTTPTrigger, newT *fv1.HTTPTrigger, kubeClient kubernetes.Interface) { if !oldT.Spec.CreateIngress && !newT.Spec.CreateIngress { return } diff --git a/pkg/router/router.go b/pkg/router/router.go index bd77bacc..8a7f34de 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -127,7 +127,7 @@ func Start(ctx context.Context, logger *zap.Logger, port int, executorURL string logger.Fatal("error connecting to kubernetes API", zap.Error(err)) } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { logger.Fatal("error waiting for CRDs", zap.Error(err)) } diff --git a/pkg/storagesvc/archivePruner.go b/pkg/storagesvc/archivePruner.go index 8293c78c..275ccac4 100644 --- a/pkg/storagesvc/archivePruner.go +++ b/pkg/storagesvc/archivePruner.go @@ -24,11 +24,12 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" ) type ArchivePruner struct { logger *zap.Logger - crdClient *crd.FissionClient + crdClient versioned.Interface archiveChan chan string stowClient *StowClient pruneInterval time.Duration diff --git a/pkg/timer/main.go b/pkg/timer/main.go index d9df1e19..91de9c23 100644 --- a/pkg/timer/main.go +++ b/pkg/timer/main.go @@ -32,7 +32,7 @@ func Start(ctx context.Context, logger *zap.Logger, routerUrl string) error { return errors.Wrap(err, "failed to get fission or kubernetes client") } - err = fissionClient.WaitForCRDs() + err = crd.WaitForCRDs(fissionClient) if err != nil { return errors.Wrap(err, "error waiting for CRDs") } diff --git a/pkg/timer/timerSync.go b/pkg/timer/timerSync.go index 547c2ae4..f0846664 100644 --- a/pkg/timer/timerSync.go +++ b/pkg/timer/timerSync.go @@ -23,19 +23,19 @@ import ( "go.uber.org/zap" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/utils" ) type ( TimerSync struct { logger *zap.Logger - fissionClient *crd.FissionClient + fissionClient versioned.Interface timer *Timer } ) -func MakeTimerSync(ctx context.Context, logger *zap.Logger, fissionClient *crd.FissionClient, timer *Timer) *TimerSync { +func MakeTimerSync(ctx context.Context, logger *zap.Logger, fissionClient versioned.Interface, timer *Timer) *TimerSync { ws := &TimerSync{ logger: logger.Named("timer_sync"), fissionClient: fissionClient,