Repository: fission/fission (language: go)

Issue:
New environment variable for executor container
**Is your feature request related to a problem? Please describe.**
Whenever user create a function with executor type new deploy and container by providing min scale property to 0, then a function name doIdleObjectReaper() runs after every 5 second(hard coded in the function) to check the idle timeout of function that user created and once there is idle timeout then doIdleObjectReaper() scale down the replica of deployment.

**Describe the solution you'd like**
So instead of calling this function doIdleObjectReaper() after every 5 second, it will be better to make this configurable in helm chart.

Create a new environment variable for executor container named IDLE_REAPER_TIMEOUT with numeric value, value will be consider in seconds. Set initial value as 5. 

Later on we can change this with the help of helm chart. Now the doIdleObjectReaper() will be call as per the value get fetch by environment variable.

**Additional context**
These changes will only be applicable for executor type as new deploy and container.
Changes needs to me make in function idleObjectReaper() for new deploy and container both.



Discussion / hints:

Fix patch:
diff --git a/charts/fission-all/templates/executor/deployment.yaml b/charts/fission-all/templates/executor/deployment.yaml
index f5152cb33a..a782df4177 100644
--- a/charts/fission-all/templates/executor/deployment.yaml
+++ b/charts/fission-all/templates/executor/deployment.yaml
@@ -57,6 +57,20 @@ spec:
           value: {{ .Values.debugEnv | quote }}
         - name: PPROF_ENABLED
           value: {{ .Values.pprof.enabled | quote }}
+        - name: OBJECT_REAPER_INTERVAL
+          value: {{ .Values.executor.objectReaperInterval | quote }}
+        {{- if .Values.executor.poolmgr.objectReaperInterval }}
+        - name: POOLMGR_OBJECT_REAPER_INTERVAL
+          value: {{ .Values.executor.poolmgr.objectReaperInterval | quote }}
+        {{- end}}
+        {{- if .Values.executor.newdeploy.objectReaperInterval }}
+        - name: NEWDEPLOY_OBJECT_REAPER_INTERVAL
+          value: {{ .Values.executor.newdeploy.objectReaperInterval | quote }}
+        {{- end}}
+        {{- if .Values.executor.container.objectReaperInterval }}
+        - name: CONTAINER_OBJECT_REAPER_INTERVAL
+          value: {{ .Values.executor.container.objectReaperInterval | quote }}
+        {{- end}}
         - name: HELM_RELEASE_NAME
           value: {{ .Release.Name | quote }}
         {{- include "opentelemtry.envs" . | indent 8 }}
diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml
index a80f2f5b4b..45721c5708 100644
--- a/charts/fission-all/values.yaml
+++ b/charts/fission-all/values.yaml
@@ -155,6 +155,26 @@ executor:
     runAsUser: 10001
     runAsGroup: 10001
 
+  ## Object Reaper
+  ## objectReaperInterval (seconds) represents GLOBAL interval to run process that reaps objects after certain idle time.
+  ## Also you can set different objectReaperInterval for specific executor type. See poolmgs/newdeploy/container section
+  ## Default: 5 (in seconds)
+  ##
+  objectReaperInterval: 5
+
+  poolmgr: {}
+    ## objectReaperInterval specific to poolmgr executor type
+    ##
+    ## objectReaperInterval: 5
+  newdeploy: {}
+    ## objectReaperInterval specific to newdeploy  executor type
+    ##
+    ## objectReaperInterval: 5
+  container: {}
+    ## objectReaperInterval specific to container executor type
+    ##
+    ## objectReaperInterval: 5
+
 ## router is responsible for routing function calls to the appropriate function.
 ##
 router:
diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go
index 8c7f214858..51509b611e 100644
--- a/pkg/executor/executortype/container/containermgr.go
+++ b/pkg/executor/executortype/container/containermgr.go
@@ -47,6 +47,7 @@ import (
 	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/executor/metrics"
 	"github.com/fission/fission/pkg/executor/reaper"
+	executorUtils "github.com/fission/fission/pkg/executor/util"
 	hpautils "github.com/fission/fission/pkg/executor/util/hpa"
 	"github.com/fission/fission/pkg/generated/clientset/versioned"
 	finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1"
@@ -84,7 +85,8 @@ type (
 		deplListerSynced k8sCache.InformerSynced
 		svcListerSynced  k8sCache.InformerSynced
 
-		hpaops *hpautils.HpaOperations
+		hpaops                     *hpautils.HpaOperations
+		objectReaperIntervalSecond time.Duration
 	}
 )
 
