Add ability to configure object reaper interval for different executor types (#2543)

Added properties to configure object reaper interval, global and specific to exec type.

OBJECT_REAPER_INTERVAL - global
NEWDEPLOY_OBJECT_REAPER_INTERVAL - for new deploy type
CONTAINER_OBJECT_REAPER_INTERVAL - for container type
POOLMGR_OBJECT_REAPER_INTERVAL - for poolmgr

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

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:

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,
+		defaultIdlePodReapTime:     1 * time.Minute,
-
+		objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeContainer, 5)) * time.Second,
-		hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
+		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) {

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,
+		defaultIdlePodReapTime:     2 * time.Minute,
-
+		objectReaperIntervalSecond: time.Duration(executorUtils.GetObjectReaperInterval(logger, fv1.ExecutorTypeNewdeploy, 5)) * time.Second,
-		hpaops: hpautils.NewHpaOperations(logger, kubernetesClient, instanceID),
+		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) {

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,
+		logger:                     gpmLogger,
-		pools:                  make(map[string]*GenericPool),
+		pools:                      make(map[string]*GenericPool),
-		kubernetesClient:       kubernetesClient,
+		kubernetesClient:           kubernetesClient,
-		metricsClient:          metricsClient,
+		metricsClient:              metricsClient,
-		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),
-		defaultIdlePodReapTime: 2 * time.Minute,
+		defaultIdlePodReapTime:     2 * time.Minute,
-		fetcherConfig:          fetcherConfig,
+		fetcherConfig:              fetcherConfig,
-		enableIstio:            enableIstio,
+		enableIstio:                enableIstio,
-		poolPodC:               poolPodC,
+		poolPodC:                   poolPodC,
-		podSpecPatch:           podSpecPatch,
+		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) {

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"
+}

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)
+	}
+}

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

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
+			}
+		})
+	}
+}