diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index bd0e7d89..8c7f2148 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -34,6 +34,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" appsinformers "k8s.io/client-go/informers/apps/v1" coreinformers "k8s.io/client-go/informers/core/v1" "k8s.io/client-go/kubernetes" @@ -141,7 +142,7 @@ func (caaf *Container) Run(ctx context.Context) { if ok := k8sCache.WaitForCacheSync(ctx.Done(), caaf.deplListerSynced, caaf.svcListerSynced); !ok { caaf.logger.Fatal("failed to wait for caches to sync") } - go caaf.idleObjectReaper() + go caaf.idleObjectReaper(ctx) } // GetTypeName returns the executor type name. @@ -168,7 +169,7 @@ func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac // GetFuncSvcFromCache returns a function service from cache; error otherwise. func (caaf *Container) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache", otelUtils.GetAttributesForFunction(fn)...) - return caaf.fsCache.GetByFunction(&fn.ObjectMeta) + return caaf.fsCache.GetByFunctionUID(fn.UID) } // DeleteFuncSvcFromCache deletes a function service from cache. @@ -703,73 +704,71 @@ func (caaf *Container) updateStatus(fn *fv1.Function, err error, message string) } // idleObjectReaper reaps objects after certain idle time -func (caaf *Container) idleObjectReaper() { - ctx := context.Background() +func (caaf *Container) idleObjectReaper(ctx context.Context) { + // calling function doIdleObjectReaper() repeatedly at given interval of time + wait.UntilWithContext(ctx, caaf.doIdleObjectReaper, time.Second*5) +} - pollSleep := 5 * time.Second - for { - time.Sleep(pollSleep) +func (caaf *Container) doIdleObjectReaper(ctx context.Context) { + funcSvcs, err := caaf.fsCache.ListOld(time.Second * 5) + if err != nil { + caaf.logger.Error("error reaping idle pods", zap.Error(err)) + return + } - funcSvcs, err := caaf.fsCache.ListOld(pollSleep) - if err != nil { - caaf.logger.Error("error reaping idle pods", zap.Error(err)) + for i := range funcSvcs { + fsvc := funcSvcs[i] + + if fsvc.Executor != fv1.ExecutorTypeContainer { continue } - for i := range funcSvcs { - fsvc := funcSvcs[i] - - if fsvc.Executor != fv1.ExecutorTypeContainer { + fn, err := caaf.fissionClient.CoreV1().Functions(fsvc.Function.Namespace).Get(ctx, fsvc.Function.Name, metav1.GetOptions{}) + if err != nil { + // CaaF manager handles the function delete event and clean cache/kubeobjs itself, + // so we ignore the not found error for functions with CaaF executor type here. + if k8sErrs.IsNotFound(err) && fsvc.Executor == fv1.ExecutorTypeContainer { continue } - - fn, err := caaf.fissionClient.CoreV1().Functions(fsvc.Function.Namespace).Get(ctx, fsvc.Function.Name, metav1.GetOptions{}) - if err != nil { - // CaaF manager handles the function delete event and clean cache/kubeobjs itself, - // so we ignore the not found error for functions with CaaF executor type here. - if k8sErrs.IsNotFound(err) && fsvc.Executor == fv1.ExecutorTypeContainer { - continue - } - caaf.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name)) - continue - } - - idlePodReapTime := caaf.defaultIdlePodReapTime - if fn.Spec.IdleTimeout != nil { - idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second - } - - if time.Since(fsvc.Atime) < idlePodReapTime { - continue - } - - go func() { - deployObj := getDeploymentObj(fsvc.KubernetesObjects) - if deployObj == nil { - caaf.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - return - } - - currentDeploy, err := caaf.kubernetesClient.AppsV1(). - Deployments(deployObj.Namespace).Get(ctx, deployObj.Name, metav1.GetOptions{}) - if err != nil { - caaf.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - return - } - - minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - - // do nothing if the current replicas is already lower than minScale - if *currentDeploy.Spec.Replicas <= minScale { - return - } - - err = caaf.scaleDeployment(ctx, deployObj.Namespace, deployObj.Name, minScale) - if err != nil { - caaf.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - } - }() + caaf.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name)) + continue } + + idlePodReapTime := caaf.defaultIdlePodReapTime + if fn.Spec.IdleTimeout != nil { + idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second + } + + if time.Since(fsvc.Atime) < idlePodReapTime { + continue + } + + go func() { + deployObj := getDeploymentObj(fsvc.KubernetesObjects) + if deployObj == nil { + caaf.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return + } + + currentDeploy, err := caaf.kubernetesClient.AppsV1(). + Deployments(deployObj.Namespace).Get(ctx, deployObj.Name, metav1.GetOptions{}) + if err != nil { + caaf.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return + } + + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + + // do nothing if the current replicas is already lower than minScale + if *currentDeploy.Spec.Replicas <= minScale { + return + } + + err = caaf.scaleDeployment(ctx, deployObj.Namespace, deployObj.Name, minScale) + if err != nil { + caaf.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + } + }() } } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 56ceb82e..2c9c8b77 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -35,6 +35,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" appsinformers "k8s.io/client-go/informers/apps/v1" coreinformers "k8s.io/client-go/informers/core/v1" "k8s.io/client-go/kubernetes" @@ -169,7 +170,7 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc // GetFuncSvcFromCache returns a function service from cache; error otherwise. func (deploy *NewDeploy) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache") - return deploy.fsCache.GetByFunction(&fn.ObjectMeta) + return deploy.fsCache.GetByFunctionUID(fn.UID) } // DeleteFuncSvcFromCache deletes a function service from cache. @@ -593,6 +594,7 @@ func (deploy *NewDeploy) updateFunction(ctx context.Context, oldFn *fv1.Function if oldFn.Spec.Environment != newFn.Spec.Environment || oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef || oldFn.Spec.Package.FunctionName != newFn.Spec.Package.FunctionName { + deploy.logger.Debug("deployment changed", zap.String("msg", "deployment changed")) deployChanged = true } @@ -766,88 +768,87 @@ func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message strin // idleObjectReaper reaps objects after certain idle time func (deploy *NewDeploy) idleObjectReaper(ctx context.Context) { - pollSleep := 5 * time.Second - for { - time.Sleep(pollSleep) + // calling function doIdleObjectReaper() repeatedly at given interval of time + wait.UntilWithContext(ctx, deploy.doIdleObjectReaper, time.Second*5) +} - envs, err := deploy.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) - if err != nil { - deploy.logger.Fatal("failed to get environment list", zap.Error(err)) - } +func (deploy *NewDeploy) doIdleObjectReaper(ctx context.Context) { + envs, err := deploy.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) + if err != nil { + deploy.logger.Fatal("failed to get environment list", zap.Error(err)) + } - envList := make(map[k8sTypes.UID]struct{}) - for _, env := range envs.Items { - envList[env.ObjectMeta.UID] = struct{}{} - } + envList := make(map[k8sTypes.UID]struct{}) + for _, env := range envs.Items { + envList[env.ObjectMeta.UID] = struct{}{} + } - funcSvcs, err := deploy.fsCache.ListOld(pollSleep) - if err != nil { - deploy.logger.Error("error reaping idle pods", zap.Error(err)) + funcSvcs, err := deploy.fsCache.ListOld(time.Second * 5) + if err != nil { + deploy.logger.Error("error reaping idle pods", zap.Error(err)) + return + } + + for i := range funcSvcs { + fsvc := funcSvcs[i] + if fsvc.Executor != fv1.ExecutorTypeNewdeploy { continue } - for i := range funcSvcs { - fsvc := funcSvcs[i] - - if fsvc.Executor != fv1.ExecutorTypeNewdeploy { - continue - } - - // For function with the environment that no longer exists, executor - // scales down the deployment as usual and prints log to notify user. - if _, ok := envList[fsvc.Environment.ObjectMeta.UID]; !ok { - deploy.logger.Warn("function environment no longer exists", - zap.String("environment", fsvc.Environment.ObjectMeta.Name), - zap.String("function", fsvc.Name)) - } - - fn, err := deploy.fissionClient.CoreV1().Functions(fsvc.Function.Namespace).Get(ctx, fsvc.Function.Name, metav1.GetOptions{}) - 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 == fv1.ExecutorTypeNewdeploy { - continue - } - deploy.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name)) - continue - } - - idlePodReapTime := deploy.defaultIdlePodReapTime - if fn.Spec.IdleTimeout != nil { - idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second - } - - if time.Since(fsvc.Atime) < idlePodReapTime { - continue - } - - go func() { - deployObj := getDeploymentObj(fsvc.KubernetesObjects) - if deployObj == nil { - deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - return - } - - currentDeploy, err := deploy.kubernetesClient.AppsV1(). - Deployments(deployObj.Namespace).Get(ctx, deployObj.Name, metav1.GetOptions{}) - if err != nil { - deploy.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - return - } - - minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - - // do nothing if the current replicas is already lower than minScale - if *currentDeploy.Spec.Replicas <= minScale { - return - } - - err = deploy.scaleDeployment(ctx, deployObj.Namespace, deployObj.Name, minScale) - if err != nil { - deploy.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - } - }() + // For function with the environment that no longer exists, executor + // scales down the deployment as usual and prints log to notify user. + if _, ok := envList[fsvc.Environment.ObjectMeta.UID]; !ok { + deploy.logger.Warn("function environment no longer exists", + zap.String("environment", fsvc.Environment.ObjectMeta.Name), + zap.String("function", fsvc.Name)) } + + fn, err := deploy.fissionClient.CoreV1().Functions(fsvc.Function.Namespace).Get(ctx, fsvc.Function.Name, metav1.GetOptions{}) + 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 == fv1.ExecutorTypeNewdeploy { + continue + } + deploy.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name)) + continue + } + + idlePodReapTime := deploy.defaultIdlePodReapTime + if fn.Spec.IdleTimeout != nil { + idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second + } + + if time.Since(fsvc.Atime) < idlePodReapTime { + continue + } + + go func() { + deployObj := getDeploymentObj(fsvc.KubernetesObjects) + if deployObj == nil { + deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return + } + + currentDeploy, err := deploy.kubernetesClient.AppsV1(). + Deployments(deployObj.Namespace).Get(ctx, deployObj.Name, metav1.GetOptions{}) + if err != nil { + deploy.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return + } + + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + + // do nothing if the current replicas is already lower than minScale + if *currentDeploy.Spec.Replicas <= minScale { + return + } + + err = deploy.scaleDeployment(ctx, deployObj.Namespace, deployObj.Name, minScale) + if err != nil { + deploy.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + } + }() } } diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index d54fa2a6..9350a5df 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -35,6 +35,7 @@ import ( "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" k8sTypes "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/watch" appsinformers "k8s.io/client-go/informers/apps/v1" coreinformers "k8s.io/client-go/informers/core/v1" @@ -169,7 +170,7 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) { gpm.poolPodC.InjectGpm(gpm) go gpm.WebsocketStartEventChecker(gpm.kubernetesClient) go gpm.NoActiveConnectionEventChecker(gpm.kubernetesClient) - go gpm.idleObjectReaper() + go gpm.idleObjectReaper(ctx) go gpm.poolPodC.Run(ctx.Done()) } @@ -569,95 +570,93 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct } // idleObjectReaper reaps objects after certain idle time -func (gpm *GenericPoolManager) idleObjectReaper() { - ctx := context.Background() - pollSleep := 5 * time.Second +func (gpm *GenericPoolManager) idleObjectReaper(ctx context.Context) { + // calling function doIdleObjectReaper() repeatedly at given interval of time + wait.UntilWithContext(ctx, gpm.doIdleObjectReaper, time.Second*5) +} - for { - time.Sleep(pollSleep) +func (gpm *GenericPoolManager) doIdleObjectReaper(ctx context.Context) { + envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) + if err != nil { + gpm.logger.Error("failed to get environment list", zap.Error(err)) + return + } - envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) - if err != nil { - gpm.logger.Error("failed to get environment list", zap.Error(err)) + envList := make(map[k8sTypes.UID]struct{}) + for _, env := range envs.Items { + envList[env.ObjectMeta.UID] = struct{}{} + } + + fns, err := gpm.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) + if err != nil { + gpm.logger.Error("failed to get environment list", zap.Error(err)) + return + } + + fnList := make(map[k8sTypes.UID]fv1.Function) + for i, fn := range fns.Items { + fnList[fn.ObjectMeta.UID] = fns.Items[i] + } + + funcSvcs, err := gpm.fsCache.ListOldForPool(time.Second * 5) + if err != nil { + gpm.logger.Error("error reaping idle pods", zap.Error(err)) + return + } + + for i := range funcSvcs { + fsvc := funcSvcs[i] + + if fsvc.Executor != fv1.ExecutorTypePoolmgr { continue } - envList := make(map[k8sTypes.UID]struct{}) - for _, env := range envs.Items { - envList[env.ObjectMeta.UID] = struct{}{} + if _, ok := gpm.fsCache.WebsocketFsvc.Load(fsvc.Name); ok { + continue + } + // For function with the environment that no longer exists, executor + // cleanups the idle pod as usual and prints log to notify user. + if _, ok := envList[fsvc.Environment.ObjectMeta.UID]; !ok { + gpm.logger.Warn("function environment no longer exists", + zap.String("environment", fsvc.Environment.ObjectMeta.Name), + zap.String("function", fsvc.Name)) } - fns, err := gpm.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(ctx, metav1.ListOptions{}) - if err != nil { - gpm.logger.Error("failed to get environment list", zap.Error(err)) + if fsvc.Environment.Spec.AllowedFunctionsPerContainer == fv1.AllowedFunctionsPerContainerInfinite { continue } - fnList := make(map[k8sTypes.UID]fv1.Function) - for i, fn := range fns.Items { - fnList[fn.ObjectMeta.UID] = fns.Items[i] + idlePodReapTime := gpm.defaultIdlePodReapTime + if fn, ok := fnList[fsvc.Function.UID]; ok { + if fn.Spec.IdleTimeout != nil { + idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second + } } - funcSvcs, err := gpm.fsCache.ListOldForPool(pollSleep) - if err != nil { - gpm.logger.Error("error reaping idle pods", zap.Error(err)) + if time.Since(fsvc.Atime) < idlePodReapTime { continue } - for i := range funcSvcs { - fsvc := funcSvcs[i] - - if fsvc.Executor != fv1.ExecutorTypePoolmgr { - continue + go func() { + deleted, err := gpm.fsCache.DeleteOldPoolCache(ctx, fsvc, idlePodReapTime) + if err != nil { + gpm.logger.Error("error deleting Kubernetes objects for function service", + zap.Error(err), + zap.Any("service", fsvc)) } - - if _, ok := gpm.fsCache.WebsocketFsvc.Load(fsvc.Name); ok { - continue - } - // For function with the environment that no longer exists, executor - // cleanups the idle pod as usual and prints log to notify user. - if _, ok := envList[fsvc.Environment.ObjectMeta.UID]; !ok { - gpm.logger.Warn("function environment no longer exists", - zap.String("environment", fsvc.Environment.ObjectMeta.Name), - zap.String("function", fsvc.Name)) - } - - if fsvc.Environment.Spec.AllowedFunctionsPerContainer == fv1.AllowedFunctionsPerContainerInfinite { - continue - } - - idlePodReapTime := gpm.defaultIdlePodReapTime - if fn, ok := fnList[fsvc.Function.UID]; ok { - if fn.Spec.IdleTimeout != nil { - idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second + if deleted { + for i := range fsvc.KubernetesObjects { + gpm.logger.Info("release idle function resources", + zap.String("function", fsvc.Function.Name), + zap.String("address", fsvc.Address), + zap.String("executor", string(fsvc.Executor)), + zap.String("pod", fsvc.Name), + ) + reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) + time.Sleep(50 * time.Millisecond) } } - - if time.Since(fsvc.Atime) < idlePodReapTime { - continue - } - - go func() { - deleted, err := gpm.fsCache.DeleteOldPoolCache(ctx, fsvc, 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.Info("release idle function resources", - zap.String("function", fsvc.Function.Name), - zap.String("address", fsvc.Address), - zap.String("executor", string(fsvc.Executor)), - zap.String("pod", fsvc.Name), - ) - reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) - time.Sleep(50 * time.Millisecond) - } - } - }() - } + }() } } diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index ae7e8d01..bbe0bf5f 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -130,10 +130,16 @@ func (fsc *FunctionServiceCache) service() { resp.error = fsc._touchByAddress(req.address) case LISTOLD: // get svcs idle for > req.age - fscs := fsc.byFunction.Copy() + fscs := fsc.byFunctionUID.Copy() funcObjects := make([]*FuncSvc, 0) for _, funcSvc := range fscs { - fsvc := funcSvc.(*FuncSvc) + mI := funcSvc.(metav1.ObjectMeta) + fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&mI)) + if err != nil { + fsc.logger.Error("error while getting service", zap.Any("error", err)) + return + } + fsvc := fsvcI.(*FuncSvc) if time.Since(fsvc.Atime) > req.age { funcObjects = append(funcObjects, fsvc) }