From b8311d43ce13d228410932a7be0c865c21ddfa8d Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Thu, 10 Oct 2019 23:32:36 +0800 Subject: [PATCH] Fix newdeploy failed to find serviceEntry in cache (#1349) When a function with executor type newdeploy got created, Newdeploy is expected to create deployment/HPA/service for it and insert serviceEntry to the cache for later use. Once clients call the function, newdeploy returns the serviceEntry to the router. However, the log shows that the newdeploy was unable to find the entry and prints "Resource not found - key 'xxx' not found". The root cause is that the informer controller instead of processing items in parallel, it dispatches XXFunc to process items one by one. So if there is any problem during the creation of the kubernetes resource, it takes a longer time to process the next item and hence the serviceEntry was not inserted before clients call the function. This PR lets the newdeploy to process items in extra goroutines instead of blocking the process loop. It's a workaround to solve the problem above, we should consider using workqueue to solve it in the future. --- pkg/controller/api.go | 4 +- pkg/executor/executor.go | 7 ++-- pkg/executor/newdeploy/newdeploymgr.go | 56 +++++++++++++++----------- 3 files changed, 37 insertions(+), 30 deletions(-) diff --git a/pkg/controller/api.go b/pkg/controller/api.go index 97cefab1..1ccf7407 100644 --- a/pkg/controller/api.go +++ b/pkg/controller/api.go @@ -20,7 +20,6 @@ import ( "fmt" "net/http" "os" - "runtime/debug" "strings" "github.com/gorilla/mux" @@ -111,11 +110,10 @@ func (api *API) respondWithSuccess(w http.ResponseWriter, resp []byte) { } func (api *API) respondWithError(w http.ResponseWriter, err error) { - debug.PrintStack() - // this error type comes with an HTTP code, so just use that se, ok := err.(*kerrors.StatusError) if ok { + api.logger.Error(err.Error(), zap.Int32("code", se.ErrStatus.Code)) http.Error(w, string(se.ErrStatus.Reason), int(se.ErrStatus.Code)) return } diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 5d247136..34571dab 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -207,6 +207,9 @@ func serveMetric(logger *zap.Logger) { // deploymgr and potential future executor types func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNamespace string, port int) error { fissionClient, kubernetesClient, _, err := crd.MakeFissionClient() + if err != nil { + return errors.Wrap(err, "failed to get kubernetes client") + } err = fissionClient.WaitForCRDs() if err != nil { @@ -219,10 +222,6 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames } restClient := fissionClient.GetCrdClient() - if err != nil { - return errors.Wrap(err, "failed to get kubernetes client") - } - fsCache := fscache.MakeFunctionServiceCache(logger) poolID := strings.ToLower(uniuri.NewLen(8)) diff --git a/pkg/executor/newdeploy/newdeploymgr.go b/pkg/executor/newdeploy/newdeploymgr.go index fa0c0c62..71e7fbb9 100644 --- a/pkg/executor/newdeploy/newdeploymgr.go +++ b/pkg/executor/newdeploy/newdeploymgr.go @@ -127,7 +127,6 @@ func MakeNewDeploy( } func (deploy *NewDeploy) Run(ctx context.Context) { - //go deploy.service() go deploy.funcController.Run(ctx.Done()) go deploy.envController.Run(ctx.Done()) go deploy.idleObjectReaper() @@ -138,33 +137,44 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll listWatch := k8sCache.NewListWatchFromClient(deploy.crdClient, "functions", metav1.NamespaceAll, fields.Everything()) store, controller := k8sCache.NewInformer(listWatch, &fv1.Function{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{ AddFunc: func(obj interface{}) { - fn := obj.(*fv1.Function) - _, err := deploy.createFunction(fn, true) - if err != nil { - deploy.logger.Error("error eager creating function", - zap.Error(err), - zap.Any("function", fn)) - } + // TODO: A workaround to process items in parallel. We should use workqueue ("k8s.io/client-go/util/workqueue") + // and worker pattern to process items instead of moving process to another goroutine. + // example: https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/job/job_controller.go + go func() { + fn := obj.(*fv1.Function) + deploy.logger.Debug("create deployment for function", zap.Any("fn", fn.Metadata), zap.Any("fnspec", fn.Spec)) + _, err := deploy.createFunction(fn, true) + if err != nil { + deploy.logger.Error("error eager creating function", + zap.Error(err), + zap.Any("function", fn)) + } + deploy.logger.Debug("end create deployment for function", zap.Any("fn", fn.Metadata), zap.Any("fnspec", fn.Spec)) + }() }, DeleteFunc: func(obj interface{}) { fn := obj.(*fv1.Function) - err := deploy.deleteFunction(fn) - if err != nil { - deploy.logger.Error("error deleting function", - zap.Error(err), - zap.Any("function", fn)) - } + go func() { + err := deploy.deleteFunction(fn) + if err != nil { + deploy.logger.Error("error deleting function", + zap.Error(err), + zap.Any("function", fn)) + } + }() }, UpdateFunc: func(oldObj interface{}, newObj interface{}) { oldFn := oldObj.(*fv1.Function) newFn := newObj.(*fv1.Function) - err := deploy.updateFunction(oldFn, newFn) - if err != nil { - deploy.logger.Error("error updating function", - zap.Error(err), - zap.Any("old_function", oldFn), - zap.Any("new_function", newFn)) - } + go func() { + err := deploy.updateFunction(oldFn, newFn) + if err != nil { + deploy.logger.Error("error updating function", + zap.Error(err), + zap.Any("old_function", oldFn), + zap.Any("new_function", newFn)) + } + }() }, }) return store, controller @@ -275,7 +285,7 @@ func (deploy *NewDeploy) createFunction(fn *fv1.Function, firstcreate bool) (*fs }) if err != nil { - e := "error updating service address entry for function" + e := "error creating k8s resources for function" deploy.logger.Error(e, zap.Error(err), zap.String("function_name", fn.Metadata.Name), @@ -315,7 +325,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache. // retrieve back the previous obj name for later use. fsvc, err := deploy.fsCache.GetByFunctionUID(fn.Metadata.UID) if err != nil { - return nil, err + return nil, errors.Wrap(err, "error getting existed function service cache") } objName = fsvc.Name }