diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index 0d120b2e..4b0f160d 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -343,6 +343,12 @@ type ( // a particular function execution should be complete. // This is optional. If not specified default value will be taken as 60s FunctionTimeout int `json:"functionTimeout,omitempty"` + + // IdleTimeout specifies the length of time that a function is idle before the + // function pod(s) are eligible for deletion. If no traffic to the function + // is detected within the idle timeout, the executor will then recycle the + // function pod(s) to release resources. + IdleTimeout *int `json:"idletimeout,omitempty"` } // InvokeStrategy is a set of controls over how the function executes. diff --git a/pkg/apis/core/v1/zz_generated.deepcopy.go b/pkg/apis/core/v1/zz_generated.deepcopy.go index fb052af1..d71d38e3 100644 --- a/pkg/apis/core/v1/zz_generated.deepcopy.go +++ b/pkg/apis/core/v1/zz_generated.deepcopy.go @@ -426,6 +426,11 @@ func (in *FunctionSpec) DeepCopyInto(out *FunctionSpec) { } in.Resources.DeepCopyInto(&out.Resources) out.InvokeStrategy = in.InvokeStrategy + if in.IdleTimeout != nil { + in, out := &in.IdleTimeout, &out.IdleTimeout + *out = new(int) + **out = **in + } return } diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 909fa0ac..8da83b44 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -74,7 +74,7 @@ type ( envStore k8sCache.Store envController k8sCache.Controller - idlePodReapTime time.Duration + defaultIdlePodReapTime time.Duration } ) @@ -112,7 +112,7 @@ func MakeNewDeploy( runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")), useIstio: enableIstio, - idlePodReapTime: 2 * time.Minute, + defaultIdlePodReapTime: 2 * time.Minute, } if nd.crdClient != nil { @@ -769,7 +769,7 @@ func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message strin // idleObjectReaper reaps objects after certain idle time func (deploy *NewDeploy) idleObjectReaper() { - pollSleep := time.Duration(deploy.idlePodReapTime) + pollSleep := 5 * time.Second for { time.Sleep(pollSleep) @@ -783,13 +783,15 @@ func (deploy *NewDeploy) idleObjectReaper() { envList[env.ObjectMeta.UID] = struct{}{} } - funcSvcs, err := deploy.fsCache.ListOld(deploy.idlePodReapTime) + funcSvcs, err := deploy.fsCache.ListOld(pollSleep) if err != nil { deploy.logger.Error("error reaping idle pods", zap.Error(err)) continue } - for _, fsvc := range funcSvcs { + for i := range funcSvcs { + fsvc := funcSvcs[i] + if fsvc.Executor != fv1.ExecutorTypeNewdeploy { continue } @@ -813,30 +815,41 @@ func (deploy *NewDeploy) idleObjectReaper() { continue } - deployObj := getDeploymentObj(fsvc.KubernetesObjects) - if deployObj == nil { - deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + idlePodReapTime := deploy.defaultIdlePodReapTime + if fn.Spec.IdleTimeout != nil { + idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second + } + + if time.Since(fsvc.Atime) < idlePodReapTime { continue } - currentDeploy, err := deploy.kubernetesClient.AppsV1(). - Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) - if err != nil { - deploy.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) - 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 + } - minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + currentDeploy, err := deploy.kubernetesClient.AppsV1(). + Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{}) + if err != nil { + deploy.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name)) + return + } - // do nothing if the current replicas is already lower than minScale - if *currentDeploy.Spec.Replicas <= minScale { - continue - } + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - err = deploy.scaleDeployment(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)) - } + // do nothing if the current replicas is already lower than minScale + if *currentDeploy.Spec.Replicas <= minScale { + return + } + + err = deploy.scaleDeployment(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/gp.go b/pkg/executor/executortype/poolmgr/gp.go index cd82f8f4..b2583c66 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -57,7 +57,6 @@ type ( namespace string // namespace to keep our resources functionNamespace string // fallback namespace for fission functions podReadyTimeout time.Duration // timeout for generic pods to become ready - idlePodReapTime time.Duration // pods unused for idlePodReapTime are deleted fsCache *fscache.FunctionServiceCache // cache funcSvc's by function, address and podname useSvc bool // create k8s service for specialized pods useIstio bool @@ -113,7 +112,6 @@ func MakeGenericPool( namespace: namespace, functionNamespace: functionNamespace, podReadyTimeout: 5 * time.Minute, // TODO make this an env param? - idlePodReapTime: 3 * time.Minute, // TODO make this configurable fsCache: fsCache, poolInstanceId: uniuri.NewLen(8), fetcherConfig: fetcherConfig, diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 42cf9499..57cf6e6c 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -77,7 +77,7 @@ type ( pkgStore k8sCache.Store pkgController k8sCache.Controller - idlePodReapTime time.Duration + defaultIdlePodReapTime time.Duration } request struct { requestType @@ -102,17 +102,17 @@ func MakeGenericPoolManager( gpmLogger := logger.Named("generic_pool_manager") gpm := &GenericPoolManager{ - logger: gpmLogger, - pools: make(map[string]*GenericPool), - kubernetesClient: kubernetesClient, - namespace: functionNamespace, - fissionClient: fissionClient, - functionEnv: cache.MakeCache(10*time.Second, 0), - fsCache: fscache.MakeFunctionServiceCache(gpmLogger), - instanceId: instanceId, - requestChannel: make(chan *request), - idlePodReapTime: 2 * time.Minute, - fetcherConfig: fetcherConfig, + logger: gpmLogger, + pools: make(map[string]*GenericPool), + kubernetesClient: kubernetesClient, + namespace: functionNamespace, + fissionClient: fissionClient, + functionEnv: cache.MakeCache(10*time.Second, 0), + fsCache: fscache.MakeFunctionServiceCache(gpmLogger), + instanceId: instanceId, + requestChannel: make(chan *request), + defaultIdlePodReapTime: 2 * time.Minute, + fetcherConfig: fetcherConfig, } go gpm.service() @@ -555,13 +555,14 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *fv1.Environment) int32 { // idleObjectReaper reaps objects after certain idle time func (gpm *GenericPoolManager) idleObjectReaper() { - pollSleep := time.Duration(gpm.idlePodReapTime) + pollSleep := 5 * time.Second for { time.Sleep(pollSleep) envs, err := gpm.fissionClient.CoreV1().Environments(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { - gpm.logger.Fatal("failed to get environment list", zap.Error(err)) + gpm.logger.Error("failed to get environment list", zap.Error(err)) + continue } envList := make(map[k8sTypes.UID]struct{}) @@ -569,7 +570,18 @@ func (gpm *GenericPoolManager) idleObjectReaper() { envList[env.ObjectMeta.UID] = struct{}{} } - funcSvcs, err := gpm.fsCache.ListOld(gpm.idlePodReapTime) + fns, err := gpm.fissionClient.CoreV1().Functions(metav1.NamespaceAll).List(metav1.ListOptions{}) + if err != nil { + gpm.logger.Error("failed to get environment list", zap.Error(err)) + continue + } + + fnList := make(map[k8sTypes.UID]fv1.Function) + for i, fn := range fns.Items { + fnList[fn.ObjectMeta.UID] = fns.Items[i] + } + + funcSvcs, err := gpm.fsCache.ListOld(pollSleep) if err != nil { gpm.logger.Error("error reaping idle pods", zap.Error(err)) continue @@ -594,8 +606,19 @@ func (gpm *GenericPoolManager) idleObjectReaper() { 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 time.Since(fsvc.Atime) < idlePodReapTime { + continue + } + go func() { - deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime) + deleted, err := gpm.fsCache.DeleteOld(fsvc, idlePodReapTime) if err != nil { gpm.logger.Error("error deleting Kubernetes objects for function service", zap.Error(err), diff --git a/pkg/fission-cli/cmd/function/command.go b/pkg/fission-cli/cmd/function/command.go index d1eae15a..b9695979 100644 --- a/pkg/fission-cli/cmd/function/command.go +++ b/pkg/fission-cli/cmd/function/command.go @@ -35,6 +35,7 @@ func Commands() *cobra.Command { flag.FnEnvName, flag.FnEntryPoint, flag.FnPkgName, flag.FnExecutorType, flag.FnCfgMap, flag.FnSecret, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, + flag.FnIdleTimeout, // TODO retired pkg & trigger related flags from function cmd flag.PkgCode, flag.PkgSrcArchive, flag.PkgDeployArchive, @@ -85,6 +86,7 @@ func Commands() *cobra.Command { flag.FnEnvName, flag.FnEntryPoint, flag.FnPkgName, flag.FnExecutorType, flag.FnSecret, flag.FnCfgMap, flag.FnSpecializationTimeout, flag.FnExecutionTimeout, + flag.FnIdleTimeout, 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 6f5265ba..32661775 100644 --- a/pkg/fission-cli/cmd/function/create.go +++ b/pkg/fission-cli/cmd/function/create.go @@ -91,9 +91,11 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { fnTimeout := input.Int(flagkey.FnExecutionTimeout) if fnTimeout <= 0 { - return errors.New("fntimeout must be greater than 0") + return errors.Errorf("--%v must be greater than 0", flagkey.FnExecutionTimeout) } + fnIdleTimeout := input.Int(flagkey.FnIdleTimeout) + pkgName := input.String(flagkey.FnPackageName) secretNames := input.StringSlice(flagkey.FnSecret) @@ -291,6 +293,7 @@ func (opts *CreateSubCommand) complete(input cli.Input) error { Resources: *resourceReq, InvokeStrategy: *invokeStrategy, FunctionTimeout: fnTimeout, + IdleTimeout: &fnIdleTimeout, }, } diff --git a/pkg/fission-cli/cmd/function/update.go b/pkg/fission-cli/cmd/function/update.go index 295bf11f..a560d160 100644 --- a/pkg/fission-cli/cmd/function/update.go +++ b/pkg/fission-cli/cmd/function/update.go @@ -150,6 +150,11 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error { function.Spec.FunctionTimeout = fnTimeout } + if input.IsSet(flagkey.FnIdleTimeout) { + fnTimeout := input.Int(flagkey.FnIdleTimeout) + function.Spec.IdleTimeout = &fnTimeout + } + if len(pkgName) == 0 { pkgName = function.Spec.Package.PackageRef.Name } diff --git a/pkg/fission-cli/flag/flag.go b/pkg/fission-cli/flag/flag.go index 3d883b34..8c277f58 100644 --- a/pkg/fission-cli/flag/flag.go +++ b/pkg/fission-cli/flag/flag.go @@ -107,6 +107,7 @@ var ( FnTestTimeout = Flag{Type: Duration, Name: flagkey.FnTestTimeout, Short: "t", Usage: "Length of time to wait for the response. If set to zero or negative number, no timeout is set", DefaultValue: 30 * time.Second} FnTestHeader = Flag{Type: StringSlice, Name: flagkey.FnTestHeader, Short: "H", Usage: "Request headers"} FnTestQuery = Flag{Type: StringSlice, Name: flagkey.FnTestQuery, Short: "q", Usage: "Request query parameters: -q key1=value1 -q key2=value2"} + FnIdleTimeout = Flag{Type: Int, Name: flagkey.FnIdleTimeout, Usage: "The length of time (in seconds) that a function is idle before pod(s) are eligible for recycling", DefaultValue: 120} HtName = Flag{Type: String, Name: flagkey.HtName, Usage: "HTTP trigger name"} HtMethod = Flag{Type: String, Name: flagkey.HtMethod, Usage: "HTTP Method: GET|POST|PUT|DELETE|HEAD", DefaultValue: http.MethodGet} diff --git a/pkg/fission-cli/flag/key/key.go b/pkg/fission-cli/flag/key/key.go index a80ae228..0a9fd1fa 100644 --- a/pkg/fission-cli/flag/key/key.go +++ b/pkg/fission-cli/flag/key/key.go @@ -61,6 +61,7 @@ const ( FnTestBody = "body" FnTestHeader = "header" FnTestQuery = "query" + FnIdleTimeout = "idletimeout" HtName = resourceName HtMethod = "method"