diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 973b33e4..794ef69a 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/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 } - - go gp.startReadyPodController() + err = gp.setupReadyPodController() + 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)) - - obj, exists, err := gp.readyPodInformer.GetIndexer().GetByKey(key) + namespace, name, err := cache.SplitMetaNamespaceKey(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 !exists { - logger.Warn("pod deleted from store", zap.String("pod", key)) - continue - } - - if !utils.IsReadyPod(obj.(*apiv1.Pod)) { + if !utils.IsReadyPod(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 { diff --git a/pkg/executor/executortype/poolmgr/readyPodController.go b/pkg/executor/executortype/poolmgr/readyPodController.go index 00ceca52..f1946d02 100644 --- a/pkg/executor/executortype/poolmgr/readyPodController.go +++ b/pkg/executor/executortype/poolmgr/readyPodController.go @@ -4,46 +4,42 @@ import ( "time" "go.uber.org/zap" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/apimachinery/pkg/labels" - informers "k8s.io/client-go/informers/core/v1" - "k8s.io/client-go/tools/cache" + k8sCache "k8s.io/client-go/tools/cache" "k8s.io/client-go/util/workqueue" + + "github.com/fission/fission/pkg/utils" ) -func (gp *GenericPool) newPodInformer() cache.SharedIndexInformer { - optionsModifier := func(options *metav1.ListOptions) { - 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{ +func (gp *GenericPool) readyPodEventHandlers() k8sCache.ResourceEventHandlerFuncs { + return k8sCache.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 } diff --git a/pkg/utils/informer.go b/pkg/utils/informer.go index 662b72ac..83015fe4 100644 --- a/pkg/utils/informer.go +++ b/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 {