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