@@ -123,9 +125,9 @@ func MakeContainer(
 		runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")),
 		useIstio:               enableIstio,
 		// Time is set slightly higher than NewDeploy as cold starts are longer for CaaF
-		defaultIdlePodReapTime: 1 * time.Minute,
-
-		hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
+		defaultIdlePodReapTime:     1 * time.Minute,
+		objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)) * time.Second,
+		hpaops:                     hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
 	}
 	caaf.deplLister = deplInformer.Lister()
 	caaf.deplListerSynced = deplInformer.Informer().HasSynced
@@ -706,7 +708,7 @@ func (caaf *Container) updateStatus(fn *fv1.Function, err error, message string)
 // idleObjectReaper reaps objects after certain idle time
 func (caaf *Container) idleObjectReaper(ctx context.Context) {
 	// calling function doIdleObjectReaper() repeatedly at given interval of time
-	wait.UntilWithContext(ctx, caaf.doIdleObjectReaper, time.Second*5)
+	wait.UntilWithContext(ctx, caaf.doIdleObjectReaper, caaf.objectReaperIntervalSecond)
 }
 
 func (caaf *Container) doIdleObjectReaper(ctx context.Context) {
diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go
index df5504f776..4abdc3c8be 100644
--- a/pkg/executor/executortype/newdeploy/newdeploymgr.go
+++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go
@@ -48,6 +48,7 @@ import (
 	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/executor/metrics"
 	"github.com/fission/fission/pkg/executor/reaper"
+	executorUtils "github.com/fission/fission/pkg/executor/util"
 	hpautils "github.com/fission/fission/pkg/executor/util/hpa"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
 	"github.com/fission/fission/pkg/generated/clientset/versioned"
@@ -88,7 +89,8 @@ type (
 
 		hpaops *hpautils.HpaOperations
 
-		podSpecPatch *apiv1.PodSpec
+		podSpecPatch               *apiv1.PodSpec
+		objectReaperIntervalSecond time.Duration
 	}
 )
 
@@ -130,9 +132,9 @@ func MakeNewDeploy(
 		runtimeImagePullPolicy: utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")),
 		useIstio:               enableIstio,
 
-		defaultIdlePodReapTime: 2 * time.Minute,
-
-		hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
+		defaultIdlePodReapTime:     2 * time.Minute,
+		objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeNewdeploy, 5)) * time.Second,
+		hpaops:                     hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
 
 		podSpecPatch: podSpecPatch,
 	}
