Fix poolmanager terminates running function pod periodically (#1435)

The pool manager keeps terminating function pod periodically even there are
traffic to the function. The root cause is that executor, poolmgr, newdeploy
manage their own functionServiceCache separately. And when router taps a
function, executor updates the access time of the function service entry in its
own cache without notifying executor types to do the update as well. Hence,
the access time of function service entry in poolmanager cache never gets updated.
Due to the access time never gets updated, the idle pod reaper in poolmanager
then thinks the function pod is in idle state and recycle it.

This PR removes the cache in executor itself, and when router tries to tap a function,
executor will call executor type to tap the function and update access time.

pkg/executor/api.go

@@ -20,13 +20,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"github.com/hashicorp/go-multierror"
-	"github.com/pkg/errors"
 	"io/ioutil"
 	"net/http"
 	"strings"
 
 	"github.com/gorilla/mux"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
 	"go.opencensus.io/plugin/ochttp"
 	"go.uber.org/zap"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
@@ -91,9 +91,15 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error
 		zap.String("function_name", fn.Metadata.Name),
 		zap.String("function_namespace", fn.Metadata.Namespace))
 
-	fsvc, err := executor.fsCache.GetByFunction(&fn.Metadata)
+	t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+	et, exists := executor.executorTypes[t]
+	if !exists {
+		return "", errors.Errorf("Unknown executor type '%v'", t)
+	}
+
+	fsvc, err := et.GetFuncSvcFromCache(fn)
 	if err == nil {
-		if executor.isValidAddress(fsvc) {
+		if et.IsValid(fsvc) {
 			// Cached, return svc address
 			return fsvc.Address, nil
 		} else {
@@ -101,7 +107,7 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error
 				zap.String("function_name", fn.Metadata.Name),
 				zap.String("function_namespace", fn.Metadata.Namespace),
 				zap.String("address", fsvc.Address))
-			executor.fsCache.DeleteEntry(fsvc)
+			et.DeleteFuncSvcFromCache(fsvc)
 		}
 	}
 
@@ -120,24 +126,7 @@ func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error
 // find funcSvc and update its atime
 // TODO: Deprecated tapService
 func (executor *Executor) tapService(w http.ResponseWriter, r *http.Request) {
-	body, err := ioutil.ReadAll(r.Body)
-	if err != nil {
-		executor.logger.Error("failed to read tap service request", zap.Error(err))
-		http.Error(w, "Failed to read request", http.StatusInternalServerError)
-		return
-	}
-	svcName := string(body)
-	svcHost := strings.TrimPrefix(svcName, "http://")
-
-	err = executor.fsCache.TouchByAddress(svcHost)
-	if err != nil {
-		executor.logger.Error("error tapping function service",
-			zap.Error(err),
-			zap.String("service", svcName),
-			zap.String("host", svcHost))
-		http.Error(w, "Not found", http.StatusNotFound)
-		return
-	}
+	// only for upgrade compatibility
 	w.WriteHeader(http.StatusOK)
 }
 
