From 3f3b11ffbfe4461c65db12266859f8a3f1e890a9 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Mon, 2 Dec 2019 17:28:19 +0800 Subject: [PATCH] Prevent deployment from rolling update due to different instance-id (#1454) The pod template is embedded inside the deployment. So if the pod annotation contains instance-id, the deployment will get updated and thus triggers a rolling update whenever a new executor starts which is unwanted. After this PR, poolmanager will patches instance-id when a pod is chosen for a function. For newdeploy, unlike poolmanager manages the lifecycle of function pod directly, newdeploy is only responsible to create the deployment so we append instance-id to top- level controller (deployment) only. --- pkg/executor/cms/cmscontroller.go | 6 +-- .../executortype/newdeploy/newdeploy.go | 7 +-- .../executortype/newdeploy/newdeploymgr.go | 46 ++----------------- pkg/executor/executortype/poolmgr/gp.go | 34 ++++++++------ pkg/executor/executortype/poolmgr/gpm.go | 8 +++- pkg/executor/util/util.go | 4 +- 6 files changed, 40 insertions(+), 65 deletions(-) diff --git a/pkg/executor/cms/cmscontroller.go b/pkg/executor/cms/cmscontroller.go index 7bfb8f27..107ae02a 100644 --- a/pkg/executor/cms/cmscontroller.go +++ b/pkg/executor/cms/cmscontroller.go @@ -85,7 +85,7 @@ func initConfigmapController(logger *zap.Logger, fissionClient *crd.FissionClien if err != nil { logger.Error("Failed to get functions related to configmap", zap.String("configmap_name", newCm.ObjectMeta.Name), zap.String("configmap_namespace", newCm.ObjectMeta.Namespace)) } - recyclePods(logger, funcs, types) + refreshPods(logger, funcs, types) } }, }) @@ -131,7 +131,7 @@ func initSecretController(logger *zap.Logger, fissionClient *crd.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, types) + refreshPods(logger, funcs, types) } }, }) @@ -157,7 +157,7 @@ func getSecretRelatedFuncs(logger *zap.Logger, m *metav1.ObjectMeta, fissionClie return relatedFunctions, nil } -func recyclePods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) { +func refreshPods(logger *zap.Logger, funcs []fv1.Function, types map[fv1.ExecutorType]executortype.ExecutorType) { for _, f := range funcs { var err error diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index 3ebd46ba..bea5f8cd 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -192,12 +192,13 @@ func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environmen if podAnnotations == nil { podAnnotations = make(map[string]string) } + + // Here, we don't append deployAnnotations to podAnnotations + // since newdeploy doesn't manager pod lifecycle directly. + if deploy.useIstio && env.Spec.AllowAccessToExternalNetwork { podAnnotations["sidecar.istio.io/inject"] = "false" } - for k, v := range deployAnnotations { - podAnnotations[k] = v - } resources := deploy.getResources(env, fn) // Set maxUnavailable and maxSurge to 20% is because we want diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index 10f17dd5..1962b79f 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -19,7 +19,6 @@ package newdeploy import ( "context" "fmt" - "math/rand" "os" "strconv" "strings" @@ -245,44 +244,6 @@ func (deploy *NewDeploy) RefreshFuncPods(logger *zap.Logger, f fv1.Function) err } func (deploy *NewDeploy) AdoptExistingResources() { - l := map[string]string{ - types.EXECUTOR_TYPE: string(fv1.ExecutorTypeNewdeploy), - } - - podList, err := deploy.kubernetesClient.CoreV1().Pods(metav1.NamespaceAll).List(metav1.ListOptions{ - LabelSelector: labels.Set(l).AsSelector().String(), - }) - if err != nil { - deploy.logger.Error("error getting pod list", zap.Error(err)) - return - } - - // Unlike poolmanager manages the lifecycle of function pod directly, - // newdeploy is only responsible to create the deployment and Kubernetes - // will handle the rest. Hence we don't need to wait for the pod patching - // process to finish. - for i := range podList.Items { - pod := &podList.Items[i] - if !utils.IsReadyPod(pod) { - continue - } - go func() { - // avoid too many requests arrive Kubernetes API server at the same time. - time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) - - patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, types.EXECUTOR_INSTANCEID_LABEL, deploy.instanceID) - pod, err = deploy.kubernetesClient.CoreV1().Pods(pod.Namespace).Patch(pod.Name, k8sTypes.StrategicMergePatchType, []byte(patch)) - if err != nil { - // just log the error since it won't affect the function serving - deploy.logger.Warn("error patching executor instance ID of pod", zap.Error(err), - zap.String("pod", pod.Name), zap.String("ns", pod.Namespace)) - return - } - deploy.logger.Info("adopt newdeploy function pod", - zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations)) - }() - } - fnList, err := deploy.fissionClient.Functions(metav1.NamespaceAll).List(metav1.ListOptions{}) if err != nil { deploy.logger.Error("error getting function list", zap.Error(err)) @@ -689,9 +650,12 @@ func (deploy *NewDeploy) updateFuncDeployment(fn *fv1.Function, env *fv1.Environ return err } - // use current replicas instead of minscale in the ExecutionStrategy. + // the resource version inside function packageRef is changed, + // so the content of fetchRequest in deployment cmd is different. + // Therefore, the deployment update will trigger a rolling update. newDeployment, err := deploy.getDeploymentSpec(fn, env, - existingDepl.Spec.Replicas, fnObjName, ns, deployLabels, deploy.getDeployAnnotations(fn.Metadata)) + existingDepl.Spec.Replicas, // use current replicas instead of minscale in the ExecutionStrategy. + fnObjName, ns, deployLabels, deploy.getDeployAnnotations(fn.Metadata)) if err != nil { deploy.updateStatus(fn, err, "failed to get new deployment spec while updating function") return err diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 8de8350d..d4becfa9 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -146,7 +146,7 @@ func MakeGenericPool( return gp, nil } -func (gp *GenericPool) getDeployLabels() map[string]string { +func (gp *GenericPool) getEnvironmentPoolLabels() map[string]string { return map[string]string{ types.EXECUTOR_TYPE: string(fv1.ExecutorTypePoolmgr), types.ENVIRONMENT_NAME: gp.env.Metadata.Name, @@ -246,6 +246,11 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro // modified, this should fail; in that case just // retry. chosenPod.ObjectMeta.Labels = newLabels + + // Append executor instance id to pod annotations to + // indicate this pod is managed by this executor. + chosenPod.ObjectMeta.Annotations = gp.getDeployAnnotations() + _, err = gp.kubernetesClient.CoreV1().Pods(gp.namespace).Update(chosenPod) if err != nil { gp.logger.Error("failed to relabel pod", zap.Error(err), zap.String("pod", chosenPod.ObjectMeta.Name)) @@ -259,7 +264,7 @@ func (gp *GenericPool) _choosePod(newLabels map[string]string) (*apiv1.Pod, erro } func (gp *GenericPool) labelsForFunction(metadata *metav1.ObjectMeta) map[string]string { - label := gp.getDeployLabels() + label := gp.getEnvironmentPoolLabels() label[types.FUNCTION_NAME] = metadata.Name label[types.FUNCTION_UID] = string(metadata.UID) label[types.FUNCTION_NAMESPACE] = metadata.Namespace // function CRD must stay within same namespace of environment CRD @@ -274,11 +279,6 @@ func (gp *GenericPool) scheduleDeletePod(name string) { // cleaned up. (We need a better solutions for both those things; log // aggregation and storage will help.) gp.logger.Error("error in pod - scheduling cleanup", zap.String("pod", name)) - // Ignore sleep here if istio feature is enabled, function pod - // will be deleted after 6 mins (terminationGracePeriodSeconds). - if !gp.useIstio { - time.Sleep(5 * time.Minute) - } gp.kubernetesClient.CoreV1().Pods(gp.namespace).Delete(name, nil) }() } @@ -351,7 +351,7 @@ func (gp *GenericPool) getPoolName() string { // A pool is a deployment of generic containers for an env. This // creates the pool but doesn't wait for any pods to be ready. func (gp *GenericPool) createPool() error { - deployLabels := gp.getDeployLabels() + deployLabels := gp.getEnvironmentPoolLabels() deployAnnotations := gp.getDeployAnnotations() // Use long terminationGracePeriodSeconds for connection draining in case that @@ -365,12 +365,15 @@ func (gp *GenericPool) createPool() error { if podAnnotations == nil { podAnnotations = make(map[string]string) } + + // Here, we don't append executor instance-id to pod annotations + // to prevent unwanted rolling updates occur. Pool manager will + // append executor instance-id to pod annotations when a pod is chosen + // for function specialization. + if gp.useIstio && gp.env.Spec.AllowAccessToExternalNetwork { podAnnotations["sidecar.istio.io/inject"] = "false" } - for k, v := range deployAnnotations { - podAnnotations[k] = v - } container, err := util.MergeContainer(&apiv1.Container{ Name: gp.env.Metadata.Name, @@ -526,7 +529,8 @@ func (gp *GenericPool) waitForReadyPod() error { func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1.Service, error) { service := apiv1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: name, + Name: name, + Labels: labels, }, Spec: apiv1.ServiceSpec{ Type: apiv1.ServiceTypeClusterIP, @@ -546,7 +550,7 @@ func (gp *GenericPool) createSvc(name string, labels map[string]string) (*apiv1. func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { gp.logger.Info("choosing pod from pool", zap.Any("function", fn.Metadata)) - newLabels := gp.labelsForFunction(&fn.Metadata) + funcLabels := gp.labelsForFunction(&fn.Metadata) if gp.useIstio { // Istio only allows accessing pod through k8s service, and requests come to @@ -589,7 +593,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac } } - pod, err := gp.choosePod(newLabels) + pod, err := gp.choosePod(funcLabels) if err != nil { return nil, err } @@ -608,7 +612,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac svcName = fmt.Sprintf("%s-%v", svcName, fn.Metadata.UID) } - svc, err := gp.createSvc(svcName, newLabels) + svc, err := gp.createSvc(svcName, funcLabels) if err != nil { gp.scheduleDeletePod(pod.ObjectMeta.Name) return nil, err diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index b86df3f5..b435450e 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -183,7 +183,7 @@ func (gpm *GenericPoolManager) TapService(svcHost string) error { // containers in it are reporting a ready status for the healthCheck. func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { for _, obj := range fsvc.KubernetesObjects { - if obj.Kind == "pod" { + if strings.ToLower(obj.Kind) == "pod" { pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{}) if err == nil && utils.IsReadyPod(pod) { // Normally, the address format is http://[pod-ip]:[port], however, if the @@ -192,7 +192,11 @@ func (gpm *GenericPoolManager) IsValid(fsvc *fscache.FuncSvc) bool { // Otherwise, we need to ensure that the address contains pod ip. if gpm.enableIstio || (!gpm.enableIstio && strings.Contains(fsvc.Address, pod.Status.PodIP)) { - gpm.logger.Debug("valid address", zap.String("address", fsvc.Address)) + gpm.logger.Debug("valid address", + zap.String("address", fsvc.Address), + zap.Any("function", fsvc.Function), + zap.String("executor", string(fsvc.Executor)), + ) return true } } diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index 64aef059..cd1700b5 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -31,7 +31,9 @@ import ( // when creating the environment deployment since kubelet will retry to // pull image until successes. func ApplyImagePullSecret(secret string, podspec apiv1.PodSpec) *apiv1.PodSpec { - podspec.ImagePullSecrets = []apiv1.LocalObjectReference{{Name: secret}} + if len(secret) > 0 { + podspec.ImagePullSecrets = []apiv1.LocalObjectReference{{Name: secret}} + } return &podspec }