Fix executor tries to create a new deployment when a function is updated (#524)

Newdeploy manager now checks the existence of function service cache by function UID before trying to create a new deployment. And return the cached fsvc directly if the cache exists.

executor/cleanup.go

@@ -117,6 +117,7 @@ func idleObjectReaper(kubeClient *kubernetes.Clientset,
 			for _, fsvc := range funcSvcs {
 
 				fn, err := fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name)
+
 				if err != nil {
 					log.Printf("Error getting function: %v", fsvc.Function.Name)
 					continue

executor/fscache/functionServiceCache.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/pkg/api"
 
 	"github.com/fission/fission"
@@ -56,8 +57,9 @@ type (
 	}
 
 	FunctionServiceCache struct {
-		byFunction *cache.Cache // function-key -> funcSvc  : map[string]*funcSvc
+		byFunction    *cache.Cache // function-key -> funcSvc  : map[string]*funcSvc
-		byAddress  *cache.Cache // address      -> function : map[string]metav1.ObjectMeta
+		byAddress     *cache.Cache // address      -> function : map[string]metav1.ObjectMeta
+		byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta
 
 		requestChannel chan *fscRequest
 	}
@@ -76,10 +78,25 @@ type (
 	}
 )
 
+func IsNotFoundError(err error) bool {
+	if fe, ok := err.(fission.Error); ok {
+		return fe.Code == fission.ErrorNotFound
+	}
+	return false
+}
+
+func IsNameExistError(err error) bool {
+	if fe, ok := err.(fission.Error); ok {
+		return fe.Code == fission.ErrorNameExists
+	}
+	return false
+}
+
 func MakeFunctionServiceCache() *FunctionServiceCache {
 	fsc := &FunctionServiceCache{
 		byFunction:     cache.MakeCache(0, 0),
 		byAddress:      cache.MakeCache(0, 0),
+		byFunctionUID:  cache.MakeCache(0, 0),
 		requestChannel: make(chan *fscRequest),
 	}
 	go fsc.service()
@@ -136,6 +153,27 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc,
 	return &fsvcCopy, nil
 }
 
+func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) {
+	mI, err := fsc.byFunctionUID.Get(uid)
+	if err != nil {
+		return nil, err
+	}
+
+	m := mI.(metav1.ObjectMeta)
+
+	fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
+	if err != nil {
+		return nil, err
+	}
+
+	// update atime
+	fsvc := fsvcI.(*FuncSvc)
+	fsvc.Atime = time.Now()
+
+	fsvcCopy := *fsvc
+	return &fsvcCopy, nil
+}
+
 func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
 	err, existing := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)
 	if err != nil {
@@ -158,16 +196,28 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
 	// because of multiple-specialization. See issue #331.
 	err, _ = fsc.byAddress.Set(fsvc.Address, *fsvc.Function)
 	if err != nil {
-		if fe, ok := err.(fission.Error); ok {
+		if IsNameExistError(err) {
-			if fe.Code == fission.ErrorNameExists {
+			err = nil
-				err = nil
-			}
 		}
 		if err != nil {
 			log.Printf("error caching fsvc: %v", err)
 		}
 		return nil, err
 	}
+
+	// Add to byFunctionUID cache. Ignore NameExists errors
+	// because of multiple-specialization. See issue #331.
+	err, _ = fsc.byFunctionUID.Set(fsvc.Function.UID, *fsvc.Function)
+	if err != nil {
+		if IsNameExistError(err) {
+			err = nil
+		}
+		if err != nil {
+			log.Printf("error caching fsvc by function uid: %v", err)
+		}
+		return nil, err
+	}
+
 	return nil, nil
 }
 
@@ -204,6 +254,7 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration)
 
 	fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
 	fsc.byAddress.Delete(fsvc.Address)
+	fsc.byFunctionUID.Delete(fsvc.Function.UID)
 
 	return true, nil
 }

executor/fscache/functionServiceCache_test.go

@@ -70,6 +70,11 @@ func TestFunctionServiceCache(t *testing.T) {
 		fsc.Log()
 		log.Panicf("Failed to get fsvc: %v", err)
 	}
+	f, err = fsc.GetByFunctionUID(fsvc.Function.UID)
+	if err != nil {
+		fsc.Log()
+		log.Panicf("Failed to get fsvc by function uid: %v", err)
+	}
 	fsvc.Atime = f.Atime
 	fsvc.Ctime = f.Ctime
 	if f.Address != fsvc.Address {
@@ -98,4 +103,10 @@ func TestFunctionServiceCache(t *testing.T) {
 		fsc.Log()
 		log.Panicf("found fsvc while expecting empty cache: %v", err)
 	}
+
+	_, err = fsc.GetByFunctionUID(fsvc.Function.UID)
+	if err == nil {
+		fsc.Log()
+		log.Panicf("found fsvc by function uid while expecting empty cache: %v", err)
+	}
 }

