From 93c5b53b777f811fa8965d93c3c935f9a4b7c9ee Mon Sep 17 00:00:00 2001 From: Vishal Date: Tue, 23 Jul 2019 06:57:43 +0530 Subject: [PATCH] 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 | 5 + pkg/executor/api.go | 2 + pkg/executor/cms/cmscontroller.go | 182 +++++++++++++++++++++++++ pkg/executor/executor.go | 9 +- pkg/executor/newdeploy/newdeploy.go | 16 +++ pkg/executor/newdeploy/newdeploymgr.go | 58 ++++++-- pkg/executor/poolmgr/gpm.go | 44 ++++++ 7 files changed, 305 insertions(+), 11 deletions(-) create mode 100644 pkg/executor/cms/cmscontroller.go diff --git a/pkg/apis/fission.io/v1/const.go b/pkg/apis/fission.io/v1/const.go index 80a159d3..18569609 100644 --- a/pkg/apis/fission.io/v1/const.go +++ b/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" ) diff --git a/pkg/executor/api.go b/pkg/executor/api.go index 8e43ab10..517cb949 100644 --- a/pkg/executor/api.go +++ b/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, }) diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go new file mode 100644 index 00000000..e31bfb0e --- /dev/null +++ b/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)) + } + } +} diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 0104f0db..d0663756 100644 --- a/pkg/executor/executor.go +++ b/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) diff --git a/pkg/executor/newdeploy/newdeploy.go b/pkg/executor/newdeploy/newdeploy.go index 780f8728..2109f4ec 100644 --- a/pkg/executor/newdeploy/newdeploy.go +++ b/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, + }, + }, }, } diff --git a/pkg/executor/newdeploy/newdeploymgr.go b/pkg/executor/newdeploy/newdeploymgr.go index 2f41f84a..e8579b51 100644 --- a/pkg/executor/newdeploy/newdeploymgr.go +++ b/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), } } diff --git a/pkg/executor/poolmgr/gpm.go b/pkg/executor/poolmgr/gpm.go index 3c42abe2..11cbdb6c 100644 --- a/pkg/executor/poolmgr/gpm.go +++ b/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