@@ -163,7 +152,16 @@ func (executor *Executor) tapServices(w http.ResponseWriter, r *http.Request) {
 	errs := &multierror.Error{}
 	for _, req := range tapSvcReqs {
 		svcHost := strings.TrimPrefix(req.ServiceUrl, "http://")
-		err = executor.fsCache.TouchByAddress(svcHost)
+
+		et, exists := executor.executorTypes[req.FnExecutorType]
+		if !exists {
+			errs = multierror.Append(errs,
+				errors.Errorf("error tapping service due to unknown executor type '%v' found",
+					req.FnExecutorType))
+			continue
+		}
+
+		err = et.TapService(svcHost)
 		if err != nil {
 			errs = multierror.Append(errs,
 				errors.Wrapf(err, "'%v' failed to tap function '%v/%v' with service url '%v'",
@@ -199,10 +197,11 @@ func (executor *Executor) Serve(port int) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	executor.ndm.Run(ctx)
-	executor.gpm.Run(ctx)
-	executor.cms.Run(ctx)
+	for _, et := range executor.executorTypes {
+		et.Run(ctx)
+	}
 
+	executor.cms.Run(ctx)
 	address := fmt.Sprintf(":%v", port)
 	err := http.ListenAndServe(address, &ochttp.Handler{
 		Handler: executor.GetHandler(),

pkg/executor/cms/cmscontroller.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/pkg/errors"
 	"go.uber.org/zap"
 	apiv1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -29,8 +30,7 @@ import (
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
-	nd "github.com/fission/fission/pkg/executor/newdeploy"
-	gpm "github.com/fission/fission/pkg/executor/poolmgr"
+	"github.com/fission/fission/pkg/executor/executortype"
 )
 
 type (
@@ -46,10 +46,10 @@ type (
 
 //MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions
 func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
-	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) *ConfigSecretController {
+	kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) *ConfigSecretController {
 	logger.Debug("Creating ConfigMap & Secret Controller")
-	_, cmcontroller := initConfigmapController(logger, fissionClient, kubernetesClient, ndm, gpm)
-	_, scontroller := initSecretController(logger, fissionClient, kubernetesClient, ndm, gpm)
+	_, cmcontroller := initConfigmapController(logger, fissionClient, kubernetesClient, types)
+	_, scontroller := initSecretController(logger, fissionClient, kubernetesClient, types)
 	cmsController := &ConfigSecretController{
 		logger:              logger,
 		configmapController: cmcontroller,
@@ -66,7 +66,7 @@ func (csController *ConfigSecretController) Run(ctx context.Context) {
 }
 
 func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClient,
-	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) {
+	kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) (cache.Store, cache.Controller) {
 	resyncPeriod := 30 * time.Second
 	listWatch := cache.NewListWatchFromClient(kubernetesClient.CoreV1().RESTClient(), "configmaps", metav1.NamespaceAll, fields.Everything())
 	store, controller := cache.NewInformer(listWatch, &apiv1.ConfigMap{}, resyncPeriod, cache.ResourceEventHandlerFuncs{
@@ -81,12 +81,11 @@ func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClien
 						zap.String("configmap_name", newCm.ObjectMeta.Name),
 						zap.String("configmap_namespace", newCm.ObjectMeta.Namespace))
 				}
-
 				funcs, err := getConfigmapRelatedFuncs(logger, &newCm.ObjectMeta, fissionClient)
 				if err != nil {
 					logger.Error("Failed to get functions related to secret", zap.String("secret_name", newCm.ObjectMeta.Name), zap.String("secret_namespace", newCm.ObjectMeta.Namespace))
 				}
-				recyclePods(logger, funcs, ndm, gpm)
+				recyclePods(logger, funcs, types)
 			}
 		},
 	})
@@ -112,7 +111,7 @@ func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionC
 }
 
 func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
-	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) {
+	kubernetesClient *kubernetes.Clientset, types map[fv1.ExecutorType]executortype.ExecutorType) (cache.Store, cache.Controller) {
 	resyncPeriod := 30 * time.Second
 	listWatch := cache.NewListWatchFromClient(kubernetesClient.CoreV1().RESTClient(), "secrets", metav1.NamespaceAll, fields.Everything())
 	store, controller := cache.NewInformer(listWatch, &apiv1.Secret{}, resyncPeriod, cache.ResourceEventHandlerFuncs{
@@ -128,14 +127,12 @@ func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
 						zap.String("configmap_namespace", newS.ObjectMeta.Namespace))
 
 				}
-
 				funcs, err := getSecretRelatedFuncs(logger, &newS.ObjectMeta, fissionClient)
 				if err != nil {
 					logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace))
 				}
-				recyclePods(logger, funcs, ndm, gpm)
+				recyclePods(logger, funcs, types)
 			}
-
 		},
 	})
 	return store, controller
@@ -160,19 +157,19 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie
 	return relatedFunctions, nil
 }
 
