Invalidate stale router cache entry with podIP's for deleted pods. (#546)

The router's cache entry for a function might become stale if the pod that had the function specialized gets deleted somehow. In such a case, we'd retry getting a new service for the function from executor and retry forwarding the user request to the newly created service.

executor/api.go

@@ -57,13 +57,27 @@ func (executor *Executor) getServiceForFunctionApi(w http.ResponseWriter, r *htt
 	w.Write([]byte(serviceName))
 }
 
+// getServiceForFunction first checks if this function's service is cached, if yes, it validates the address.
+// if it's a valid address, just returns it.
+// else, invalidates its cache entry and makes a new request to create a service for this function and finally responds
+// with new address or error.
+//
+// checking for the validity of the address causes a little more over-head than desired. but, it ensures that
+// stale addresses are not returned to the router.
+// To make it optimal, plan is to add an eager cache invalidator function that watches for pod deletion events and
+// invalidates the cache entry if the pod address was cached.
 func (executor *Executor) getServiceForFunction(m *metav1.ObjectMeta) (string, error) {
 	// Check function -> svc cache
 	log.Printf("[%v] Checking for cached function service", m.Name)
 	fsvc, err := executor.fsCache.GetByFunction(m)
 	if err == nil {
-		// Cached, return svc address
-		return fsvc.Address, nil
+		if executor.isValidAddress(fsvc) {
+			// Cached, return svc address
+			return fsvc.Address, nil
+		} else {
+			log.Printf("[%v] Deleting cache entry for invalid address : %s", m.Name, fsvc.Address)
+			executor.fsCache.DeleteEntry(fsvc)
+		}
 	}
 
 	respChan := make(chan *createFuncServiceResponse)

executor/executor.go

@@ -119,6 +119,15 @@ func (executor *Executor) serveCreateFuncServices() {
 	}
 }
 
+func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fission.ExecutorType, error) {
+	fn, err := executor.fissionClient.Functions(meta.Namespace).Get(meta.Name)
+	if err != nil {
+		return "", err
+	}
+	return fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType, nil
+
+}
+
 func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
 	log.Printf("[%v] No cached function service found, creating one", meta.Name)
 
@@ -129,14 +138,12 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs
 		return nil, err
 	}
 
-	fn, err := executor.fissionClient.
-		Functions(meta.Namespace).
-		Get(meta.Name)
+	executorType, err := executor.getFunctionExecutorType(meta)
 	if err != nil {
 		return nil, err
 	}
 
-	switch fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType {
+	switch executorType {
 	case fission.ExecutorTypeNewdeploy:
 		fs, err := executor.ndm.GetFuncSvc(meta)
 		return fs, err
@@ -182,6 +189,15 @@ func (executor *Executor) getFunctionEnv(m *metav1.ObjectMeta) (*crd.Environment
 	return env, nil
 }
 
+// isValidAddress invokes isValidService or isValidPod depending on the type of executor
+func (executor *Executor) isValidAddress(fsvc *fscache.FuncSvc) bool {
+	if fsvc.Executor == fscache.NEWDEPLOY {
+		return executor.ndm.IsValidService(fsvc.Address)
+	} else {
+		return executor.gpm.IsValidPod(fsvc.KubernetesObjects, fsvc.Address)
+	}
+}
+
 func dumpStackTrace() {
 	debug.PrintStack()
 }

executor/fscache/functionServiceCache.go

@@ -247,14 +247,18 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error {
 	return nil
 }
 
+func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
+	fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
+	fsc.byAddress.Delete(fsvc.Address)
+	fsc.byFunctionUID.Delete(fsvc.Function.UID)
+}
+
 func (fsc *FunctionServiceCache) DeleteOld(fsvc *FuncSvc, minAge time.Duration) (bool, error) {
 	if time.Since(fsvc.Atime) < minAge {
 		return false, nil
 	}
 
-	fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
-	fsc.byAddress.Delete(fsvc.Address)
-	fsc.byFunctionUID.Delete(fsvc.Function.UID)
+	fsc.DeleteEntry(fsvc)
 
 	return true, nil
 }

executor/newdeploy/newdeploymgr.go

@@ -22,6 +22,7 @@ import (
 	"log"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -553,3 +554,18 @@ func (deploy *NewDeploy) updateKubeObjRefRV(fsvc *fscache.FuncSvc, objKind strin
 func updateStatus(fn *crd.Function, err error, message string) {
 	log.Printf(message, err)
 }
+
+// IsValidService does a get on the service address to ensure it's a valid service. returns true if it is, else false.
+func (deploy *NewDeploy) IsValidService(svc string) bool {
+	service := strings.Split(svc, ".")
+	if len(service) == 0 {
+		return false
+	}
+	svcObj, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(service[0], metav1.GetOptions{})
+	if err == nil {
+		log.Printf("Valid service address : %s", svcObj.Spec.ClusterIP)
+		return true
+	}
+
+	return false
+}

executor/poolmgr/gpm.go

@@ -21,10 +21,12 @@ import (
 	"log"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/pkg/api"
 	k8sCache "k8s.io/client-go/tools/cache"
 
 	"github.com/fission/fission"
@@ -217,3 +219,18 @@ func (gpm *GenericPoolManager) getEnvPoolsize(env *crd.Environment) int32 {
 	}
 	return poolsize
 }
+
+// IsValidPod checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the
+// containers in it are reporting a ready status for the healthCheck.
+func (gpm *GenericPoolManager) IsValidPod(kubeObjects []api.ObjectReference, podAddress string) bool {
+	for _, obj := range kubeObjects {
+		if obj.Kind == "pod" {
+			pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(obj.Name, metav1.GetOptions{})
+			if err == nil && strings.Contains(podAddress, pod.Status.PodIP) && fission.IsReadyPod(pod) {
+				log.Printf("Valid pod address : %s", podAddress)
+				return true
+			}
+		}
+	}
+	return false
+}