diff --git a/charts/fission-all/templates/deployment.yaml b/charts/fission-all/templates/deployment.yaml index e49d4655..b3cabc7c 100644 --- a/charts/fission-all/templates/deployment.yaml +++ b/charts/fission-all/templates/deployment.yaml @@ -221,6 +221,8 @@ spec: value: "{{ .Values.pullPolicy }}" - name: ADOPT_EXISTING_RESOURCES value: {{ .Values.executor.adoptExistingResources | default false | quote }} + - name: POD_READY_TIMEOUT + value: {{ .Values.executor.podReadyTimeout | default false | quote }} - name: ENABLE_ISTIO value: "{{ .Values.enableIstio }}" - name: TRACE_JAEGER_COLLECTOR_ENDPOINT diff --git a/charts/fission-all/templates/router.yaml b/charts/fission-all/templates/router.yaml index 5568ca03..fc9c6e26 100644 --- a/charts/fission-all/templates/router.yaml +++ b/charts/fission-all/templates/router.yaml @@ -53,6 +53,8 @@ spec: value: {{ .Values.router.svcAddressMaxRetries | default 5 | quote }} - name: ROUTER_SVC_ADDRESS_UPDATE_TIMEOUT value: {{ .Values.router.svcAddressUpdateTimeout | default "30s" | quote }} + - name: ROUTER_UNTAP_SERVICE_TIMEOUT + value: {{ .Values.router.unTapServiceTimeout | default "3600s" | quote }} - name: TRACE_JAEGER_COLLECTOR_ENDPOINT value: "{{ .Values.traceCollectorEndpoint }}" - name: TRACING_SAMPLING_RATE diff --git a/charts/fission-all/values.yaml b/charts/fission-all/values.yaml index 4001600e..7715f419 100644 --- a/charts/fission-all/values.yaml +++ b/charts/fission-all/values.yaml @@ -90,13 +90,14 @@ logger: executor: adoptExistingResources: false + podReadyTimeout: 300s ## Router config router: deployAsDaemonSet: false svcAddressMaxRetries: 5 svcAddressUpdateTimeout: 30s - + unTapServiceTimeout: 3600s ## Display endpoint access logs ## To be aware of enabling logging endpoint access log, it increases ## router resource utilization when under heavy workloads. diff --git a/charts/fission-core/templates/deployment.yaml b/charts/fission-core/templates/deployment.yaml index 0ae8449b..463a9911 100644 --- a/charts/fission-core/templates/deployment.yaml +++ b/charts/fission-core/templates/deployment.yaml @@ -221,6 +221,8 @@ spec: value: {{ .Values.traceSamplingRate | default "0.5" | quote }} - name: ADOPT_EXISTING_RESOURCES value: {{ .Values.executor.adoptExistingResources | default false | quote }} + - name: POD_READY_TIMEOUT + value: {{ .Values.executor.podReadyTimeout | default false | quote }} - name: ENABLE_ISTIO value: "{{ .Values.enableIstio }}" - name: FETCHER_MINCPU diff --git a/charts/fission-core/templates/router.yaml b/charts/fission-core/templates/router.yaml index 5568ca03..fc9c6e26 100644 --- a/charts/fission-core/templates/router.yaml +++ b/charts/fission-core/templates/router.yaml @@ -53,6 +53,8 @@ spec: value: {{ .Values.router.svcAddressMaxRetries | default 5 | quote }} - name: ROUTER_SVC_ADDRESS_UPDATE_TIMEOUT value: {{ .Values.router.svcAddressUpdateTimeout | default "30s" | quote }} + - name: ROUTER_UNTAP_SERVICE_TIMEOUT + value: {{ .Values.router.unTapServiceTimeout | default "3600s" | quote }} - name: TRACE_JAEGER_COLLECTOR_ENDPOINT value: "{{ .Values.traceCollectorEndpoint }}" - name: TRACING_SAMPLING_RATE diff --git a/charts/fission-core/values.yaml b/charts/fission-core/values.yaml index 743d86e2..9233db7b 100644 --- a/charts/fission-core/values.yaml +++ b/charts/fission-core/values.yaml @@ -58,13 +58,14 @@ fetcher: executor: adoptExistingResources: false + podReadyTimeout: 300s ## Router config router: deployAsDaemonSet: false svcAddressMaxRetries: 5 svcAddressUpdateTimeout: 30s - + unTapServiceTimeout: 3600s ## Display endpoint access logs ## To be aware of enabling logging endpoint access log, it increases ## router resource utilization when under heavy workloads. diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 0ba9962a..dba61bde 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -211,18 +211,8 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { http.Error(w, "Failed to parse request", http.StatusBadRequest) return } - - fn, err := executor.fissionClient.CoreV1().Functions(tapSvcReq.FnMetadata.Namespace).Get(tapSvcReq.FnMetadata.Name, metav1.GetOptions{}) - if err != nil { - if k8serrors.IsNotFound(err) { - http.Error(w, "Failed to find function", http.StatusNotFound) - } else { - http.Error(w, "Failed to get function", http.StatusInternalServerError) - } - return - } - - t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType + key := fmt.Sprintf("%v_%v", tapSvcReq.FnMetadata.UID, tapSvcReq.FnMetadata.ResourceVersion) + t := tapSvcReq.FnExecutorType if t != fv1.ExecutorTypePoolmgr { msg := fmt.Sprintf("Unknown executor type '%v'", t) http.Error(w, msg, http.StatusBadRequest) @@ -231,7 +221,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) { et := executor.executorTypes[t] - et.UnTapService(fn, tapSvcReq.ServiceUrl) + et.UnTapService(key, tapSvcReq.ServiceUrl) w.WriteHeader(http.StatusOK) } diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 7a83fd53..2e042f1d 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -46,7 +46,7 @@ type ExecutorType interface { TapService(serviceUrl string) error // UnTapService updates the isActive to false - UnTapService(fn *fv1.Function, svcHost string) + UnTapService(key string, svcHost string) // IsValid returns true if a function service is valid. Different executor types // use distinct ways to examine the function service. diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 177c08d6..a85a9fc6 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -153,7 +153,7 @@ func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { deploy.fsCache.DeleteEntry(fsvc) } -func (deploy *NewDeploy) UnTapService(fn *fv1.Function, svcHost string) { +func (deploy *NewDeploy) UnTapService(key string, svcHost string) { // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 4cddcb6b..03c4891b 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -20,7 +20,6 @@ import ( "context" "encoding/json" "fmt" - "math/rand" "net" "os" "strings" @@ -96,6 +95,16 @@ func MakeGenericPool( gpLogger := logger.Named("generic_pool") + podReadyTimeoutStr := os.Getenv("POD_READY_TIMEOUT") + podReadyTimeout, err := time.ParseDuration(podReadyTimeoutStr) + if err != nil { + podReadyTimeout = 300 * time.Second + gpLogger.Error("failed to parse pod ready timeout duration from 'POD_READY_TIMEOUT' - set to the default value", + zap.Error(err), + zap.String("value", podReadyTimeoutStr), + zap.Duration("default", podReadyTimeout)) + } + gpLogger.Info("creating pool", zap.Any("environment", env.ObjectMeta)) ctx, stopCh := context.WithCancel(context.Background()) @@ -111,7 +120,7 @@ func MakeGenericPool( kubernetesClient: kubernetesClient, namespace: namespace, functionNamespace: functionNamespace, - podReadyTimeout: 5 * time.Minute, // TODO make this an env param? + podReadyTimeout: podReadyTimeout, fsCache: fsCache, poolInstanceId: uniuri.NewLen(8), fetcherConfig: fetcherConfig, @@ -124,7 +133,7 @@ func MakeGenericPool( gp.runtimeImagePullPolicy = utils.GetImagePullPolicy(os.Getenv("RUNTIME_IMAGE_PULL_POLICY")) // create fetcher SA in this ns, if not already created - err := fetcherConfig.SetupServiceAccount(gp.kubernetesClient, gp.namespace, nil) + err = fetcherConfig.SetupServiceAccount(gp.kubernetesClient, gp.namespace, nil) if err != nil { return nil, errors.Wrapf(err, "error creating fetcher service account in namespace %q", gp.namespace) } @@ -202,6 +211,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro // Get pods; filter the ones that are ready podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List( metav1.ListOptions{ + FieldSelector: "status.phase=Running", LabelSelector: labels.Set( gp.deployment.Spec.Selector.MatchLabels).AsSelector().String(), }) @@ -219,6 +229,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro // add it to the list of ready pods readyPods = append(readyPods, &pod) + break } gp.logger.Info("found ready pods", zap.Any("labels", newLabels), @@ -237,7 +248,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro // Pick a ready pod. For now just choose randomly; // ideally we'd care about which node it's running on, // and make a good scheduling decision. - chosenPod := readyPods[rand.Intn(len(readyPods))] + chosenPod := readyPods[0] if gp.env.Spec.AllowedFunctionsPerContainer != fv1.AllowedFunctionsPerContainerInfinite { // Relabel. If the pod already got picked and diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 718ff9e4..f1f67f5b 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -173,8 +173,8 @@ func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) { gpm.fsCache.DeleteFunctionSvc(fsvc) } -func (gpm *GenericPoolManager) UnTapService(fn *fv1.Function, svcHost string) { - gpm.fsCache.MarkAvailable(fn, svcHost) +func (gpm *GenericPoolManager) UnTapService(key string, svcHost string) { + gpm.fsCache.MarkAvailable(key, svcHost) } func (gpm *GenericPoolManager) GetTotalAvailable(fn *fv1.Function) int { diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 8bf84b5e..2e4699de 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -218,8 +218,8 @@ func (fsc *FunctionServiceCache) GetTotalAvailable(m *metav1.ObjectMeta) int { return fsc.connFunctionCache.GetTotalAvailable(crd.CacheKey(m)) } -func (fsc *FunctionServiceCache) MarkAvailable(fn *fv1.Function, svcHost string) { - fsc.connFunctionCache.MarkAvailable(crd.CacheKey(&fn.ObjectMeta), svcHost) +func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) { + fsc.connFunctionCache.MarkAvailable(key, svcHost) } func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) { diff --git a/pkg/executor/fscache/functionServiceCache_test.go b/pkg/executor/fscache/functionServiceCache_test.go index 7d0d2600..2b16a675 100644 --- a/pkg/executor/fscache/functionServiceCache_test.go +++ b/pkg/executor/fscache/functionServiceCache_test.go @@ -189,7 +189,8 @@ func TestFunctionServiceNewCache(t *testing.T) { logger.Panic(fmt.Sprintln("active instances not matched expected 1, found ", active)) } - fsc.MarkAvailable(fn, fsvc.Address) + key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion) + fsc.MarkAvailable(key, fsvc.Address) if fsc.GetTotalAvailable(fsvc.Function) != 0 { log.Panicln("active instances not matched") diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 37612f41..08ced339 100644 --- a/pkg/router/functionHandler.go +++ b/pkg/router/functionHandler.go @@ -59,6 +59,7 @@ type ( isDebugEnv bool svcAddrUpdateThrottler *throttler.Throttler functionTimeoutMap map[k8stypes.UID]int + unTapServiceTimeout time.Duration } tsRoundTripperParams struct { @@ -209,7 +210,9 @@ func (roundTripper *RetryingRoundTripper) RoundTrip(req *http.Request) (*http.Re // } } if roundTripper.funcHandler.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr { - defer roundTripper.funcHandler.unTapService(roundTripper.funcHandler.function, roundTripper.serviceUrl) + defer func(fn *fv1.Function, serviceUrl *url.URL) { + go roundTripper.funcHandler.unTapService(fn, serviceUrl) + }(roundTripper.funcHandler.function, roundTripper.serviceUrl) } // modify the request to reflect the service url @@ -500,7 +503,7 @@ func (roundTripper RetryingRoundTripper) addForwardedHostHeader(req *http.Reques // unTapservice marks the serviceURL in executor's cache as inactive, so that it can be reused func (fh functionHandler) unTapService(fn *fv1.Function, serviceUrl *url.URL) error { fh.logger.Info("UnTapService Called") - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), fh.unTapServiceTimeout) defer cancel() err := fh.executor.UnTapService(ctx, fn.ObjectMeta, fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, serviceUrl) if err != nil { diff --git a/pkg/router/httpTriggers.go b/pkg/router/httpTriggers.go index f3b2e7d7..37ddf2ce 100644 --- a/pkg/router/httpTriggers.go +++ b/pkg/router/httpTriggers.go @@ -57,10 +57,11 @@ type HTTPTriggerSet struct { tsRoundTripperParams *tsRoundTripperParams isDebugEnv bool svcAddrUpdateThrottler *throttler.Throttler + unTapServiceTimeout time.Duration } func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionClient *crd.FissionClient, - kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient rest.Interface, params *tsRoundTripperParams, isDebugEnv bool, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { + kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient rest.Interface, params *tsRoundTripperParams, isDebugEnv bool, unTapServiceTimeout time.Duration, actionThrottler *throttler.Throttler) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) { httpTriggerSet := &HTTPTriggerSet{ logger: logger.Named("http_trigger_set"), @@ -74,6 +75,7 @@ func makeHTTPTriggerSet(logger *zap.Logger, fmap *functionServiceMap, fissionCli tsRoundTripperParams: params, isDebugEnv: isDebugEnv, svcAddrUpdateThrottler: actionThrottler, + unTapServiceTimeout: unTapServiceTimeout, } var tStore, fnStore k8sCache.Store var tController, fnController k8sCache.Controller @@ -148,6 +150,7 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router isDebugEnv: ts.isDebugEnv, svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler, functionTimeoutMap: fnTimeoutMap, + unTapServiceTimeout: ts.unTapServiceTimeout, } // The functionHandler for HTTP trigger with fn reference type "FunctionReferenceTypeFunctionName", @@ -196,6 +199,7 @@ func (ts *HTTPTriggerSet) getRouter(fnTimeoutMap map[types.UID]int) *mux.Router isDebugEnv: ts.isDebugEnv, svcAddrUpdateThrottler: ts.svcAddrUpdateThrottler, functionTimeoutMap: fnTimeoutMap, + unTapServiceTimeout: ts.unTapServiceTimeout, } muxRouter.HandleFunc(utils.UrlForFunction(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace), fh.handler) } diff --git a/pkg/router/router.go b/pkg/router/router.go index cea5f8b8..1684a9ce 100644 --- a/pkg/router/router.go +++ b/pkg/router/router.go @@ -200,6 +200,17 @@ func Start(logger *zap.Logger, port int, executorUrl string) { zap.Duration("default", svcAddrUpdateTimeout)) } + // unTapServiceTimeout is the timeout used as timeout in the request context of unTapService + unTapServiceTimeoutstr := os.Getenv("ROUTER_UNTAP_SERVICE_TIMEOUT") + unTapServiceTimeout, err := time.ParseDuration(unTapServiceTimeoutstr) + if err != nil { + unTapServiceTimeout = 3600 * time.Second + logger.Error("failed to parse unTap service timeout duration from 'ROUTER_UNTAP_SERVICE_TIMEOUT' - set to the default value", + zap.Error(err), + zap.String("value", unTapServiceTimeoutstr), + zap.Duration("default", unTapServiceTimeout)) + } + tracingSamplingRateStr := os.Getenv("TRACING_SAMPLING_RATE") tracingSamplingRate, err := strconv.ParseFloat(tracingSamplingRateStr, 64) if err != nil { @@ -227,7 +238,7 @@ func Start(logger *zap.Logger, port int, executorUrl string) { keepAliveTime: keepAliveTime, maxRetries: maxRetries, svcAddrRetryCount: svcAddrRetryCount, - }, isDebugEnv, throttler.MakeThrottler(svcAddrUpdateTimeout)) + }, isDebugEnv, unTapServiceTimeout, throttler.MakeThrottler(svcAddrUpdateTimeout)) resolver := makeFunctionReferenceResolver(fnStore)