-func recyclePods(logger *zap.Logger, funcs []fv1.Function, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) {
+func recyclePods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) {
 	for _, f := range funcs {
 		var err error
 
-		switch f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType {
-		case fv1.ExecutorTypeNewdeploy:
-			err = ndm.RefreshFuncPods(logger, f)
-		case fv1.ExecutorTypePoolmgr:
-			err = gpm.RefreshFuncPods(logger, f)
+		et, exists := types[f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType]
+		if exists {
+			err = et.RefreshFuncPods(logger, f)
+		} else {
+			err = errors.Errorf("Unknown executor type '%v'", f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType)
 		}
 
 		if err != nil {
-			logger.Error("Failed to recycle pods for function after configmap changed",
+			logger.Error("Failed to recycle pods for function after configmap/secret changed",
 				zap.Error(err),
 				zap.Any("function", f))
 		}

pkg/executor/executor.go

@@ -32,9 +32,10 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/executor/cms"
+	"github.com/fission/fission/pkg/executor/executortype"
+	"github.com/fission/fission/pkg/executor/executortype/newdeploy"
+	"github.com/fission/fission/pkg/executor/executortype/poolmgr"
 	"github.com/fission/fission/pkg/executor/fscache"
-	"github.com/fission/fission/pkg/executor/newdeploy"
-	"github.com/fission/fission/pkg/executor/poolmgr"
 	"github.com/fission/fission/pkg/executor/reaper"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
 )
@@ -43,12 +44,10 @@ type (
 	Executor struct {
 		logger *zap.Logger
 
-		gpm *poolmgr.GenericPoolManager
-		ndm *newdeploy.NewDeploy
-		cms *cms.ConfigSecretController
+		executorTypes map[fv1.ExecutorType]executortype.ExecutorType
+		cms           *cms.ConfigSecretController
 
 		fissionClient *crd.FissionClient
-		fsCache       *fscache.FunctionServiceCache
 
 		requestChan chan *createFuncServiceRequest
 		fsCreateWg  map[string]*sync.WaitGroup
@@ -64,21 +63,20 @@ type (
 	}
 )
 
-func MakeExecutor(logger *zap.Logger, gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, cms *cms.ConfigSecretController, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor {
+func MakeExecutor(logger *zap.Logger, cms *cms.ConfigSecretController,
+	fissionClient *crd.FissionClient, types map[fv1.ExecutorType]executortype.ExecutorType) (*Executor, error) {
 	executor := &Executor{
 		logger:        logger.Named("executor"),
-		gpm:           gpm,
-		ndm:           ndm,
 		cms:           cms,
 		fissionClient: fissionClient,
-		fsCache:       fsCache,
+		executorTypes: types,
 
 		requestChan: make(chan *createFuncServiceRequest),
 		fsCreateWg:  make(map[string]*sync.WaitGroup),
 	}
 	go executor.serveCreateFuncServices()
 
-	return executor
+	return executor, nil
 }
 
 // All non-cached function service requests go through this goroutine
@@ -115,6 +113,7 @@ func (executor *Executor) serveCreateFuncServices() {
 				buffer := 10 // add some buffer time for specialization
 				fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(),
 					time.Duration(req.function.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout+buffer)*time.Second)
+				defer cancel()
 
 				fsvc, err := executor.createServiceForFunction(fnSpecializationTimeoutContext, req.function)
 				req.respChan <- &createFuncServiceResponse{
@@ -122,8 +121,6 @@ func (executor *Executor) serveCreateFuncServices() {
 					err:     err,
 				}
 				delete(executor.fsCreateWg, crd.CacheKey(fnMetadata))
-
-				cancel()
 				wg.Done()
 			}()
 		} else {
@@ -134,7 +131,7 @@ func (executor *Executor) serveCreateFuncServices() {
 				wg.Wait()
 
 				// get the function service from the cache
-				fsvc, err := executor.fsCache.GetByFunction(fnMetadata)
+				fsvc, err := executor.getFunctionServiceFromCache(req.function)
 
 				// fsCache return error when the entry does not exist/expire.
 				// It normally happened if there are multiple requests are
@@ -155,18 +152,13 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.
 		zap.String("function_name", fn.Metadata.Name),
 		zap.String("function_namespace", fn.Metadata.Namespace))
 
-	executorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
-
-	var fsvc *fscache.FuncSvc
-	var fsvcErr error
-
-	switch executorType {
-	case fv1.ExecutorTypeNewdeploy:
-		fsvc, fsvcErr = executor.ndm.GetFuncSvc(ctx, fn)
-	default:
-		fsvc, fsvcErr = executor.gpm.GetFuncSvc(ctx, fn)
+	t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+	e, ok := executor.executorTypes[t]
+	if !ok {
+		return nil, errors.Errorf("Unknown executor type '%v'", t)
 	}
 
+	fsvc, fsvcErr := e.GetFuncSvc(ctx, fn)
 	if fsvcErr != nil {
 		e := "error creating service for function"
 		executor.logger.Error(e,
@@ -174,25 +166,18 @@ func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.
 			zap.String("function_name", fn.Metadata.Name),
 			zap.String("function_namespace", fn.Metadata.Namespace))
 		fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%s] %s", fn.Metadata.Name, e))
-	} else if fsvc != nil {
-		_, err := executor.fsCache.Add(*fsvc)
-		if err != nil {
-			return nil, err
-		}
 	}
 
-	executor.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID))
-
 	return fsvc, fsvcErr
 }
 
-// isValidAddress invokes isValidService or isValidPod depending on the type of executor
-func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool {
-	if fsvc.Executor == fscache.NEWDEPLOY {
-		return executor.ndm.IsValid(fsvc)
-	} else {
-		return executor.gpm.IsValid(fsvc)
+func (executor *Executor) getFunctionServiceFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+	e, ok := executor.executorTypes[t]
+	if !ok {
+		return nil, errors.Errorf("Unknown executor type '%v'", t)
 	}
+	return e.GetFuncSvcFromCache(fn)
 }
 
 func serveMetric(logger *zap.Logger) {
@@ -223,7 +208,6 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
 	}
 
 	restClient := fissionClient.GetCrdClient()
-	fsCache := fscache.MakeFunctionServiceCache(logger)
 
 	poolID := strings.ToLower(uniuri.NewLen(8))
 	reaper.CleanupOldExecutorObjects(logger, kubernetesClient, poolID)
@@ -239,9 +223,16 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
 		fissionClient, kubernetesClient, restClient,
 		functionNamespace, fetcherConfig, poolID)
 
-	cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, ndm, gpm)
+	executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType)
+	executorTypes[gpm.GetTypeName()] = gpm
+	executorTypes[ndm.GetTypeName()] = ndm
 
-	api := MakeExecutor(logger, gpm, ndm, cms, fissionClient, fsCache)
+	cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, executorTypes)
+
+	api, err := MakeExecutor(logger, cms, fissionClient, executorTypes)
+	if err != nil {
+		return err
+	}
 
 	go api.Serve(port)
 	go serveMetric(logger)

