diff --git a/pkg/apis/fission.io/v1/const.go b/pkg/apis/fission.io/v1/const.go index 671cb61c..1b731a6e 100644 --- a/pkg/apis/fission.io/v1/const.go +++ b/pkg/apis/fission.io/v1/const.go @@ -54,8 +54,8 @@ const ( ) const ( - ExecutorTypePoolmgr = "poolmgr" - ExecutorTypeNewdeploy = "newdeploy" + ExecutorTypePoolmgr ExecutorType = "poolmgr" + ExecutorTypeNewdeploy ExecutorType = "newdeploy" ) const ( diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 2cb57934..7f4ad86d 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -20,13 +20,13 @@ import ( "context" "encoding/json" "fmt" - "github.com/hashicorp/go-multierror" - "github.com/pkg/errors" "io/ioutil" "net/http" "strings" "github.com/gorilla/mux" + "github.com/hashicorp/go-multierror" + "github.com/pkg/errors" "go.opencensus.io/plugin/ochttp" "go.uber.org/zap" k8serrors "k8s.io/apimachinery/pkg/api/errors" @@ -91,9 +91,15 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error zap.String("function_name", fn.Metadata.Name), zap.String("function_namespace", fn.Metadata.Namespace)) - fsvc, err := executor.fsCache.GetByFunction(&fn.Metadata) + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + et, exists := executor.executorTypes[t] + if !exists { + return "", errors.Errorf("Unknown executor type '%v'", t) + } + + fsvc, err := et.GetFuncSvcFromCache(fn) if err == nil { - if executor.isValidAddress(fsvc) { + if et.IsValid(fsvc) { // Cached, return svc address return fsvc.Address, nil } else { @@ -101,7 +107,7 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error zap.String("function_name", fn.Metadata.Name), zap.String("function_namespace", fn.Metadata.Namespace), zap.String("address", fsvc.Address)) - executor.fsCache.DeleteEntry(fsvc) + et.DeleteFuncSvcFromCache(fsvc) } } @@ -120,24 +126,7 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error // find funcSvc and update its atime // TODO: Deprecated tapService func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) { - body, err := ioutil.ReadAll(r.Body) - if err != nil { - executor.logger.Error("failed to read tap service request", zap.Error(err)) - http.Error(w, "Failed to read request", http.StatusInternalServerError) - return - } - svcName := string(body) - svcHost := strings.TrimPrefix(svcName, "http://") - - err = executor.fsCache.TouchByAddress(svcHost) - if err != nil { - executor.logger.Error("error tapping function service", - zap.Error(err), - zap.String("service", svcName), - zap.String("host", svcHost)) - http.Error(w, "Not found", http.StatusNotFound) - return - } + // only for upgrade compatibility w.WriteHeader(http.StatusOK) } @@ -163,7 +152,16 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) { errs := &multierror.Error{} for _, req := range tapSvcReqs { svcHost := strings.TrimPrefix(req.ServiceUrl, "http://") - err = executor.fsCache.TouchByAddress(svcHost) + + et, exists := executor.executorTypes[req.FnExecutorType] + if !exists { + errs = multierror.Append(errs, + errors.Errorf("error tapping service due to unknown executor type '%v' found", + req.FnExecutorType)) + continue + } + + err = et.TapService(svcHost) if err != nil { errs = multierror.Append(errs, errors.Wrapf(err, "'%v' failed to tap function '%v/%v' with service url '%v'", @@ -199,10 +197,11 @@ func (executor *Executor) Serve(port int) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - executor.ndm.Run(ctx) - executor.gpm.Run(ctx) - executor.cms.Run(ctx) + for _, et := range executor.executorTypes { + et.Run(ctx) + } + executor.cms.Run(ctx) address := fmt.Sprintf(":%v", port) err := http.ListenAndServe(address, &ochttp.Handler{ Handler: executor.GetHandler(), diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index e88041a0..2df4b2f8 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -20,6 +20,7 @@ import ( "context" "time" + "github.com/pkg/errors" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -29,8 +30,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" - nd "github.com/fission/fission/pkg/executor/newdeploy" - gpm "github.com/fission/fission/pkg/executor/poolmgr" + "github.com/fission/fission/pkg/executor/executortype" ) type ( @@ -46,10 +46,10 @@ type ( //MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) *ConfigSecretController { + kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) *ConfigSecretController { logger.Debug("Creating ConfigMap & Secret Controller") - _, cmcontroller := initConfigmapController(logger, fissionClient, kubernetesClient, ndm, gpm) - _, scontroller := initSecretController(logger, fissionClient, kubernetesClient, ndm, gpm) + _, cmcontroller := initConfigmapController(logger, fissionClient, kubernetesClient, types) + _, scontroller := initSecretController(logger, fissionClient, kubernetesClient, types) cmsController := &ConfigSecretController{ logger: logger, configmapController: cmcontroller, @@ -66,7 +66,7 @@ func (csController *ConfigSecretController) Run(ctx context.Context) { } func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) { + kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) (cache.Store, cache.Controller) { resyncPeriod := 30 * time.Second listWatch := cache.NewListWatchFromClient(kubernetesClient.CoreV1().RESTClient(), "configmaps", metav1.NamespaceAll, fields.Everything()) store, controller := cache.NewInformer(listWatch, &apiv1.ConfigMap{}, resyncPeriod, cache.ResourceEventHandlerFuncs{ @@ -81,12 +81,11 @@ func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClien zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace)) } - funcs, err := getConfigmapRelatedFuncs(logger, &newCm.ObjectMeta, fissionClient) if err != nil { logger.Error("Failed to get functions related to secret", zap.String("secret_name", newCm.ObjectMeta.Name), zap.String("secret_namespace", newCm.ObjectMeta.Namespace)) } - recyclePods(logger, funcs, ndm, gpm) + recyclePods(logger, funcs, types) } }, }) @@ -112,7 +111,7 @@ func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionC } func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient, - kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) { + kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) (cache.Store, cache.Controller) { resyncPeriod := 30 * time.Second listWatch := cache.NewListWatchFromClient(kubernetesClient.CoreV1().RESTClient(), "secrets", metav1.NamespaceAll, fields.Everything()) store, controller := cache.NewInformer(listWatch, &apiv1.Secret{}, resyncPeriod, cache.ResourceEventHandlerFuncs{ @@ -128,14 +127,12 @@ func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient, zap.String("configmap_namespace", newS.ObjectMeta.Namespace)) } - funcs, err := getSecretRelatedFuncs(logger, &newS.ObjectMeta, fissionClient) if err != nil { logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace)) } - recyclePods(logger, funcs, ndm, gpm) + recyclePods(logger, funcs, types) } - }, }) return store, controller @@ -160,19 +157,19 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie return relatedFunctions, nil } -func recyclePods(logger *zap.Logger, funcs []fv1.Function, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) { +func recyclePods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) { for _, f := range funcs { var err error - switch f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType { - case fv1.ExecutorTypeNewdeploy: - err = ndm.RefreshFuncPods(logger, f) - case fv1.ExecutorTypePoolmgr: - err = gpm.RefreshFuncPods(logger, f) + et, exists := types[f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType] + if exists { + err = et.RefreshFuncPods(logger, f) + } else { + err = errors.Errorf("Unknown executor type '%v'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType) } if err != nil { - logger.Error("Failed to recycle pods for function after configmap changed", + logger.Error("Failed to recycle pods for function after configmap/secret changed", zap.Error(err), zap.Any("function", f)) } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 3306f2d3..f6701f98 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -32,9 +32,10 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/cms" + "github.com/fission/fission/pkg/executor/executortype" + "github.com/fission/fission/pkg/executor/executortype/newdeploy" + "github.com/fission/fission/pkg/executor/executortype/poolmgr" "github.com/fission/fission/pkg/executor/fscache" - "github.com/fission/fission/pkg/executor/newdeploy" - "github.com/fission/fission/pkg/executor/poolmgr" "github.com/fission/fission/pkg/executor/reaper" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" ) @@ -43,12 +44,10 @@ type ( Executor struct { logger *zap.Logger - gpm *poolmgr.GenericPoolManager - ndm *newdeploy.NewDeploy - cms *cms.ConfigSecretController + executorTypes map[fv1.ExecutorType]executortype.ExecutorType + cms *cms.ConfigSecretController fissionClient *crd.FissionClient - fsCache *fscache.FunctionServiceCache requestChan chan *createFuncServiceRequest fsCreateWg map[string]*sync.WaitGroup @@ -64,21 +63,20 @@ type ( } ) -func MakeExecutor(logger *zap.Logger, gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, cms *cms.ConfigSecretController, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor { +func MakeExecutor(logger *zap.Logger, cms *cms.ConfigSecretController, + fissionClient *crd.FissionClient, types map[fv1.ExecutorType]executortype.ExecutorType) (*Executor, error) { executor := &Executor{ logger: logger.Named("executor"), - gpm: gpm, - ndm: ndm, cms: cms, fissionClient: fissionClient, - fsCache: fsCache, + executorTypes: types, requestChan: make(chan *createFuncServiceRequest), fsCreateWg: make(map[string]*sync.WaitGroup), } go executor.serveCreateFuncServices() - return executor + return executor, nil } // All non-cached function service requests go through this goroutine @@ -115,6 +113,7 @@ func (executor *Executor) serveCreateFuncServices() { buffer := 10 // add some buffer time for specialization fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(), time.Duration(req.function.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout+buffer)*time.Second) + defer cancel() fsvc, err := executor.createServiceForFunction(fnSpecializationTimeoutContext, req.function) req.respChan <- &createFuncServiceResponse{ @@ -122,8 +121,6 @@ func (executor *Executor) serveCreateFuncServices() { err: err, } delete(executor.fsCreateWg, crd.CacheKey(fnMetadata)) - - cancel() wg.Done() }() } else { @@ -134,7 +131,7 @@ func (executor *Executor) serveCreateFuncServices() { wg.Wait() // get the function service from the cache - fsvc, err := executor.fsCache.GetByFunction(fnMetadata) + fsvc, err := executor.getFunctionServiceFromCache(req.function) // fsCache return error when the entry does not exist/expire. // It normally happened if there are multiple requests are @@ -155,18 +152,13 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. zap.String("function_name", fn.Metadata.Name), zap.String("function_namespace", fn.Metadata.Namespace)) - executorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType - - var fsvc *fscache.FuncSvc - var fsvcErr error - - switch executorType { - case fv1.ExecutorTypeNewdeploy: - fsvc, fsvcErr = executor.ndm.GetFuncSvc(ctx, fn) - default: - fsvc, fsvcErr = executor.gpm.GetFuncSvc(ctx, fn) + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + e, ok := executor.executorTypes[t] + if !ok { + return nil, errors.Errorf("Unknown executor type '%v'", t) } + fsvc, fsvcErr := e.GetFuncSvc(ctx, fn) if fsvcErr != nil { e := "error creating service for function" executor.logger.Error(e, @@ -174,25 +166,18 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1. zap.String("function_name", fn.Metadata.Name), zap.String("function_namespace", fn.Metadata.Namespace)) fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%s] %s", fn.Metadata.Name, e)) - } else if fsvc != nil { - _, err := executor.fsCache.Add(*fsvc) - if err != nil { - return nil, err - } } - executor.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID)) - return fsvc, fsvcErr } -// isValidAddress invokes isValidService or isValidPod depending on the type of executor -func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool { - if fsvc.Executor == fscache.NEWDEPLOY { - return executor.ndm.IsValid(fsvc) - } else { - return executor.gpm.IsValid(fsvc) +func (executor *Executor) getFunctionServiceFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { + t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + e, ok := executor.executorTypes[t] + if !ok { + return nil, errors.Errorf("Unknown executor type '%v'", t) } + return e.GetFuncSvcFromCache(fn) } func serveMetric(logger *zap.Logger) { @@ -223,7 +208,6 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames } restClient := fissionClient.GetCrdClient() - fsCache := fscache.MakeFunctionServiceCache(logger) poolID := strings.ToLower(uniuri.NewLen(8)) reaper.CleanupOldExecutorObjects(logger, kubernetesClient, poolID) @@ -239,9 +223,16 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames fissionClient, kubernetesClient, restClient, functionNamespace, fetcherConfig, poolID) - cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, ndm, gpm) + executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType) + executorTypes[gpm.GetTypeName()] = gpm + executorTypes[ndm.GetTypeName()] = ndm - api := MakeExecutor(logger, gpm, ndm, cms, fissionClient, fsCache) + cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, executorTypes) + + api, err := MakeExecutor(logger, cms, fissionClient, executorTypes) + if err != nil { + return err + } go api.Serve(port) go serveMetric(logger) diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go new file mode 100644 index 00000000..d5879071 --- /dev/null +++ b/pkg/executor/executortype/executortype.go @@ -0,0 +1,53 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package executortype + +import ( + "context" + "go.uber.org/zap" + + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" + "github.com/fission/fission/pkg/executor/fscache" +) + +type ExecutorType interface { + // Run runs background job. + Run(context.Context) + + // GetTypeName returns the name of executor type + GetTypeName() fv1.ExecutorType + + // GetFuncSvc specializes function pod(s) and returns a service URL for the function. + GetFuncSvc(context.Context, *fv1.Function) (*fscache.FuncSvc, error) + + // GetFuncSvcFromCache retrieves function service from cache. + GetFuncSvcFromCache(*fv1.Function) (*fscache.FuncSvc, error) + + // DeleteFuncSvcFromCache deletes function service entry in cache. + DeleteFuncSvcFromCache(*fscache.FuncSvc) + + // TapService updates the access time of function service entry to + // avoid idle pod reaper recycles pods. + TapService(serviceUrl string) error + + // IsValid returns true if a function service is valid. Different executor types + // use distinct ways to examine the function service. + IsValid(*fscache.FuncSvc) bool + + // RefreshFuncPods refreshes function pods if the secrets/configmaps pods reference to get updated. + RefreshFuncPods(*zap.Logger, fv1.Function) error +} diff --git a/pkg/executor/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go similarity index 100% rename from pkg/executor/newdeploy/newdeploy.go rename to pkg/executor/executortype/newdeploy/newdeploy.go diff --git a/pkg/executor/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go similarity index 96% rename from pkg/executor/newdeploy/newdeploymgr.go rename to pkg/executor/executortype/newdeploy/newdeploymgr.go index e00c01dd..7bad6b81 100644 --- a/pkg/executor/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -24,8 +24,6 @@ import ( "strings" "time" - "github.com/fission/fission/pkg/throttler" - "github.com/fission/fission/pkg/utils" multierror "github.com/hashicorp/go-multierror" "github.com/pkg/errors" "go.uber.org/zap" @@ -42,11 +40,16 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" + "github.com/fission/fission/pkg/executor/executortype" "github.com/fission/fission/pkg/executor/fscache" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" + "github.com/fission/fission/pkg/throttler" "github.com/fission/fission/pkg/types" + "github.com/fission/fission/pkg/utils" ) +var _ executortype.ExecutorType = &NewDeploy{} + type ( NewDeploy struct { logger *zap.Logger @@ -82,7 +85,7 @@ func MakeNewDeploy( namespace string, fetcherConfig *fetcherConfig.Config, instanceID string, -) *NewDeploy { +) executortype.ExecutorType { enableIstio := false if len(os.Getenv("ENABLE_ISTIO")) > 0 { istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO")) @@ -130,6 +133,105 @@ func (deploy *NewDeploy) Run(ctx context.Context) { go deploy.idleObjectReaper() } +func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType { + return fv1.ExecutorTypeNewdeploy +} + +func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + return deploy.createFunction(fn, false) +} + +func (deploy *NewDeploy) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { + return deploy.fsCache.GetByFunction(&fn.Metadata) +} + +func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { + deploy.fsCache.DeleteEntry(fsvc) +} + +func (deploy *NewDeploy) TapService(svcHost string) error { + err := deploy.fsCache.TouchByAddress(svcHost) + if err != nil { + return err + } + return nil +} + +// IsValid does a get on the service address to ensure it's a valid service, then +// scale deployment to 1 replica if there are no available replicas for function. +// Return true if no error occurs, return false otherwise. +func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { + service := strings.Split(fsvc.Address, ".") + if len(service) == 0 { + return false + } + + _, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{}) + if err != nil { + deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err)) + return false + } + + deployObj := getDeploymentObj(fsvc.KubernetesObjects) + if deployObj == nil { + deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name)) + return false + } + + currentDeploy, err := deploy.kubernetesClient.AppsV1(). + Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) + if err != nil { + deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return false + } + + // return directly when available replicas > 0 + if currentDeploy.Status.AvailableReplicas > 0 { + return true + } + + return false +} + +// RefreshFuncPods deleted pods related to the function so that new pods are replenished +func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { + + env, err := deploy.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) + if err != nil { + return err + } + + funcLabels := deploy.getDeployLabels(f.Metadata, metav1.ObjectMeta{ + Name: f.Spec.Environment.Name, + Namespace: f.Spec.Environment.Namespace, + UID: env.Metadata.UID, + }) + + dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{ + LabelSelector: labels.Set(funcLabels).AsSelector().String(), + }) + + if err != nil { + return err + } + + patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%s"}]}]}}}}`, + f.Metadata.Name, + fv1.LastUpdateTimestamp, + time.Now().String()) + + // Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition + for _, deployment := range dep.Items { + _, err := deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(deployment.ObjectMeta.Name, + k8sTypes.StrategicMergePatchType, + []byte(patch)) + if err != nil { + return err + } + } + return nil +} + func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controller) { resyncPeriod := 30 * time.Second listWatch := k8sCache.NewListWatchFromClient(deploy.crdClient, "functions", metav1.NamespaceAll, fields.Everything()) @@ -223,49 +325,6 @@ func (deploy *NewDeploy) getEnvFunctions(m *metav1.ObjectMeta) []fv1.Function { return relatedFunctions } -func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - return deploy.createFunction(fn, false) -} - -// RefreshFuncPods deleted pods related to the function so that new pods are replenished -func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { - - env, err := deploy.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) - if err != nil { - return err - } - - funcLabels := deploy.getDeployLabels(f.Metadata, metav1.ObjectMeta{ - Name: f.Spec.Environment.Name, - Namespace: f.Spec.Environment.Namespace, - UID: env.Metadata.UID, - }) - - dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{ - LabelSelector: labels.Set(funcLabels).AsSelector().String(), - }) - - if err != nil { - return err - } - - patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%s"}]}]}}}}`, - f.Metadata.Name, - fv1.LastUpdateTimestamp, - time.Now().String()) - - // Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition - for _, deployment := range dep.Items { - _, err := deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(deployment.ObjectMeta.Name, - k8sTypes.StrategicMergePatchType, - []byte(patch)) - if err != nil { - return err - } - } - return nil -} - func (deploy *NewDeploy) createFunction(fn *fv1.Function, firstcreate bool) (*fscache.FuncSvc, error) { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { return nil, nil @@ -385,7 +444,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache. Environment: env, Address: svcAddress, KubernetesObjects: kubeObjRefs, - Executor: fscache.NEWDEPLOY, + Executor: fv1.ExecutorTypeNewdeploy, } _, err = deploy.fsCache.Add(*fsvc) @@ -393,6 +452,9 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache. deploy.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function)) return fsvc, err } + + deploy.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID)) + return fsvc, nil } @@ -602,7 +664,7 @@ func (deploy *NewDeploy) getObjName(fn *fv1.Function) string { func (deploy *NewDeploy) getDeployLabels(fnMeta metav1.ObjectMeta, envMeta metav1.ObjectMeta) map[string]string { return map[string]string{ types.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID, - types.EXECUTOR_TYPE: fv1.ExecutorTypeNewdeploy, + types.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy), types.ENVIRONMENT_NAME: envMeta.Name, types.ENVIRONMENT_NAMESPACE: envMeta.Namespace, types.ENVIRONMENT_UID: string(envMeta.UID), @@ -618,42 +680,6 @@ func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message strin deploy.logger.Error("function status update", zap.Error(err), zap.Any("function", fn), zap.String("message", message)) } -// IsValid does a get on the service address to ensure it's a valid service, then -// scale deployment to 1 replica if there are no available replicas for function. -// Return true if no error occurs, return false otherwise. -func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool { - service := strings.Split(fsvc.Address, ".") - if len(service) == 0 { - return false - } - - _, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{}) - if err != nil { - deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err)) - return false - } - - deployObj := getDeploymentObj(fsvc.KubernetesObjects) - if deployObj == nil { - deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name)) - return false - } - - currentDeploy, err := deploy.kubernetesClient.AppsV1(). - Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) - if err != nil { - deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - return false - } - - // return directly when available replicas > 0 - if currentDeploy.Status.AvailableReplicas > 0 { - return true - } - - return false -} - // idleObjectReaper reaps objects after certain idle time func (deploy *NewDeploy) idleObjectReaper() { @@ -678,7 +704,7 @@ func (deploy *NewDeploy) idleObjectReaper() { } for _, fsvc := range funcSvcs { - if fsvc.Executor != fscache.NEWDEPLOY { + if fsvc.Executor != fv1.ExecutorTypeNewdeploy { continue } @@ -694,7 +720,7 @@ func (deploy *NewDeploy) idleObjectReaper() { if err != nil { // Newdeploy manager handles the function delete event and clean cache/kubeobjs itself, // so we ignore the not found error for functions with newdeploy executor type here. - if k8sErrs.IsNotFound(err) && fsvc.Executor == fscache.NEWDEPLOY { + if k8sErrs.IsNotFound(err) && fsvc.Executor == fv1.ExecutorTypeNewdeploy { continue } deploy.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name)) diff --git a/pkg/executor/poolmgr/funcwatcher.go b/pkg/executor/executortype/poolmgr/funcwatcher.go similarity index 100% rename from pkg/executor/poolmgr/funcwatcher.go rename to pkg/executor/executortype/poolmgr/funcwatcher.go diff --git a/pkg/executor/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go similarity index 97% rename from pkg/executor/poolmgr/gp.go rename to pkg/executor/executortype/poolmgr/gp.go index 46160e34..c9481f86 100644 --- a/pkg/executor/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -144,7 +144,7 @@ func MakeGenericPool( func (gp *GenericPool) getDeployLabels() map[string]string { return map[string]string{ fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceId, - types.EXECUTOR_TYPE: fv1.ExecutorTypePoolmgr, + types.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr), types.ENVIRONMENT_NAME: gp.env.Metadata.Name, types.ENVIRONMENT_NAMESPACE: gp.env.Metadata.Namespace, types.ENVIRONMENT_UID: string(gp.env.Metadata.UID), @@ -471,14 +471,17 @@ func (gp *GenericPool) waitForReadyPod() error { // Since even single pod is not ready, choosing the first pod to inspect is a good approximation. In future this can be done better pod := podList.Items[0] - multierr := &multierror.Error{} + errs := &multierror.Error{} for _, cStatus := range pod.Status.ContainerStatuses { if !cStatus.Ready { - multierr = multierror.Append(multierr, errors.New(fmt.Sprintf("%v: %v", cStatus.State.Waiting.Reason, cStatus.State.Waiting.Message))) + errs = multierror.Append(errs, errors.New(fmt.Sprintf("%v: %v", cStatus.State.Waiting.Reason, cStatus.State.Waiting.Message))) } } - return errors.Wrapf(multierr, "Timeout: waited too long for pod of deployment %v in namespace %v to be ready", - gp.deployment.ObjectMeta.Name, gp.namespace) + if errs.ErrorOrNil() != nil { + return errors.Wrapf(errs, "Timeout: waited too long for pod of deployment %v in namespace %v to be ready", + gp.deployment.ObjectMeta.Name, gp.namespace) + } + return nil } time.Sleep(1000 * time.Millisecond) } @@ -614,7 +617,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac Environment: gp.env, Address: svcHost, KubernetesObjects: kubeObjRefs, - Executor: fscache.POOLMGR, + Executor: fv1.ExecutorTypePoolmgr, Ctime: time.Now(), Atime: time.Now(), } @@ -623,6 +626,9 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac if err != nil { return nil, err } + + gp.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID)) + return fsvc, nil } diff --git a/pkg/executor/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go similarity index 88% rename from pkg/executor/poolmgr/gpm.go rename to pkg/executor/executortype/poolmgr/gpm.go index 3075e502..84c35070 100644 --- a/pkg/executor/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -35,12 +35,15 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/cache" "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/reaper" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" "github.com/fission/fission/pkg/types" ) +var _ executortype.ExecutorType = &GenericPoolManager{} + type requestType int const ( @@ -90,7 +93,7 @@ func MakeGenericPoolManager( kubernetesClient *kubernetes.Clientset, functionNamespace string, fetcherConfig *fetcherConfig.Config, - instanceId string) *GenericPoolManager { + instanceId string) executortype.ExecutorType { gpmLogger := logger.Named("generic_pool_manager") @@ -132,6 +135,67 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) { go gpm.idleObjectReaper() } +func (gpm *GenericPoolManager) GetTypeName() fv1.ExecutorType { + return fv1.ExecutorTypePoolmgr +} + +func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + // from Func -> get Env + gpm.logger.Debug("getting environment for function", zap.String("function", fn.Metadata.Name)) + env, err := gpm.getFunctionEnv(fn) + if err != nil { + return nil, err + } + + pool, err := gpm.getPool(env) + if err != nil { + return nil, err + } + + // from GenericPool -> get one function container + // (this also adds to the cache) + gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name)) + return pool.getFuncSvc(ctx, fn) +} + +func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { + return gpm.fsCache.GetByFunction(&fn.Metadata) +} + +func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { + gpm.fsCache.DeleteEntry(fsvc) +} + +func (gpm *GenericPoolManager) TapService(svcHost string) error { + err := gpm.fsCache.TouchByAddress(svcHost) + if err != nil { + return err + } + return nil +} + +// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the +// containers in it are reporting a ready status for the healthCheck. +func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { + for _, obj := range fsvc.KubernetesObjects { + if obj.Kind == "pod" { + pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{}) + if err == nil && utils.IsReadyPod(pod) { + // Normally, the address format is http://[pod-ip]:[port], however, if the + // Istio is enabled the address format changes to http://[svc-name]:[port]. + // So if the Istio is enabled and pod is in ready state, we return true directly; + // Otherwise, we need to ensure that the address contains pod ip. + if gpm.enableIstio || + (!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) { + gpm.logger.Debug("valid address", zap.String("address", fsvc.Address)) + return true + } + } + } + } + return false +} + func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error { env, err := gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name) @@ -246,25 +310,6 @@ func (gpm *GenericPoolManager) cleanupPools(envs []fv1.Environment) { } } -func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - // from Func -> get Env - gpm.logger.Debug("getting environment for function", zap.String("function", fn.Metadata.Name)) - env, err := gpm.getFunctionEnv(fn) - if err != nil { - return nil, err - } - - pool, err := gpm.getPool(env) - if err != nil { - return nil, err - } - - // from GenericPool -> get one function container - // (this also adds to the cache) - gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name)) - return pool.getFuncSvc(ctx, fn) -} - func (gpm *GenericPoolManager) getFunctionEnv(fn *fv1.Function) (*fv1.Environment, error) { var env *fv1.Environment @@ -335,28 +380,6 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *fv1.Environment) int32 { return poolsize } -// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the -// containers in it are reporting a ready status for the healthCheck. -func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { - for _, obj := range fsvc.KubernetesObjects { - if obj.Kind == "pod" { - pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{}) - if err == nil && utils.IsReadyPod(pod) { - // Normally, the address format is http://[pod-ip]:[port], however, if the - // Istio is enabled the address format changes to http://[svc-name]:[port]. - // So if the Istio is enabled and pod is in ready state, we return true directly; - // Otherwise, we need to ensure that the address contains pod ip. - if gpm.enableIstio || - (!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) { - gpm.logger.Debug("valid address", zap.String("address", fsvc.Address)) - return true - } - } - } - } - return false -} - // idleObjectReaper reaps objects after certain idle time func (gpm *GenericPoolManager) idleObjectReaper() { @@ -380,8 +403,10 @@ func (gpm *GenericPoolManager) idleObjectReaper() { continue } - for _, fsvc := range funcSvcs { - if fsvc.Executor != fscache.POOLMGR { + for i := range funcSvcs { + fsvc := funcSvcs[i] + + if fsvc.Executor != fv1.ExecutorTypePoolmgr { continue } @@ -397,20 +422,23 @@ func (gpm *GenericPoolManager) idleObjectReaper() { continue } - deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime) - if err != nil { - gpm.logger.Error("error deleting Kubernetes objects for function service", - zap.Error(err), - zap.Any("service", fsvc)) - } - - if !deleted { - continue - } - - for _, kubeobj := range fsvc.KubernetesObjects { - reaper.CleanupKubeObject(gpm.logger, gpm.kubernetesClient, &kubeobj) - } + go func() { + deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime) + if err != nil { + gpm.logger.Error("error deleting Kubernetes objects for function service", + zap.Error(err), + zap.Any("service", fsvc)) + } + if deleted { + for i := range fsvc.KubernetesObjects { + gpm.logger.Debug("release idle function resources", + zap.String("function", fsvc.Name), zap.String("address", fsvc.Address), + zap.String("executor", string(fsvc.Executor))) + reaper.CleanupKubeObject(gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) + time.Sleep(50 * time.Millisecond) + } + } + }() } } } diff --git a/pkg/executor/poolmgr/packagewatcher.go b/pkg/executor/executortype/poolmgr/packagewatcher.go similarity index 100% rename from pkg/executor/poolmgr/packagewatcher.go rename to pkg/executor/executortype/poolmgr/packagewatcher.go diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 85554e1d..ddb12457 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -33,7 +33,8 @@ import ( ) type fscRequestType int -type executorType int + +//type executorType int const ( TOUCH fscRequestType = iota @@ -41,11 +42,6 @@ const ( LOG ) -const ( - POOLMGR executorType = iota - NEWDEPLOY -) - type ( FuncSvc struct { Name string // Name of object @@ -53,7 +49,7 @@ type ( Environment *fv1.Environment // function's environment Address string // Host:Port or IP:Port that the function's service can be reached at. KubernetesObjects []apiv1.ObjectReference // Kubernetes Objects (within the function namespace) - Executor executorType + Executor fv1.ExecutorType Ctime time.Time Atime time.Time diff --git a/pkg/executor/reaper/reaper.go b/pkg/executor/reaper/reaper.go index 5d343556..5f3ad061 100644 --- a/pkg/executor/reaper/reaper.go +++ b/pkg/executor/reaper/reaper.go @@ -26,6 +26,7 @@ import ( meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/types" ) @@ -294,7 +295,7 @@ func CleanupRoleBindings(logger *zap.Logger, client *kubernetes.Clientset, fissi break } - if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == types.ExecutorTypeNewdeploy { + if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { ndmFunc = true break } diff --git a/pkg/fission-cli/cmd/function/create.go b/pkg/fission-cli/cmd/function/create.go index 1d91fa79..6add5898 100644 --- a/pkg/fission-cli/cmd/function/create.go +++ b/pkg/fission-cli/cmd/function/create.go @@ -36,7 +36,6 @@ import ( "github.com/fission/fission/pkg/fission-cli/console" flagkey "github.com/fission/fission/pkg/fission-cli/flag/key" "github.com/fission/fission/pkg/fission-cli/util" - "github.com/fission/fission/pkg/types" ) const ( @@ -363,18 +362,17 @@ func (opts *CreateSubCommand) run(input cli.Input) error { } func getInvokeStrategy(input cli.Input, existingInvokeStrategy *fv1.InvokeStrategy) (strategy *fv1.InvokeStrategy, err error) { - var fnExecutor, newFnExecutor fv1.ExecutorType switch input.String(flagkey.FnExecutorType) { case "": fallthrough - case types.ExecutorTypePoolmgr: - newFnExecutor = types.ExecutorTypePoolmgr - case types.ExecutorTypeNewdeploy: - newFnExecutor = types.ExecutorTypeNewdeploy + case string(fv1.ExecutorTypePoolmgr): + newFnExecutor = fv1.ExecutorTypePoolmgr + case string(fv1.ExecutorTypeNewdeploy): + newFnExecutor = fv1.ExecutorTypeNewdeploy default: - return nil, errors.New("executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'") + return nil, errors.Errorf("executor type must be one of '%v' or '%v'", fv1.ExecutorTypePoolmgr, fv1.ExecutorTypeNewdeploy) } if existingInvokeStrategy != nil { @@ -388,11 +386,11 @@ func getInvokeStrategy(input cli.Input, existingInvokeStrategy *fv1.InvokeStrate fnExecutor = newFnExecutor } - if input.IsSet(flagkey.FnSpecializationTimeout) && fnExecutor != types.ExecutorTypeNewdeploy { + if input.IsSet(flagkey.FnSpecializationTimeout) && fnExecutor != fv1.ExecutorTypeNewdeploy { return nil, errors.Errorf("%v flag is only applicable for newdeploy type of executor", flagkey.FnSpecializationTimeout) } - if fnExecutor == types.ExecutorTypePoolmgr { + if fnExecutor == fv1.ExecutorTypePoolmgr { if input.IsSet(flagkey.RuntimeTargetcpu) || input.IsSet(flagkey.ReplicasMinscale) || input.IsSet(flagkey.ReplicasMaxscale) { return nil, errors.New("to set target CPU or min/max scale for function, please specify \"--executortype newdeploy\"") } @@ -403,7 +401,7 @@ func getInvokeStrategy(input cli.Input, existingInvokeStrategy *fv1.InvokeStrate strategy = &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, ExecutionStrategy: fv1.ExecutionStrategy{ - ExecutorType: types.ExecutorTypePoolmgr, + ExecutorType: fv1.ExecutorTypePoolmgr, }, } } else { @@ -413,7 +411,7 @@ func getInvokeStrategy(input cli.Input, existingInvokeStrategy *fv1.InvokeStrate maxScale := minScale specializationTimeout := fv1.DefaultSpecializationTimeOut - if existingInvokeStrategy != nil && existingInvokeStrategy.ExecutionStrategy.ExecutorType == types.ExecutorTypeNewdeploy { + if existingInvokeStrategy != nil && existingInvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { minScale = existingInvokeStrategy.ExecutionStrategy.MinScale maxScale = existingInvokeStrategy.ExecutionStrategy.MaxScale targetCPU = existingInvokeStrategy.ExecutionStrategy.TargetCPUPercent diff --git a/pkg/fission-cli/cmd/function/function_test.go b/pkg/fission-cli/cmd/function/function_test.go index 5ab8c248..b50a83b2 100644 --- a/pkg/fission-cli/cmd/function/function_test.go +++ b/pkg/fission-cli/cmd/function/function_test.go @@ -48,7 +48,7 @@ func TestGetInvokeStrategy(t *testing.T) { }, { // case: executor type set to poolmgr - testArgs: map[string]interface{}{flagkey.FnExecutorType: fv1.ExecutorTypePoolmgr}, + testArgs: map[string]interface{}{flagkey.FnExecutorType: string(fv1.ExecutorTypePoolmgr)}, existingInvokeStrategy: nil, expectedResult: &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, @@ -60,7 +60,7 @@ func TestGetInvokeStrategy(t *testing.T) { }, { // case: executor type set to newdeploy - testArgs: map[string]interface{}{flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy}, + testArgs: map[string]interface{}{flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy)}, existingInvokeStrategy: nil, expectedResult: &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, @@ -76,7 +76,7 @@ func TestGetInvokeStrategy(t *testing.T) { }, { // case: executor type change from poolmgr to newdeploy - testArgs: map[string]interface{}{flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy}, + testArgs: map[string]interface{}{flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy)}, existingInvokeStrategy: &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, ExecutionStrategy: fv1.ExecutionStrategy{ @@ -97,7 +97,7 @@ func TestGetInvokeStrategy(t *testing.T) { }, { // case: executor type change from newdeploy to poolmgr - testArgs: map[string]interface{}{flagkey.FnExecutorType: fv1.ExecutorTypePoolmgr}, + testArgs: map[string]interface{}{flagkey.FnExecutorType: string(fv1.ExecutorTypePoolmgr)}, existingInvokeStrategy: &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, ExecutionStrategy: fv1.ExecutionStrategy{ @@ -119,7 +119,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: minscale < maxscale testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMinscale: 2, flagkey.ReplicasMaxscale: 3, }, @@ -139,7 +139,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: minscale > maxscale testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMinscale: 5, flagkey.ReplicasMaxscale: 3, }, @@ -150,7 +150,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: maxscale not specified testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMinscale: 5, }, existingInvokeStrategy: nil, @@ -160,7 +160,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: minscale not specified testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMaxscale: 3, }, existingInvokeStrategy: nil, @@ -179,7 +179,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: maxscale set to 0 testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMaxscale: 0, }, existingInvokeStrategy: nil, @@ -189,7 +189,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: maxscale set to 9 when existing is 5 testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.ReplicasMaxscale: 9, }, existingInvokeStrategy: &fv1.InvokeStrategy{ @@ -217,7 +217,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: change nothing for existing strategy testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), }, existingInvokeStrategy: &fv1.InvokeStrategy{ StrategyType: fv1.StrategyTypeExecution, @@ -244,7 +244,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: set target cpu percentage testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.RuntimeTargetcpu: 50, }, existingInvokeStrategy: nil, @@ -263,7 +263,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: change target cpu percentage testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.RuntimeTargetcpu: 20, }, existingInvokeStrategy: &fv1.InvokeStrategy{ @@ -291,7 +291,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: change specializationtimeout testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.FnSpecializationTimeout: 200, }, existingInvokeStrategy: &fv1.InvokeStrategy{ @@ -318,7 +318,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: specializationtimeout should not work for poolmgr testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypePoolmgr, + flagkey.FnExecutorType: string(fv1.ExecutorTypePoolmgr), flagkey.FnSpecializationTimeout: 10, }, existingInvokeStrategy: nil, @@ -328,7 +328,7 @@ func TestGetInvokeStrategy(t *testing.T) { { // case: specializationtimeout should not be less than 120 testArgs: map[string]interface{}{ - flagkey.FnExecutorType: fv1.ExecutorTypeNewdeploy, + flagkey.FnExecutorType: string(fv1.ExecutorTypeNewdeploy), flagkey.FnSpecializationTimeout: 90, }, existingInvokeStrategy: nil, diff --git a/pkg/fission-cli/cmd/function/update.go b/pkg/fission-cli/cmd/function/update.go index 1c541601..a6b9a21c 100644 --- a/pkg/fission-cli/cmd/function/update.go +++ b/pkg/fission-cli/cmd/function/update.go @@ -30,7 +30,6 @@ import ( "github.com/fission/fission/pkg/fission-cli/console" flagkey "github.com/fission/fission/pkg/fission-cli/flag/key" "github.com/fission/fission/pkg/fission-cli/util" - "github.com/fission/fission/pkg/types" ) type UpdateSubCommand struct { @@ -171,7 +170,7 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { function.Spec.InvokeStrategy = *strategy if input.IsSet(flagkey.FnSpecializationTimeout) { - if strategy.ExecutionStrategy.ExecutorType != types.ExecutorTypeNewdeploy { + if strategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { return errors.Errorf("--%v flag is only applicable for newdeploy type of executor", flagkey.FnSpecializationTimeout) } diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index de8ffd94..bc6c08c5 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -24,7 +24,6 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" flagkey "github.com/fission/fission/pkg/fission-cli/flag/key" - "github.com/fission/fission/pkg/types" ) type ( @@ -94,7 +93,7 @@ var ( FnBuildCmd = Flag{Type: String, Name: flagkey.FnBuildCmd, Usage: "Package build command for builder to run with"} FnSecret = Flag{Type: StringSlice, Name: flagkey.FnSecret, Usage: "Function access to secret, should be present in the same namespace as the function. You can provide multiple secrets using multiple --secrets flags. In the case of fn update the the secrets will be replaced by the provided list of secrets."} FnCfgMap = Flag{Type: StringSlice, Name: flagkey.FnCfgMap, Usage: "Function access to configmap, should be present in the same namespace as the function. You can provide multiple configmaps using multiple --configmap flags. In case of fn update the configmaps will be replaced by the provided list of configmaps."} - FnExecutorType = Flag{Type: String, Name: flagkey.FnExecutorType, Usage: "Executor type for execution; one of 'poolmgr', 'newdeploy'", DefaultValue: types.ExecutorTypePoolmgr} + FnExecutorType = Flag{Type: String, Name: flagkey.FnExecutorType, Usage: "Executor type for execution; one of 'poolmgr', 'newdeploy'", DefaultValue: string(fv1.ExecutorTypePoolmgr)} FnExecutionTimeout = Flag{Type: Int, Name: flagkey.FnExecutionTimeout, Aliases: []string{"ft"}, Usage: "Maximum time for a request to wait for the response from the function", DefaultValue: 60} FnLogPod = Flag{Type: String, Name: flagkey.FnLogPod, Usage: "Function pod name (use the latest pod name if unspecified)"} FnLogFollow = Flag{Type: Bool, Name: flagkey.FnLogFollow, Short: "f", Usage: "Specify if the logs should be streamed"} diff --git a/pkg/types/types.go b/pkg/types/types.go index b8d96ea7..852fa152 100644 --- a/pkg/types/types.go +++ b/pkg/types/types.go @@ -130,15 +130,6 @@ const ( EXECUTOR_TYPE = "executorType" ) -const ( - ExecutorTypePoolmgr = fv1.ExecutorTypePoolmgr - ExecutorTypeNewdeploy = fv1.ExecutorTypeNewdeploy -) - -const ( - StrategyTypeExecution = fv1.StrategyTypeExecution -) - const ( SharedVolumeUserfunc = fv1.SharedVolumeUserfunc SharedVolumePackages = fv1.SharedVolumePackages diff --git a/test/tests/test_fn_update/test_idle_objects_reaper.sh b/test/tests/test_fn_update/test_idle_objects_reaper.sh index b29ab717..4362546e 100755 --- a/test/tests/test_fn_update/test_idle_objects_reaper.sh +++ b/test/tests/test_fn_update/test_idle_objects_reaper.sh @@ -41,7 +41,7 @@ timeout 60 bash -c "test_fn ${fn}-gpm 'world'" log "Waiting for idle pod reaper to recycle resources" # the LIST_OLD function list fsvc older than 2 mins # so in worst case, we need to wait for up to 4 mins + some buffer -sleep 260 +sleep 300 # The replicas of function deployment should be 0 due to minScale = 0 ndDeployReplicas=$(kubectl -n $FUNCTION_NAMESPACE get deploy -l functionName=${fn}-nd -ojsonpath='{.items[0].spec.replicas}')