From 51b264e8ca82ae58759d0e6b70c681791479b617 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Tue, 26 Nov 2019 01:16:30 +0800 Subject: [PATCH] Fix poolmanager terminates running function pod periodically (#1435) The pool manager keeps terminating function pod periodically even there are traffic to the function. The root cause is that executor, poolmgr, newdeploy manage their own functionServiceCache separately. And when router taps a function, executor updates the access time of the function service entry in its own cache without notifying executor types to do the update as well. Hence, the access time of function service entry in poolmanager cache never gets updated. Due to the access time never gets updated, the idle pod reaper in poolmanager then thinks the function pod is in idle state and recycle it. This PR removes the cache in executor itself, and when router tries to tap a function, executor will call executor type to tap the function and update access time. --- pkg/apis/fission.io/v1/const.go | 4 +- pkg/executor/api.go | 53 +++-- pkg/executor/cms/cmscontroller.go | 35 ++-- pkg/executor/executor.go | 71 +++---- pkg/executor/executortype/executortype.go | 53 +++++ .../{ => executortype}/newdeploy/newdeploy.go | 0 .../newdeploy/newdeploymgr.go | 198 ++++++++++-------- .../{ => executortype}/poolmgr/funcwatcher.go | 0 pkg/executor/{ => executortype}/poolmgr/gp.go | 18 +- .../{ => executortype}/poolmgr/gpm.go | 144 ++++++++----- .../poolmgr/packagewatcher.go | 0 pkg/executor/fscache/functionServiceCache.go | 10 +- pkg/executor/reaper/reaper.go | 3 +- pkg/fission-cli/cmd/function/create.go | 20 +- pkg/fission-cli/cmd/function/function_test.go | 32 +-- pkg/fission-cli/cmd/function/update.go | 3 +- pkg/fission-cli/flag/flag.go | 3 +- pkg/types/types.go | 9 - .../test_idle_objects_reaper.sh | 2 +- 19 files changed, 371 insertions(+), 287 deletions(-) create mode 100644 pkg/executor/executortype/executortype.go rename pkg/executor/{ => executortype}/newdeploy/newdeploy.go (100%) rename pkg/executor/{ => executortype}/newdeploy/newdeploymgr.go (96%) rename pkg/executor/{ => executortype}/poolmgr/funcwatcher.go (100%) rename pkg/executor/{ => executortype}/poolmgr/gp.go (97%) rename pkg/executor/{ => executortype}/poolmgr/gpm.go (88%) rename pkg/executor/{ => executortype}/poolmgr/packagewatcher.go (100%) 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}')