Repository: fission/fission (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.


Pull request description:
Add ability to configure object reaper interval
<!--  Thanks for sending a pull request! We request you provide detailed description as much as possible. -->

## Description
<!--- Describe your changes in detail. -->
<!-- Typically try to give details of what, why and how of the PR changes. -->
Added properties to configure object reaper interval, global and specific to exec type.

`OBJECT_REAPER_INTERVAL` - global
`NEWDEPLOY_OBJECT_REAPER_INTERVAL` - for `newdeploy` type
`CONTAINER_OBJECT_REAPER_INTERVAL` - for `container` type
`POOLMGR_OBJECT_REAPER_INTERVAL` - for `poolmgr`
## Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #2464

## Testing
<!--- Please describe in detail how you tested your changes. -->

## Checklist:
<!-- Please tick following checkboxes as per your understanding. -->
- [x] I ran tests as well as code linting locally to verify my changes. 
- [x] I have done manual verification of my changes, changes working as expected.
- [x] I have added new tests to cover my changes.
- [x] My changes follow contributing guidelines of Fission.
- [x] I have signed all of my commits.


Patch:
diff --git a/charts/fission-all/templates/executor/deployment.yaml b/charts/fission-all/templates/executor/deployment.yaml
index f5152cb3..a782df41 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 a80f2f5b..45721c57 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 8c7f2148..51509b61 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 df5504f7..4abdc3c8 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 aa983b1a..db637baa 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 df513505..05840348 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 07df91ff..df316433 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 60314af4..78c7fcb5 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 125a3558..a26bff0f 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
+			}
+		})
+	}
+}