Function update after change in env (#1116)

If an environment image changes then the new deployment function also should be updated with the newer image

executor/newdeploy/newdeploymgr.go

@@ -69,6 +69,9 @@ type (
 		funcStore      k8sCache.Store
 		funcController k8sCache.Controller
 
+		envStore      k8sCache.Store
+		envController k8sCache.Controller
+
 		idlePodReapTime time.Duration
 	}
 )
@@ -128,6 +131,10 @@ func MakeNewDeploy(
 		fnStore, fnController := nd.initFuncController()
 		nd.funcStore = fnStore
 		nd.funcController = fnController
+
+		envStore, envController := nd.initEnvController()
+		nd.envStore = envStore
+		nd.envController = envController
 	}
 
 	return nd
@@ -136,6 +143,7 @@ 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()
 }
 
@@ -176,6 +184,49 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll
 	return store, controller
 }
 
+func (deploy *NewDeploy) initEnvController() (k8sCache.Store, k8sCache.Controller) {
+	resyncPeriod := 30 * time.Second
+	listWatch := k8sCache.NewListWatchFromClient(deploy.crdClient, "environments", metav1.NamespaceAll, fields.Everything())
+	store, controller := k8sCache.NewInformer(listWatch, &crd.Environment{}, resyncPeriod, k8sCache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+			newEnv := newObj.(*crd.Environment)
+			oldEnv := oldObj.(*crd.Environment)
+			// Currently only an image update in environment calls for function's deployment recreation. In future there might be more attributes which would want to do it
+			if oldEnv.Spec.Runtime.Image != newEnv.Spec.Runtime.Image {
+				deploy.logger.Info("Updating all function of the environment that changed, old env:", zap.Any("environment", oldEnv))
+				funcs := deploy.getEnvFunctions(&newEnv.Metadata)
+				for _, f := range funcs {
+					function, err := deploy.fissionClient.Functions(f.Metadata.Namespace).Get(f.Metadata.Name)
+					if err != nil {
+						deploy.logger.Error("Error getting function", zap.Error(err), zap.Any("function", function))
+					}
+					err = deploy.updateFuncDeployment(function, newEnv)
+					if err != nil {
+						deploy.logger.Error("Error updating function", zap.Error(err), zap.Any("function", function))
+					}
+				}
+			}
+		},
+	})
+	return store, controller
+}
+
+func (deploy *NewDeploy) getEnvFunctions(m *metav1.ObjectMeta) []crd.Function {
+	funcList, err := deploy.fissionClient.Functions(m.Namespace).List(metav1.ListOptions{})
+	if err != nil {
+		deploy.logger.Error("Error getting functions for env", zap.Error(err), zap.Any("environment", m))
+	}
+	relatedFunctions := make([]crd.Function, 0)
+	for _, f := range funcList.Items {
+		if (f.Spec.Environment.Name == m.Name) && (f.Spec.Environment.Namespace == m.Namespace) {
+			relatedFunctions = append(relatedFunctions, f)
+		}
+	}
+	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 {
@@ -345,13 +396,6 @@ func (deploy *NewDeploy) updateFunction(oldFn *crd.Function, newFn *crd.Function
 		return err
 	}
 
-	fsvc, err := deploy.fsCache.GetByFunctionUID(newFn.Metadata.UID)
-	if err != nil {
-		err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", oldFn)
-		return err
-	}
-
-	fnObjName := fsvc.Name
 	deployChanged := false
 
 	if oldFn.Spec.InvokeStrategy != newFn.Spec.InvokeStrategy {
@@ -363,7 +407,13 @@ func (deploy *NewDeploy) updateFunction(oldFn *crd.Function, newFn *crd.Function
 			ns = newFn.Metadata.Namespace
 		}
 
-		hpa, err := deploy.getHpa(ns, fnObjName)
+		fsvc, err := deploy.fsCache.GetByFunctionUID(newFn.Metadata.UID)
+		if err != nil {
+			err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", oldFn)
+			return err
+		}
+
+		hpa, err := deploy.getHpa(ns, fsvc.Name)
 		if err != nil {
 			deploy.updateStatus(oldFn, err, "error getting HPA while updating function")
 			return err
@@ -432,27 +482,41 @@ func (deploy *NewDeploy) updateFunction(oldFn *crd.Function, newFn *crd.Function
 			deploy.updateStatus(oldFn, err, "failed to get environment while updating function")
 			return err
 		}
+		return deploy.updateFuncDeployment(newFn, env)
+	}
 
-		deployLabels := deploy.getDeployLabels(oldFn, env)
+	return nil
-		deploy.logger.Info("updating deployment due to function update", zap.String("deployment", fnObjName), zap.String("function", newFn.Metadata.Name))
+}
-		newDeployment, err := deploy.getDeploymentSpec(newFn, env, fnObjName, deployLabels)
-		if err != nil {
-			deploy.updateStatus(oldFn, err, "failed to get new deployment spec while updating function")
-			return err
-		}
 
-		// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+func (deploy *NewDeploy) updateFuncDeployment(fn *crd.Function, env *crd.Environment) error {
-		// deployment of the function in fission-function ns
-		ns := deploy.namespace
-		if newFn.Metadata.Namespace != metav1.NamespaceDefault {
-			ns = newFn.Metadata.Namespace
-		}
 
-		err = deploy.updateDeployment(newDeployment, ns)
+	fsvc, err := deploy.fsCache.GetByFunctionUID(fn.Metadata.UID)
-		if err != nil {
+	if err != nil {
-			deploy.updateStatus(oldFn, err, "failed to update deployment while updating function")
+		err = errors.Wrapf(err, "error updating function due to unable to find function service cache: %v", fn)
-			return err
+		return err
-		}
+	}
+	fnObjName := fsvc.Name
+
+	deployLabels := deploy.getDeployLabels(fn, env)
+	deploy.logger.Info("updating deployment due to function/environment update", zap.String("deployment", fnObjName), zap.Any("function", fn.Metadata.Name))
+
+	newDeployment, err := deploy.getDeploymentSpec(fn, env, fnObjName, deployLabels)
+	if err != nil {
+		deploy.updateStatus(fn, err, "failed to get new deployment spec while updating function")
+		return err
+	}
+
+	// to support backward compatibility, if the function was created in default ns, we fall back to creating the
+	// deployment of the function in fission-function ns
+	ns := deploy.namespace
+	if fn.Metadata.Namespace != metav1.NamespaceDefault {
+		ns = fn.Metadata.Namespace
+	}
+
+	err = deploy.updateDeployment(newDeployment, ns)
+	if err != nil {
+		deploy.updateStatus(fn, err, "failed to update deployment while updating function")
+		return err
 	}
 
 	return nil