@@ -769,7 +771,7 @@ 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) {
 	// calling function doIdleObjectReaper() repeatedly at given interval of time
-	wait.UntilWithContext(ctx, deploy.doIdleObjectReaper, time.Second*5)
+	wait.UntilWithContext(ctx, deploy.doIdleObjectReaper, deploy.objectReaperIntervalSecond)
 }
 
 func (deploy *NewDeploy) doIdleObjectReaper(ctx context.Context) {
diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go
index aa983b1aa0..db637baaeb 100644
--- a/pkg/executor/executortype/poolmgr/gpm.go
+++ b/pkg/executor/executortype/poolmgr/gpm.go
@@ -50,6 +50,7 @@ import (
 	"github.com/fission/fission/pkg/executor/executortype"
 	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/executor/reaper"
+	executorUtils "github.com/fission/fission/pkg/executor/util"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
 	"github.com/fission/fission/pkg/generated/clientset/versioned"
 	finformerv1 "github.com/fission/fission/pkg/generated/informers/externalversions/core/v1"
@@ -94,7 +95,8 @@ type (
 
 		poolPodC *PoolPodController
 
-		podSpecPatch *apiv1.PodSpec
+		podSpecPatch               *apiv1.PodSpec
+		objectReaperIntervalSecond time.Duration
 	}
 	request struct {
 		requestType
@@ -140,21 +142,22 @@ func MakeGenericPoolManager(
 		enableIstio, funcInformer, pkgInformer, envInformer, rsInformer, podInformer)
 
 	gpm := &GenericPoolManager{
-		logger:                 gpmLogger,
-		pools:                  make(map[string]*GenericPool),
-		kubernetesClient:       kubernetesClient,
-		metricsClient:          metricsClient,
-		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,
-		enableIstio:            enableIstio,
-		poolPodC:               poolPodC,
-		podSpecPatch:           podSpecPatch,
+		logger:                     gpmLogger,
+		pools:                      make(map[string]*GenericPool),
+		kubernetesClient:           kubernetesClient,
+		metricsClient:              metricsClient,
+		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,
+		enableIstio:                enableIstio,
+		poolPodC:                   poolPodC,
+		podSpecPatch:               podSpecPatch,
+		objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypePoolmgr, 5)) * time.Second,
 	}
 	gpm.podLister = podInformer.Lister()
 	gpm.podListerSynced = podInformer.Informer().HasSynced
@@ -572,7 +575,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct
 // idleObjectReaper reaps objects after certain idle time
 func (gpm *GenericPoolManager) idleObjectReaper(ctx context.Context) {
 	// calling function doIdleObjectReaper() repeatedly at given interval of time
-	wait.UntilWithContext(ctx, gpm.doIdleObjectReaper, time.Second*5)
+	wait.UntilWithContext(ctx, gpm.doIdleObjectReaper, gpm.objectReaperIntervalSecond)
 }
 
 func (gpm *GenericPoolManager) doIdleObjectReaper(ctx context.Context) {
diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go
index df513505f0..0584034835 100644
--- a/pkg/executor/util/util.go
+++ b/pkg/executor/util/util.go
@@ -19,15 +19,19 @@ package util
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
+	"go.uber.org/zap"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	"sigs.k8s.io/yaml"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/utils"
 )
 
 // ApplyImagePullSecret applies image pull secret to the give pod spec.
@@ -128,3 +132,26 @@ func GetSpecFromConfigMap(ctx context.Context, kubeClient kubernetes.Interface,
 
 	return &additionalSpec, err
 }
+
+func GetObjectReaperInterval(logger *zap.Logger, executorType fv1.ExecutorType, defaultReaperInterval uint) uint {
+
+	// TODO think about migration to executor package as const.
+	globalEnvVarName := "OBJECT_REAPER_INTERVAL"
+
+	executorTypeEnvVarName := getExecutorEnvVarName(executorType)
+	keys := []string{executorTypeEnvVarName, globalEnvVarName}
+	for _, k := range keys {
+		interval, err := utils.GetUIntValueFromEnv(k)
+		if err != nil {
+			logger.Debug(fmt.Sprintf("Failed to parse %s", k))
+		} else {
+			return interval
+		}
+	}
+
+	return defaultReaperInterval
+}
+
+func getExecutorEnvVarName(executor fv1.ExecutorType) string {
+	return strings.ToUpper(string(executor)) + "_OBJECT_REAPER_INTERVAL"
+}
diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go
index 07df91ff71..df316433a5 100644
--- a/pkg/utils/utils.go
+++ b/pkg/utils/utils.go
@@ -26,6 +26,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 
 	"github.com/mholt/archiver/v3"
@@ -225,3 +226,23 @@ func GetCurrentNamespace() (string, error) {
 	}
 	return string(body), nil
 }
+
+func GetStringValueFromEnv(envVar string) (string, error) {
+	v := os.Getenv(envVar)
+	if v == "" {
+		return v, errors.New(fmt.Sprintf("Еnvironment variable %s empty", envVar))
+	}
+	return v, nil
+}
+
+func GetUIntValueFromEnv(envVar string) (uint, error) {
+	s, err := GetStringValueFromEnv(envVar)
+	if err != nil {
+		return 0, err
+	}
+	value, err := strconv.ParseUint(s, 10, 32)
+	if err != nil {
+		return 0, err
+	}
+	return uint(value), nil
+}