pkg/executor/executortype/executortype.go

@@ -0,0 +1,53 @@
+/*
+Copyright 2019 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package executortype
+
+import (
+	"context"
+	"go.uber.org/zap"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+	"github.com/fission/fission/pkg/executor/fscache"
+)
+
+type ExecutorType interface {
+	// Run runs background job.
+	Run(context.Context)
+
+	// GetTypeName returns the name of executor type
+	GetTypeName() fv1.ExecutorType
+
+	// GetFuncSvc specializes function pod(s) and returns a service URL for the function.
+	GetFuncSvc(context.Context, *fv1.Function) (*fscache.FuncSvc, error)
+
+	// GetFuncSvcFromCache retrieves function service from cache.
+	GetFuncSvcFromCache(*fv1.Function) (*fscache.FuncSvc, error)
+
+	// DeleteFuncSvcFromCache deletes function service entry in cache.
+	DeleteFuncSvcFromCache(*fscache.FuncSvc)
+
+	// TapService updates the access time of function service entry to
+	// avoid idle pod reaper recycles pods.
+	TapService(serviceUrl string) error
+
+	// IsValid returns true if a function service is valid. Different executor types
+	// use distinct ways to examine the function service.
+	IsValid(*fscache.FuncSvc) bool
+
+	// RefreshFuncPods refreshes function pods if the secrets/configmaps pods reference to get updated.
+	RefreshFuncPods(*zap.Logger, fv1.Function) error
+}

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -24,8 +24,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/fission/fission/pkg/throttler"
-	"github.com/fission/fission/pkg/utils"
 	multierror "github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
@@ -42,11 +40,16 @@ import (
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
+	"github.com/fission/fission/pkg/executor/executortype"
 	"github.com/fission/fission/pkg/executor/fscache"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
+	"github.com/fission/fission/pkg/throttler"
 	"github.com/fission/fission/pkg/types"
+	"github.com/fission/fission/pkg/utils"
 )
 
+var _ executortype.ExecutorType = &NewDeploy{}
+
 type (
 	NewDeploy struct {
 		logger *zap.Logger
@@ -82,7 +85,7 @@ func MakeNewDeploy(
 	namespace string,
 	fetcherConfig *fetcherConfig.Config,
 	instanceID string,
-) *NewDeploy {
+) executortype.ExecutorType {
 	enableIstio := false
 	if len(os.Getenv("ENABLE_ISTIO")) > 0 {
 		istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO"))
@@ -130,6 +133,105 @@ func (deploy *NewDeploy) Run(ctx context.Context) {
 	go deploy.idleObjectReaper()
 }
 
+func (deploy *NewDeploy) GetTypeName() fv1.ExecutorType {
+	return fv1.ExecutorTypeNewdeploy
+}
+
+func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
+	return deploy.createFunction(fn, false)
+}
+
+func (deploy *NewDeploy) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	return deploy.fsCache.GetByFunction(&fn.Metadata)
+}
+
+func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
+	deploy.fsCache.DeleteEntry(fsvc)
+}
+
+func (deploy *NewDeploy) TapService(svcHost string) error {
+	err := deploy.fsCache.TouchByAddress(svcHost)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// IsValid does a get on the service address to ensure it's a valid service, then
+// scale deployment to 1 replica if there are no available replicas for function.
+// Return true if no error occurs, return false otherwise.
+func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
+	service := strings.Split(fsvc.Address, ".")
+	if len(service) == 0 {
+		return false
+	}
+
+	_, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{})
+	if err != nil {
+		deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err))
+		return false
+	}
+
+	deployObj := getDeploymentObj(fsvc.KubernetesObjects)
+	if deployObj == nil {
+		deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name))
+		return false
+	}
+
+	currentDeploy, err := deploy.kubernetesClient.AppsV1().
+		Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{})
+	if err != nil {
+		deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+		return false
+	}
+
+	// return directly when available replicas > 0
+	if currentDeploy.Status.AvailableReplicas > 0 {
+		return true
+	}
+
+	return false
+}
+
+// RefreshFuncPods deleted pods related to the function so that new pods are replenished
+func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
+
+	env, err := deploy.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
+	if err != nil {
+		return err
+	}
+
+	funcLabels := deploy.getDeployLabels(f.Metadata, metav1.ObjectMeta{
+		Name:      f.Spec.Environment.Name,
+		Namespace: f.Spec.Environment.Namespace,
+		UID:       env.Metadata.UID,
+	})
+
+	dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{
+		LabelSelector: labels.Set(funcLabels).AsSelector().String(),
+	})
+
+	if err != nil {
+		return err
+	}
+
+	patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%s"}]}]}}}}`,
+		f.Metadata.Name,
+		fv1.LastUpdateTimestamp,
+		time.Now().String())
+
+	// Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition
+	for _, deployment := range dep.Items {
+		_, err := deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(deployment.ObjectMeta.Name,
+			k8sTypes.StrategicMergePatchType,
+			[]byte(patch))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controller) {
 	resyncPeriod := 30 * time.Second
 	listWatch := k8sCache.NewListWatchFromClient(deploy.crdClient, "functions", metav1.NamespaceAll, fields.Everything())
@@ -223,49 +325,6 @@ func (deploy *NewDeploy) getEnvFunctions(m *metav1.ObjectMeta) []fv1.Function {
 	return relatedFunctions
 }
 
-func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	return deploy.createFunction(fn, false)
-}
-
-// RefreshFuncPods deleted pods related to the function so that new pods are replenished
-func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
-
-	env, err := deploy.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
-	if err != nil {
-		return err
-	}
-
-	funcLabels := deploy.getDeployLabels(f.Metadata, metav1.ObjectMeta{
-		Name:      f.Spec.Environment.Name,
-		Namespace: f.Spec.Environment.Namespace,
-		UID:       env.Metadata.UID,
-	})
-
-	dep, err := deploy.kubernetesClient.AppsV1().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{
-		LabelSelector: labels.Set(funcLabels).AsSelector().String(),
-	})
-
-	if err != nil {
-		return err
-	}
-
-	patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%s"}]}]}}}}`,
-		f.Metadata.Name,
-		fv1.LastUpdateTimestamp,
-		time.Now().String())
-
-	// Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition
-	for _, deployment := range dep.Items {
-		_, err := deploy.kubernetesClient.AppsV1().Deployments(deployment.ObjectMeta.Namespace).Patch(deployment.ObjectMeta.Name,
-			k8sTypes.StrategicMergePatchType,
-			[]byte(patch))
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 func (deploy *NewDeploy) createFunction(fn *fv1.Function, firstcreate bool) (*fscache.FuncSvc, error) {
 	if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy {
 		return nil, nil
@@ -385,7 +444,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache.
 		Environment:       env,
 		Address:           svcAddress,
 		KubernetesObjects: kubeObjRefs,
-		Executor:          fscache.NEWDEPLOY,
+		Executor:          fv1.ExecutorTypeNewdeploy,
 	}
 
 	_, err = deploy.fsCache.Add(*fsvc)
@@ -393,6 +452,9 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache.
 		deploy.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function))
 		return fsvc, err
 	}
+
+	deploy.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID))
+
 	return fsvc, nil
 }
 
@@ -602,7 +664,7 @@ func (deploy *NewDeploy) getObjName(fn *fv1.Function) string {
 func (deploy *NewDeploy) getDeployLabels(fnMeta metav1.ObjectMeta, envMeta metav1.ObjectMeta) map[string]string {
 	return map[string]string{
 		types.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID,
-		types.EXECUTOR_TYPE:             fv1.ExecutorTypeNewdeploy,
+		types.EXECUTOR_TYPE:             string(fv1.ExecutorTypeNewdeploy),
 		types.ENVIRONMENT_NAME:          envMeta.Name,
 		types.ENVIRONMENT_NAMESPACE:     envMeta.Namespace,
 		types.ENVIRONMENT_UID:           string(envMeta.UID),
@@ -618,42 +680,6 @@ func (deploy *NewDeploy) updateStatus(fn *fv1.Function, err error, message strin
 	deploy.logger.Error("function status update", zap.Error(err), zap.Any("function", fn), zap.String("message", message))
 }
 
-// IsValid does a get on the service address to ensure it's a valid service, then
-// scale deployment to 1 replica if there are no available replicas for function.
-// Return true if no error occurs, return false otherwise.
-func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
-	service := strings.Split(fsvc.Address, ".")
-	if len(service) == 0 {
-		return false
-	}
-
-	_, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{})
-	if err != nil {
-		deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err))
-		return false
-	}
-
-	deployObj := getDeploymentObj(fsvc.KubernetesObjects)
-	if deployObj == nil {
-		deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name))
-		return false
-	}
-
-	currentDeploy, err := deploy.kubernetesClient.AppsV1().
-		Deployments(deployObj.Namespace).Get(deployObj.Name, metav1.GetOptions{})
-	if err != nil {
-		deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
-		return false
-	}
-
-	// return directly when available replicas > 0
-	if currentDeploy.Status.AvailableReplicas > 0 {
-		return true
-	}
-
-	return false
-}
-
 // idleObjectReaper reaps objects after certain idle time
 func (deploy *NewDeploy) idleObjectReaper() {
 
@@ -678,7 +704,7 @@ func (deploy *NewDeploy) idleObjectReaper() {
 		}
 
 		for _, fsvc := range funcSvcs {
-			if fsvc.Executor != fscache.NEWDEPLOY {
+			if fsvc.Executor != fv1.ExecutorTypeNewdeploy {
 				continue
 			}
 
@@ -694,7 +720,7 @@ func (deploy *NewDeploy) idleObjectReaper() {
 			if err != nil {
 				// Newdeploy manager handles the function delete event and clean cache/kubeobjs itself,
 				// so we ignore the not found error for functions with newdeploy executor type here.
-				if k8sErrs.IsNotFound(err) && fsvc.Executor == fscache.NEWDEPLOY {
+				if k8sErrs.IsNotFound(err) && fsvc.Executor == fv1.ExecutorTypeNewdeploy {
 					continue
 				}
 				deploy.logger.Error("error getting function", zap.Error(err), zap.String("function", fsvc.Function.Name))

pkg/executor/executortype/poolmgr/gp.go

@@ -144,7 +144,7 @@ func MakeGenericPool(
 func (gp *GenericPool) getDeployLabels() map[string]string {
 	return map[string]string{
 		fv1.EXECUTOR_INSTANCEID_LABEL: gp.instanceId,
-		types.EXECUTOR_TYPE:           fv1.ExecutorTypePoolmgr,
+		types.EXECUTOR_TYPE:           string(fv1.ExecutorTypePoolmgr),
 		types.ENVIRONMENT_NAME:        gp.env.Metadata.Name,
 		types.ENVIRONMENT_NAMESPACE:   gp.env.Metadata.Namespace,
 		types.ENVIRONMENT_UID:         string(gp.env.Metadata.UID),
@@ -471,14 +471,17 @@ func (gp *GenericPool) waitForReadyPod() error {
 
 			// Since even single pod is not ready, choosing the first pod to inspect is a good approximation. In future this can be done better
 			pod := podList.Items[0]
-			multierr := &multierror.Error{}
+			errs := &multierror.Error{}
 			for _, cStatus := range pod.Status.ContainerStatuses {
 				if !cStatus.Ready {
-					multierr = multierror.Append(multierr, errors.New(fmt.Sprintf("%v: %v", cStatus.State.Waiting.Reason, cStatus.State.Waiting.Message)))
+					errs = multierror.Append(errs, errors.New(fmt.Sprintf("%v: %v", cStatus.State.Waiting.Reason, cStatus.State.Waiting.Message)))
 				}
 			}
-			return errors.Wrapf(multierr, "Timeout: waited too long for pod of deployment %v in namespace %v to be ready",
-				gp.deployment.ObjectMeta.Name, gp.namespace)
+			if errs.ErrorOrNil() != nil {
+				return errors.Wrapf(errs, "Timeout: waited too long for pod of deployment %v in namespace %v to be ready",
+					gp.deployment.ObjectMeta.Name, gp.namespace)
+			}
+			return nil
 		}
 		time.Sleep(1000 * time.Millisecond)
 	}
@@ -614,7 +617,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 		Environment:       gp.env,
 		Address:           svcHost,
 		KubernetesObjects: kubeObjRefs,
-		Executor:          fscache.POOLMGR,
+		Executor:          fv1.ExecutorTypePoolmgr,
 		Ctime:             time.Now(),
 		Atime:             time.Now(),
 	}
@@ -623,6 +626,9 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 	if err != nil {
 		return nil, err
 	}
+
+	gp.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID))
+
 	return fsvc, nil
 }
 

pkg/executor/executortype/poolmgr/gpm.go

@@ -35,12 +35,15 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/cache"
 	"github.com/fission/fission/pkg/crd"
+	"github.com/fission/fission/pkg/executor/executortype"
 	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/executor/reaper"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
 	"github.com/fission/fission/pkg/types"
 )
 
+var _ executortype.ExecutorType = &GenericPoolManager{}
+
 type requestType int
 
 const (
@@ -90,7 +93,7 @@ func MakeGenericPoolManager(
 	kubernetesClient *kubernetes.Clientset,
 	functionNamespace string,
 	fetcherConfig *fetcherConfig.Config,
-	instanceId string) *GenericPoolManager {
+	instanceId string) executortype.ExecutorType {
 
 	gpmLogger := logger.Named("generic_pool_manager")
 
@@ -132,6 +135,67 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) {
 	go gpm.idleObjectReaper()
 }
 
+func (gpm *GenericPoolManager) GetTypeName() fv1.ExecutorType {
+	return fv1.ExecutorTypePoolmgr
+}
+
+func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
+	// from Func -> get Env
+	gpm.logger.Debug("getting environment for function", zap.String("function", fn.Metadata.Name))
+	env, err := gpm.getFunctionEnv(fn)
+	if err != nil {
+		return nil, err
+	}
+
+	pool, err := gpm.getPool(env)
+	if err != nil {
+		return nil, err
+	}
+
+	// from GenericPool -> get one function container
+	// (this also adds to the cache)
+	gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name))
+	return pool.getFuncSvc(ctx, fn)
+}
+
+func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
+	return gpm.fsCache.GetByFunction(&fn.Metadata)
+}
+
+func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
+	gpm.fsCache.DeleteEntry(fsvc)
+}
+
+func (gpm *GenericPoolManager) TapService(svcHost string) error {
+	err := gpm.fsCache.TouchByAddress(svcHost)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the
+// containers in it are reporting a ready status for the healthCheck.
+func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
+	for _, obj := range fsvc.KubernetesObjects {
+		if obj.Kind == "pod" {
+			pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{})
+			if err == nil && utils.IsReadyPod(pod) {
+				// Normally, the address format is http://[pod-ip]:[port], however, if the
+				// Istio is enabled the address format changes to http://[svc-name]:[port].
+				// So if the Istio is enabled and pod is in ready state, we return true directly;
+				// Otherwise, we need to ensure that the address contains pod ip.
+				if gpm.enableIstio ||
+					(!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) {
+					gpm.logger.Debug("valid address", zap.String("address", fsvc.Address))
+					return true
+				}
+			}
+		}
+	}
+	return false
+}
+
 func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
 
 	env, err := gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
@@ -246,25 +310,6 @@ func (gpm *GenericPoolManager) cleanupPools(envs []fv1.Environment) {
 	}
 }
 
-func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	// from Func -> get Env
-	gpm.logger.Debug("getting environment for function", zap.String("function", fn.Metadata.Name))
-	env, err := gpm.getFunctionEnv(fn)
-	if err != nil {
-		return nil, err
-	}
-
-	pool, err := gpm.getPool(env)
-	if err != nil {
-		return nil, err
-	}
-
-	// from GenericPool -> get one function container
-	// (this also adds to the cache)
-	gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name))
-	return pool.getFuncSvc(ctx, fn)
-}
-
 func (gpm *GenericPoolManager) getFunctionEnv(fn *fv1.Function) (*fv1.Environment, error) {
 	var env *fv1.Environment
 
@@ -335,28 +380,6 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *fv1.Environment) int32 {
 	return poolsize
 }
 
-// IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the
-// containers in it are reporting a ready status for the healthCheck.
-func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool {
-	for _, obj := range fsvc.KubernetesObjects {
-		if obj.Kind == "pod" {
-			pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{})
-			if err == nil && utils.IsReadyPod(pod) {
-				// Normally, the address format is http://[pod-ip]:[port], however, if the
-				// Istio is enabled the address format changes to http://[svc-name]:[port].
-				// So if the Istio is enabled and pod is in ready state, we return true directly;
-				// Otherwise, we need to ensure that the address contains pod ip.
-				if gpm.enableIstio ||
-					(!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) {
-					gpm.logger.Debug("valid address", zap.String("address", fsvc.Address))
-					return true
-				}
-			}
-		}
-	}
-	return false
-}
-
 // idleObjectReaper reaps objects after certain idle time
 func (gpm *GenericPoolManager) idleObjectReaper() {
 
@@ -380,8 +403,10 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 			continue
 		}
 
-		for _, fsvc := range funcSvcs {
-			if fsvc.Executor != fscache.POOLMGR {
+		for i := range funcSvcs {
+			fsvc := funcSvcs[i]
+
+			if fsvc.Executor != fv1.ExecutorTypePoolmgr {
 				continue
 			}
 
@@ -397,20 +422,23 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 				continue
 			}
 
-			deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime)
-			if err != nil {
-				gpm.logger.Error("error deleting Kubernetes objects for function service",
-					zap.Error(err),
-					zap.Any("service", fsvc))
-			}
-
-			if !deleted {
-				continue
-			}
-
-			for _, kubeobj := range fsvc.KubernetesObjects {
-				reaper.CleanupKubeObject(gpm.logger, gpm.kubernetesClient, &kubeobj)
-			}
+			go func() {
+				deleted, err := gpm.fsCache.DeleteOld(fsvc, gpm.idlePodReapTime)
+				if err != nil {
+					gpm.logger.Error("error deleting Kubernetes objects for function service",
+						zap.Error(err),
+						zap.Any("service", fsvc))
+				}
+				if deleted {
+					for i := range fsvc.KubernetesObjects {
+						gpm.logger.Debug("release idle function resources",
+							zap.String("function", fsvc.Name), zap.String("address", fsvc.Address),
+							zap.String("executor", string(fsvc.Executor)))
+						reaper.CleanupKubeObject(gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i])
+						time.Sleep(50 * time.Millisecond)
+					}
+				}
+			}()
 		}
 	}
 }

pkg/executor/fscache/functionServiceCache.go

@@ -33,7 +33,8 @@ import (
 )
 
 type fscRequestType int
-type executorType int
+
+//type executorType int
 
 const (
 	TOUCH fscRequestType = iota
@@ -41,11 +42,6 @@ const (
 	LOG
 )
 
-const (
-	POOLMGR executorType = iota
-	NEWDEPLOY
-)
-
 type (
 	FuncSvc struct {
 		Name              string                  // Name of object
@@ -53,7 +49,7 @@ type (
 		Environment       *fv1.Environment        // function's environment
 		Address           string                  // Host:Port or IP:Port that the function's service can be reached at.
 		KubernetesObjects []apiv1.ObjectReference // Kubernetes Objects (within the function namespace)
-		Executor          executorType
+		Executor          fv1.ExecutorType
 
 		Ctime time.Time
 		Atime time.Time

pkg/executor/reaper/reaper.go

@@ -26,6 +26,7 @@ import (
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/types"
 )
@@ -294,7 +295,7 @@ func CleanupRoleBindings(logger *zap.Logger, client *kubernetes.Clientset, fissi
 						break
 					}
 
-					if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == types.ExecutorTypeNewdeploy {
+					if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy {
 						ndmFunc = true
 						break
 					}