Add a SharedIndexInformer for services and deployments to NewDeploy executor (#2061)

* Add a SharedIndexInformer for services and deployments to NewDeploy executor.

This brings the NewDeploy executor behaviour into sync with GenericPoolManager behaviour
by caching Kubernetes services and deployments used in per-request function validation.

* Create informers in executer by executor label

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

Co-authored-by: James Sinclair <irgeek@btc.com.au>

pkg/executor/executor.go

@@ -257,15 +257,21 @@ func StartExecutor(logger *zap.Logger, functionNamespace string, envBuilderNames
 
 	logger.Info("Starting executor", zap.String("instanceID", executorInstanceID))
 
-	gpm := poolmgr.MakeGenericPoolManager(
+	gpm, err := poolmgr.MakeGenericPoolManager(
 		logger,
 		fissionClient, kubernetesClient, metricsClient,
 		functionNamespace, fetcherConfig, executorInstanceID)
+	if err != nil {
+		return errors.Wrap(err, "pool manager creation faied")
+	}
 
-	ndm := newdeploy.MakeNewDeploy(
+	ndm, err := newdeploy.MakeNewDeploy(
 		logger,
 		fissionClient, kubernetesClient, fissionClient.CoreV1().RESTClient(),
 		functionNamespace, fetcherConfig, executorInstanceID)
+	if err != nil {
+		return errors.Wrap(err, "new deploy manager creation faied")
+	}
 
 	executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType)
 	executorTypes[gpm.GetTypeName()] = gpm

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -28,6 +28,7 @@ import (
 	multierror "github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
+	appsv1 "k8s.io/api/apps/v1"
 	autoscalingv1 "k8s.io/api/autoscaling/v1"
 	apiv1 "k8s.io/api/core/v1"
 	k8sErrs "k8s.io/apimachinery/pkg/api/errors"
@@ -72,8 +73,10 @@ type (
 		funcStore      k8sCache.Store
 		funcController k8sCache.Controller
 
-		envStore      k8sCache.Store
+		envStore           k8sCache.Store
-		envController k8sCache.Controller
+		envController      k8sCache.Controller
+		serviceInformer    k8sCache.SharedIndexInformer
+		deploymentInformer k8sCache.SharedIndexInformer
 
 		defaultIdlePodReapTime time.Duration
 	}
@@ -88,7 +91,7 @@ func MakeNewDeploy(
 	namespace string,
 	fetcherConfig *fetcherConfig.Config,
 	instanceID string,
-) executortype.ExecutorType {
+) (executortype.ExecutorType, error) {
 	enableIstio := false
 	if len(os.Getenv("ENABLE_ISTIO")) > 0 {
 		istio, err := strconv.ParseBool(os.Getenv("ENABLE_ISTIO"))
@@ -127,13 +130,21 @@ func MakeNewDeploy(
 		nd.envController = envController
 	}
 
-	return nd
+	informerFactory, err := utils.GetInformerFacoryByExecutor(nd.kubernetesClient, fv1.ExecutorTypePoolmgr)
+	if err != nil {
+		return nil, err
+	}
+	nd.serviceInformer = informerFactory.Core().V1().Services().Informer()
+	nd.deploymentInformer = informerFactory.Apps().V1().Deployments().Informer()
+	return nd, nil
 }
 
 // Run start the function and environment controller along with an object reaper.
 func (deploy *NewDeploy) Run(ctx context.Context) {
 	go deploy.funcController.Run(ctx.Done())
 	go deploy.envController.Run(ctx.Done())
+	go deploy.serviceInformer.Run(ctx.Done())
+	go deploy.deploymentInformer.Run(ctx.Done())
 	go deploy.idleObjectReaper()
 }
 
@@ -180,41 +191,86 @@ func (deploy *NewDeploy) TapService(svcHost string) error {
 	return nil
 }
 
+func getCachedItem(obj apiv1.ObjectReference, informer k8sCache.SharedIndexInformer) (item interface{}, exists bool, err error) {
+	store := informer.GetStore()
+
+	item, exists, err = store.Get(obj)
+	if err != nil || !exists {
+		item, exists, err = store.GetByKey(fmt.Sprintf("%s/%s", obj.Namespace, obj.Name))
+	}
+
+	return item, exists, err
+}
+
+func (deploy *NewDeploy) getServiceInfo(obj apiv1.ObjectReference) (*apiv1.Service, error) {
+	item, exists, err := getCachedItem(obj, deploy.serviceInformer)
+
+	if err != nil || !exists {
+		deploy.logger.Debug(
+			"Falling back to getting service info from k8s API -- this may cause performace issues for your function.",
+			zap.Bool("exists", exists),
+			zap.Error(err),
+		)
+		service, err := deploy.kubernetesClient.CoreV1().Services(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
+		return service, err
+	}
+
+	service := item.(*apiv1.Service)
+	return service, nil
+}
+
+func (deploy *NewDeploy) getDeploymentInfo(obj apiv1.ObjectReference) (*appsv1.Deployment, error) {
+	item, exists, err := getCachedItem(obj, deploy.deploymentInformer)
+
+	if err != nil || !exists {
+		deploy.logger.Debug(
+			"Falling back to getting deployment info from k8s API -- this may cause performace issues for your function.",
+			zap.Bool("exists", exists),
+			zap.Error(err),
+		)
+		deployment, err := deploy.kubernetesClient.AppsV1().Deployments(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
+		return deployment, err
+	}
+
+	deployment := item.(*appsv1.Deployment)
+	return deployment, nil
+}
+
 // IsValid does a get on the service address to ensure it's a valid service, then
 // scale deployment to 1 replica if there are no available replicas for function.
 // Return true if no error occurs, return false otherwise.
 func (deploy *NewDeploy) IsValid(fsvc *fscache.FuncSvc) bool {
-	service := strings.Split(fsvc.Address, ".")
+	if len(strings.Split(fsvc.Address, ".")) == 0 {
-	if len(service) == 0 {
 		return false
 	}
 
-	_, err := deploy.kubernetesClient.CoreV1().Services(service[1]).Get(context.TODO(), service[0], metav1.GetOptions{})
+	for _, obj := range fsvc.KubernetesObjects {
-	if err != nil {
+		if strings.ToLower(obj.Kind) == "service" {
-		if !k8sErrs.IsNotFound(err) {
+			_, err := deploy.getServiceInfo(obj)
-			deploy.logger.Error("error validating function service address", zap.String("function", fsvc.Function.Name), zap.Error(err))
+			if err != nil {
+				if !k8sErrs.IsNotFound(err) {
+					deploy.logger.Error("error validating function service", zap.String("function", fsvc.Function.Name), zap.Error(err))
+				}
+				return false
+			}
+
 		}
-		return false
 	}
 
-	deployObj := getDeploymentObj(fsvc.KubernetesObjects)
+	for _, obj := range fsvc.KubernetesObjects {
-	if deployObj == nil {
+		if strings.ToLower(obj.Kind) == "deployment" {
-		deploy.logger.Error("deployment obj for function does not exist", zap.String("function", fsvc.Function.Name))
+			currentDeploy, err := deploy.getDeploymentInfo(obj)
-		return false
+			if err != nil {
-	}
+				if !k8sErrs.IsNotFound(err) {
-
+					deploy.logger.Error("error validating function deployment", zap.String("function", fsvc.Function.Name), zap.Error(err))
-	currentDeploy, err := deploy.kubernetesClient.AppsV1().
+				}
-		Deployments(deployObj.Namespace).Get(context.TODO(), deployObj.Name, metav1.GetOptions{})
+				return false
-	if err != nil {
+			}
-		if !k8sErrs.IsNotFound(err) {
+			// return directly when available replicas > 0
-			deploy.logger.Error("error validating function deployment", zap.Error(err), zap.String("function", fsvc.Function.Name))
+			if currentDeploy.Status.AvailableReplicas > 0 {
+				return true
+			}
 		}
-		return false
-	}
-
-	// return directly when available replicas > 0
-	if currentDeploy.Status.AvailableReplicas > 0 {
-		return true
 	}
 
 	return false

pkg/executor/executortype/poolmgr/gp.go

@@ -36,7 +36,6 @@ import (
 	k8sErrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -174,7 +173,7 @@ func (gp *GenericPool) getDeployAnnotations() map[string]string {
 
 func (gp *GenericPool) updateCPUUtilizationSvc() {
 	for {
-		podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.namespace).List(context.TODO(), v1.ListOptions{
+		podMetricsList, err := gp.metricsClient.MetricsV1beta1().PodMetricses(gp.namespace).List(context.TODO(), metav1.ListOptions{
 			LabelSelector: "managed=false",
 		})
 

pkg/executor/executortype/poolmgr/gpm.go

@@ -35,10 +35,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
-	k8sInformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	k8sCache "k8s.io/client-go/tools/cache"
-	k8scache "k8s.io/client-go/tools/cache"
 	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
@@ -105,7 +103,7 @@ func MakeGenericPoolManager(
 	metricsClient *metricsclient.Clientset,
 	functionNamespace string,
 	fetcherConfig *fetcherConfig.Config,
-	instanceID string) executortype.ExecutorType {
+	instanceID string) (executortype.ExecutorType, error) {
 
 	gpmLogger := logger.Named("generic_pool_manager")
 
@@ -139,10 +137,12 @@ func MakeGenericPoolManager(
 
 	gpm.pkgStore, gpm.pkgController = gpm.makePkgController(gpm.fissionClient, gpm.kubernetesClient, gpm.namespace)
 
-	informerFactory := k8sInformers.NewSharedInformerFactoryWithOptions(kubernetesClient, 0, k8sInformers.WithNamespace(gpm.namespace))
+	informerFactory, err := utils.GetInformerFacoryByExecutor(gpm.kubernetesClient, fv1.ExecutorTypePoolmgr)
+	if err != nil {
+		return nil, err
+	}
 	gpm.podInformer = informerFactory.Core().V1().Pods().Informer()
-
+	return gpm, nil
-	return gpm
 }
 
 func (gpm *GenericPoolManager) Run(ctx context.Context) {
@@ -210,7 +210,7 @@ func (gpm *GenericPoolManager) getPodInfo(obj apiv1.ObjectReference) (*apiv1.Pod
 		item, exists, err = store.GetByKey(fmt.Sprintf("%s/%s", obj.Namespace, obj.Name))
 	}
 
-	if !exists {
+	if err != nil || !exists {
 		gpm.logger.Debug("Falling back to getting pod info from k8s API -- this may cause performace issues for your function.")
 		pod, err := gpm.kubernetesClient.CoreV1().Pods(obj.Namespace).Get(context.TODO(), obj.Name, metav1.GetOptions{})
 		return pod, err
@@ -701,8 +701,8 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 // WebsocketStartEventChecker checks if the pod has emitted a websocket connection start event
 func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes.Clientset) {
 
-	informer := k8scache.NewSharedInformer(
+	informer := k8sCache.NewSharedInformer(
-		&k8scache.ListWatch{
+		&k8sCache.ListWatch{
 			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 				options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=WsConnectionStarted"
 				return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(context.TODO(), options)
@@ -718,7 +718,7 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes
 
 	stopper := make(chan struct{})
 	defer close(stopper)
-	informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{
+	informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			mObj := obj.(metav1.Object)
 			gpm.logger.Info("Websocket event detected for pod",
@@ -737,8 +737,8 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes
 // NoActiveConnectionEventChecker checks if the pod has emitted an inactive event
 func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubernetes.Clientset) {
 
-	informer := k8scache.NewSharedInformer(
+	informer := k8sCache.NewSharedInformer(
-		&k8scache.ListWatch{
+		&k8sCache.ListWatch{
 			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
 				options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=NoActiveConnections"
 				return kubeClient.CoreV1().Events(apiv1.NamespaceAll).List(context.TODO(), options)
@@ -754,7 +754,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern
 
 	stopper := make(chan struct{})
 	defer close(stopper)
-	informer.AddEventHandler(k8scache.ResourceEventHandlerFuncs{
+	informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			mObj := obj.(metav1.Object)
 			gpm.logger.Info("Inactive event detected for pod",

pkg/poolcache/poolcache.go

@@ -30,7 +30,6 @@ type requestType int
 const (
 	getValue requestType = iota
 	listAvailableValue
-	getTotalAvailable
 	setValue
 	markAvailable
 	deleteValue

pkg/utils/informer.go

@@ -0,0 +1,24 @@
+package utils
+
+import (
+	v1 "github.com/fission/fission/pkg/apis/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/selection"
+	k8sInformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+)
+
+func GetInformerFacoryByExecutor(client *kubernetes.Clientset, executorType v1.ExecutorType) (k8sInformers.SharedInformerFactory, error) {
+	executorLabel, err := labels.NewRequirement(v1.EXECUTOR_TYPE, selection.DoubleEquals, []string{string(executorType)})
+	if err != nil {
+		return nil, err
+	}
+	labelSelector := labels.NewSelector()
+	labelSelector.Add(*executorLabel)
+	informerFactory := k8sInformers.NewSharedInformerFactoryWithOptions(client, 0,
+		k8sInformers.WithTweakListOptions(func(options *metav1.ListOptions) {
+			options.LabelSelector = labelSelector.String()
+		}))
+	return informerFactory, nil
+}