Improvements from scale testing (#1812)

Poolmanager when tested at high load had some issues and this PR fixes one set of them which were found so far. 

Co-authored-by: Vishal <vishal-biyani@users.noreply.github.com>

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

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

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.

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

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

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.

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

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.

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

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

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 {

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) {

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")

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 {

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

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)