Adding Concurrency in Pool Manager (#1698)

Concurrency in the pool manager allows specializing pods concurrently based on a specified limit. 

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

pkg/executor/api.go

@@ -61,6 +61,28 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
 		return
 	}
 
+	t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+	et, exists := executor.executorTypes[t]
+	if !exists {
+		http.Error(w, fmt.Sprintf("Unknown executor type '%v'", t), http.StatusNotFound)
+		return
+	}
+
+	executor.logger.Debug(fmt.Sprintf("active instances: %v", et.GetTotalAvailable(fn)))
+	conncurrency := fn.Spec.Concurrency
+	if conncurrency == 0 {
+		// set to default conncurrency
+		conncurrency = 5
+		executor.logger.Debug(fmt.Sprintf("concurrency specified in function: %v", fn.Spec.Concurrency))
+		executor.logger.Debug("setting concurrency to 5")
+	}
+	if t == fv1.ExecutorTypePoolmgr && et.GetTotalAvailable(fn) >= conncurrency {
+		errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, fn.Spec.Concurrency)
+		executor.logger.Error("error occured", zap.String("error", errMsg))
+		http.Error(w, errMsg, http.StatusTooManyRequests)
+		return
+	}
+
 	serviceName, err := executor.getServiceForFunction(fn)
 	if err != nil {
 		code, msg := ferror.GetHTTPError(err)
@@ -85,17 +107,13 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
 // To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and
 // invalidates the cache entry if the pod address was cached.
 func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error) {
+	t := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
+	et := executor.executorTypes[t]
 	// Check function -> svc cache
 	executor.logger.Debug("checking for cached function service",
 		zap.String("function_name", fn.ObjectMeta.Name),
 		zap.String("function_namespace", fn.ObjectMeta.Namespace))
 
-	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 et.IsValid(fsvc) {
@@ -181,12 +199,50 @@ func (executor *Executor) healthHandler(w http.ResponseWriter, r *http.Request)
 	w.WriteHeader(http.StatusOK)
 }
 
+func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		http.Error(w, "Failed to read request", http.StatusInternalServerError)
+		return
+	}
+	tapSvcReq := client.TapServiceRequest{}
+	err = json.Unmarshal(body, &tapSvcReq)
+	if err != nil {
+		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
+	if t != fv1.ExecutorTypePoolmgr {
+		msg := fmt.Sprintf("Unknown executor type '%v'", t)
+		http.Error(w, msg, http.StatusBadRequest)
+		return
+	}
+
+	et := executor.executorTypes[t]
+
+	et.UnTapService(fn, tapSvcReq.ServiceUrl)
+
+	w.WriteHeader(http.StatusOK)
+}
+
 func (executor *Executor) GetHandler() http.Handler {
 	r := mux.NewRouter()
 	r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST")
 	r.HandleFunc("/v2/tapService", executor.tapService).Methods("POST") // for backward compatibility
 	r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST")
 	r.HandleFunc("/healthz", executor.healthHandler).Methods("GET")
+	r.HandleFunc("/v2/unTapService", executor.unTapService).Methods("POST")
 	return r
 }
 

pkg/executor/client/client.go

@@ -91,6 +91,32 @@ func (c *Client) GetServiceForFunction(ctx context.Context, metadata *metav1.Obj
 	return string(svcName), nil
 }
 
