Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use Clientset interface instead of type for Fission/kubernetes clients (#2416)

Browse Source 
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 <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2022-04-21 16:49:52 +05:30
committed by GitHub
parent 3bdbeb6c87
commit 2a43213387
39 changed files with 123 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/fission-bundle/mqtrigger/mqtrigger.go
+1 -1
View File
@@ -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")
}
cmd/preupgradechecks/checks.go
+4 -3
View File
@@ -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
}
pkg/buildermgr/buildermgr.go
+1 -1
View File
@@ -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")
}
pkg/buildermgr/common.go
+3 -3
View File
@@ -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) {
pkg/buildermgr/envwatcher.go
+5 -5
View File
@@ -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 {
pkg/buildermgr/pkgwatcher.go
+5 -5
View File
@@ -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()},
}
pkg/canaryconfigmgr/canaryConfigMgr.go
+4 -4
View File
@@ -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")
pkg/controller/api.go
+3 -3
View File
@@ -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
pkg/controller/config.go
+3 -3
View File
@@ -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
pkg/controller/controller.go
+1 -1
View File
@@ -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))
}
pkg/controller/functionApi.go
+1 -1
View File
@@ -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 {
pkg/crd/client.go
+7 -17
View File
@@ -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)
pkg/crd/crd.go
+1 -1
View File
@@ -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",
pkg/executor/cms/cmhandler.go
+4 -4
View File
@@ -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{}) {},
pkg/executor/cms/cmscontroller.go
+4 -4
View File
@@ -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")
pkg/executor/cms/secrethandler.go
+4 -4
View File
@@ -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{}) {},
pkg/executor/executor.go
+4 -3
View File
@@ -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")
}
pkg/executor/executor_test.go
+4 -4
View File
@@ -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)
}
pkg/executor/executortype/container/common.go
+1 -1
View File
@@ -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 {
pkg/executor/executortype/container/containermgr.go
+5 -5
View File
@@ -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,
pkg/executor/executortype/newdeploy/newdeploy.go
+1 -1
View File
@@ -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 {
pkg/executor/executortype/newdeploy/newdeploymgr.go
+5 -5
View File
@@ -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,
pkg/executor/reaper/reaper.go
+2 -2
View File
@@ -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)
pkg/executor/util/util.go
+1 -1
View File
@@ -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
pkg/fetcher/fetcher.go
+4 -3
View File
@@ -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(
pkg/fission-cli/cmd/support/resources/kubernetes.go
+7 -7
View File
@@ -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,
pkg/fission-cli/util/util.go
+1 -1
View File
@@ -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")
pkg/healthcheck/healthcheck.go
+2 -2
View File
@@ -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
}
pkg/kubewatcher/kubewatcher.go
+5 -5
View File
@@ -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"),
pkg/kubewatcher/main.go
+1 -1
View File
@@ -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")
}
pkg/kubewatcher/watchSync.go
+3 -3
View File
@@ -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,
pkg/mqtrigger/mqtmanager.go
+3 -3
View File
@@ -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),
pkg/mqtrigger/scalermanager.go
+8 -8
View File
@@ -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,
pkg/router/httpTriggers.go
+5 -5
View File
@@ -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"),
pkg/router/ingress.go
+3 -3
View File
@@ -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
}
pkg/router/router.go
+1 -1
View File
@@ -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))
}
pkg/storagesvc/archivePruner.go
+2 -1
View File
@@ -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
pkg/timer/main.go
+1 -1
View File
@@ -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")
}
pkg/timer/timerSync.go
+3 -3
View File
@@ -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,