From ee623d31b2585f8507d3507ad05148ee8c39bc8d Mon Sep 17 00:00:00 2001 From: Shubham Bansal <62992590+shubham-bansal96@users.noreply.github.com> Date: Mon, 5 Dec 2022 13:06:47 +0530 Subject: [PATCH] Check pods events via infomer in user configured namespaces (#2653) * informer changes for event checker in multi namespace * run informers in wait group --- pkg/executor/executortype/poolmgr/gpm.go | 138 ++++++++++------------- pkg/utils/informer.go | 27 +++++ 2 files changed, 84 insertions(+), 81 deletions(-) diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index 3cc48157..1267ed0b 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -33,10 +33,8 @@ import ( k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" - "k8s.io/apimachinery/pkg/runtime" k8sTypes "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/wait" - "k8s.io/apimachinery/pkg/watch" k8sInformers "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" corelisters "k8s.io/client-go/listers/core/v1" @@ -171,10 +169,12 @@ func MakeGenericPoolManager(ctx context.Context, } func (gpm *GenericPoolManager) Run(ctx context.Context) { + waitSynced := make([]k8sCache.InformerSynced, 0) for _, podListerSynced := range gpm.podListerSynced { - if ok := k8sCache.WaitForCacheSync(ctx.Done(), podListerSynced); !ok { - gpm.logger.Fatal("failed to wait for caches to sync") - } + waitSynced = append(waitSynced, podListerSynced) + } + if ok := k8sCache.WaitForCacheSync(ctx.Done(), waitSynced...); !ok { + gpm.logger.Fatal("failed to wait for caches to sync") } go gpm.service() gpm.poolPodC.InjectGpm(gpm) @@ -680,93 +680,69 @@ func (gpm *GenericPoolManager) doIdleObjectReaper(ctx context.Context) { // WebsocketStartEventChecker checks if the pod has emitted a websocket connection start event func (gpm *GenericPoolManager) WebsocketStartEventChecker(ctx context.Context, kubeClient kubernetes.Interface) { - - informer := k8sCache.NewSharedInformer( - &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(ctx, options) - }, - WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=WsConnectionStarted" - return kubeClient.CoreV1().Events(apiv1.NamespaceAll).Watch(ctx, options) - }, - }, - &apiv1.Event{}, - 0, - ) - stopper := make(chan struct{}) defer close(stopper) - informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - mObj := obj.(metav1.Object) - gpm.logger.Info("Websocket event detected for pod", - zap.String("Pod name", mObj.GetName())) - podName := strings.SplitAfter(mObj.GetName(), ".") - if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok { - fsvc, ok := fsvc.(*fscache.FuncSvc) - if !ok { - gpm.logger.Error("could not convert item from PodToFsvc") - return + var wg wait.Group + for _, informer := range utils.GetInformerEventChecker(ctx, kubeClient, "WsConnectionStarted") { + informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + mObj := obj.(metav1.Object) + gpm.logger.Info("Websocket event detected for pod", + zap.String("Pod name", mObj.GetName())) + + podName := strings.SplitAfter(mObj.GetName(), ".") + if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok { + fsvc, ok := fsvc.(*fscache.FuncSvc) + if !ok { + gpm.logger.Error("could not convert item from PodToFsvc") + return + } + gpm.fsCache.WebsocketFsvc.Store(fsvc.Name, true) } - gpm.fsCache.WebsocketFsvc.Store(fsvc.Name, true) - } - }, - }) - informer.Run(stopper) - + }, + }) + wg.StartWithChannel(stopper, informer.Run) + } + wg.Wait() } // NoActiveConnectionEventChecker checks if the pod has emitted an inactive event func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(ctx context.Context, kubeClient kubernetes.Interface) { - - informer := k8sCache.NewSharedInformer( - &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(ctx, options) - }, - WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { - options.FieldSelector = "involvedObject.kind=Pod,type=Normal,reason=NoActiveConnections" - return kubeClient.CoreV1().Events(apiv1.NamespaceAll).Watch(ctx, options) - }, - }, - &apiv1.Event{}, - 0, - ) - stopper := make(chan struct{}) defer close(stopper) - informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - mObj := obj.(metav1.Object) - gpm.logger.Info("Inactive event detected for pod", - zap.String("Pod name", mObj.GetName())) - podName := strings.SplitAfter(mObj.GetName(), ".") - if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok { - fsvc, ok := fsvc.(*fscache.FuncSvc) - if !ok { - gpm.logger.Error("could not convert value from PodToFsvc") - return + var wg wait.Group + for _, informer := range utils.GetInformerEventChecker(ctx, kubeClient, "WsConnectionStarted") { + informer.AddEventHandler(k8sCache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + mObj := obj.(metav1.Object) + gpm.logger.Info("Inactive event detected for pod", + zap.String("Pod name", mObj.GetName())) + + podName := strings.SplitAfter(mObj.GetName(), ".") + if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok { + fsvc, ok := fsvc.(*fscache.FuncSvc) + if !ok { + gpm.logger.Error("could not convert value from PodToFsvc") + return + } + gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) + for i := range fsvc.KubernetesObjects { + gpm.logger.Info("release idle function resources due to inactivity", + zap.String("function", fsvc.Function.Name), + zap.String("address", fsvc.Address), + zap.String("executor", string(fsvc.Executor)), + zap.String("pod", fsvc.Name), + ) + reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) + time.Sleep(50 * time.Millisecond) + } } - gpm.fsCache.DeleteFunctionSvc(ctx, fsvc) - for i := range fsvc.KubernetesObjects { - gpm.logger.Info("release idle function resources due to inactivity", - zap.String("function", fsvc.Function.Name), - zap.String("address", fsvc.Address), - zap.String("executor", string(fsvc.Executor)), - zap.String("pod", fsvc.Name), - ) - reaper.CleanupKubeObject(ctx, gpm.logger, gpm.kubernetesClient, &fsvc.KubernetesObjects[i]) - time.Sleep(50 * time.Millisecond) - } - } - - }, - }) - informer.Run(stopper) + }, + }) + wg.StartWithChannel(stopper, informer.Run) + } + wg.Wait() } diff --git a/pkg/utils/informer.go b/pkg/utils/informer.go index 1f8097ea..c9f27cf5 100644 --- a/pkg/utils/informer.go +++ b/pkg/utils/informer.go @@ -1,11 +1,16 @@ package utils import ( + "context" + "fmt" "time" + apiv1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/selection" + "k8s.io/apimachinery/pkg/watch" k8sInformers "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" "k8s.io/client-go/tools/cache" @@ -69,6 +74,28 @@ func GetK8sInformersForNamespaces(client kubernetes.Interface, defaultSync time. return informers } +func GetInformerEventChecker(ctx context.Context, client kubernetes.Interface, reason string) map[string]cache.SharedInformer { + informers := make(map[string]cache.SharedInformer) + namespaces := DefaultNSResolver() + for _, ns := range namespaces.FissionNSWithOptions(WithBuilderNs(), WithFunctionNs(), WithDefaultNs()) { + informers[ns] = cache.NewSharedInformer( + &cache.ListWatch{ + ListFunc: func(options metav1.ListOptions) (runtime.Object, error) { + options.FieldSelector = fmt.Sprintf("involvedObject.kind=Pod,type=Normal,reason=%s", reason) + return client.CoreV1().Events(ns).List(ctx, options) + }, + WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) { + options.FieldSelector = fmt.Sprintf("involvedObject.kind=Pod,type=Normal,reason=%s", reason) + return client.CoreV1().Events(ns).Watch(ctx, options) + }, + }, + &apiv1.Event{}, + 0, + ) + } + return informers +} + func GetInformerFactoryByExecutor(client kubernetes.Interface, labels labels.Selector, defaultResync time.Duration) map[string]k8sInformers.SharedInformerFactory { informerFactory := make(map[string]k8sInformers.SharedInformerFactory)