+func (c *Client) UnTapService(ctx context.Context, fnMeta metav1.ObjectMeta, executorType fv1.ExecutorType, serviceUrl *url.URL) error {
+	url := c.executorUrl + "/v2/unTapService"
+	tapSvc := TapServiceRequest{
+		FnMetadata:     fnMeta,
+		FnExecutorType: executorType,
+		ServiceUrl:     strings.TrimPrefix(serviceUrl.String(), "http://"),
+	}
+
+	body, err := json.Marshal(tapSvc)
+	if err != nil {
+		return errors.Wrap(err, "could not marshal request body for getting service for function")
+	}
+
+	resp, err := ctxhttp.Post(ctx, c.httpClient, url, "application/json", bytes.NewReader(body))
+	if err != nil {
+		return errors.Wrap(err, "error posting to getting service for function")
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return ferror.MakeErrorFromHTTP(resp)
+	}
+
+	return nil
+}
+
 func (c *Client) service() {
 	ticker := time.NewTicker(time.Second * 5)
 	for {

pkg/executor/executor.go

@@ -99,6 +99,31 @@ func (executor *Executor) serveCreateFuncServices() {
 		req := <-executor.requestChan
 		fnMetadata := &req.function.ObjectMeta
 
+		if req.function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypePoolmgr {
+			go func() {
+				buffer := 10 // add some buffer time for specialization
+				specializationTimeout := req.function.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout
+
+				// set minimum specialization timeout to avoid illegal input and
+				// compatibility problem when applying old spec file that doesn't
+				// have specialization timeout field.
+				if specializationTimeout < fv1.DefaultSpecializationTimeOut {
+					specializationTimeout = fv1.DefaultSpecializationTimeOut
+				}
+
+				fnSpecializationTimeoutContext, cancel := context.WithTimeout(context.Background(),
+					time.Duration(specializationTimeout+buffer)*time.Second)
+				defer cancel()
+
+				fsvc, err := executor.createServiceForFunction(fnSpecializationTimeoutContext, req.function)
+				req.respChan <- &createFuncServiceResponse{
+					funcSvc: fsvc,
+					err:     err,
+				}
+			}()
+			continue
+		}
+
 		// Cache miss -- is this first one to request the func?
 		wg, found := executor.fsCreateWg[crd.CacheKey(fnMetadata)]
 		if !found {

pkg/executor/executortype/executortype.go

@@ -18,6 +18,7 @@ package executortype
 
 import (
 	"context"
+
 	"go.uber.org/zap"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
@@ -44,6 +45,9 @@ type ExecutorType interface {
 	// avoid idle pod reaper recycles pods.
 	TapService(serviceUrl string) error
 
+	// UnTapService updates the isActive to false
+	UnTapService(fn *fv1.Function, svcHost string)
+
 	// IsValid returns true if a function service is valid. Different executor types
 	// use distinct ways to examine the function service.
 	IsValid(*fscache.FuncSvc) bool
@@ -56,4 +60,7 @@ type ExecutorType interface {
 
 	// CleanupOldExecutorObjects cleans up resources created by old executor instances
 	CleanupOldExecutorObjects()
+
+	// getTotalAvailable returns total active instances of particular function
+	GetTotalAvailable(*fv1.Function) int
 }

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -153,6 +153,15 @@ func (deploy *NewDeploy) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
 	deploy.fsCache.DeleteEntry(fsvc)
 }
 
+func (deploy *NewDeploy) UnTapService(fn *fv1.Function, svcHost string) {
+	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
+}
+
+func (deploy *NewDeploy) GetTotalAvailable(fn *fv1.Function) int {
+	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
+	return 0
+}
+
 func (deploy *NewDeploy) TapService(svcHost string) error {
 	err := deploy.fsCache.TouchByAddress(svcHost)
 	if err != nil {

pkg/executor/executortype/poolmgr/gp.go

@@ -703,10 +703,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 		Atime:             time.Now(),
 	}
 
-	_, err = gp.fsCache.Add(*fsvc)
-	if err != nil {
-		return nil, err
-	}
+	gp.fsCache.AddFunc(*fsvc)
 
 	gp.fsCache.IncreaseColdStarts(fn.ObjectMeta.Name, string(fn.ObjectMeta.UID))
 

pkg/executor/executortype/poolmgr/gpm.go

@@ -166,11 +166,19 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function)
 }
 
 func (gpm *GenericPoolManager) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) {
-	return gpm.fsCache.GetByFunction(&fn.ObjectMeta)
+	return gpm.fsCache.GetFuncSvc(&fn.ObjectMeta)
 }
 
 func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(fsvc *fscache.FuncSvc) {
-	gpm.fsCache.DeleteEntry(fsvc)
+	gpm.fsCache.DeleteFunctionSvc(fsvc)
+}
+
+func (gpm *GenericPoolManager) UnTapService(fn *fv1.Function, svcHost string) {
+	gpm.fsCache.MarkAvailable(fn, svcHost)
+}
+
+func (gpm *GenericPoolManager) GetTotalAvailable(fn *fv1.Function) int {
+	return gpm.fsCache.GetTotalAvailable(&fn.ObjectMeta)
 }
 
 func (gpm *GenericPoolManager) TapService(svcHost string) error {
@@ -581,7 +589,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 			fnList[fn.ObjectMeta.UID] = fns.Items[i]
 		}
 
-		funcSvcs, err := gpm.fsCache.ListOld(pollSleep)
+		funcSvcs, err := gpm.fsCache.ListOldForPool(pollSleep)
 		if err != nil {
 			gpm.logger.Error("error reaping idle pods", zap.Error(err))
 			continue
@@ -618,7 +626,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 			}
 
 			go func() {
-				deleted, err := gpm.fsCache.DeleteOld(fsvc, idlePodReapTime)
+				deleted, err := gpm.fsCache.DeleteOldPoolCache(fsvc, idlePodReapTime)
 				if err != nil {
 					gpm.logger.Error("error deleting Kubernetes objects for function service",
 						zap.Error(err),

pkg/executor/fscache/functionServiceCache.go

@@ -30,6 +30,7 @@ import (
 	"github.com/fission/fission/pkg/cache"
 	"github.com/fission/fission/pkg/crd"
 	ferror "github.com/fission/fission/pkg/error"
+	poolcache "github.com/fission/fission/pkg/newcache"
 )
 
 type fscRequestType int
@@ -40,6 +41,7 @@ const (
 	TOUCH fscRequestType = iota
 	LISTOLD
 	LOG
+	LISTOLDPOOL
 )
 
 type (
@@ -56,10 +58,11 @@ type (
 	}
 
 	FunctionServiceCache struct {
-		logger        *zap.Logger
-		byFunction    *cache.Cache // function-key -> funcSvc  : map[string]*funcSvc
-		byAddress     *cache.Cache // address      -> function : map[string]metav1.ObjectMeta
-		byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta
+		logger            *zap.Logger
+		byFunction        *cache.Cache     // function-key -> funcSvc  : map[string]*funcSvc
+		byAddress         *cache.Cache     // address      -> function : map[string]metav1.ObjectMeta
+		byFunctionUID     *cache.Cache     // function uid -> function : map[string]metav1.ObjectMeta
+		connFunctionCache *poolcache.Cache // function-key -> funcSvc : map[string]*funcSvc
 
 		requestChannel chan *fscRequest
 	}
@@ -91,11 +94,12 @@ func IsNameExistError(err error) bool {
 
 func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache {
 	fsc := &FunctionServiceCache{
-		logger:         logger.Named("function_service_cache"),
-		byFunction:     cache.MakeCache(0, 0),
-		byAddress:      cache.MakeCache(0, 0),
-		byFunctionUID:  cache.MakeCache(0, 0),
-		requestChannel: make(chan *fscRequest),
+		logger:            logger.Named("function_service_cache"),
+		byFunction:        cache.MakeCache(0, 0),
+		byAddress:         cache.MakeCache(0, 0),
+		byFunctionUID:     cache.MakeCache(0, 0),
+		connFunctionCache: poolcache.NewPoolCache(),
+		requestChannel:    make(chan *fscRequest),
 	}
 	go fsc.service()
 	return fsc
@@ -131,6 +135,17 @@ func (fsc *FunctionServiceCache) service() {
 				}
 			}
 			fsc.logger.Info("function service cache", zap.Int("item_count", len(funcCopy)), zap.Strings("cache", info))
+		case LISTOLDPOOL:
+			fscs := fsc.connFunctionCache.ListValue()
+			funcObjects := make([]*FuncSvc, 0)
+			for _, funcSvc := range fscs {
+				fsvc := funcSvc.(*FuncSvc)
+				if time.Since(fsvc.Atime) > req.age {
+					funcObjects = append(funcObjects, fsvc)
+				}
+			}
+			resp.objects = funcObjects
+
 		}
 		req.responseChannel <- resp
 	}
@@ -152,6 +167,23 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc,
 	return &fsvcCopy, nil
 }
 
+func (fsc *FunctionServiceCache) GetFuncSvc(m *metav1.ObjectMeta) (*FuncSvc, error) {
+	key := crd.CacheKey(m)
+
+	fsvcI, err := fsc.connFunctionCache.GetValue(key)
+	if err != nil {
+		fsc.logger.Info("Not found in Cache")
+		return nil, err
+	}
+
+	// update atime
+	fsvc := fsvcI.(*FuncSvc)
+	fsvc.Atime = time.Now()
+
+	fsvcCopy := *fsvc
+	return &fsvcCopy, nil
+}
+
 func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) {
 	mI, err := fsc.byFunctionUID.Get(uid)
 	if err != nil {
@@ -173,6 +205,23 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
 	return &fsvcCopy, nil
 }
 
+func (fsc *FunctionServiceCache) AddFunc(fsvc FuncSvc) {
+	fsc.connFunctionCache.SetValue(crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc)
+	now := time.Now()
+	fsvc.Ctime = now
+	fsvc.Atime = now
+
+	fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), true)
+}
+
+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) Add(fsvc FuncSvc) (*FuncSvc, error) {
 	existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)
 	if err != nil {
@@ -255,6 +304,10 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
 	fsc.setFuncAlive(fsvc.Function.Name, string(fsvc.Function.UID), false)
 }
 
+func (fsc *FunctionServiceCache) DeleteFunctionSvc(fsvc *FuncSvc) {
+	fsc.connFunctionCache.DeleteValue(crd.CacheKey(fsvc.Function), fsvc.Address)
+}
+
 func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) {
 	if time.Since(fsvc.Atime) < minAge {
 		return false, nil
@@ -265,6 +318,16 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration)
 	return true, nil
 }
 
+func (fsc *FunctionServiceCache) DeleteOldPoolCache(fsvc *FuncSvc, minAge time.Duration) (bool, error) {
+	if time.Since(fsvc.Atime) < minAge {
+		return false, nil
+	}
+
+	fsc.DeleteFunctionSvc(fsvc)
+
+	return true, nil
+}
+
 func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) {
 	responseChannel := make(chan *fscResponse)
 	fsc.requestChannel <- &fscRequest{
@@ -276,6 +339,17 @@ func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error)
 	return resp.objects, resp.error
 }
 
