Poolmanager wait for function specialization timeout when specializing a pod (#1392)

pkg/executor/api.go

@@ -27,8 +27,10 @@ import (
 	"github.com/gorilla/mux"
 	"go.opencensus.io/plugin/ochttp"
 	"go.uber.org/zap"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	ferror "github.com/fission/fission/pkg/error"
 )
 
@@ -47,7 +49,17 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
 		return
 	}
 
-	serviceName, err := executor.getServiceForFunction(r.Context(), &m)
+	fn, err := executor.fissionClient.Functions(m.Namespace).Get(m.Name)
+	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
+	}
+
+	serviceName, err := executor.getServiceForFunction(fn)
 	if err != nil {
 		code, msg := ferror.GetHTTPError(err)
 		executor.logger.Error("error getting service for function",
@@ -70,21 +82,21 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
 // stale addresses are not returned to the router.
 // 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(ctx context.Context, m *metav1.ObjectMeta) (string, error) {
+func (executor *Executor) getServiceForFunction(fn *fv1.Function) (string, error) {
 	// Check function -> svc cache
 	executor.logger.Debug("checking for cached function service",
-		zap.String("function_name", m.Name),
+		zap.String("function_name", fn.Metadata.Name),
-		zap.String("function_namespace", m.Namespace))
+		zap.String("function_namespace", fn.Metadata.Namespace))
 
-	fsvc, err := executor.fsCache.GetByFunction(m)
+	fsvc, err := executor.fsCache.GetByFunction(&fn.Metadata)
 	if err == nil {
 		if executor.isValidAddress(fsvc) {
 			// Cached, return svc address
 			return fsvc.Address, nil
 		} else {
 			executor.logger.Debug("deleting cache entry for invalid address",
-				zap.String("function_name", m.Name),
+				zap.String("function_name", fn.Metadata.Name),
-				zap.String("function_namespace", m.Namespace),
+				zap.String("function_namespace", fn.Metadata.Namespace),
 				zap.String("address", fsvc.Address))
 			executor.fsCache.DeleteEntry(fsvc)
 		}
@@ -92,8 +104,7 @@ func (executor *Executor) getServiceForFunction(ctx context.Context, m *metav1.O
 
 	respChan := make(chan *createFuncServiceResponse)
 	executor.requestChan <- &createFuncServiceRequest{
-		ctx:      ctx,
+		function: fn,
-		funcMeta: m,
 		respChan: respChan,
 	}
 	resp := <-respChan

pkg/executor/executor.go

@@ -28,7 +28,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
@@ -55,8 +54,7 @@ type (
 		fsCreateWg  map[string]*sync.WaitGroup
 	}
 	createFuncServiceRequest struct {
-		ctx      context.Context
+		function *fv1.Function
-		funcMeta *metav1.ObjectMeta
 		respChan chan *createFuncServiceResponse
 	}
 
@@ -92,43 +90,57 @@ func MakeExecutor(logger *zap.Logger, gpm *poolmgr.GenericPoolManager, ndm *newd
 func (executor *Executor) serveCreateFuncServices() {
 	for {
 		req := <-executor.requestChan
-		m := req.funcMeta
+		fnMetadata := &req.function.Metadata
 
 		// Cache miss -- is this first one to request the func?
-		wg, found := executor.fsCreateWg[crd.CacheKey(m)]
+		wg, found := executor.fsCreateWg[crd.CacheKey(fnMetadata)]
 		if !found {
 			// create a waitgroup for other requests for
 			// the same function to wait on
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
-			executor.fsCreateWg[crd.CacheKey(m)] = wg
+			executor.fsCreateWg[crd.CacheKey(fnMetadata)] = wg
 
 			// launch a goroutine for each request, to parallelize
 			// the specialization of different functions
 			go func() {
-				fsvc, err := executor.createServiceForFunction(req.ctx, m)
+				// Control overall specialization time by setting function
+				// specialization time to context. The reason not to use
+				// context from router requests is because a request maybe
+				// canceled for unknown reasons and let executor keeps
+				// spawning pods that never finish specialization process.
+				// Also, even a request failed, a specialized function pod
+				// still can serve other subsequent requests.
+
+				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)
+
+				fsvc, err := executor.createServiceForFunction(fnSpecializationTimeoutContext, req.function)
 				req.respChan <- &createFuncServiceResponse{
 					funcSvc: fsvc,
 					err:     err,
 				}
-				delete(executor.fsCreateWg, crd.CacheKey(m))
+				delete(executor.fsCreateWg, crd.CacheKey(fnMetadata))
+
+				cancel()
 				wg.Done()
 			}()
 		} else {
 			// There's an existing request for this function, wait for it to finish
 			go func() {
 				executor.logger.Debug("waiting for concurrent request for the same function",
-					zap.Any("function", m))
+					zap.Any("function", fnMetadata))
 				wg.Wait()
 
 				// get the function service from the cache
-				fsvc, err := executor.fsCache.GetByFunction(m)
+				fsvc, err := executor.fsCache.GetByFunction(fnMetadata)
 
 				// fsCache return error when the entry does not exist/expire.
 				// It normally happened if there are multiple requests are
 				// waiting for the same function and executor failed to cre-
 				// ate service for function.
-				err = errors.Wrapf(err, "error getting service for function %v in namespace %v", m.Name, m.Namespace)
+				err = errors.Wrapf(err, "error getting service for function %v in namespace %v", fnMetadata.Name, fnMetadata.Namespace)
 				req.respChan <- &createFuncServiceResponse{
 					funcSvc: fsvc,
 					err:     err,
@@ -138,49 +150,38 @@ func (executor *Executor) serveCreateFuncServices() {
 	}
 }
 
-func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fv1.ExecutorType, error) {
+func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	fn, err := executor.fissionClient.Functions(meta.Namespace).Get(meta.Name)
-	if err != nil {
-		return "", err
-	}
-	return fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, nil
-}
-
-func (executor *Executor) createServiceForFunction(ctx context.Context, meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
 	executor.logger.Debug("no cached function service found, creating one",
-		zap.String("function_name", meta.Name),
+		zap.String("function_name", fn.Metadata.Name),
-		zap.String("function_namespace", meta.Namespace))
+		zap.String("function_namespace", fn.Metadata.Namespace))
 
-	executorType, err := executor.getFunctionExecutorType(meta)
+	executorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType
-	if err != nil {
-		return nil, err
-	}
 
 	var fsvc *fscache.FuncSvc
 	var fsvcErr error
 
 	switch executorType {
 	case fv1.ExecutorTypeNewdeploy:
-		fsvc, fsvcErr = executor.ndm.GetFuncSvc(ctx, meta)
+		fsvc, fsvcErr = executor.ndm.GetFuncSvc(ctx, fn)
 	default:
-		fsvc, fsvcErr = executor.gpm.GetFuncSvc(ctx, meta)
+		fsvc, fsvcErr = executor.gpm.GetFuncSvc(ctx, fn)
 	}
 
 	if fsvcErr != nil {
 		e := "error creating service for function"
 		executor.logger.Error(e,
 			zap.Error(fsvcErr),
-			zap.String("function_name", meta.Name),
+			zap.String("function_name", fn.Metadata.Name),
-			zap.String("function_namespace", meta.Namespace))
+			zap.String("function_namespace", fn.Metadata.Namespace))
-		fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%s] %s", meta.Name, e))
+		fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%s] %s", fn.Metadata.Name, e))
 	} else if fsvc != nil {
-		_, err = executor.fsCache.Add(*fsvc)
+		_, err := executor.fsCache.Add(*fsvc)
 		if err != nil {
 			return nil, err
 		}
 	}
 
-	executor.fsCache.IncreaseColdStarts(meta.Name, string(meta.UID))
+	executor.fsCache.IncreaseColdStarts(fn.Metadata.Name, string(fn.Metadata.UID))
 
 	return fsvc, fsvcErr
 }

pkg/executor/newdeploy/newdeploy.go

@@ -425,7 +425,7 @@ func (deploy *NewDeploy) waitForDeploy(depl *appsv1.Deployment, replicas int32,
 		if err != nil {
 			return nil, err
 		}
-		//TODO check for imagePullerror
+		// TODO check for imagePullerror
 		// use AvailableReplicas here is better than ReadyReplicas
 		// since the pods may not be able to serve network traffic yet.
 		if latestDepl.Status.AvailableReplicas >= replicas {

pkg/executor/newdeploy/newdeploymgr.go

@@ -224,11 +224,7 @@ func (deploy *NewDeploy) getEnvFunctions(m *metav1.ObjectMeta) []fv1.Function {
 	return relatedFunctions
 }
 
-func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
+func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name)
-	if err != nil {
-		return nil, err
-	}
 	return deploy.createFunction(fn, false)
 }
 

pkg/executor/poolmgr/gp.go

@@ -298,7 +298,7 @@ func (gp *GenericPool) getFetcherUrl(podIP string) string {
 // specializePod chooses a pod, copies the required user-defined function to that pod
 // (via fetcher), and calls the function-run container to load it, resulting in a
 // specialized pod.
-func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, metadata *metav1.ObjectMeta) error {
+func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, fn *fv1.Function) error {
 	// for fetcher we don't need to create a service, just talk to the pod directly
 	podIP := pod.Status.PodIP
 	if len(podIP) == 0 {
@@ -306,28 +306,21 @@ func (gp *GenericPool) specializePod(ctx context.Context, pod *apiv1.Pod, metada
 	}
 	// specialize pod with service
 	if gp.useIstio {
-		svc := utils.GetFunctionIstioServiceName(metadata.Name, metadata.Namespace)
+		svc := utils.GetFunctionIstioServiceName(fn.Metadata.Name, fn.Metadata.Namespace)
 		podIP = fmt.Sprintf("%v.%v", svc, gp.namespace)
 	}
 
 	// tell fetcher to get the function.
 	fetcherUrl := gp.getFetcherUrl(podIP)
-	gp.logger.Info("calling fetcher to copy function", zap.String("function", metadata.Name), zap.String("url", fetcherUrl))
+	gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.Metadata.Name), zap.String("url", fetcherUrl))
-
-	fn, err := gp.fissionClient.
-		Functions(metadata.Namespace).
-		Get(metadata.Name)
-	if err != nil {
-		return err
-	}
 
 	specializeReq := gp.fetcherConfig.NewSpecializeRequest(fn, gp.env)
 
-	gp.logger.Info("specializing pod", zap.String("function", metadata.Name))
+	gp.logger.Info("specializing pod", zap.String("function", fn.Metadata.Name))
 
 	// Fetcher will download user function to share volume of pod, and
 	// invoke environment specialize api for pod specialization.
-	err = fetcherClient.MakeClient(gp.logger, fetcherUrl).Specialize(ctx, &specializeReq)
+	err := fetcherClient.MakeClient(gp.logger, fetcherUrl).Specialize(ctx, &specializeReq)
 	if err != nil {
 		return err
 	}
@@ -508,9 +501,9 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.
 	return svc, err
 }
 
-func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
+func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
-	gp.logger.Info("choosing pod from pool", zap.String("function", m.Name))
+	gp.logger.Info("choosing pod from pool", zap.Any("function", fn.Metadata))
-	newLabels := gp.labelsForFunction(m)
+	newLabels := gp.labelsForFunction(&fn.Metadata)
 
 	if gp.useIstio {
 		// Istio only allows accessing pod through k8s service, and requests come to
@@ -536,8 +529,8 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f
 		// and make sure that there is only one pod behind the service
 
 		sel := map[string]string{
-			"functionName": m.Name,
+			"functionName": fn.Metadata.Name,
-			"functionUid":  string(m.UID),
+			"functionUid":  string(fn.Metadata.UID),
 		}
 		podList, err := gp.kubernetesClient.CoreV1().Pods(gp.namespace).List(metav1.ListOptions{
 			LabelSelector: labels.Set(sel).AsSelector().String(),
@@ -558,22 +551,21 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f
 		return nil, err
 	}
 
-	err = gp.specializePod(ctx, pod, m)
+	err = gp.specializePod(ctx, pod, fn)
 	if err != nil {
 		gp.scheduleDeletePod(pod.ObjectMeta.Name)
 		return nil, err
 	}
-	gp.logger.Info("specialized pod", zap.String("pod", pod.ObjectMeta.Name), zap.String("function", m.Name))
+	gp.logger.Info("specialized pod", zap.String("pod", pod.ObjectMeta.Name), zap.Any("function", fn.Metadata))
 
 	var svcHost string
 	if gp.useSvc && !gp.useIstio {
-		svcName := fmt.Sprintf("svc-%v", m.Name)
+		svcName := fmt.Sprintf("svc-%v", fn.Metadata.Name)
-		if len(m.UID) > 0 {
+		if len(fn.Metadata.UID) > 0 {
-			svcName = fmt.Sprintf("%s-%v", svcName, m.UID)
+			svcName = fmt.Sprintf("%s-%v", svcName, fn.Metadata.UID)
 		}
 
-		labels := gp.labelsForFunction(m)
+		svc, err := gp.createSvc(svcName, newLabels)
-		svc, err := gp.createSvc(svcName, labels)
 		if err != nil {
 			gp.scheduleDeletePod(pod.ObjectMeta.Name)
 			return nil, err
@@ -587,7 +579,7 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f
 		// namespace-qualified hostname
 		svcHost = fmt.Sprintf("%v.%v", svcName, gp.namespace)
 	} else if gp.useIstio {
-		svc := utils.GetFunctionIstioServiceName(m.Name, m.Namespace)
+		svc := utils.GetFunctionIstioServiceName(fn.Metadata.Name, fn.Metadata.Namespace)
 		svcHost = fmt.Sprintf("%v.%v:8888", svc, gp.namespace)
 	} else {
 		svcHost = fmt.Sprintf("%v:8888", pod.Status.PodIP)
@@ -596,8 +588,8 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f
 	gp.logger.Info("specialized pod",
 		zap.String("pod", pod.ObjectMeta.Name),
 		zap.String("podNamespace", pod.ObjectMeta.Namespace),
-		zap.String("function", m.Name),
+		zap.String("function", fn.Metadata.Name),
-		zap.String("functionNamespace", m.Namespace),
+		zap.String("functionNamespace", fn.Metadata.Namespace),
 		zap.String("specialization_host", svcHost))
 
 	kubeObjRefs := []apiv1.ObjectReference{
@@ -611,9 +603,10 @@ func (gp *GenericPool) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta) (*f
 		},
 	}
 
+	m := fn.Metadata // only cache necessary part
 	fsvc := &fscache.FuncSvc{
 		Name:              pod.ObjectMeta.Name,
-		Function:          m,
+		Function:          &m,
 		Environment:       gp.env,
 		Address:           svcHost,
 		KubernetesObjects: kubeObjRefs,

pkg/executor/poolmgr/gpm.go

@@ -139,7 +139,7 @@ func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Functio
 		return err
 	}
 
-	gp, err := gpm.GetPool(env)
+	gp, err := gpm.getPool(env)
 	if err != nil {
 		return err
 	}
@@ -228,7 +228,7 @@ func (gpm *GenericPoolManager) service() {
 	}
 }
 
-func (gpm *GenericPoolManager) GetPool(env *fv1.Environment) (*GenericPool, error) {
+func (gpm *GenericPoolManager) getPool(env *fv1.Environment) (*GenericPool, error) {
 	c := make(chan *response)
 	gpm.requestChannel <- &request{
 		requestType:     GET_POOL,
@@ -239,61 +239,58 @@ func (gpm *GenericPoolManager) GetPool(env *fv1.Environment) (*GenericPool, erro
 	return resp.pool, resp.error
 }
 
-func (gpm *GenericPoolManager) CleanupPools(envs []fv1.Environment) {
+func (gpm *GenericPoolManager) cleanupPools(envs []fv1.Environment) {
 	gpm.requestChannel <- &request{
 		requestType: CLEANUP_POOLS,
 		envList:     envs,
 	}
 }
 
-func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, metadata *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
+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", metadata.Name))
+	gpm.logger.Debug("getting environment for function", zap.String("function", fn.Metadata.Name))
-	env, err := gpm.getFunctionEnv(metadata)
+	env, err := gpm.getFunctionEnv(fn)
 	if err != nil {
 		return nil, err
 	}
 
-	pool, err := gpm.GetPool(env)
+	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", metadata.Name))
+	gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name))
-	return pool.GetFuncSvc(ctx, metadata)
+	return pool.getFuncSvc(ctx, fn)
 }
 
-func (gpm *GenericPoolManager) getFunctionEnv(m *metav1.ObjectMeta) (*fv1.Environment, error) {
+func (gpm *GenericPoolManager) getFunctionEnv(fn *fv1.Function) (*fv1.Environment, error) {
 	var env *fv1.Environment
 
 	// Cached ?
-	result, err := gpm.functionEnv.Get(crd.CacheKey(m))
+	// TODO: the cache should be able to search by <env name, fn namespace> instead of function metadata.
+	result, err := gpm.functionEnv.Get(crd.CacheKey(&fn.Metadata))
 	if err == nil {
 		env = result.(*fv1.Environment)
 		return env, nil
 	}
 
-	// Cache miss -- get func from controller
+	// Get env from controller
-	f, err := gpm.fissionClient.Functions(m.Namespace).Get(m.Name)
+	env, err = gpm.fissionClient.Environments(fn.Spec.Environment.Namespace).Get(fn.Spec.Environment.Name)
-	if err != nil {
-		return nil, err
-	}
-
-	// Get env from metadata
-	env, err = gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
 	if err != nil {
 		return nil, err
 	}
 
 	// cache for future lookups
-	gpm.functionEnv.Set(crd.CacheKey(m), env)
+	m := fn.Metadata
+	gpm.functionEnv.Set(crd.CacheKey(&m), env)
 
 	return env, nil
 }
 
 func (gpm *GenericPoolManager) eagerPoolCreator() {
-	pollSleep := time.Duration(2 * time.Second)
+	pollSleep := 2 * time.Second
 	for {
 		// get list of envs from controller
 		envs, err := gpm.fissionClient.Environments(metav1.NamespaceAll).List(metav1.ListOptions{})
@@ -303,7 +300,7 @@ func (gpm *GenericPoolManager) eagerPoolCreator() {
 				time.Sleep(5 * time.Second)
 				continue
 			}
-			gpm.logger.Fatal("failed to get environment list", zap.Error(err))
+			gpm.logger.Error("failed to get environment list", zap.Error(err))
 		}
 
 		// Create pools for all envs.  TODO: we should make this a bit less eager, only
@@ -314,7 +311,7 @@ func (gpm *GenericPoolManager) eagerPoolCreator() {
 			env := envs.Items[i]
 			// Create pool only if poolsize greater than zero
 			if gpm.getEnvPoolsize(&env) > 0 {
-				_, err := gpm.GetPool(&envs.Items[i])
+				_, err := gpm.getPool(&envs.Items[i])
 				if err != nil {
 					gpm.logger.Error("eager-create pool failed", zap.Error(err))
 				}
@@ -322,7 +319,7 @@ func (gpm *GenericPoolManager) eagerPoolCreator() {
 		}
 
 		// Clean up pools whose env was deleted
-		gpm.CleanupPools(envs.Items)
+		gpm.cleanupPools(envs.Items)
 		time.Sleep(pollSleep)
 	}
 }

pkg/router/functionHandler.go

@@ -584,7 +584,6 @@ func (fh *functionHandler) getServiceEntry() (serviceUrl *url.URL, serviceUrlFro
 	}
 
 	// cache miss or nil entry in cache
-
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()