Fix executor does not reap specialized function pod when env no longer exists (#633)

executor/cleanup.go

@@ -22,9 +22,9 @@ import (
 	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/api/errors"
 	meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/pkg/api"
 
@@ -104,52 +104,46 @@ func idleObjectReaper(kubeClient *kubernetes.Clientset,
 			log.Fatalf("Failed to get environment list: %v", err)
 		}
 
+		envList := make(map[types.UID]struct{})
 		for i := range envs.Items {
 			env := envs.Items[i]
-			if env.Spec.AllowedFunctionsPerContainer == fission.AllowedFunctionsPerContainerInfinite {
+			envList[env.Metadata.UID] = struct{}{}
-				continue
+		}
+
+		funcSvcs, err := fsCache.ListOld(idlePodReapTime)
+		if err != nil {
+			log.Printf("Error reaping idle pods: %v", err)
+			continue
+		}
+
+		for _, fsvc := range funcSvcs {
+			if _, ok := envList[fsvc.Environment.Metadata.UID]; !ok {
+				log.Printf("Environment %v for function %v no longer exists",
+					fsvc.Environment.Metadata.Name, fsvc.Name)
 			}
-			funcSvcs, err := fsCache.ListOld(&env.Metadata, idlePodReapTime)
+
-			if err != nil {
+			if fsvc.Environment.Spec.AllowedFunctionsPerContainer == fission.AllowedFunctionsPerContainerInfinite {
-				log.Printf("Error reaping idle pods: %v", err)
 				continue
 			}
 
-			for _, fsvc := range funcSvcs {
+			// Newdeploy manager handles the function delete event and clean cache/kubeobjs itself,
-
+			// so we ignore the function service cache with newdepoy executor type here.
-				fn, err := fissionClient.Functions(fsvc.Function.Namespace).Get(fsvc.Function.Name)
+			if fsvc.Executor != fscache.NEWDEPLOY {
-				if err == nil {
+				deleted, err := fsCache.DeleteOld(fsvc, idlePodReapTime)
-					// Ignore functions of NewDeploy ExecutorType with MinScale > 0
+				if err != nil {
-					if fn.Spec.InvokeStrategy.ExecutionStrategy.MinScale > 0 &&
+					log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", fsvc, err)
-						fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypeNewdeploy {
+					log.Printf("Object Name| Object Kind | Object Namespace")
-						continue
+					for _, kubeobj := range fsvc.KubernetesObjects {
+						log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace)
 					}
 				}
 
-				// Return errors not equal to "is not found" error
+				if !deleted {
-				if err != nil && !errors.IsNotFound(err) {
-					log.Printf("Error getting function: %v", fsvc.Function.Name)
 					continue
 				}
 
-				// Newdeploy manager handles the function delete event and clean cache/kubeobjs itself,
+				for _, kubeobj := range fsvc.KubernetesObjects {
-				// so we ignore the function service cache with newdepoy executor type here.
+					deleteKubeobject(kubeClient, &kubeobj)
-				if fsvc.Executor != fscache.NEWDEPLOY {
-					deleted, err := fsCache.DeleteOld(fsvc, idlePodReapTime)
-					if err != nil {
-						log.Printf("Error deleting Kubernetes objects for fsvc '%v': %v", fsvc, err)
-						log.Printf("Object Name| Object Kind | Object Namespace")
-						for _, kubeobj := range fsvc.KubernetesObjects {
-							log.Printf("%v | %v | %v", kubeobj.Name, kubeobj.Kind, kubeobj.Namespace)
-						}
-					}
-
-					if !deleted {
-						continue
-					}
-					for _, kubeobj := range fsvc.KubernetesObjects {
-						deleteKubeobject(kubeClient, &kubeobj)
-					}
 				}
 			}
 		}

executor/fscache/functionServiceCache.go

@@ -68,7 +68,6 @@ type (
 		address           string
 		kubernetesObjects []api.ObjectReference
 		age               time.Duration
-		env               *metav1.ObjectMeta // used for ListOld
 		responseChannel   chan *fscResponse
 	}
 	fscResponse struct {
@@ -117,8 +116,7 @@ func (fsc *FunctionServiceCache) service() {
 			funcObjects := make([]*FuncSvc, 0)
 			for _, funcSvc := range fscs {
 				fsvc := funcSvc.(*FuncSvc)
-				if fsvc.Environment.Metadata.UID == req.env.UID &&
+				if time.Since(fsvc.Atime) > req.age {
-					time.Since(fsvc.Atime) > req.age {
 					funcObjects = append(funcObjects, fsvc)
 				}
 			}
@@ -263,12 +261,11 @@ func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration)
 	return true, nil
 }
 
-func (fsc *FunctionServiceCache) ListOld(env *metav1.ObjectMeta, age time.Duration) ([]*FuncSvc, error) {
+func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error) {
 	responseChannel := make(chan *fscResponse)
 	fsc.requestChannel <- &fscRequest{
 		requestType:     LISTOLD,
 		age:             age,
-		env:             env,
 		responseChannel: responseChannel,
 	}
 	resp := <-responseChannel