From 47aaa85108eeb11d3c5a5d21779d5d3da380fea2 Mon Sep 17 00:00:00 2001 From: Ta-Ching Chen Date: Fri, 29 Nov 2019 14:18:23 +0800 Subject: [PATCH] Improve executor bootstrap speed (#1446) --- pkg/apis/fission.io/v1/const.go | 1 - pkg/executor/executor.go | 14 ++- .../executortype/newdeploy/newdeploy.go | 51 ++++++----- .../executortype/newdeploy/newdeploymgr.go | 46 +++++----- pkg/executor/executortype/poolmgr/gp.go | 14 +-- pkg/executor/executortype/poolmgr/gpm.go | 31 +++---- pkg/executor/reaper/reaper.go | 88 ++++++++----------- pkg/executor/util/util.go | 15 ++++ pkg/types/types.go | 1 - 9 files changed, 134 insertions(+), 127 deletions(-) diff --git a/pkg/apis/fission.io/v1/const.go b/pkg/apis/fission.io/v1/const.go index 1b731a6e..46215576 100644 --- a/pkg/apis/fission.io/v1/const.go +++ b/pkg/apis/fission.io/v1/const.go @@ -18,7 +18,6 @@ package v1 const ( EXECUTOR_INSTANCEID_LABEL string = "executorInstanceId" - POOLMGR_INSTANCEID_LABEL string = "poolmgrInstanceId" DEFAULT_FUNCTION_TIMEOUT int = 60 ) diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index 0350a02c..bdace374 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -37,6 +37,7 @@ import ( "github.com/fission/fission/pkg/executor/executortype/poolmgr" "github.com/fission/fission/pkg/executor/fscache" "github.com/fission/fission/pkg/executor/reaper" + "github.com/fission/fission/pkg/executor/util" fetcherConfig "github.com/fission/fission/pkg/fetcher/config" ) @@ -74,6 +75,12 @@ func MakeExecutor(logger *zap.Logger, cms *cms.ConfigSecretController, requestChan: make(chan *createFuncServiceRequest), fsCreateWg: make(map[string]*sync.WaitGroup), } + for _, et := range types { + go func(et executortype.ExecutorType) { + et.Run(context.Background()) + }(et) + } + go cms.Run(context.Background()) go executor.serveCreateFuncServices() return executor, nil @@ -241,21 +248,20 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames go func(et executortype.ExecutorType) { defer wg.Done() et.AdoptOrphanResources() - et.Run(context.Background()) }(et) } - wg.Wait() + // set hard timeout for resource adoption + util.WaitTimeout(wg, 30*time.Second) cms := cms.MakeConfigSecretController(logger, fissionClient, kubernetesClient, executorTypes) - cms.Run(context.Background()) api, err := MakeExecutor(logger, cms, fissionClient, executorTypes) if err != nil { return err } - go reaper.CleanupRoleBindings(logger, kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30) go reaper.CleanupOldExecutorObjects(logger, kubernetesClient, executorInstanceID) + go reaper.CleanupRoleBindings(logger, kubernetesClient, fissionClient, functionNamespace, envBuilderNamespace, time.Minute*30) go api.Serve(port) go serveMetric(logger) diff --git a/pkg/executor/executortype/newdeploy/newdeploy.go b/pkg/executor/executortype/newdeploy/newdeploy.go index 9fd4fbaa..7a634c92 100644 --- a/pkg/executor/executortype/newdeploy/newdeploy.go +++ b/pkg/executor/executortype/newdeploy/newdeploy.go @@ -44,22 +44,22 @@ const ( ) func (deploy *NewDeploy) createOrGetDeployment(fn *fv1.Function, env *fv1.Environment, - deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string, firstcreate bool) (*appsv1.Deployment, error) { + deployName string, deployLabels map[string]string, deployAnnotations map[string]string, deployNamespace string) (*appsv1.Deployment, error) { - minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) specializationTimeout := int(fn.Spec.InvokeStrategy.ExecutionStrategy.SpecializationTimeout) + minScale := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) - // If it's not the first time creation and minscale is 0 means that all pods for function were recycled, - // in such cases we need set minscale to 1 for router to serve requests. - if !firstcreate && minScale <= 0 { + // Always scale to at least one pod when createOrGetDeployment + // is called. The idleObjectReaper will scale-in the deployment + // later if no requests to the function. + if minScale <= 0 { minScale = 1 } - waitForDeploy := minScale > 0 - existingDepl, err := deploy.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(deployName, metav1.GetOptions{}) if err == nil { - if existingDepl.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { + // Try to adopt orphan deployment created by the old executor. + if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, types.EXECUTOR_INSTANCEID_LABEL, deploy.instanceID) existingDepl, err = deploy.kubernetesClient.AppsV1().Deployments(deployNamespace).Patch(deployName, k8sTypes.StrategicMergePatchType, []byte(patch)) if err != nil { @@ -67,19 +67,21 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *fv1.Function, env *fv1.Enviro zap.String("deploy", deployName), zap.String("ns", deployNamespace)) return nil, err } + // In this case, we just return without waiting for it for fast bootstraping. + return existingDepl, nil } - if waitForDeploy { - if *existingDepl.Spec.Replicas < minScale { - err = deploy.scaleDeployment(existingDepl.Namespace, existingDepl.Name, minScale) - if err != nil { - deploy.logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.Metadata.Name)) - return nil, err - } - } - if existingDepl.Status.AvailableReplicas < minScale { - existingDepl, err = deploy.waitForDeploy(existingDepl, minScale, specializationTimeout) + + if *existingDepl.Spec.Replicas < minScale { + err = deploy.scaleDeployment(existingDepl.Namespace, existingDepl.Name, minScale) + if err != nil { + deploy.logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.Metadata.Name)) + return nil, err } } + if existingDepl.Status.AvailableReplicas < minScale { + existingDepl, err = deploy.waitForDeploy(existingDepl, minScale, specializationTimeout) + } + return existingDepl, err } else if k8s_err.IsNotFound(err) { err := deploy.setupRBACObjs(deployNamespace, fn) @@ -87,7 +89,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *fv1.Function, env *fv1.Enviro return nil, err } - deployment, err := deploy.getDeploymentSpec(fn, env, deployName, deployNamespace, deployLabels, deployAnnotations) + deployment, err := deploy.getDeploymentSpec(fn, env, &minScale, deployName, deployNamespace, deployLabels, deployAnnotations) if err != nil { return nil, err } @@ -102,7 +104,7 @@ func (deploy *NewDeploy) createOrGetDeployment(fn *fv1.Function, env *fv1.Enviro return nil, err } - if waitForDeploy { + if minScale > 0 { depl, err = deploy.waitForDeploy(depl, minScale, specializationTimeout) } @@ -168,10 +170,13 @@ func (deploy *NewDeploy) deleteDeployment(ns string, name string) error { }) } -func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environment, +func (deploy *NewDeploy) getDeploymentSpec(fn *fv1.Function, env *fv1.Environment, targetReplicas *int32, deployName string, deployNamespace string, deployLabels map[string]string, deployAnnotations map[string]string) (*appsv1.Deployment, error) { replicas := int32(fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale) + if targetReplicas != nil { + replicas = *targetReplicas + } gracePeriodSeconds := int64(6 * 60) if env.Spec.TerminationGracePeriod > 0 { @@ -351,7 +356,7 @@ func (deploy *NewDeploy) createOrGetHpa(hpaName string, execStrategy *fv1.Execut existingHpa, err := deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Get(hpaName, metav1.GetOptions{}) if err == nil { - if existingHpa.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { + if existingHpa.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, types.EXECUTOR_INSTANCEID_LABEL, deploy.instanceID) existingHpa, err = deploy.kubernetesClient.AutoscalingV1().HorizontalPodAutoscalers(depl.ObjectMeta.Namespace).Patch(hpaName, k8sTypes.StrategicMergePatchType, []byte(patch)) if err != nil { @@ -407,7 +412,7 @@ func (deploy *NewDeploy) deleteHpa(ns string, name string) error { func (deploy *NewDeploy) createOrGetSvc(deployLabels map[string]string, deployAnnotations map[string]string, svcName string, svcNamespace string) (*apiv1.Service, error) { existingSvc, err := deploy.kubernetesClient.CoreV1().Services(svcNamespace).Get(svcName, metav1.GetOptions{}) if err == nil { - if existingSvc.Labels[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { + if existingSvc.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != deploy.instanceID { patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, types.EXECUTOR_INSTANCEID_LABEL, deploy.instanceID) existingSvc, err = deploy.kubernetesClient.CoreV1().Services(svcNamespace).Patch(svcName, k8sTypes.StrategicMergePatchType, []byte(patch)) if err != nil { diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index aa81d22c..097cdcca 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -143,7 +143,7 @@ func (deploy *NewDeploy) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fsc // TODO: client-go doesn't support to pass in context. // Once it supports context, we should change the signature of method. // https://github.com/kubernetes/kubernetes/issues/46503 - return deploy.createFunction(fn, false) + return deploy.createFunction(fn) } func (deploy *NewDeploy) GetFuncSvcFromCache(fn *fv1.Function) (*fscache.FuncSvc, error) { @@ -249,24 +249,21 @@ func (deploy *NewDeploy) AdoptOrphanResources() { 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 } - podWG := &sync.WaitGroup{} - + // 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 } - - podWG.Add(1) go func() { - defer podWG.Done() - // avoid too many requests arrive Kubernetes API server at the same time. time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) @@ -278,7 +275,6 @@ func (deploy *NewDeploy) AdoptOrphanResources() { 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)) }() @@ -290,16 +286,16 @@ func (deploy *NewDeploy) AdoptOrphanResources() { return } - deployWG := &sync.WaitGroup{} + wg := &sync.WaitGroup{} for i := range fnList.Items { fn := &fnList.Items[i] if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fv1.ExecutorTypeNewdeploy { - deployWG.Add(1) + wg.Add(1) go func() { - defer deployWG.Done() + defer wg.Done() - _, err = deploy.fnCreate(fn, true) + _, err = deploy.fnCreate(fn) if err != nil { deploy.logger.Warn("failed to adopt resources for function", zap.Error(err)) return @@ -309,8 +305,7 @@ func (deploy *NewDeploy) AdoptOrphanResources() { } } - podWG.Wait() - deployWG.Wait() + wg.Wait() } func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controller) { @@ -324,7 +319,7 @@ func (deploy *NewDeploy) initFuncController() (k8sCache.Store, k8sCache.Controll go func() { fn := obj.(*fv1.Function) deploy.logger.Debug("create deployment for function", zap.Any("fn", fn.Metadata), zap.Any("fnspec", fn.Spec)) - _, err := deploy.createFunction(fn, true) + _, err := deploy.createFunction(fn) if err != nil { deploy.logger.Error("error eager creating function", zap.Error(err), @@ -406,14 +401,14 @@ func (deploy *NewDeploy) getEnvFunctions(m *metav1.ObjectMeta) []fv1.Function { return relatedFunctions } -func (deploy *NewDeploy) createFunction(fn *fv1.Function, firstcreate bool) (*fscache.FuncSvc, error) { +func (deploy *NewDeploy) createFunction(fn *fv1.Function) (*fscache.FuncSvc, error) { if fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fv1.ExecutorTypeNewdeploy { return nil, nil } fsvcObj, err := deploy.throttler.RunOnce(string(fn.Metadata.UID), func(ableToCreate bool) (interface{}, error) { if ableToCreate { - return deploy.fnCreate(fn, firstcreate) + return deploy.fnCreate(fn) } return deploy.fsCache.GetByFunctionUID(fn.Metadata.UID) }) @@ -446,7 +441,7 @@ func (deploy *NewDeploy) deleteFunction(fn *fv1.Function) error { return err } -func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache.FuncSvc, error) { +func (deploy *NewDeploy) fnCreate(fn *fv1.Function) (*fscache.FuncSvc, error) { env, err := deploy.fissionClient. Environments(fn.Spec.Environment.Namespace). Get(fn.Spec.Environment.Name) @@ -478,7 +473,7 @@ func (deploy *NewDeploy) fnCreate(fn *fv1.Function, firstcreate bool) (*fscache. } svcAddress := fmt.Sprintf("%v.%v", svc.Name, svc.Namespace) - depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, deployAnnotations, ns, firstcreate) + depl, err := deploy.createOrGetDeployment(fn, env, objName, deployLabels, deployAnnotations, ns) if err != nil { deploy.logger.Error("error creating deployment", zap.Error(err), zap.String("deployment", objName)) go deploy.cleanupNewdeploy(ns, objName) @@ -567,7 +562,7 @@ func (deploy *NewDeploy) updateFunction(oldFn *fv1.Function, newFn *fv1.Function deploy.logger.Info("function type changed to new deployment, creating resources", zap.Any("old_function", oldFn.Metadata), zap.Any("new_function", newFn.Metadata)) - _, err := deploy.createFunction(newFn, true) + _, err := deploy.createFunction(newFn) if err != nil { deploy.updateStatus(oldFn, err, "error changing the function's type to newdeploy") } @@ -686,7 +681,14 @@ func (deploy *NewDeploy) updateFuncDeployment(fn *fv1.Function, env *fv1.Environ ns = fn.Metadata.Namespace } - newDeployment, err := deploy.getDeploymentSpec(fn, env, fnObjName, ns, deployLabels, deploy.getDeployAnnotations(fn.Metadata)) + existingDepl, err := deploy.kubernetesClient.AppsV1().Deployments(ns).Get(fnObjName, metav1.GetOptions{}) + if err != nil { + return err + } + + // use current replicas instead of minscale in the ExecutionStrategy. + newDeployment, err := deploy.getDeploymentSpec(fn, env, + existingDepl.Spec.Replicas, 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 f983991a..028fe84b 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -456,20 +456,20 @@ func (gp *GenericPool) createPool() error { deployment.Spec.Template.Spec = *newPodSpec } - _, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Get(deployment.Name, metav1.GetOptions{}) + depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Get(deployment.Name, metav1.GetOptions{}) if err == nil { - patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gp.instanceId) - depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Patch(deployment.Name, k8sTypes.StrategicMergePatchType, []byte(patch)) - if err == nil { - gp.deployment = depl - return nil + if depl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != gp.instanceId { + patch := fmt.Sprintf(`{"metadata":{"annotations":{"%v":"%v"}}}`, fv1.EXECUTOR_INSTANCEID_LABEL, gp.instanceId) + depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Patch(deployment.Name, k8sTypes.StrategicMergePatchType, []byte(patch)) } + gp.deployment = depl + return err } else if !k8sErrs.IsNotFound(err) { gp.logger.Error("error getting deployment in kubernetes", zap.Error(err), zap.String("deployment", deployment.Name)) return err } - depl, err := gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Create(deployment) + depl, err = gp.kubernetesClient.AppsV1().Deployments(gp.namespace).Create(deployment) if err != nil { gp.logger.Error("error creating deployment in kubernetes", zap.Error(err), zap.String("deployment", deployment.Name)) return err diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index b4c7cc41..2c73d6cf 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -251,15 +251,15 @@ func (gpm *GenericPoolManager) AdoptOrphanResources() { } envMap := make(map[string]fv1.Environment, len(envs.Items)) - envWG := &sync.WaitGroup{} + wg := &sync.WaitGroup{} for i := range envs.Items { env := envs.Items[i] if gpm.getEnvPoolsize(&env) > 0 { - envWG.Add(1) + wg.Add(1) go func() { - defer envWG.Done() + defer wg.Done() _, err := gpm.getPool(&env) if err != nil { gpm.logger.Error("adopt pool failed", zap.Error(err)) @@ -285,17 +285,15 @@ func (gpm *GenericPoolManager) AdoptOrphanResources() { return } - podWG := &sync.WaitGroup{} - for i := range podList.Items { pod := &podList.Items[i] if !utils.IsReadyPod(pod) { continue } - podWG.Add(1) + wg.Add(1) go func() { - defer podWG.Done() + defer wg.Done() // avoid too many requests arrive Kubernetes API server at the same time. time.Sleep(time.Duration(rand.Intn(30)) * time.Millisecond) @@ -324,7 +322,7 @@ func (gpm *GenericPoolManager) AdoptOrphanResources() { env, ok8 := envMap[fmt.Sprintf("%v/%v", envNS, envName)] if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6 && ok7 && ok8) { - gpm.logger.Warn("failed to adopt pod for function due to lack necessary information", + gpm.logger.Warn("failed to adopt pod for function due to lack of necessary information", zap.String("pod", pod.Name), zap.Any("labels", pod.Labels), zap.Any("annotations", pod.Annotations), zap.String("env", env.Metadata.Name)) return @@ -355,16 +353,14 @@ func (gpm *GenericPoolManager) AdoptOrphanResources() { Atime: time.Now(), } - // If fsvc already exists we just skip the duplicate one. And let reaper to recycle the duplicate pod. - // This is for the case that there are multiple function pods for the same function due to unknown reason. - _, err := gpm.fsCache.GetByFunction(fsvc.Function) - if err == nil { - return - } - _, err = gpm.fsCache.Add(fsvc) if err != nil { - gpm.logger.Warn("failed to adopt pod for function", zap.Error(err), zap.String("pod", pod.Name)) + // If fsvc already exists we just skip the duplicate one. And let reaper to recycle the duplicate pods. + // This is for the case that there are multiple function pods for the same function due to unknown reason. + if !fscache.IsNameExistError(err) { + gpm.logger.Warn("failed to adopt pod for function", zap.Error(err), zap.String("pod", pod.Name)) + } + return } @@ -373,8 +369,7 @@ func (gpm *GenericPoolManager) AdoptOrphanResources() { }() } - envWG.Wait() - podWG.Wait() + wg.Wait() } func (gpm *GenericPoolManager) service() { diff --git a/pkg/executor/reaper/reaper.go b/pkg/executor/reaper/reaper.go index a37b132b..fe237710 100644 --- a/pkg/executor/reaper/reaper.go +++ b/pkg/executor/reaper/reaper.go @@ -20,7 +20,7 @@ import ( "strings" "time" - "github.com/fission/fission/pkg/utils" + "github.com/hashicorp/go-multierror" "go.uber.org/zap" apiv1 "k8s.io/api/core/v1" meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -29,6 +29,7 @@ import ( fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" "github.com/fission/fission/pkg/crd" "github.com/fission/fission/pkg/types" + "github.com/fission/fission/pkg/utils" ) var ( @@ -38,45 +39,39 @@ var ( // CleanupOldExecutorObjects cleans up resources created by old executor instances func CleanupOldExecutorObjects(logger *zap.Logger, kubernetesClient *kubernetes.Clientset, instanceId string) { - err := cleanup(logger, kubernetesClient, instanceId) - if err != nil { - // TODO retry reaper; logged and ignored for now - logger.Error("Failed to cleanup old executor objects", zap.Error(err)) - } -} + errs := &multierror.Error{} -func cleanup(logger *zap.Logger, client *kubernetes.Clientset, instanceId string) error { - // Pods might still be running user functions, so we give them + err := cleanupHpa(logger, kubernetesClient, instanceId) + if err != nil { + errs = multierror.Append(errs, err) + } + + err = cleanupDeployments(logger, kubernetesClient, instanceId) + if err != nil { + errs = multierror.Append(errs, err) + } + + // Pods might be running user functions, so we give them // a few minutes before terminating them. This time is the // maximum function runtime, plus the time a router might // still route to an old instance, i.e. router cache expiry // time. time.Sleep(6 * time.Minute) - err := cleanupServices(logger, client, instanceId) + err = cleanupServices(logger, kubernetesClient, instanceId) if err != nil { - return err + errs = multierror.Append(errs, err) } - err = cleanupHpa(logger, client, instanceId) + err = cleanupPods(logger, kubernetesClient, instanceId) if err != nil { - return err + errs = multierror.Append(errs, err) } - // Deployments are used for idle pools and can be cleaned up - // immediately. (We should "adopt" these instead of creating - // a new pool.) - err = cleanupDeployments(logger, client, instanceId) - if err != nil { - return err + if errs.ErrorOrNil() != nil { + // TODO retry reaper; logged and ignored for now + logger.Error("Failed to cleanup old executor objects", zap.Error(err)) } - - err = cleanupPods(logger, client, instanceId) - if err != nil { - return err - } - - return nil } // CleanupKubeObject deletes given kubernetes object @@ -119,20 +114,11 @@ func cleanupDeployments(logger *zap.Logger, client *kubernetes.Clientset, instan } for _, dep := range deploymentList.Items { id, ok := dep.ObjectMeta.Annotations[types.EXECUTOR_INSTANCEID_LABEL] - if ok && id != instanceId { - logger.Debug("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) - err := client.AppsV1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt) - if err != nil { - logger.Error("error cleaning up deployment", - zap.Error(err), - zap.String("deployment_name", dep.ObjectMeta.Name), - zap.String("deployment_namespace", dep.ObjectMeta.Namespace)) - } - // ignore err + if !ok { + // Backward compatibility with older label name + id, ok = dep.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] } - // Backward compatibility with older label name - pid, pok := dep.ObjectMeta.Annotations[types.POOLMGR_INSTANCEID_LABEL] - if pok && pid != instanceId { + if ok && id != instanceId { logger.Debug("cleaning up deployment", zap.String("deployment", dep.ObjectMeta.Name)) err := client.AppsV1().Deployments(dep.ObjectMeta.Namespace).Delete(dep.ObjectMeta.Name, &delOpt) if err != nil { @@ -154,6 +140,10 @@ func cleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st } for _, pod := range podList.Items { id, ok := pod.ObjectMeta.Annotations[types.EXECUTOR_INSTANCEID_LABEL] + if !ok { + // Backward compatibility with older label name + id, ok = pod.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] + } if ok && id != instanceId { logger.Debug("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil) @@ -165,18 +155,6 @@ func cleanupPods(logger *zap.Logger, client *kubernetes.Clientset, instanceId st } // ignore err } - // Backward compatibility with older label name - pid, pok := pod.ObjectMeta.Annotations[types.POOLMGR_INSTANCEID_LABEL] - if pok && pid != instanceId { - logger.Debug("cleaning up pod", zap.String("pod", pod.ObjectMeta.Name)) - err := client.CoreV1().Pods(pod.ObjectMeta.Namespace).Delete(pod.ObjectMeta.Name, nil) - if err != nil { - logger.Error("error cleaning up pod", - zap.Error(err), - zap.String("pod_name", pod.ObjectMeta.Name), - zap.String("pod_namespace", pod.ObjectMeta.Namespace)) - } - } } return nil } @@ -188,6 +166,10 @@ func cleanupServices(logger *zap.Logger, client *kubernetes.Clientset, instanceI } for _, svc := range svcList.Items { id, ok := svc.ObjectMeta.Annotations[types.EXECUTOR_INSTANCEID_LABEL] + if !ok { + // Backward compatibility with older label name + id, ok = svc.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] + } if ok && id != instanceId { logger.Debug("cleaning up service", zap.String("service", svc.ObjectMeta.Name)) err := client.CoreV1().Services(svc.ObjectMeta.Namespace).Delete(svc.ObjectMeta.Name, nil) @@ -211,6 +193,10 @@ func cleanupHpa(logger *zap.Logger, client *kubernetes.Clientset, instanceId str for _, hpa := range hpaList.Items { id, ok := hpa.ObjectMeta.Annotations[types.EXECUTOR_INSTANCEID_LABEL] + if !ok { + // Backward compatibility with older label name + id, ok = hpa.ObjectMeta.Labels[types.EXECUTOR_INSTANCEID_LABEL] + } if ok && id != instanceId { logger.Debug("cleaning up HPA", zap.String("hpa", hpa.ObjectMeta.Name)) err := client.AutoscalingV1().HorizontalPodAutoscalers(hpa.ObjectMeta.Namespace).Delete(hpa.ObjectMeta.Name, nil) diff --git a/pkg/executor/util/util.go b/pkg/executor/util/util.go index 40d92eee..64aef059 100644 --- a/pkg/executor/util/util.go +++ b/pkg/executor/util/util.go @@ -17,6 +17,9 @@ limitations under the License. package util import ( + "sync" + "time" + apiv1 "k8s.io/api/core/v1" ) @@ -31,3 +34,15 @@ func ApplyImagePullSecret(secret string, podspec apiv1.PodSpec) *apiv1.PodSpec { podspec.ImagePullSecrets = []apiv1.LocalObjectReference{{Name: secret}} return &podspec } + +func WaitTimeout(wg *sync.WaitGroup, timeout time.Duration) { + waitCh := make(chan struct{}) + go func() { + defer close(waitCh) + wg.Wait() + }() + select { + case <-waitCh: + case <-time.After(timeout): + } +} diff --git a/pkg/types/types.go b/pkg/types/types.go index 7b5804ac..fbb7dace 100644 --- a/pkg/types/types.go +++ b/pkg/types/types.go @@ -91,7 +91,6 @@ const ( ) const EXECUTOR_INSTANCEID_LABEL = fv1.EXECUTOR_INSTANCEID_LABEL -const POOLMGR_INSTANCEID_LABEL = fv1.POOLMGR_INSTANCEID_LABEL const ( ChecksumTypeSHA256 = fv1.ChecksumTypeSHA256