Test patch:
diff --git a/pkg/executor/util/util_test.go b/pkg/executor/util/util_test.go
index 60314af493..78c7fcb591 100644
--- a/pkg/executor/util/util_test.go
+++ b/pkg/executor/util/util_test.go
@@ -18,12 +18,17 @@ package util
 
 import (
 	"context"
+	"fmt"
+	"os"
 	"reflect"
 	"testing"
 
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
+
+	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/utils/loggerfactory"
 )
 
 func TestGetSpecFromConfigMap(t *testing.T) {
@@ -112,3 +117,52 @@ securityContext:
 		})
 	}
 }
+
+func TestGetObjectReaperInterval(t *testing.T) {
+	logger := loggerfactory.GetLogger()
+
+	var want uint
+
+	// Test default reaper interval
+	want = 1
+	got := GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, want)
+	if want != got {
+		t.Fatalf(`Get default ObjectReaperInterval failed. Want %d, Got %d`, want, got)
+	}
+
+	// Test when only specific reaper interval set
+	want = 2
+	os.Setenv("CONTAINER_OBJECT_REAPER_INTERVAL", fmt.Sprint(want))
+	os.Unsetenv("OBJECT_REAPER_INTERVAL")
+	got = GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)
+	if want != got {
+		t.Fatalf(`%d %d`, want, got)
+	}
+
+	// Test when only global reaper interval set
+	want = 3
+	os.Unsetenv("CONTAINER_OBJECT_REAPER_INTERVAL")
+	os.Setenv("OBJECT_REAPER_INTERVAL", fmt.Sprint(want))
+	got = GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)
+	if want != got {
+		t.Fatalf(`%d %d`, want, got)
+	}
+
+	// Test when broken specific reaper interval set
+	want = 4
+	os.Setenv("CONTAINER_OBJECT_REAPER_INTERVAL", "just some string!")
+	os.Unsetenv("OBJECT_REAPER_INTERVAL")
+	got = GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, want)
+	if want != got {
+		t.Fatalf(`%d %d`, want, got)
+	}
+
+	// Test when empty specific reaper interval set
+	want = 5
+	os.Setenv("CONTAINER_OBJECT_REAPER_INTERVAL", "")
+	os.Unsetenv("OBJECT_REAPER_INTERVAL")
+	got = GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)
+	if want != got {
+		t.Fatalf(`%d %d`, want, got)
+	}
+}
diff --git a/pkg/utils/utils_test.go b/pkg/utils/utils_test.go
index 125a355833..a26bff0fb4 100644
--- a/pkg/utils/utils_test.go
+++ b/pkg/utils/utils_test.go
@@ -19,6 +19,7 @@ package utils
 import (
 	"bytes"
 	"io"
+	"os"
 	"reflect"
 	"testing"
 
@@ -81,3 +82,83 @@ func TestGetChecksum(t *testing.T) {
 		})
 	}
 }
+
+func TestGetStringValueFromEnv(t *testing.T) {
+	varName := "TEST_VAR"
+	tests := []struct {
+		name    string
+		value   string
+		want    string
+		wantErr bool
+	}{
+		{
+			name:    "empty string case",
+			value:   "",
+			want:    "",
+			wantErr: true,
+		},
+		{
+			name:    "string case",
+			value:   "test sting",
+			want:    "test sting",
+			wantErr: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			os.Setenv(varName, tt.value)
+			got, err := GetStringValueFromEnv(varName)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetStringValueFromEnv() error = %v, wantErr %v, got %s", err, tt.wantErr, got)
+				return
+			}
+		})
+	}
+
+}
+
+func TestGetUIntValueFromEnv(t *testing.T) {
+	varName := "TEST_VAR"
+	tests := []struct {
+		name    string
+		value   string
+		want    uint
+		wantErr bool
+	}{
+		{
+			name:    "empty string case",
+			value:   "",
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name:    "string case",
+			value:   "test sting",
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name:    "not uint case",
+			value:   "-100",
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name:    "uint case",
+			value:   "7",
+			want:    7,
+			wantErr: false,
+		}}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			os.Setenv(varName, tt.value)
+			got, err := GetUIntValueFromEnv(varName)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetUIntValueFromEnv() error = %v, wantErr %v, got %d", err, tt.wantErr, got)
+				return
+			}
+		})
+	}
+}