Function Update if config/secret changes (#1224)

* Config and secret change to invoke function update
* Added recycle function for pool manager as well - where it recycles the specialized pods
* Switched to rolling update of pods and env variable based change instead of deleting pods in new deploy executor

pkg/apis/fission.io/v1/const.go

@@ -21,6 +21,11 @@ const (
 	POOLMGR_INSTANCEID_LABEL  string = "poolmgrInstanceId"
 )
 
+const (
+	//LastUpdateTimestamp env variable is used for updating configmaps and secrets in pods
+	LastUpdateTimestamp string = "LASTUPDATE_TIMESTAMP"
+)
+
 const (
 	ChecksumTypeSHA256 ChecksumType = "sha256"
 )

pkg/executor/api.go

@@ -137,6 +137,7 @@ func (executor *Executor) Serve(port int) {
 	defer cancel()
 	executor.ndm.Run(ctx)
 	executor.gpm.Run(ctx)
+	executor.cms.Run(ctx)
 
 	r := mux.NewRouter()
 	r.HandleFunc("/v2/getServiceForFunction", executor.getServiceForFunctionApi).Methods("POST")
@@ -144,6 +145,7 @@ func (executor *Executor) Serve(port int) {
 	r.HandleFunc("/healthz", executor.healthHandler).Methods("GET")
 
 	address := fmt.Sprintf(":%v", port)
+
 	err := http.ListenAndServe(address, &ochttp.Handler{
 		Handler: r,
 	})

pkg/executor/cms/cmscontroller.go

@@ -0,0 +1,182 @@
+/*
+Copyright 2016 The Fission Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cms
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+	apiv1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
+	"github.com/fission/fission/pkg/crd"
+	nd "github.com/fission/fission/pkg/executor/newdeploy"
+	gpm "github.com/fission/fission/pkg/executor/poolmgr"
+)
+
+type (
+	ConfigSecretController struct {
+		logger *zap.Logger
+
+		configmapController cache.Controller
+		secretController    cache.Controller
+
+		fissionClient *crd.FissionClient
+	}
+)
+
+//MakeConfigSecretController makes a controller for configmaps and secrets which changes related functions
+func MakeConfigSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) *ConfigSecretController {
+	logger.Debug("Creating ConfigMap & Secret Controller")
+	_, cmcontroller := initConfigmapController(logger, fissionClient, kubernetesClient, ndm, gpm)
+	_, scontroller := initSecretController(logger, fissionClient, kubernetesClient, ndm, gpm)
+	cmsController := &ConfigSecretController{
+		logger:              logger,
+		configmapController: cmcontroller,
+		secretController:    scontroller,
+		fissionClient:       fissionClient,
+	}
+	return cmsController
+}
+
+//Run runs the controllers for configmaps and secrets
+func (csController *ConfigSecretController) Run(ctx context.Context) {
+	go csController.configmapController.Run(ctx.Done())
+	go csController.secretController.Run(ctx.Done())
+}
+
+func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) {
+	resyncPeriod := 30 * time.Second
+	listWatch := cache.NewListWatchFromClient(kubernetesClient.Core().RESTClient(), "configmaps", metav1.NamespaceAll, fields.Everything())
+	store, controller := cache.NewInformer(listWatch, &apiv1.ConfigMap{}, resyncPeriod, cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+			oldCm := oldObj.(*apiv1.ConfigMap)
+			newCm := newObj.(*apiv1.ConfigMap)
+			if oldCm.ObjectMeta.ResourceVersion != newCm.ObjectMeta.ResourceVersion {
+				if newCm.ObjectMeta.Namespace != "kube-system" {
+					logger.Debug("Configmap changed",
+						zap.String("configmap_name", newCm.ObjectMeta.Name),
+						zap.String("configmap_namespace", newCm.ObjectMeta.Namespace))
+
+				}
+
+				funcs, err := getConfigmapRelatedFuncs(logger, &newCm.ObjectMeta, fissionClient)
+				if err != nil {
+					logger.Error("Failed to get functions related to secret", zap.String("secret_name", newCm.ObjectMeta.Name), zap.String("secret_namespace", newCm.ObjectMeta.Namespace))
+				}
+				recyclePods(logger, funcs, ndm, gpm)
+			}
+
+		},
+	})
+	return store, controller
+}
+
+func getConfigmapRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
+	funcList, err := fissionClient.Functions(metav1.NamespaceAll).List(metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	// In future a cache that populates at start and is updated on changes might be better solution
+	relatedFunctions := make([]fv1.Function, 0)
+	for _, f := range funcList.Items {
+		for _, cm := range f.Spec.ConfigMaps {
+			if (cm.Name == m.Name) && (cm.Namespace == m.Namespace) {
+				relatedFunctions = append(relatedFunctions, f)
+				break
+			}
+		}
+	}
+	return relatedFunctions, nil
+}
+
+func initSecretController(logger *zap.Logger, fissionClient *crd.FissionClient,
+	kubernetesClient *kubernetes.Clientset, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) (cache.Store, cache.Controller) {
+	resyncPeriod := 30 * time.Second
+	listWatch := cache.NewListWatchFromClient(kubernetesClient.Core().RESTClient(), "secrets", metav1.NamespaceAll, fields.Everything())
+	store, controller := cache.NewInformer(listWatch, &apiv1.Secret{}, resyncPeriod, cache.ResourceEventHandlerFuncs{
+		AddFunc:    func(obj interface{}) {},
+		DeleteFunc: func(obj interface{}) {},
+		UpdateFunc: func(oldObj interface{}, newObj interface{}) {
+			oldS := oldObj.(*apiv1.Secret)
+			newS := newObj.(*apiv1.Secret)
+			if oldS.ObjectMeta.ResourceVersion != newS.ObjectMeta.ResourceVersion {
+				if newS.ObjectMeta.Namespace != "kube-system" {
+					logger.Debug("Secret changed",
+						zap.String("configmap_name", newS.ObjectMeta.Name),
+						zap.String("configmap_namespace", newS.ObjectMeta.Namespace))
+
+				}
+
+				funcs, err := getSecretRelatedFuncs(logger, &newS.ObjectMeta, fissionClient)
+				if err != nil {
+					logger.Error("Failed to get functions related to secret", zap.String("secret_name", newS.ObjectMeta.Name), zap.String("secret_namespace", newS.ObjectMeta.Namespace))
+				}
+				recyclePods(logger, funcs, ndm, gpm)
+			}
+
+		},
+	})
+	return store, controller
+
+}
+
+func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClient *crd.FissionClient) ([]fv1.Function, error) {
+	funcList, err := fissionClient.Functions(metav1.NamespaceAll).List(metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	// In future a cache that populates at start and is updated on changes might be better solution
+	relatedFunctions := make([]fv1.Function, 0)
+	for _, f := range funcList.Items {
+		for _, secret := range f.Spec.Secrets {
+			if (secret.Name == m.Name) && (secret.Namespace == m.Namespace) {
+				relatedFunctions = append(relatedFunctions, f)
+				break
+			}
+		}
+	}
+	return relatedFunctions, nil
+}
+
+func recyclePods(logger *zap.Logger, funcs []fv1.Function, ndm *nd.NewDeploy, gpm *gpm.GenericPoolManager) {
+	for _, f := range funcs {
+		var err error
+
+		switch f.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType {
+		case fv1.ExecutorTypeNewdeploy:
+			err = ndm.RefreshFuncPods(logger, f)
+		case fv1.ExecutorTypePoolmgr:
+			err = gpm.RefreshFuncPods(logger, f)
+		}
+
+		if err != nil {
+			logger.Error("Failed to recycle pods for function after configmap changed",
+				zap.Error(err),
+				zap.Any("function", f))
+		}
+	}
+}

pkg/executor/executor.go

@@ -32,6 +32,7 @@ import (
 
 	fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
 	"github.com/fission/fission/pkg/crd"
+	"github.com/fission/fission/pkg/executor/cms"
 	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/executor/newdeploy"
 	"github.com/fission/fission/pkg/executor/poolmgr"
@@ -45,6 +46,7 @@ type (
 
 		gpm *poolmgr.GenericPoolManager
 		ndm *newdeploy.NewDeploy
+		cms *cms.ConfigSecretController
 
 		fissionClient *crd.FissionClient
 		fsCache       *fscache.FunctionServiceCache
@@ -64,11 +66,12 @@ type (
 	}
 )
 
-func MakeExecutor(logger *zap.Logger, gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor {
+func MakeExecutor(logger *zap.Logger, gpm *poolmgr.GenericPoolManager, ndm *newdeploy.NewDeploy, cms *cms.ConfigSecretController, fissionClient *crd.FissionClient, fsCache *fscache.FunctionServiceCache) *Executor {
 	executor := &Executor{
 		logger:        logger.Named("executor"),
 		gpm:           gpm,
 		ndm:           ndm,
+		cms:           cms,
 		fissionClient: fissionClient,
 		fsCache:       fsCache,
 
@@ -238,7 +241,9 @@ func StartExecutor(logger *zap.Logger, fissionNamespace string, functionNamespac
 		fissionClient, kubernetesClient, restClient,
 		functionNamespace, fetcherConfig, poolID)
 
-	api := MakeExecutor(logger, gpm, ndm, fissionClient, fsCache)
+	cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, ndm, gpm)
+
+	api := MakeExecutor(logger, gpm, ndm, cms, fissionClient, fsCache)
 
 	go api.Serve(port)
 	go serveMetric(logger)

pkg/executor/newdeploy/newdeploy.go

@@ -185,6 +185,9 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen
 	}
 	resources := deploy.getResources(env, fn)
 
+	maxUnavailable := intstr.FromString("20%")
+	maxSurge := intstr.FromString("100%")
+
 	deployment := &v1beta1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   deployName,
@@ -217,6 +220,12 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen
 									},
 								},
 							},
+							Env: []apiv1.EnvVar{
+								{
+									Name:  fv1.LastUpdateTimestamp,
+									Value: time.Now().String(),
+								},
+							},
 							// https://istio.io/docs/setup/kubernetes/additional-setup/requirements/
 							Ports: []apiv1.ContainerPort{
 								{
@@ -231,6 +240,13 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen
 					TerminationGracePeriodSeconds: &gracePeriodSeconds,
 				},
 			},
+			Strategy: v1beta1.DeploymentStrategy{
+				Type: v1beta1.RollingUpdateDeploymentStrategyType,
+				RollingUpdate: &v1beta1.RollingUpdateDeployment{
+					MaxUnavailable: &maxUnavailable,
+					MaxSurge:       &maxSurge,
+				},
+			},
 		},
 	}
 

pkg/executor/newdeploy/newdeploymgr.go

@@ -35,6 +35,7 @@ import (
 	k8sErrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/labels"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
@@ -220,6 +221,45 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, metadata *metav1.Object
 	return deploy.createFunction(fn, false)
 }
 
+// RefreshFuncPods deleted pods related to the function so that new pods are replenished
+func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
+
+	env, err := deploy.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
+	if err != nil {
+		return err
+	}
+
+	funcLabels := deploy.getDeployLabels(f.Metadata, metav1.ObjectMeta{
+		Name:      f.Spec.Environment.Name,
+		Namespace: f.Spec.Environment.Namespace,
+		UID:       env.Metadata.UID,
+	})
+
+	dep, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(metav1.NamespaceAll).List(metav1.ListOptions{
+		LabelSelector: labels.Set(funcLabels).AsSelector().String(),
+	})
+
+	if err != nil {
+		return err
+	}
+
+	patch := fmt.Sprintf(`{"spec" : {"template": {"spec":{"containers":[{"name": "%s", "env":[{"name": "%s", "value": "%s"}]}]}}}}`,
+		f.Metadata.Name,
+		fv1.LastUpdateTimestamp,
+		time.Now().String())
+
+	// Ideally there should be only one deployment but for now we rely on label/selector to ensure that condition
+	for _, deployment := range dep.Items {
+		_, err := deploy.kubernetesClient.ExtensionsV1beta1().Deployments(deployment.ObjectMeta.Namespace).Patch(deployment.ObjectMeta.Name,
+			k8sTypes.StrategicMergePatchType,
+			[]byte(patch))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (deploy *NewDeploy) createFunction(fn *fv1.Function, firstcreate bool) (*fscache.FuncSvc, error) {
 	if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy {
 		return nil, nil
@@ -276,7 +316,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache.
 			objName = fsvc.Name
 		}
 	}
-	deployLabels := deploy.getDeployLabels(fn, env)
+	deployLabels := deploy.getDeployLabels(fn.Metadata, env.Metadata)
 
 	// 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
@@ -491,7 +531,7 @@ func (deploy *NewDeploy) updateFuncDeployment(fn *fv1.Function, env *fv1.Environ
 	}
 	fnObjName := fsvc.Name
 
-	deployLabels := deploy.getDeployLabels(fn, env)
+	deployLabels := deploy.getDeployLabels(fn.Metadata, env.Metadata)
 	deploy.logger.Info("updating deployment due to function/environment update",
 		zap.String("deployment", fnObjName), zap.Any("function", fn.Metadata.Name))
 
@@ -557,16 +597,16 @@ func (deploy *NewDeploy) getObjName(fn *fv1.Function) string {
 	return strings.ToLower(fmt.Sprintf("newdeploy-%v-%v-%v", fn.Metadata.Name, fn.Metadata.Namespace, uniuri.NewLen(8)))
 }
 
-func (deploy *NewDeploy) getDeployLabels(fn *fv1.Function, env *fv1.Environment) map[string]string {
+func (deploy *NewDeploy) getDeployLabels(fnMeta metav1.ObjectMeta, envMeta metav1.ObjectMeta) map[string]string {
 	return map[string]string{
 		types.EXECUTOR_INSTANCEID_LABEL: deploy.instanceID,
 		types.EXECUTOR_TYPE:             fv1.ExecutorTypeNewdeploy,
-		types.ENVIRONMENT_NAME:          env.Metadata.Name,
-		types.ENVIRONMENT_NAMESPACE:     env.Metadata.Namespace,
-		types.ENVIRONMENT_UID:           string(env.Metadata.UID),
-		types.FUNCTION_NAME:             fn.Metadata.Name,
-		types.FUNCTION_NAMESPACE:        fn.Metadata.Namespace,
-		types.FUNCTION_UID:              string(fn.Metadata.UID),
+		types.ENVIRONMENT_NAME:          envMeta.Name,
+		types.ENVIRONMENT_NAMESPACE:     envMeta.Namespace,
+		types.ENVIRONMENT_UID:           string(envMeta.UID),
+		types.FUNCTION_NAME:             fnMeta.Name,
+		types.FUNCTION_NAMESPACE:        fnMeta.Namespace,
+		types.FUNCTION_UID:              string(fnMeta.UID),
 	}
 }
 

pkg/executor/poolmgr/gpm.go

@@ -25,7 +25,9 @@ import (
 
 	"github.com/fission/fission/pkg/utils"
 	"go.uber.org/zap"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	k8sCache "k8s.io/client-go/tools/cache"
@@ -130,6 +132,48 @@ func (gpm *GenericPoolManager) Run(ctx context.Context) {
 	go gpm.idleObjectReaper()
 }
 
+func (gpm *GenericPoolManager) RefreshFuncPods(logger *zap.Logger, f fv1.Function) error {
+
+	env, err := gpm.fissionClient.Environments(f.Spec.Environment.Namespace).Get(f.Spec.Environment.Name)
+	if err != nil {
+		return err
+	}
+
+	gp, err := gpm.GetPool(env)
+	if err != nil {
+		return err
+	}
+
+	funcSvc, err := gp.fsCache.GetByFunction(&f.Metadata)
+	if err != nil {
+		return err
+	}
+
+	gp.fsCache.DeleteEntry(funcSvc)
+
+	funcLabels := gp.labelsForFunction(&f.Metadata)
+
+	podList, err := gpm.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(metav1.ListOptions{
+		LabelSelector: labels.Set(funcLabels).AsSelector().String(),
+	})
+
+	if err != nil {
+		return err
+	}
+
+	for _, po := range podList.Items {
+		err := gpm.kubernetesClient.CoreV1().Pods(po.ObjectMeta.Namespace).Delete(po.ObjectMeta.Name, &metav1.DeleteOptions{})
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (gpm *GenericPoolManager) service() {
 	for {
 		req := <-gpm.requestChannel