From c7d0c09f010afe52344c9d7aa7a5d044d78b6f3c Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Sun, 10 Nov 2019 00:01:19 +0800 Subject: [PATCH] Poolmanager wait for function specialization timeout when specializing a pod (#1392) --- pkg/executor/api.go | 29 +++++++---- pkg/executor/executor.go | 67 +++++++++++++------------- pkg/executor/newdeploy/newdeploy.go | 2 +- pkg/executor/newdeploy/newdeploymgr.go | 6 +-- pkg/executor/poolmgr/gp.go | 49 ++++++++----------- pkg/executor/poolmgr/gpm.go | 45 ++++++++--------- pkg/router/functionHandler.go | 1 - 7 files changed, 98 insertions(+), 101 deletions(-) diff --git a/pkg/executor/api.go b/pkg/executor/api.go index a2d0f488..ff486644 100644 --- a/pkg/executor/api.go +++ b/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_namespace", m.Namespace)) + zap.String("function_name", fn.Metadata.Name), + 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_namespace", m.Namespace), + zap.String("function_name", fn.Metadata.Name), + 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, - funcMeta: m, + function: fn, respChan: respChan, } resp := <-respChan diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 34571dab..3306f2d3 100644 --- a/pkg/executor/executor.go +++ b/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 - funcMeta *metav1.ObjectMeta + function *fv1.Function 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) { - 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) { +func (executor *Executor) createServiceForFunction(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { executor.logger.Debug("no cached function service found, creating one", - zap.String("function_name", meta.Name), - zap.String("function_namespace", meta.Namespace)) + zap.String("function_name", fn.Metadata.Name), + zap.String("function_namespace", fn.Metadata.Namespace)) - executorType, err := executor.getFunctionExecutorType(meta) - if err != nil { - return nil, err - } + executorType := fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType 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_namespace", meta.Namespace)) - fsvcErr = errors.Wrap(fsvcErr, fmt.Sprintf("[%s] %s", meta.Name, e)) + 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) + _, 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 } diff --git a/pkg/executor/newdeploy/newdeploy.go b/pkg/executor/newdeploy/newdeploy.go index 8ab24d46..1271549e 100644 --- a/pkg/executor/newdeploy/newdeploy.go +++ b/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 { diff --git a/pkg/executor/newdeploy/newdeploymgr.go b/pkg/executor/newdeploy/newdeploymgr.go index 5583b2e7..5d00ed31 100644 --- a/pkg/executor/newdeploy/newdeploymgr.go +++ b/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) { - fn, err := deploy.fissionClient.Functions(metadata.Namespace).Get(metadata.Name) - if err != nil { - return nil, err - } +func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { return deploy.createFunction(fn, false) } diff --git a/pkg/executor/poolmgr/gp.go b/pkg/executor/poolmgr/gp.go index f90657d8..b4d64546 100644 --- a/pkg/executor/poolmgr/gp.go +++ b/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)) - - fn, err := gp.fissionClient. - Functions(metadata.Namespace). - Get(metadata.Name) - if err != nil { - return err - } + gp.logger.Info("calling fetcher to copy function", zap.String("function", fn.Metadata.Name), zap.String("url", fetcherUrl)) 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) { - gp.logger.Info("choosing pod from pool", zap.String("function", m.Name)) - newLabels := gp.labelsForFunction(m) +func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { + gp.logger.Info("choosing pod from pool", zap.Any("function", fn.Metadata)) + 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, - "functionUid": string(m.UID), + "functionName": fn.Metadata.Name, + "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) - if len(m.UID) > 0 { - svcName = fmt.Sprintf("%s-%v", svcName, m.UID) + svcName := fmt.Sprintf("svc-%v", fn.Metadata.Name) + if len(fn.Metadata.UID) > 0 { + svcName = fmt.Sprintf("%s-%v", svcName, fn.Metadata.UID) } - labels := gp.labelsForFunction(m) - svc, err := gp.createSvc(svcName, labels) + svc, err := gp.createSvc(svcName, newLabels) 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("functionNamespace", m.Namespace), + zap.String("function", fn.Metadata.Name), + 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, diff --git a/pkg/executor/poolmgr/gpm.go b/pkg/executor/poolmgr/gpm.go index 11cbdb6c..2501130d 100644 --- a/pkg/executor/poolmgr/gpm.go +++ b/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)) - env, err := gpm.getFunctionEnv(metadata) + 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) + 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)) - return pool.GetFuncSvc(ctx, metadata) + gpm.logger.Debug("getting function service from pool", zap.String("function", fn.Metadata.Name)) + 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 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 - f, err := gpm.fissionClient.Functions(m.Namespace).Get(m.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) + // Get env from controller + env, err = gpm.fissionClient.Environments(fn.Spec.Environment.Namespace).Get(fn.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) } } diff --git a/pkg/router/functionHandler.go b/pkg/router/functionHandler.go index 1e05d9cb..0d16bb48 100644 --- a/pkg/router/functionHandler.go +++ b/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()