executor/newdeploy/newdeploy.go

@@ -58,7 +58,10 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro
 	}
 
 	existingDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(deployName, metav1.GetOptions{})
-	if err == nil && existingDepl.Status.ReadyReplicas >= replicas {
+	if err == nil {
+		if existingDepl.Status.ReadyReplicas < replicas {
+			existingDepl, err = deploy.waitForDeploy(existingDepl, replicas)
+		}
 		return existingDepl, err
 	}
 
@@ -75,18 +78,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *crd.Function, env *crd.Enviro
 			return nil, err
 		}
 
-		for i := 0; i < 120; i++ {
+		return deploy.waitForDeploy(depl, replicas)
-			latestDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(depl.Name, metav1.GetOptions{})
-			if err != nil {
-				return nil, err
-			}
-			//TODO check for imagePullerror
-			if latestDepl.Status.ReadyReplicas == replicas {
-				return latestDepl, err
-			}
-			time.Sleep(time.Second)
-		}
-		return nil, errors.New("failed to create deployment within timeout window")
 	}
 
 	return nil, err
@@ -459,3 +451,18 @@ func (deploy *NewDeploy) deleteSvc(ns string, name string) error {
 	}
 	return nil
 }
+
+func (deploy *NewDeploy) waitForDeploy(depl *v1beta1.Deployment, replicas int32) (*v1beta1.Deployment, error) {
+	for i := 0; i < 120; i++ {
+		latestDepl, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deploy.namespace).Get(depl.Name, metav1.GetOptions{})
+		if err != nil {
+			return nil, err
+		}
+		//TODO check for imagePullerror
+		if latestDepl.Status.ReadyReplicas >= replicas {
+			return latestDepl, err
+		}
+		time.Sleep(time.Second)
+	}
+	return nil, errors.New("failed to create deployment within timeout window")
+}

executor/newdeploy/newdeploymgr.go

@@ -77,6 +77,7 @@ type (
 
 const (
 	FnCreate requestType = iota
+	FnUpdate
 	FnDelete
 )
 
@@ -191,11 +192,27 @@ func (deploy *NewDeploy) GetFuncSvc(metadata *metav1.ObjectMeta) (*fscache.FuncS
 	if err != nil {
 		return nil, err
 	}
+
+	fsvc, err := deploy.fsCache.GetByFunctionUID(metadata.UID)
+
+	// If the function service cache exists, means
+	// the kubeObjects of function are created before.
+	// In this case, return cached fsvc.
+	if err == nil {
+		return fsvc, nil
+	}
+
+	if !fscache.IsNotFoundError(err) {
+		log.Printf("error getting function service by uid: %v", err)
+		return nil, err
+	}
+
 	deploy.requestChannel <- &fnRequest{
 		fn:              fn,
 		reqType:         FnCreate,
 		responseChannel: c,
 	}
+
 	resp := <-c
 	if resp.error != nil {
 		return nil, resp.error
@@ -224,6 +241,19 @@ func (deploy *NewDeploy) createFunction(fn *crd.Function) {
 	}
 }
 
+func (deploy *NewDeploy) updateFunction(fn *crd.Function) {
+	c := make(chan *fnResponse)
+	deploy.requestChannel <- &fnRequest{
+		fn:              fn,
+		reqType:         FnUpdate,
+		responseChannel: c,
+	}
+	resp := <-c
+	if resp.error != nil {
+		log.Printf("Error eager updating function: %v", resp.error)
+	}
+}
+
 func (deploy *NewDeploy) deleteFunction(fn *crd.Function) {
 	if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
 		c := make(chan *fnResponse)
@@ -336,7 +366,7 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
 		return
 	}
 
-	changed := false
+	deployChanged := false
 
 	if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy {
 
@@ -366,59 +396,63 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
 			return
 		}
 
+		hpaChanged := false
+
 		if newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale {
 			replicas := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MinScale)
 			hpa.Spec.MinReplicas = &replicas
-			changed = true // Will start deployment update
+			deployChanged = true
+			hpaChanged = true
 		}
 
 		if newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale != oldFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale {
 			hpa.Spec.MaxReplicas = int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.MaxScale)
+			hpaChanged = true
 		}
 
 		if newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent != oldFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent {
 			targetCpupercent := int32(newFn.Spec.InvokeStrategy.ExecutionStrategy.TargetCPUPercent)
 			hpa.Spec.TargetCPUUtilizationPercentage = &targetCpupercent
+			hpaChanged = true
 		}
 
