Issue #2258 Reafctor ready pod conroller to use lister and cache sync checks (#2259)

Reactored ready pod controller code to user lister and cache sync checks,
so that we avoid querying lister if cache is not synced in choodPod function.
Also, as noticied in #2258 we were initializing workqueue in goroutine
which was causing nil pointer reference. We have moved it out of goroutine
and kept specific parts in goroutine.

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

pkg/executor/executortype/poolmgr/gp.go

@@ -38,6 +38,7 @@ import (
 	k8sTypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	metricsclient "k8s.io/metrics/pkg/client/clientset/versioned"
@@ -70,7 +71,8 @@ type (
 		fissionClient            *crd.FissionClient
 		fetcherConfig            *fetcherConfig.Config
 		stopReadyPodControllerCh chan struct{}
-		readyPodInformer         cache.SharedIndexInformer
+		readyPodLister           corelisters.PodLister
+		readyPodListerSynced     cache.InformerSynced
 		readyPodQueue            workqueue.DelayingInterface
 		poolInstanceID           string // small random string to uniquify pod names
 		instanceID               string // poolmgr instance id
@@ -145,8 +147,10 @@ func (gp *GenericPool) setup(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-
+	err = gp.setupReadyPodController()
-	go gp.startReadyPodController()
+	if err != nil {
+		return err
+	}
 	go gp.updateCPUUtilizationSvc()
 	return nil
 }
@@ -228,6 +232,10 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin
 	startTime := time.Now()
 	expoDelay := 100 * time.Millisecond
 	logger := otelUtils.LoggerWithTraceID(ctx, gp.logger)
+	if !cache.WaitForCacheSync(ctx.Done(), gp.readyPodListerSynced) {
+		logger.Error("timed out waiting for ready pod lister synced")
+		return "", nil, errors.New("ready pod lister not synced")
+	}
 	for {
 		// Retries took too long, error out.
 		if time.Since(startTime) > gp.podReadyTimeout {
@@ -246,26 +254,25 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin
 		}
 		key = item.(string)
 		logger.Debug("got key from the queue", zap.String("key", key))
-
+		namespace, name, err := cache.SplitMetaNamespaceKey(key)
-		obj, exists, err := gp.readyPodInformer.GetIndexer().GetByKey(key)
+		if err != nil {
+			logger.Error("error splitting key", zap.Error(err), zap.String("key", key))
+			gp.readyPodQueue.Done(key)
+			return "", nil, err
+		}
+		pod, err := gp.readyPodLister.Pods(namespace).Get(name)
 		if err != nil {
 			logger.Error("fetching object from store failed", zap.String("key", key), zap.Error(err))
 			return "", nil, err
 		}
-
+		if !utils.IsReadyPod(pod) {
-		if !exists {
-			logger.Warn("pod deleted from store", zap.String("pod", key))
-			continue
-		}
-
-		if !utils.IsReadyPod(obj.(*apiv1.Pod)) {
 			logger.Warn("pod not ready, pod will be checked again", zap.String("key", key), zap.Duration("delay", expoDelay))
 			gp.readyPodQueue.Done(key)
 			gp.readyPodQueue.AddAfter(key, expoDelay)
 			expoDelay *= 2
 			continue
 		}
-		chosenPod = obj.(*apiv1.Pod).DeepCopy()
+		chosenPod = pod.DeepCopy()
 		otelUtils.SpanTrackEvent(ctx, "foundPod", otelUtils.GetAttributesForPod(chosenPod)...)
 
 		if gp.env.Spec.AllowedFunctionsPerContainer != fv1.AllowedFunctionsPerContainerInfinite {

pkg/executor/executortype/poolmgr/readyPodController.go

@@ -4,46 +4,42 @@ import (
 	"time"
 
 	"go.uber.org/zap"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sCache "k8s.io/client-go/tools/cache"
-	"k8s.io/apimachinery/pkg/labels"
-	informers "k8s.io/client-go/informers/core/v1"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
+
+	"github.com/fission/fission/pkg/utils"
 )
 
-func (gp *GenericPool) newPodInformer() cache.SharedIndexInformer {
+func (gp *GenericPool) readyPodEventHandlers() k8sCache.ResourceEventHandlerFuncs {
-	optionsModifier := func(options *metav1.ListOptions) {
+	return k8sCache.ResourceEventHandlerFuncs{
-		options.LabelSelector = labels.Set(
-			gp.deployment.Spec.Selector.MatchLabels).AsSelector().String()
-		options.FieldSelector = "status.phase=Running"
-	}
-	return informers.NewFilteredPodInformer(gp.kubernetesClient, gp.namespace, 0, nil, optionsModifier)
-}
-
-func (gp *GenericPool) startReadyPodController() {
-	// create the pod watcher to filter by labels
-	// Filtering pod by phase=Running. In some cases the pod can be in
-	// different state than Running, for example Kubernetes sets a
-	// pod to Termination while k8s waits for the grace period of
-	// the pod, even if all the containers are in Ready state.
-	gp.readyPodQueue = workqueue.NewDelayingQueue()
-	gp.readyPodInformer = gp.newPodInformer()
-	gp.readyPodInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
-			key, err := cache.MetaNamespaceKeyFunc(obj)
+			key, err := k8sCache.MetaNamespaceKeyFunc(obj)
 			if err == nil {
 				gp.readyPodQueue.AddAfter(key, 100*time.Millisecond)
 				gp.logger.Debug("add func called", zap.String("key", key))
 			}
 		},
 		DeleteFunc: func(obj interface{}) {
-			key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+			key, err := k8sCache.DeletionHandlingMetaNamespaceKeyFunc(obj)
 			if err == nil {
 				gp.readyPodQueue.Done(key)
 				gp.logger.Debug("delete func called", zap.String("key", key))
 			}
 		},
-	})
+	}
-	go gp.readyPodInformer.Run(gp.stopReadyPodControllerCh)
+}
-	gp.logger.Info("readyPod controller started", zap.String("env", gp.env.ObjectMeta.Name), zap.String("envID", string(gp.env.ObjectMeta.UID)))
+
+func (gp *GenericPool) setupReadyPodController() error {
+	gp.readyPodQueue = workqueue.NewDelayingQueue()
+	informerFactory, err := utils.GetInformerFactoryByReadyPod(gp.kubernetesClient, gp.namespace, gp.deployment.Spec.Selector)
+	if err != nil {
+		return err
+	}
+	podInformer := informerFactory.Core().V1().Pods()
+	gp.readyPodLister = podInformer.Lister()
+	gp.readyPodListerSynced = podInformer.Informer().HasSynced
+	podInformer.Informer().AddEventHandler(gp.readyPodEventHandlers())
+	go podInformer.Informer().Run(gp.stopReadyPodControllerCh)
+	gp.logger.Info("readyPod controller started", zap.String("env", gp.env.ObjectMeta.Name), zap.String("envID", string(gp.env.ObjectMeta.UID)))
+	return nil
 }

pkg/utils/informer.go

@@ -13,6 +13,16 @@ import (
 	v1 "github.com/fission/fission/pkg/apis/core/v1"
 )
 
+func GetInformerFactoryByReadyPod(client *kubernetes.Clientset, namespace string, labelSelector *metav1.LabelSelector) (k8sInformers.SharedInformerFactory, error) {
+	informerFactory := k8sInformers.NewSharedInformerFactoryWithOptions(client, 0,
+		k8sInformers.WithNamespace(namespace),
+		k8sInformers.WithTweakListOptions(func(options *metav1.ListOptions) {
+			options.LabelSelector = labels.Set(labelSelector.MatchLabels).AsSelector().String()
+			options.FieldSelector = "status.phase=Running"
+		}))
+	return informerFactory, nil
+}
+
 func GetInformerFactoryByExecutor(client *kubernetes.Clientset, executorType v1.ExecutorType, defaultResync time.Duration) (k8sInformers.SharedInformerFactory, error) {
 	executorLabel, err := labels.NewRequirement(v1.EXECUTOR_TYPE, selection.DoubleEquals, []string{string(executorType)})
 	if err != nil {