diff --git a/Makefile b/Makefile index ba4599e6..91573278 100644 --- a/Makefile +++ b/Makefile @@ -51,7 +51,7 @@ test-run: code-checks ### Binaries build-fission-cli: - @GOOS=$(GOOS) GOARCH=$(GOARCH) GOAMD64=$(GOAMD64) GORELEASER_CURRENT_TAG=$(VERSION) goreleaser build --snapshot --rm-dist --single-target --id fission-cli + @GOOS=$(GOOS) GOARCH=$(GOARCH) GOAMD64=$(GOAMD64) GORELEASER_CURRENT_TAG=$(VERSION) goreleaser build --snapshot --clean --single-target --id fission-cli install-fission-cli: # TODO: Fix this hack, replace v1 with GOAMD64 @@ -111,7 +111,7 @@ generate-crd-ref-docs: install-crd-ref-docs all-generators: codegen generate-crds generate-swagger-doc generate-cli-docs generate-crd-ref-docs skaffold-prebuild: - @GOOS=linux GOARCH=amd64 GORELEASER_CURRENT_TAG=$(VERSION) goreleaser build --snapshot --rm-dist --single-target + @GOOS=linux GOARCH=amd64 GORELEASER_CURRENT_TAG=$(VERSION) goreleaser build --snapshot --clean --single-target @cp -v cmd/builder/Dockerfile dist/builder_linux_amd64_v1/Dockerfile @cp -v cmd/fetcher/Dockerfile dist/fetcher_linux_amd64_v1/Dockerfile @cp -v cmd/fission-bundle/Dockerfile dist/fission-bundle_linux_amd64_v1/Dockerfile diff --git a/crds/v1/fission.io_functions.yaml b/crds/v1/fission.io_functions.yaml index 93b8f999..1d6c43a5 100644 --- a/crds/v1/fission.io_functions.yaml +++ b/crds/v1/fission.io_functions.yaml @@ -8047,6 +8047,11 @@ spec: More info: https://kubernetes.io/docs/concepts/configuration/manage-resources-containers/' type: object type: object + retainPods: + description: RetainPods specifies the number of specialized pods that + should be retained after serving requests This is optional. If not + specified default value will be taken as 0 + type: integer secrets: description: Reference to a list of secrets. items: diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index a1a858de..b94614b5 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -399,6 +399,11 @@ type ( // +optional OnceOnly bool `json:"onceOnly,omitempty"` + // RetainPods specifies the number of specialized pods that should be retained after serving requests + // This is optional. If not specified default value will be taken as 0 + // +optional + RetainPods int `json:"retainPods,omitempty"` + // Podspec specifies podspec to use for executor type container based functions // Different arguments mentioned for container based function are populated inside a pod. // +optional @@ -876,6 +881,10 @@ func (fn Function) GetConcurrency() int { return fn.Spec.Concurrency } +func (fn Function) GetRetainPods() int { + return fn.Spec.RetainPods +} + func (fn Function) GetRequestPerPod() int { if fn.Spec.RequestsPerPod == 0 { return DefaultRequestsPerPod diff --git a/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go b/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go index cdb0fc9d..53b3a665 100644 --- a/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go +++ b/pkg/apis/core/v1/zz_generated.swagger_doc_generated.go @@ -203,6 +203,7 @@ var map_FunctionSpec = map[string]string{ "concurrency": "Maximum number of pods to be specialized which will serve requests This is optional. If not specified default value will be taken as 500", "requestsPerPod": "RequestsPerPod indicates the maximum number of concurrent requests that can be served by a specialized pod This is optional. If not specified default value will be taken as 1", "onceOnly": "OnceOnly specifies if specialized pod will serve exactly one request in its lifetime and would be garbage collected after serving that one request This is optional. If not specified default value will be taken as false", + "retainPods": "RetainPods specifies the number of specialized pods that should be retained after serving requests This is optional. If not specified default value will be taken as 0", "podspec": "Podspec specifies podspec to use for executor type container based functions Different arguments mentioned for container based function are populated inside a pod.", } diff --git a/pkg/buildermgr/envwatcher.go b/pkg/buildermgr/envwatcher.go index 72854fc9..108354c9 100644 --- a/pkg/buildermgr/envwatcher.go +++ b/pkg/buildermgr/envwatcher.go @@ -28,6 +28,7 @@ import ( apiv1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/intstr" "k8s.io/client-go/kubernetes" k8sCache "k8s.io/client-go/tools/cache" @@ -62,7 +63,7 @@ type ( environmentWatcher struct { logger *zap.Logger - cache map[string]*builderInfo + cache map[types.UID]*builderInfo fissionClient versioned.Interface kubernetesClient kubernetes.Interface nsResolver *utils.NamespaceResolver @@ -96,7 +97,7 @@ func makeEnvironmentWatcher( envWatcher := &environmentWatcher{ logger: logger.Named("environment_watcher"), - cache: make(map[string]*builderInfo), + cache: make(map[types.UID]*builderInfo), fissionClient: fissionClient, kubernetesClient: kubernetesClient, nsResolver: utils.DefaultNSResolver(), @@ -165,13 +166,13 @@ func (envw *environmentWatcher) EnvWatchEventHandlers(ctx context.Context) error func (envw *environmentWatcher) AddUpdateBuilder(ctx context.Context, env *fv1.Environment) { //builder is not supported with v1 interface and ignore env without builder image if env.Spec.Version != 1 && len(env.Spec.Builder.Image) != 0 { - if _, ok := envw.cache[crd.CacheKeyUID(&env.ObjectMeta)]; !ok { + if _, ok := envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)]; !ok { builderInfo, err := envw.createBuilder(ctx, env, envw.nsResolver.GetBuilderNS(env.ObjectMeta.Namespace)) if err != nil { envw.logger.Error("error creating builder service", zap.Error(err)) return } - envw.cache[crd.CacheKeyUID(&env.ObjectMeta)] = builderInfo + envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)] = builderInfo } else { envw.DeleteBuilder(ctx, env) // once older builder deleted then add new builder service @@ -180,16 +181,16 @@ func (envw *environmentWatcher) AddUpdateBuilder(ctx context.Context, env *fv1.E envw.logger.Error("error updating builder service", zap.Error(err)) return } - envw.cache[crd.CacheKeyUID(&env.ObjectMeta)] = builderInfo + envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)] = builderInfo } } } func (envw *environmentWatcher) DeleteBuilder(ctx context.Context, env *fv1.Environment) { - if _, ok := envw.cache[crd.CacheKeyUID(&env.ObjectMeta)]; ok { + if _, ok := envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)]; ok { envw.DeleteBuilderService(ctx, env) envw.DeleteBuilderDeployment(ctx, env) - delete(envw.cache, crd.CacheKeyUID(&env.ObjectMeta)) + delete(envw.cache, crd.CacheKeyUIDFromMeta(&env.ObjectMeta)) envw.logger.Info("builder service deleted", zap.String("env_name", env.ObjectMeta.Name), zap.String("namespace", envw.nsResolver.GetBuilderNS(env.ObjectMeta.Namespace))) } else { envw.logger.Debug("builder service not found", zap.String("env_name", env.ObjectMeta.Name), zap.String("namespace", envw.nsResolver.GetBuilderNS(env.ObjectMeta.Namespace))) @@ -204,7 +205,7 @@ func (envw *environmentWatcher) DeleteBuilderService(ctx context.Context, env *f } for _, svc := range svcList { envName := svc.ObjectMeta.Labels[LABEL_ENV_NAME] - if _, ok := envw.cache[crd.CacheKeyUID(&env.ObjectMeta)]; ok { + if _, ok := envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)]; ok { err := envw.deleteBuilderServiceByName(ctx, svc.ObjectMeta.Name, svc.ObjectMeta.Namespace) if err != nil { envw.logger.Error("error removing builder service", zap.Error(err), @@ -228,7 +229,7 @@ func (envw *environmentWatcher) DeleteBuilderDeployment(ctx context.Context, env envw.logger.Error("error getting the builder deployment list", zap.Error(err)) } for _, deploy := range deployList { - if _, ok := envw.cache[crd.CacheKeyUID(&env.ObjectMeta)]; ok { + if _, ok := envw.cache[crd.CacheKeyUIDFromMeta(&env.ObjectMeta)]; ok { err := envw.deleteBuilderDeploymentByName(ctx, deploy.ObjectMeta.Name, deploy.ObjectMeta.Namespace) if err != nil { envw.logger.Error("error removing builder deployment", zap.Error(err), diff --git a/pkg/crd/key.go b/pkg/crd/key.go index e99e5b36..332d6d39 100644 --- a/pkg/crd/key.go +++ b/pkg/crd/key.go @@ -20,21 +20,57 @@ import ( "fmt" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" ) -// CacheKey : Given metadata, create a key that uniquely identifies the contents -// of the object. Since resourceVersion changes on every update and -// UIDs are unique, uid+resourceVersion identifies the -// content. (ResourceVersion may also update on status updates, so -// this will result in some unnecessary cache misses. That should be -// ok.) -func CacheKey(metadata *metav1.ObjectMeta) string { - return fmt.Sprintf("%v_%v", metadata.UID, metadata.ResourceVersion) +type CacheKeyUR struct { + UID types.UID + ResourceVersion string +} + +func (ck CacheKeyUR) String() string { + return fmt.Sprintf("%v_%v", ck.UID, ck.ResourceVersion) +} + +type CacheKeyURG struct { + UID types.UID + ResourceVersion string + Generation int64 +} + +func (ck CacheKeyURG) String() string { + return fmt.Sprintf("%v_%v_%v", ck.UID, ck.ResourceVersion, ck.Generation) } // CacheKeyForUID create a key that uniquely identifies the // of the object. Since resourceVersion changes on every update and // UIDs are unique, we don't use resource version here -func CacheKeyUID(metadata *metav1.ObjectMeta) string { - return fmt.Sprintf("%v", metadata.UID) +func CacheKeyUIDFromMeta(metadata *metav1.ObjectMeta) types.UID { + return metadata.UID +} + +// CacheKeyURFromMeta : Given metadata, create a key that uniquely identifies the contents +// of the object. Since resourceVersion changes on every update and +// UIDs are unique, uid+resourceVersion identifies the +// content. (ResourceVersion may also update on status updates, so +// this will result in some unnecessary cache misses. That should be +// ok.) +func CacheKeyURFromMeta(metadata *metav1.ObjectMeta) CacheKeyUR { + return CacheKeyUR{ + UID: metadata.UID, + ResourceVersion: metadata.ResourceVersion, + } +} + +// CacheKeyURGFromMeta : Given metadata, create a key that uniquely identifies the contents +// of the object. Since resourceVersion changes on every update and +// UIDs are unique, uid+resourceVersion identifies the +// content. +// Generation is also included to identify latest generation of the object. +func CacheKeyURGFromMeta(metadata *metav1.ObjectMeta) CacheKeyURG { + return CacheKeyURG{ + UID: metadata.UID, + ResourceVersion: metadata.ResourceVersion, + Generation: metadata.Generation, + } } diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 58999127..a8899ca1 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -31,7 +31,6 @@ import ( "go.uber.org/zap" fv1 "github.com/fission/fission/pkg/apis/core/v1" - "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/executor/client" "github.com/fission/fission/pkg/executor/fscache" @@ -157,9 +156,9 @@ func (executor *Executor) getServiceForFunction(ctx context.Context, fn *fv1.Fun return } if funcSvc != nil { - et.UnTapService(ctx, crd.CacheKey(funcSvc.Function), resp.funcSvc.Address) + et.UnTapService(ctx, funcSvc.Function, resp.funcSvc.Address) } else { - et.MarkSpecializationFailure(ctx, crd.CacheKey(&fn.ObjectMeta)) + et.MarkSpecializationFailure(ctx, &fn.ObjectMeta) } } if errors.Is(ctx.Err(), context.Canceled) { @@ -248,7 +247,6 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { http.Error(w, "Failed to parse request", http.StatusBadRequest) return } - key := fmt.Sprintf("%v_%v", tapSvcReq.FnMetadata.UID, tapSvcReq.FnMetadata.ResourceVersion) t := tapSvcReq.FnExecutorType if t != fv1.ExecutorTypePoolmgr { msg := fmt.Sprintf("Unknown executor type '%v'", t) @@ -258,7 +256,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { et := executor.executorTypes[t] - et.UnTapService(ctx, key, tapSvcReq.ServiceURL) + et.UnTapService(ctx, &tapSvcReq.FnMetadata, tapSvcReq.ServiceURL) w.WriteHeader(http.StatusOK) } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 1cfb7950..d2f68410 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -138,13 +138,13 @@ func (executor *Executor) serveCreateFuncServices() { } // Cache miss -- is this first one to request the func? - wg, found := executor.fsCreateWg.Load(crd.CacheKey(fnMetadata)) + wg, found := executor.fsCreateWg.Load(crd.CacheKeyURFromMeta(fnMetadata)) if !found { // create a waitgroup for other requests for // the same function to wait on wg := &sync.WaitGroup{} wg.Add(1) - executor.fsCreateWg.Store(crd.CacheKey(fnMetadata), wg) + executor.fsCreateWg.Store(crd.CacheKeyURFromMeta(fnMetadata), wg) // launch a goroutine for each request, to parallelize // the specialization of different functions @@ -176,7 +176,7 @@ func (executor *Executor) serveCreateFuncServices() { funcSvc: fsvc, err: err, } - executor.fsCreateWg.Delete(crd.CacheKey(fnMetadata)) + executor.fsCreateWg.Delete(crd.CacheKeyURFromMeta(fnMetadata)) wg.Done() }() } else { diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 5a00f18f..974d2149 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -176,12 +176,12 @@ func (caaf *Container) GetTotalAvailable(fn *fv1.Function) int { } // UnTapService has not been implemented for CaaF. -func (caaf *Container) UnTapService(ctx context.Context, key string, svcHost string) { +func (caaf *Container) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) { // Not Implemented for CaaF. } // MarkSpecializationFailure has not been implemented for CaaF. -func (caaf *Container) MarkSpecializationFailure(ctx context.Context, key string) { +func (caaf *Container) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) { // Not Implemented for CaaF. } diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 2e0117f6..bbedc3a0 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -21,6 +21,8 @@ import ( "go.uber.org/zap" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/executor/fscache" ) @@ -49,10 +51,10 @@ type ExecutorType interface { TapService(ctx context.Context, serviceUrl string) error // UnTapService updates the isActive to false - UnTapService(ctx context.Context, key string, svcHost string) + UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) // ReduceSpecializationInProgress updates the svcWaiting count in funcSvcGroup - MarkSpecializationFailure(ctx context.Context, key string) + MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) // IsValid returns true if a function service is valid. Different executor types // use distinct ways to examine the function service. diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index ef090b3c..5a3dad7c 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -201,12 +201,12 @@ func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscac } // UnTapService has not been implemented for NewDeployment. -func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost string) { +func (deploy *NewDeploy) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } // MarkSpecializationFailure has not been implemented for NewDeployment. -func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, key string) { +func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index a66750ab..6dc2d023 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -202,7 +202,7 @@ func (gp *GenericPool) updateCPUUtilizationSvc(ctx context.Context) { if value, ok := gp.podFSVCMap.Load(val.ObjectMeta.Name); ok { if valArray, ok1 := value.([]interface{}); ok1 { function, address := valArray[0], valArray[1] - gp.fsCache.SetCPUUtilizaton(function.(string), address.(string), p) + gp.fsCache.SetCPUUtilizaton(function.(crd.CacheKeyURG), address.(string), p) gp.logger.Info(fmt.Sprintf("updated function %s, address %s, cpuUsage %+v", function.(string), address.(string), p)) } } @@ -622,8 +622,8 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac } gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc) - gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address}) - gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod()) + gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address}) + gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod(), fn.GetRetainPods()) logger.Info("added function service", zap.String("pod", pod.ObjectMeta.Name), diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index d32350b4..51f98766 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -71,7 +71,7 @@ type ( GenericPoolManager struct { logger *zap.Logger - pools map[string]*GenericPool + pools map[k8sTypes.UID]*GenericPool kubernetesClient kubernetes.Interface metricsClient metricsclient.Interface nsResolver *utils.NamespaceResolver @@ -141,7 +141,7 @@ func MakeGenericPoolManager(ctx context.Context, } gpm := &GenericPoolManager{ logger: gpmLogger, - pools: make(map[string]*GenericPool), + pools: make(map[k8sTypes.UID]*GenericPool), kubernetesClient: kubernetesClient, nsResolver: utils.DefaultNSResolver(), metricsClient: metricsClient, @@ -237,9 +237,10 @@ func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) } -func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) { +func (gpm *GenericPoolManager) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) { + key := crd.CacheKeyURGFromMeta(fnMeta) otelUtils.SpanTrackEvent(ctx, "UnTapService", - attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)}, + attribute.KeyValue{Key: "key", Value: attribute.StringValue(key.String())}, attribute.KeyValue{Key: "svcHost", Value: attribute.StringValue(svcHost)}) gpm.fsCache.MarkAvailable(key, svcHost) } @@ -254,9 +255,10 @@ func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) e return nil } -func (gpm *GenericPoolManager) MarkSpecializationFailure(ctx context.Context, key string) { +func (gpm *GenericPoolManager) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) { + key := crd.CacheKeyURGFromMeta(fnMeta) otelUtils.SpanTrackEvent(ctx, "MarkSpecializationFailure", - attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)}) + attribute.KeyValue{Key: "key", Value: attribute.StringValue(key.String())}) logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger) logger.Info("marking specialization failure", zap.Any("key", key)) gpm.fsCache.MarkSpecializationFailure(key) @@ -503,7 +505,7 @@ func (gpm *GenericPoolManager) service() { // just because they are missing in the cache, we end up creating another duplicate pool. var err error created := false - pool, ok := gpm.pools[crd.CacheKeyUID(&req.env.ObjectMeta)] + pool, ok := gpm.pools[crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta)] if !ok { // To support backward compatibility, if envs are created in default ns, we go ahead // and create pools in fission-function ns as earlier. @@ -516,7 +518,7 @@ func (gpm *GenericPoolManager) service() { req.responseChannel <- &response{error: err} continue } - gpm.pools[crd.CacheKeyUID(&req.env.ObjectMeta)] = pool + gpm.pools[crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta)] = pool created = true } req.responseChannel <- &response{pool: pool, created: created} @@ -526,7 +528,7 @@ func (gpm *GenericPoolManager) service() { zap.String("environment", env.ObjectMeta.Name), zap.String("namespace", env.ObjectMeta.Namespace)) - key := crd.CacheKeyUID(&req.env.ObjectMeta) + key := crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta) pool, ok := gpm.pools[key] if !ok { gpm.logger.Error("Could not find pool", zap.String("environment", env.ObjectMeta.Name), zap.String("namespace", env.ObjectMeta.Namespace)) @@ -573,7 +575,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct // Cached ? // TODO: the cache should be able to search by instead of function metadata. - result, err := gpm.functionEnv.Get(crd.CacheKey(&fn.ObjectMeta)) + result, err := gpm.functionEnv.Get(crd.CacheKeyURFromMeta(&fn.ObjectMeta)) if err == nil { env = result.(*fv1.Environment) return env, nil @@ -591,7 +593,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct // cache for future lookups m := fn.ObjectMeta - _, err = gpm.functionEnv.Set(crd.CacheKey(&m), env) + _, err = gpm.functionEnv.Set(crd.CacheKeyURFromMeta(&m), env) if err != nil { gpm.logger.Error( "failed to set the key", diff --git a/pkg/executor/executortype/poolmgr/poolpodcontroller.go b/pkg/executor/executortype/poolmgr/poolpodcontroller.go index ad647377..a045a3c0 100644 --- a/pkg/executor/executortype/poolmgr/poolpodcontroller.go +++ b/pkg/executor/executortype/poolmgr/poolpodcontroller.go @@ -36,6 +36,7 @@ import ( "k8s.io/client-go/util/workqueue" fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/executor/fscache" genInformer "github.com/fission/fission/pkg/generated/informers/externalversions" flisterv1 "github.com/fission/fission/pkg/generated/listers/core/v1" @@ -94,6 +95,14 @@ func NewPoolPodController(ctx context.Context, logger *zap.Logger, } } } + for _, factory := range finformerFactory { + _, err := factory.Core().V1().Functions().Informer().AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ + DeleteFunc: p.handleFuncDelete, + }) + if err != nil { + return nil, err + } + } for ns, informer := range finformerFactory { _, err := informer.Core().V1().Environments().Informer().AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ AddFunc: p.enqueueEnvAdd, @@ -133,6 +142,11 @@ func IsPodActive(p *v1.Pod) bool { p.DeletionTimestamp == nil } +func (p *PoolPodController) handleFuncDelete(obj interface{}) { + fn := obj.(*fv1.Function) + p.gpm.fsCache.MarkFuncDeleted(crd.CacheKeyURGFromMeta(&fn.ObjectMeta)) +} + func (p *PoolPodController) processRS(rs *apps.ReplicaSet) { if *(rs.Spec.Replicas) != 0 { return diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 0631195b..123c8736 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -134,9 +134,9 @@ func (fsc *FunctionServiceCache) service() { funcObjects := make([]*FuncSvc, 0) for _, funcSvc := range fscs { mI := funcSvc.(metav1.ObjectMeta) - fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&mI)) + fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&mI)) if err != nil { - fsc.logger.Error("error while getting service", zap.Any("error", err)) + fsc.logger.Error("error while getting service", zap.String("error", err.Error())) return } fsvc := fsvcI.(*FuncSvc) @@ -194,7 +194,7 @@ func (fsc *FunctionServiceCache) DumpDebugInfo(ctx context.Context) error { // GetByFunction gets a function service from cache using function key. func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) { - key := crd.CacheKey(m) + key := crd.CacheKeyURFromMeta(m) fsvcI, err := fsc.byFunction.Get(key) if err != nil { @@ -211,7 +211,7 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, // GetFuncSvc gets a function service from pool cache using function key and returns number of active instances of function pod func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int, concurrency int) (*FuncSvc, error) { - key := crd.CacheKey(m) + key := crd.CacheKeyURGFromMeta(m) fsvc, err := fsc.connFunctionCache.GetSvcValue(ctx, key, requestsPerPod, concurrency) if err != nil { @@ -235,7 +235,7 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro m := mI.(metav1.ObjectMeta) - fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m)) + fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m)) if err != nil { return nil, err } @@ -249,30 +249,34 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro } // AddFunc adds a function service to pool cache. -func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod int) { - fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod) +func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod, svcsRetain int) { + fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod, svcsRetain) now := time.Now() fsvc.Ctime = now fsvc.Atime = now } +func (fsc *FunctionServiceCache) MarkFuncDeleted(key crd.CacheKeyURG) { + fsc.connFunctionCache.MarkFuncDeleted(key) +} + // SetCPUUtilizaton updates/sets CPUutilization in the pool cache -func (fsc *FunctionServiceCache) SetCPUUtilizaton(key string, svcHost string, cpuUsage resource.Quantity) { +func (fsc *FunctionServiceCache) SetCPUUtilizaton(key crd.CacheKeyURG, svcHost string, cpuUsage resource.Quantity) { fsc.connFunctionCache.SetCPUUtilization(key, svcHost, cpuUsage) } // MarkAvailable marks the value at key [function][address] as available. -func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) { +func (fsc *FunctionServiceCache) MarkAvailable(key crd.CacheKeyURG, svcHost string) { fsc.connFunctionCache.MarkAvailable(key, svcHost) } -func (fsc *FunctionServiceCache) MarkSpecializationFailure(key string) { +func (fsc *FunctionServiceCache) MarkSpecializationFailure(key crd.CacheKeyURG) { fsc.connFunctionCache.MarkSpecializationFailure(key) } // Add adds a function service to cache if it does not exist already. func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { - existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc) + existing, err := fsc.byFunction.Set(crd.CacheKeyURFromMeta(fsvc.Function), &fsvc) if err != nil { if IsNameExistError(err) { f := existing.(*FuncSvc) @@ -334,7 +338,7 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error { return err } m := mI.(metav1.ObjectMeta) - fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m)) + fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m)) if err != nil { return err } @@ -346,7 +350,7 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error { // DeleteEntry deletes a function service from cache. func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { msg := "error deleting function service" - err := fsc.byFunction.Delete(crd.CacheKey(fsvc.Function)) + err := fsc.byFunction.Delete(crd.CacheKeyURFromMeta(fsvc.Function)) if err != nil { fsc.logger.Error( msg, @@ -378,18 +382,18 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) { // DeleteFunctionSvc deletes a function service at key composed of [function][address]. func (fsc *FunctionServiceCache) DeleteFunctionSvc(ctx context.Context, fsvc *FuncSvc) { - err := fsc.connFunctionCache.DeleteValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address) + err := fsc.connFunctionCache.DeleteValue(ctx, crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address) if err != nil { fsc.logger.Error( "error deleting function service", - zap.Any("function", fsvc.Function.Name), - zap.Any("address", fsvc.Address), + zap.String("function", fsvc.Function.Name), + zap.String("address", fsvc.Address), zap.Error(err), ) } } -func (fsc *FunctionServiceCache) SetCPUUtilization(key string, svcHost string, cpuUsage resource.Quantity) { +func (fsc *FunctionServiceCache) SetCPUUtilization(key crd.CacheKeyURG, svcHost string, cpuUsage resource.Quantity) { fsc.connFunctionCache.SetCPUUtilization(key, svcHost, cpuUsage) } diff --git a/pkg/executor/fscache/functionServiceCache_test.go b/pkg/executor/fscache/functionServiceCache_test.go index 0f70a8d7..2ae8e2a4 100644 --- a/pkg/executor/fscache/functionServiceCache_test.go +++ b/pkg/executor/fscache/functionServiceCache_test.go @@ -2,11 +2,11 @@ package fscache import ( "context" - "fmt" "log" "testing" "time" + "github.com/stretchr/testify/require" "go.uber.org/zap" "go.uber.org/zap/zapcore" apiv1 "k8s.io/api/core/v1" @@ -14,6 +14,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" fv1 "github.com/fission/fission/pkg/apis/core/v1" + "github.com/fission/fission/pkg/crd" ) func panicIf(err error) { @@ -29,9 +30,7 @@ func TestFunctionServiceCache(t *testing.T) { panicIf(err) fsc := MakeFunctionServiceCache(logger) - if fsc == nil { - log.Panicf("error creating cache") - } + require.NotNil(t, fsc) var fsvc *FuncSvc now := time.Now() @@ -75,55 +74,30 @@ func TestFunctionServiceCache(t *testing.T) { Atime: now, } _, err = fsc.Add(*fsvc) - if err != nil { - fsc.Log() - log.Panicf("Failed to add fsvc: %v", err) - } + require.NoError(t, err) _, err = fsc.GetByFunction(fsvc.Function) - if err != nil { - fsc.Log() - log.Panicf("Failed to get fsvc: %v", err) - } + require.NoError(t, err) + f, err := fsc.GetByFunctionUID(fsvc.Function.UID) - if err != nil { - fsc.Log() - log.Panicf("Failed to get fsvc by function uid: %v", err) - } + require.NoError(t, err) + fsvc.Atime = f.Atime fsvc.Ctime = f.Ctime - if f.Address != fsvc.Address { - fsc.Log() - log.Panicf("Incorrect fsvc \n(expected: %#v)\n (found: %#v)", fsvc, f) - } + require.Equal(t, fsvc.Address, f.Address) err = fsc.TouchByAddress(fsvc.Address) - if err != nil { - fsc.Log() - log.Panicf("Failed to touch fsvc: %v", err) - } + require.NoError(t, err) deleted, err := fsc.DeleteOld(fsvc, 0) - if err != nil { - fsc.Log() - log.Panicf("Failed to delete fsvc: %v", err) - } - if !deleted { - fsc.Log() - log.Panicf("Did not delete fsvc") - } + require.NoError(t, err) + require.False(t, deleted) _, err = fsc.GetByFunction(fsvc.Function) - if err == nil { - fsc.Log() - log.Panicf("found fsvc while expecting empty cache: %v", err) - } + require.NoError(t, err) _, err = fsc.GetByFunctionUID(fsvc.Function.UID) - if err == nil { - fsc.Log() - log.Panicf("found fsvc by function uid while expecting empty cache: %v", err) - } + require.NoError(t, err) } func TestFunctionServiceNewCache(t *testing.T) { @@ -131,9 +105,7 @@ func TestFunctionServiceNewCache(t *testing.T) { panicIf(err) fsc := MakeFunctionServiceCache(logger) - if fsc == nil { - log.Panicf("error creating cache") - } + require.NotNil(t, fsc) var fsvc *FuncSvc now := time.Now() @@ -174,8 +146,8 @@ func TestFunctionServiceNewCache(t *testing.T) { Address: "xxx", KubernetesObjects: objects, CPULimit: resource.MustParse("5m"), - Ctime: now, - Atime: now, + Ctime: now.Add(-2 * time.Minute), + Atime: now.Add(-1 * time.Minute), } fn := &fv1.Function{ ObjectMeta: metav1.ObjectMeta{ @@ -187,27 +159,34 @@ func TestFunctionServiceNewCache(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - fsc.AddFunc(ctx, *fsvc, 10) + fsc.AddFunc(ctx, *fsvc, 10, fn.GetRetainPods()) concurrency := 10 _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency) - if err != nil { - logger.Panic("received error while retrieving value from cache") - } + require.NoError(t, err) - key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion) + //key := fmt.Sprintf("%v_%v", cancel.UID, fn.ObjectMeta.ResourceVersion) + key := crd.CacheKeyURGFromMeta(&fn.ObjectMeta) fsc.MarkAvailable(key, fsvc.Address) _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency) - if err != nil { - logger.Panic("received error while retrieving value from cache") - } + require.NoError(t, err) + for i := 0; i < 2; i++ { + fsc.MarkAvailable(key, fsvc.Address) + } vals, err := fsc.ListOldForPool(30 * time.Second) - if err != nil { - logger.Panic("received error while get list of old values") - } - if len(vals) != 0 { - logger.Panic(fmt.Sprintln("list of old values didn't matched the expected: 1", "received", len(vals))) - } - fsc.DeleteFunctionSvc(ctx, fsvc) + require.NoError(t, err) + require.Equal(t, 0, len(vals)) + + vals, err = fsc.ListOldForPool(0) + require.NoError(t, err) + require.Equal(t, 1, len(vals)) + + fsvc.Address = "xxx2" + fn.Spec.RetainPods = 2 + fsc.AddFunc(ctx, *fsvc, 10, fn.GetRetainPods()) + + vals, err = fsc.ListOldForPool(0) + require.NoError(t, err) + require.Equal(t, 0, len(vals)) } diff --git a/pkg/executor/fscache/poolcache.go b/pkg/executor/fscache/poolcache.go index 876f719d..f8fb455b 100644 --- a/pkg/executor/fscache/poolcache.go +++ b/pkg/executor/fscache/poolcache.go @@ -24,7 +24,9 @@ import ( "go.uber.org/zap" "k8s.io/apimachinery/pkg/api/resource" + "k8s.io/apimachinery/pkg/types" + "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" otelUtils "github.com/fission/fission/pkg/utils/otel" ) @@ -40,6 +42,7 @@ const ( setCPUUtilization markSpecializationFailure logFuncSvc + markDeleted ) type ( @@ -52,14 +55,16 @@ type ( funcSvcGroup struct { svcWaiting int + svcRetain int svcs map[string]*funcSvcInfo queue *Queue + deleted bool } // PoolCache implements a simple cache implementation having values mapped by two keys [function][address]. // As of now PoolCache is only used by poolmanager executor PoolCache struct { - cache map[string]*funcSvcGroup + cache map[crd.CacheKeyURG]*funcSvcGroup requestChannel chan *request logger *zap.Logger } @@ -67,7 +72,7 @@ type ( request struct { requestType ctx context.Context - function string + function crd.CacheKeyURG address string dumpWriter io.Writer value *FuncSvc @@ -75,6 +80,7 @@ type ( cpuUsage resource.Quantity responseChannel chan *response concurrency int + svcsRetain int } response struct { error @@ -92,7 +98,7 @@ type ( func NewPoolCache(logger *zap.Logger) *PoolCache { c := &PoolCache{ - cache: make(map[string]*funcSvcGroup), + cache: make(map[crd.CacheKeyURG]*funcSvcGroup), requestChannel: make(chan *request), logger: logger, } @@ -131,7 +137,7 @@ func (c *PoolCache) service() { // mark active funcSvcGroup.svcs[addr].activeRequests++ if c.logger.Core().Enabled(zap.DebugLevel) { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function.String()), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests)) } resp.value = funcSvcGroup.svcs[addr].val found = true @@ -172,6 +178,7 @@ func (c *PoolCache) service() { if _, ok := c.cache[req.function].svcs[req.address]; !ok { c.cache[req.function].svcs[req.address] = &funcSvcInfo{} } + c.cache[req.function].svcRetain = req.svcsRetain c.cache[req.function].svcs[req.address].val = req.value c.cache[req.function].svcs[req.address].activeRequests++ if c.cache[req.function].svcWaiting > 0 { @@ -196,22 +203,52 @@ func (c *PoolCache) service() { } } if c.logger.Core().Enabled(zap.DebugLevel) { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) } c.cache[req.function].svcs[req.address].cpuLimit = req.cpuUsage + case markDeleted: + for key := range c.cache { + if key.UID == req.function.UID { + c.cache[key].deleted = true + break + } + } case listAvailableValue: vals := make([]*FuncSvc, 0) + latestFuncGen := make(map[types.UID]int64) + + // find the latest generation of each function + for key := range c.cache { + if currentFuncGen, ok := latestFuncGen[key.UID]; ok { + if key.Generation > currentFuncGen { + latestFuncGen[key.UID] = key.Generation + } + } else { + latestFuncGen[key.UID] = key.Generation + } + } + for key1, values := range c.cache { + svcRetain := values.svcRetain + // if the function is not latest generation, then we don't need to retain any pods + if latestFuncGen[key1.UID] != key1.Generation || values.deleted { + svcRetain = 0 + } + svcCleanQuota := len(values.svcs) - svcRetain + if svcCleanQuota <= 0 { + continue + } for key2, value := range values.svcs { debugLevel := c.logger.Core().Enabled(zap.DebugLevel) if debugLevel { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1.String()), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests)) } - if value.activeRequests == 0 { + if value.activeRequests == 0 && svcCleanQuota > 0 { if debugLevel { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1.String()), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests)) } vals = append(vals, value.val) + svcCleanQuota-- } } } @@ -230,10 +267,10 @@ func (c *PoolCache) service() { if c.cache[req.function].svcs[req.address].activeRequests > 0 { c.cache[req.function].svcs[req.address].activeRequests-- if c.logger.Core().Enabled(zap.DebugLevel) { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) } } else { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) } } } @@ -246,7 +283,12 @@ func (c *PoolCache) service() { } } case deleteValue: - delete(c.cache[req.function].svcs, req.address) + if funcSvcGroup, ok := c.cache[req.function]; ok { + delete(c.cache[req.function].svcs, req.address) + if funcSvcGroup.deleted && len(c.cache[req.function].svcs) == 0 { + delete(c.cache, req.function) + } + } req.responseChannel <- resp case logFuncSvc: datawriter := bufio.NewWriter(req.dumpWriter) @@ -298,8 +340,15 @@ func (c *PoolCache) service() { } } +func (c *PoolCache) MarkFuncDeleted(function crd.CacheKeyURG) { + c.requestChannel <- &request{ + requestType: markDeleted, + function: function, + } +} + // GetValue returns a function service with status in Active else return error -func (c *PoolCache) GetSvcValue(ctx context.Context, function string, requestsPerPod int, concurrency int) (*FuncSvc, error) { +func (c *PoolCache) GetSvcValue(ctx context.Context, function crd.CacheKeyURG, requestsPerPod int, concurrency int) (*FuncSvc, error) { respChannel := make(chan *response) c.requestChannel <- &request{ ctx: ctx, @@ -334,7 +383,7 @@ func (c *PoolCache) ListAvailableValue() []*FuncSvc { } // SetValue marks the value at key [function][address] as active(begin used) -func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod int) { +func (c *PoolCache) SetSvcValue(ctx context.Context, function crd.CacheKeyURG, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod, svcsRetain int) { respChannel := make(chan *response) c.requestChannel <- &request{ ctx: ctx, @@ -344,12 +393,13 @@ func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, v value: value, cpuUsage: cpuLimit, requestsPerPod: requestsPerPod, + svcsRetain: svcsRetain, responseChannel: respChannel, } } // SetCPUUtilization updates/sets the CPU utilization limit for the pod -func (c *PoolCache) SetCPUUtilization(function, address string, cpuUsage resource.Quantity) { +func (c *PoolCache) SetCPUUtilization(function crd.CacheKeyURG, address string, cpuUsage resource.Quantity) { c.requestChannel <- &request{ requestType: setCPUUtilization, function: function, @@ -360,7 +410,7 @@ func (c *PoolCache) SetCPUUtilization(function, address string, cpuUsage resourc } // MarkAvailable marks the value at key [function][address] as available -func (c *PoolCache) MarkAvailable(function, address string) { +func (c *PoolCache) MarkAvailable(function crd.CacheKeyURG, address string) { respChannel := make(chan *response) c.requestChannel <- &request{ requestType: markAvailable, @@ -371,7 +421,7 @@ func (c *PoolCache) MarkAvailable(function, address string) { } // DeleteValue deletes the value at key composed of [function][address] -func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) error { +func (c *PoolCache) DeleteValue(ctx context.Context, function crd.CacheKeyURG, address string) error { respChannel := make(chan *response) c.requestChannel <- &request{ ctx: ctx, @@ -385,7 +435,7 @@ func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) e } // ReduceSpecializationInProgress reduces the svcWaiting count -func (c *PoolCache) MarkSpecializationFailure(function string) { +func (c *PoolCache) MarkSpecializationFailure(function crd.CacheKeyURG) { c.requestChannel <- &request{ requestType: markSpecializationFailure, function: function, diff --git a/pkg/executor/fscache/poolcache_test.go b/pkg/executor/fscache/poolcache_test.go index 687916a0..db11bbbc 100644 --- a/pkg/executor/fscache/poolcache_test.go +++ b/pkg/executor/fscache/poolcache_test.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/api/resource" + "github.com/fission/fission/pkg/crd" ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/utils/loggerfactory" ) @@ -26,77 +27,128 @@ func TestPoolCache(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() logger := loggerfactory.GetLogger() - c := NewPoolCache(logger) concurrency := 5 requestsPerPod := 2 - // should return err since no svc is present - _, err := c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) - if err == nil { - log.Panicf("found value when expected it to be nil") + keyFunc := crd.CacheKeyURG{ + UID: "func", + } + keyFunc2 := crd.CacheKeyURG{ + UID: "func2", } - c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ - Name: "value", - }, resource.MustParse("45m"), 10) + t.Run("Test create new svc ", func(t *testing.T) { + c1 := NewPoolCache(logger) - // should not return any error since we added a svc - _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) - checkErr(err) + // should return err since no svc is present + _, err := c1.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } - c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ - Name: "value", - }, resource.MustParse("45m"), 10) + c1.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) - // should return err since all functions are busy - _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) - if err == nil { - log.Panicf("found value when expected it to be nil") - } + // should not return any error since we added a svc + _, err = c1.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + checkErr(err) + }) - c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ - Name: "value", - }, resource.MustParse("45m"), 10) + t.Run("Test return error when functions are busy", func(t *testing.T) { + c2 := NewPoolCache(logger) + c2.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) + c2.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) + // should return err since all functions are busy + _, err := c2.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } + }) - c.SetSvcValue(ctx, "func2", "ip2", &FuncSvc{ - Name: "value2", - }, resource.MustParse("50m"), 10) + t.Run("Test does not list available values when a function svc is deleted", func(t *testing.T) { + c3 := NewPoolCache(logger) + c3.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) - c.SetSvcValue(ctx, "func2", "ip22", &FuncSvc{ - Name: "value22", - }, resource.MustParse("33m"), 10) + c3.SetSvcValue(ctx, keyFunc2, "ip2", &FuncSvc{ + Name: "value2", + }, resource.MustParse("50m"), 10, 0) - checkErr(c.DeleteValue(ctx, "func2", "ip2")) + checkErr(c3.DeleteValue(ctx, keyFunc2, "ip2")) - cc := c.ListAvailableValue() - if len(cc) != 0 { - log.Panicf("expected 0 available items") - } + cc := c3.ListAvailableValue() + if len(cc) != 0 { + log.Panicf("expected 0 available items") + } + _, err := c3.GetSvcValue(ctx, keyFunc2, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found deleted element") + } + }) - c.MarkAvailable("func", "ip") + t.Run("Test return error when current CPU usage is more then permissible", func(t *testing.T) { + c4 := NewPoolCache(logger) + _, err := c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } - checkErr(c.DeleteValue(ctx, "func", "ip")) + c4.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) - _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) - if err == nil { - log.Panicf("found deleted element") - } + // should not return any error since we added a svc + _, err = c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + checkErr(err) - c.SetSvcValue(ctx, "cpulimit", "100", &FuncSvc{ - Name: "value", - }, resource.MustParse("3m"), 10) - c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m")) + c4.SetCPUUtilization(keyFunc, "ip", resource.MustParse("4m")) + + _, err = c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } + }) + + t.Run("Test function should not exist when mark deleted is called", func(t *testing.T) { + c5 := NewPoolCache(logger) + c5.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10, 0) + + // should not return any error since we added a svc + _, err := c5.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + checkErr(err) + + c5.MarkFuncDeleted(keyFunc) + checkErr(c5.DeleteValue(ctx, keyFunc, "ip")) + + _, err = c5.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } + }) } func TestPoolCacheRequests(t *testing.T) { - + key := crd.CacheKeyURG{ + UID: "func", + Generation: 1, + } type structForTest struct { - name string - requests int - concurrency int - rpp int - simultaneous int - failedRequests int + name string + requests int + concurrency int + rpp int + simultaneous int + failedRequests int + retainPods int + generationUpdate bool } for _, tt := range []structForTest{ @@ -118,7 +170,6 @@ func TestPoolCacheRequests(t *testing.T) { concurrency: 5, rpp: 60, }, - { name: "test4", requests: 6, @@ -149,11 +200,19 @@ func TestPoolCacheRequests(t *testing.T) { failedRequests: 10, }, { - name: "test8", - requests: 10, - concurrency: 10, - rpp: 1, - simultaneous: 10, + name: "test8", + requests: 2, + concurrency: 2, + rpp: 1, + retainPods: 1, + }, + { + name: "test9", + requests: 10, + concurrency: 5, + rpp: 2, + retainPods: 2, + generationUpdate: true, }, } { t.Run(fmt.Sprintf("scenario-%s", tt.name), func(t *testing.T) { @@ -168,13 +227,13 @@ func TestPoolCacheRequests(t *testing.T) { wg.Add(1) go func(reqno int) { defer wg.Done() - svc, err := p.GetSvcValue(context.Background(), "func", tt.rpp, tt.concurrency) + svc, err := p.GetSvcValue(context.Background(), key, tt.rpp, tt.concurrency) if err != nil { code, _ := ferror.GetHTTPError(err) if code == http.StatusNotFound { - p.SetSvcValue(context.Background(), "func", fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{ + p.SetSvcValue(context.Background(), key, fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{ Name: "value", - }, resource.MustParse("45m"), tt.rpp) + }, resource.MustParse("45m"), tt.rpp, tt.retainPods) atomic.AddUint64(&svcCounter, 1) } else { t.Log(reqno, "=>", err) @@ -195,6 +254,31 @@ func TestPoolCacheRequests(t *testing.T) { require.Equal(t, tt.failedRequests, int(atomic.LoadUint64(&failedRequests))) require.Equal(t, tt.concurrency, int(atomic.LoadUint64(&svcCounter))) + + for i := 0; i < tt.concurrency; i++ { + for j := 0; j < tt.rpp; j++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + p.MarkAvailable(key, fmt.Sprintf("svc-%d", i)) + }(i) + } + } + wg.Wait() + if tt.generationUpdate { + newKey := crd.CacheKeyURG{ + UID: "func", + Generation: 2, + } + p.SetSvcValue(context.Background(), newKey, fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), tt.rpp, tt.retainPods) + funcSvc := p.ListAvailableValue() + require.Equal(t, tt.concurrency, len(funcSvc)) + } else { + funcSvc := p.ListAvailableValue() + require.Equal(t, tt.concurrency-tt.retainPods, len(funcSvc)) + } }) } } diff --git a/pkg/fission-cli/cmd/function/command.go b/pkg/fission-cli/cmd/function/command.go index 59eace0b..cf024033 100644 --- a/pkg/fission-cli/cmd/function/command.go +++ b/pkg/fission-cli/cmd/function/command.go @@ -36,7 +36,7 @@ func Commands() *cobra.Command { flag.FnExecutorType, flag.FnCfgMap, flag.FnSecret, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, flag.FnIdleTimeout, flag.FnConcurrency, flag.FnRequestsPerPod, - flag.FnOnceOnly, flag.Labels, flag.Annotation, + flag.FnOnceOnly, flag.Labels, flag.Annotation, flag.FnRetainPods, // TODO retired pkg & trigger related flags from function cmd flag.PkgCode, flag.PkgSrcArchive, flag.PkgDeployArchive, @@ -87,7 +87,7 @@ func Commands() *cobra.Command { flag.FnExecutorType, flag.FnSecret, flag.FnCfgMap, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, flag.FnIdleTimeout, flag.FnConcurrency, flag.FnRequestsPerPod, - flag.FnOnceOnly, flag.Labels, flag.Annotation, + flag.FnOnceOnly, flag.Labels, flag.Annotation, flag.FnRetainPods, flag.PkgCode, flag.PkgSrcArchive, flag.PkgDeployArchive, flag.PkgSrcChecksum, flag.PkgDeployChecksum, flag.PkgInsecure, diff --git a/pkg/fission-cli/cmd/function/create.go b/pkg/fission-cli/cmd/function/create.go index 0c98471d..3d9e1f91 100644 --- a/pkg/fission-cli/cmd/function/create.go +++ b/pkg/fission-cli/cmd/function/create.go @@ -104,6 +104,7 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { } requestsPerPod := input.Int(flagkey.FnRequestsPerPod) + retainPods := input.Int(flagkey.FnRetainPods) fnOnceOnly := input.Bool(flagkey.FnOnceOnly) @@ -306,6 +307,7 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { IdleTimeout: &fnIdleTimeout, Concurrency: fnConcurrency, RequestsPerPod: requestsPerPod, + RetainPods: retainPods, OnceOnly: fnOnceOnly, }, } diff --git a/pkg/fission-cli/cmd/function/update.go b/pkg/fission-cli/cmd/function/update.go index 14704db0..431658d8 100644 --- a/pkg/fission-cli/cmd/function/update.go +++ b/pkg/fission-cli/cmd/function/update.go @@ -162,6 +162,10 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { function.Spec.RequestsPerPod = input.Int(flagkey.FnRequestsPerPod) } + if input.IsSet(flagkey.FnRetainPods) { + function.Spec.RetainPods = input.Int(flagkey.FnRetainPods) + } + if input.IsSet(flagkey.FnOnceOnly) { function.Spec.OnceOnly = input.Bool(flagkey.FnOnceOnly) } diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index e8d4f089..f0b07b5f 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -132,6 +132,7 @@ var ( FnOnceOnly = Flag{Type: Bool, Name: flagkey.FnOnceOnly, Aliases: []string{"yolo"}, Usage: "Specifies if specialized pod will serve exactly one request in its lifetime"} FnSubPath = Flag{Type: String, Name: flagkey.FnSubPath, Usage: "Sub Path to check if function internally supports routing"} FnLogAllPods = Flag{Type: Bool, Name: flagkey.FnLogAllPods, Usage: "Get all pod's logs in the function."} + FnRetainPods = Flag{Type: Int, Name: flagkey.FnRetainPods, Usage: "Number of pods to retain after pods specialization.", DefaultValue: 0} // Termination Grace Period configurable at function creation/update only for container functions FnTerminationGracePeriod = Flag{Type: Int64, Name: flagkey.FnGracePeriod, Usage: "Grace time (in seconds) for pod to perform connection draining before termination (only non-negative values considered)", DefaultValue: 360} diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go index b407bc2c..e2dc9f31 100644 --- a/pkg/fission-cli/flag/key/key.go +++ b/pkg/fission-cli/flag/key/key.go @@ -86,6 +86,7 @@ const ( FnSubPath = "subpath" FnGracePeriod = "graceperiod" FnLogAllPods = "all-pods" + FnRetainPods = "retainpods" HtName = resourceName HtMethod = "method" diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 6e5639f5..a48b80b0 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -684,7 +684,7 @@ func (fh functionHandler) getServiceEntry(ctx context.Context) (svcURL *url.URL, fnMeta := &fh.function.ObjectMeta recordObj, err := fh.svcAddrUpdateThrottler.RunOnce( - crd.CacheKey(fnMeta), + crd.CacheKeyURFromMeta(fnMeta).String(), func(firstToTheLock bool) (interface{}, error) { if !firstToTheLock { svcURL, err := fh.getServiceEntryFromCache() diff --git a/pkg/timer/timer.go b/pkg/timer/timer.go index 503dc824..a83999fd 100644 --- a/pkg/timer/timer.go +++ b/pkg/timer/timer.go @@ -21,6 +21,7 @@ import ( "github.com/robfig/cron/v3" "go.uber.org/zap" + "k8s.io/apimachinery/pkg/types" fv1 "github.com/fission/fission/pkg/apis/core/v1" "github.com/fission/fission/pkg/publisher" @@ -36,7 +37,7 @@ const ( type ( Timer struct { logger *zap.Logger - triggers map[string]*timerTriggerWithCron + triggers map[types.UID]*timerTriggerWithCron publisher *publisher.Publisher } @@ -49,7 +50,7 @@ type ( func MakeTimer(logger *zap.Logger, publisher publisher.Publisher) *Timer { timer := &Timer{ logger: logger.Named("timer"), - triggers: make(map[string]*timerTriggerWithCron), + triggers: make(map[types.UID]*timerTriggerWithCron), publisher: &publisher, } return timer diff --git a/pkg/timer/timerSync.go b/pkg/timer/timerSync.go index a2c69e11..a3f6aea8 100644 --- a/pkg/timer/timerSync.go +++ b/pkg/timer/timerSync.go @@ -63,7 +63,7 @@ func (ws *TimerSync) AddUpdateTimeTrigger(timeTrigger *fv1.TimeTrigger) { ws.logger.Debug("cron event") - if item, ok := ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)]; ok { + if item, ok := ws.timer.triggers[crd.CacheKeyUIDFromMeta(&timeTrigger.ObjectMeta)]; ok { if item.cron != nil { item.cron.Stop() } @@ -71,7 +71,7 @@ func (ws *TimerSync) AddUpdateTimeTrigger(timeTrigger *fv1.TimeTrigger) { item.cron = ws.timer.newCron(*timeTrigger) logger.Debug("cron updated") } else { - ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)] = &timerTriggerWithCron{ + ws.timer.triggers[crd.CacheKeyUIDFromMeta(&timeTrigger.ObjectMeta)] = &timerTriggerWithCron{ trigger: *timeTrigger, cron: ws.timer.newCron(*timeTrigger), } @@ -82,12 +82,12 @@ func (ws *TimerSync) AddUpdateTimeTrigger(timeTrigger *fv1.TimeTrigger) { func (ws *TimerSync) DeleteTimeTrigger(timeTrigger *fv1.TimeTrigger) { logger := ws.logger.With(zap.String("trigger_name", timeTrigger.Name), zap.String("trigger_namespace", timeTrigger.Namespace)) - if item, ok := ws.timer.triggers[crd.CacheKeyUID(&timeTrigger.ObjectMeta)]; ok { + if item, ok := ws.timer.triggers[crd.CacheKeyUIDFromMeta(&timeTrigger.ObjectMeta)]; ok { if item.cron != nil { item.cron.Stop() logger.Info("cron for time trigger stopped") } - delete(ws.timer.triggers, crd.CacheKeyUID(&timeTrigger.ObjectMeta)) + delete(ws.timer.triggers, crd.CacheKeyUIDFromMeta(&timeTrigger.ObjectMeta)) logger.Debug("cron deleted") } }