Support Function-level idle timeout setting (#1538)

This PR allows users to specify the idle timeout setting at Function-level.
Fix #1050

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.

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
 }
 

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)
+			idlePodReapTime := deploy.defaultIdlePodReapTime
-			if deployObj == nil {
+			if fn.Spec.IdleTimeout != nil {
-				deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+				idlePodReapTime = time.Duration(*fn.Spec.IdleTimeout) * time.Second
+			}
+
+			if time.Since(fsvc.Atime) < idlePodReapTime {
 				continue
 			}
 
-			currentDeploy, err := deploy.kubernetesClient.AppsV1().
+			go func() {
-				Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{})
+				deployObj := getDeploymentObj(fsvc.KubernetesObjects)
-			if err != nil {
+				if deployObj == nil {
-				deploy.logger.Error("error getting function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+					deploy.logger.Error("error finding function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
-				continue
+					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
+				minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
-			if *currentDeploy.Spec.Replicas <= minScale {
-				continue
-			}
 
-			err = deploy.scaleDeployment(deployObj.Namespace, deployObj.Name, minScale)
+				// do nothing if the current replicas is already lower than minScale
-			if err != nil {
+				if *currentDeploy.Spec.Replicas <= minScale {
-				deploy.logger.Error("error scaling down function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+					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))
+				}
+			}()
 		}
 	}
 }

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,

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,
+		logger:                 gpmLogger,
-		pools:            make(map[string]*GenericPool),
+		pools:                  make(map[string]*GenericPool),
-		kubernetesClient: kubernetesClient,
+		kubernetesClient:       kubernetesClient,
-		namespace:        functionNamespace,
+		namespace:              functionNamespace,
-		fissionClient:    fissionClient,
+		fissionClient:          fissionClient,
-		functionEnv:      cache.MakeCache(10*time.Second, 0),
+		functionEnv:            cache.MakeCache(10*time.Second, 0),
-		fsCache:          fscache.MakeFunctionServiceCache(gpmLogger),
+		fsCache:                fscache.MakeFunctionServiceCache(gpmLogger),
-		instanceId:       instanceId,
+		instanceId:             instanceId,
-		requestChannel:   make(chan *request),
+		requestChannel:         make(chan *request),
-		idlePodReapTime:  2 * time.Minute,
+		defaultIdlePodReapTime: 2 * time.Minute,
-		fetcherConfig:    fetcherConfig,
+		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),

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,

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,
 		},
 	}
 

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
 	}

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}

pkg/fission-cli/flag/key/key.go

@@ -61,6 +61,7 @@ const (
 	FnTestBody              = "body"
 	FnTestHeader            = "header"
 	FnTestQuery             = "query"
+	FnIdleTimeout           = "idletimeout"
 
 	HtName              = resourceName
 	HtMethod            = "method"