+func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, error) {
+	responseChannel := make(chan *fscResponse)
+	fsc.requestChannel <- &fscRequest{
+		requestType:     LISTOLDPOOL,
+		age:             age,
+		responseChannel: responseChannel,
+	}
+	resp := <-responseChannel
+	return resp.objects, resp.error
+}
+
 func (fsc *FunctionServiceCache) Log() {
 	fsc.logger.Info("--- FunctionService Cache Contents")
 	responseChannel := make(chan *fscResponse)

pkg/executor/fscache/functionServiceCache_test.go

@@ -1,6 +1,7 @@
 package fscache
 
 import (
+	"fmt"
 	"log"
 	"testing"
 	"time"
@@ -122,3 +123,89 @@ func TestFunctionServiceCache(t *testing.T) {
 		log.Panicf("found fsvc by function uid while expecting empty cache: %v", err)
 	}
 }
+
+func TestFunctionServiceNewCache(t *testing.T) {
+	logger, err := zap.NewDevelopment()
+	panicIf(err)
+
+	fsc := MakeFunctionServiceCache(logger)
+	if fsc == nil {
+		log.Panicf("error creating cache")
+	}
+
+	var fsvc *FuncSvc
+	now := time.Now()
+
+	objects := []apiv1.ObjectReference{
+		{
+			Kind:       "pod",
+			Name:       "xxx",
+			APIVersion: "v1",
+			Namespace:  "fission-function",
+		},
+		{
+			Kind:       "pod",
+			Name:       "xxx2",
+			APIVersion: "v1",
+			Namespace:  "fission-function",
+		},
+	}
+
+	fsvc = &FuncSvc{
+		Function: &metav1.ObjectMeta{
+			Name: "foo",
+			UID:  "1212",
+		},
+		Environment: &fv1.Environment{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "foo-env",
+				UID:  "2323",
+			},
+			Spec: fv1.EnvironmentSpec{
+				Version: 1,
+				Runtime: fv1.Runtime{
+					Image: "fission/foo-env",
+				},
+				Builder: fv1.Builder{},
+			},
+		},
+		Address:           "xxx",
+		KubernetesObjects: objects,
+		Ctime:             now,
+		Atime:             now,
+	}
+
+	fn := &fv1.Function{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "foo",
+			UID:  "1212",
+		},
+	}
+
+	fsc.AddFunc(*fsvc)
+
+	active := fsc.GetTotalAvailable(fsvc.Function)
+	if active != 1 {
+		logger.Panic(fmt.Sprintln("active instances not matched expected 1, found ", active))
+	}
+
+	fsc.MarkAvailable(fn, fsvc.Address)
+
+	if fsc.GetTotalAvailable(fsvc.Function) != 0 {
+		log.Panicln("active instances not matched")
+	}
+
+	_, err = fsc.GetFuncSvc(fsvc.Function)
+	if err != nil {
+		logger.Panic("received error while retrieving value from cache")
+	}
+
+	vals, err := fsc.ListOldForPool(30 * time.Second)
+	if err != nil {
+		logger.Panic("received error while get list of old values")
+	}
+	if len(vals) != 0 {
+		logger.Panic(fmt.Sprintln("list of old values didn't matched the expected: 1", "received", len(vals)))
+	}
+	fsc.DeleteFunctionSvc(fsvc)
+}