Improve executor bootstrap speed (#1446)

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)
 

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 {

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

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

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() {

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)

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):
+	}
+}