-		err = deploy.updateHpa(hpa)
+		if hpaChanged {
-		if err != nil {
+			err := deploy.updateHpa(hpa)
-			updateStatus(oldFn, err, "error updating HPA while updating function")
+			if err != nil {
-			return
+				updateStatus(oldFn, err, "error updating HPA while updating function")
+				return
+			}
 		}
 	}
 
-	if oldFn.Spec.Environment != newFn.Spec.Environment {
+	if oldFn.Spec.Environment != newFn.Spec.Environment ||
-		changed = true
+		oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef {
-	}
+		deployChanged = true
-
-	if oldFn.Spec.Package.PackageRef != newFn.Spec.Package.PackageRef {
-		changed = true
 	}
 
 	// If length of slice has changed then no need to check individual elements
 	if len(oldFn.Spec.Secrets) != len(newFn.Spec.Secrets) {
-		changed = true
+		deployChanged = true
 	} else {
 		for i, newSecret := range newFn.Spec.Secrets {
 			if newSecret != oldFn.Spec.Secrets[i] {
-				changed = true
+				deployChanged = true
 				break
 			}
 		}
 	}
 	if len(oldFn.Spec.ConfigMaps) != len(newFn.Spec.ConfigMaps) {
-		changed = true
+		deployChanged = true
 	} else {
 		for i, newConfig := range newFn.Spec.ConfigMaps {
 			if newConfig != oldFn.Spec.ConfigMaps[i] {
-				changed = true
+				deployChanged = true
 				break
 			}
 		}
 	}
 
-	if changed == true {
+	if deployChanged == true {
 		env, err := deploy.fissionClient.Environments(newFn.Spec.Environment.Namespace).
 			Get(newFn.Spec.Environment.Name)
 		if err != nil {
@@ -438,7 +472,6 @@ func (deploy *NewDeploy) fnUpdate(oldFn *crd.Function, newFn *crd.Function) {
 			updateStatus(oldFn, err, "failed to update deployment while updating function")
 			return
 		}
-		return
 	}
 }
 
@@ -501,6 +534,20 @@ func (deploy *NewDeploy) getDeployLabels(fn *crd.Function, env *crd.Environment)
 	}
 }
 
+// updateKubeObjRefRV update the resource version of kubeObjectRef with
+// given kind and return error if failed to find the reference.
+func (deploy *NewDeploy) updateKubeObjRefRV(fsvc *fscache.FuncSvc, objKind string, rv string) error {
+	kubeObjs := fsvc.KubernetesObjects
+	for i, obj := range kubeObjs {
+		if obj.Kind == objKind {
+			kubeObjs[i].ResourceVersion = rv
+			return nil
+		}
+	}
+	fsvc.KubernetesObjects = kubeObjs
+	return errors.New(fmt.Sprintf("error finding kubernetes object reference with kind: %v", objKind))
+}
+
 // updateStatus is a function which updates status of update.
 // Current implementation only logs messages, in future it will update function status
 func updateStatus(fn *crd.Function, err error, message string) {

test/test_utils.sh

@@ -176,7 +176,7 @@ helm_install_fission() {
     helm list -q|xargs -I@ bash -c "helm_uninstall_fission @"
 
     # deleting ns does take a while after command is issued
-    while `kubectl get ns| grep "fission-builder"`
+    while kubectl get ns| grep "fission-builder"
     do
         sleep 5
     done

test/tests/test_backend_nd_update.sh

@@ -56,12 +56,19 @@ update_fn() {
 }
 
 test_fn() {
-    log "Doing an HTTP GET on the function's route"
+    echo "Doing an HTTP GET on the function's route"
-    response0=$(curl http://$FISSION_ROUTER/$1)
+    echo "Checking for valid response"
 
-    log "Checking for valid response"
+    while true; do
-    echo $response0 | grep -i $2
+      response0=$(curl http://$FISSION_ROUTER/$1)
+      echo $response0 | grep -i $2
+      if [[ $? -eq 0 ]]; then
+        break
+      fi
+      sleep 1
+    done
 }
+export -f test_fn
 
 # This test only tests one path of execution: updating package and checking results of function
 # There might be potential future tests where one can test changes in:
@@ -81,10 +88,10 @@ main() {
     create_env $env
     create_fn $fn_name $env
     create_route $fn_name
-    test_fn $fn_name "world"
+    timeout 60 bash -c "test_fn $fn_name 'world'"
     update_archive
     update_fn $fn_name $env
-    test_fn $fn_name "fission"
+    timeout 60 bash -c "test_fn $fn_name 'fission'"
     log "Update function for new deployment executor passed"
 }