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

@@ -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
 	}

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))

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
 	}