Merge pull request #56 from kubescape/feature/finalizer-watcher

Feature/finalizer watcher

Author: amitschendel

pkg/collector/collector.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/kubescape/kapprofiler/pkg/eventsink"
 	"github.com/kubescape/kapprofiler/pkg/tracing"
+	"github.com/kubescape/kapprofiler/pkg/watcher"
 
 	"golang.org/x/exp/slices"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -64,7 +65,7 @@ type CollectorManager struct {
 	config CollectorManagerConfig
 
 	// Pod finalizer watcher
-	podFinalizerControl chan struct{}
+	podFinalizerWatcher watcher.WatcherInterface
 
 	// Pod finalizer state table
 	podFinalizerState map[string]*PodProfileFinalizerState

pkg/collector/pod_finalizer.go

@@ -8,14 +8,13 @@ import (
 	"sync"
 	"time"
 
+	"github.com/kubescape/kapprofiler/pkg/watcher"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	apitypes "k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/dynamic/dynamicinformer"
-	"k8s.io/client-go/tools/cache"
 )
 
 type PodProfileFinalizerState struct {
@@ -34,33 +33,31 @@ func (cm *CollectorManager) StartFinalizerWatcher() {
 	cm.podFinalizerStateMutex = &sync.Mutex{}
 	// Initialize map
 	cm.podFinalizerState = make(map[string]*PodProfileFinalizerState)
+	// Initialize watcher
+	cm.podFinalizerWatcher = watcher.NewWatcher(cm.dynamicClient, false)
 
-	// Initialize factory and informer
-	factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(cm.dynamicClient, 0, metav1.NamespaceAll, func(lo *metav1.ListOptions) {
-		lo.FieldSelector = "spec.nodeName=" + cm.config.NodeName
-	})
-	// Informer for Pods
-	informer := factory.ForResource(schema.GroupVersionResource{
-		Group:    "",
-		Version:  "v1",
-		Resource: "pods",
-	}).Informer()
-
-	// Add event handlers to informer
-	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
+	// Start watcher
+	err := cm.podFinalizerWatcher.Start(watcher.WatchNotifyFunctions{
+		AddFunc: func(obj *unstructured.Unstructured) {
 			cm.handlePodAddEvent(obj)
 		},
-		UpdateFunc: func(oldObj, newObj interface{}) {
-			cm.handlePodUpdateEvent(oldObj, newObj)
+		UpdateFunc: func(obj *unstructured.Unstructured) {
+			cm.handlePodUpdateEvent(obj)
 		},
-		DeleteFunc: func(obj interface{}) {
+		DeleteFunc: func(obj *unstructured.Unstructured) {
 			cm.handlePodDeleteEvent(obj)
 		},
-	})
+	}, schema.GroupVersionResource{
+		Group:    "",
+		Version:  "v1",
+		Resource: "pods",
+	}, metav1.ListOptions{})
 
-	// Run the informer
-	go informer.Run(cm.podFinalizerControl)
+	if err != nil {
+		log.Printf("Error starting watcher: %v", err)
+		cm.podFinalizerWatcher = nil
+		return
+	}
 }
 
 func generateTableKey(obj metav1.Object) string {
@@ -69,8 +66,6 @@ func generateTableKey(obj metav1.Object) string {
 
 func (cm *CollectorManager) handlePodAddEvent(obj interface{}) {
 	// Add pod to finalizer map
-
-	// Convert object to Pod
 	pod, err := ConvertInterfaceToPod(obj)
 	if err != nil {
 		log.Printf("the interface is not a Pod %v", err)
@@ -103,59 +98,44 @@ func (cm *CollectorManager) handlePodAddEvent(obj interface{}) {
 			cm.startFinalizationTimer(pod)
 		}
 	}
-
 }
 
-func (cm *CollectorManager) handlePodUpdateEvent(oldObj, newObj interface{}) {
-	// Need to access the status of the old and new pod to check if the pod became ready
-
+func (cm *CollectorManager) handlePodUpdateEvent(obj interface{}) {
 	// Convert interface to Pod object
-	oldPod, err := ConvertInterfaceToPod(oldObj)
-	if err != nil {
-		log.Printf("the interface is not a Pod %v", err)
-		return
-	}
-	newPod, err := ConvertInterfaceToPod(newObj)
+	pod, err := ConvertInterfaceToPod(obj)
 	if err != nil {
 		log.Printf("the interface is not a Pod %v", err)
 		return
 	}
 
 	// Check if recoding
 	cm.podFinalizerStateMutex.Lock()
-	finalizerState, ok := cm.podFinalizerState[generateTableKey(oldPod)]
+	finalizerState, ok := cm.podFinalizerState[generateTableKey(&pod.ObjectMeta)]
 	if !ok || !finalizerState.Recording {
 		// Discard
 		cm.podFinalizerStateMutex.Unlock()
 		return
 	}
 	cm.podFinalizerStateMutex.Unlock()
 
-	// Check old pod status
-	oldPodReady := false
-	for _, condition := range oldPod.Status.Conditions {
-		if condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue {
-			oldPodReady = true
-		}
-	}
-
-	newPodReady := false
-	for _, condition := range newPod.Status.Conditions {
+	// Check pod status
+	podReady := false
+	for _, condition := range pod.Status.Conditions {
 		if condition.Type == v1.PodReady && condition.Status == v1.ConditionTrue {
-			newPodReady = true
+			podReady = true
 		}
 	}
 
-	if newPodReady && !oldPodReady {
+	if podReady {
 		// Pod became ready, add finalizer
 		// Get mutex
 		cm.podFinalizerStateMutex.Lock()
 		defer cm.podFinalizerStateMutex.Unlock()
 
 		// Check if pod is in map
-		podState, ok := cm.podFinalizerState[generateTableKey(newPod)]
+		podState, ok := cm.podFinalizerState[generateTableKey(&pod.ObjectMeta)]
 		if !ok {
-			log.Printf("Pod %s in namespace %s not in finalizer map", newPod.GetName(), newPod.GetNamespace())
+			log.Printf("Pod %s in namespace %s not in finalizer map", pod.GetName(), pod.GetNamespace())
 			return
 		}
 
@@ -166,16 +146,16 @@ func (cm *CollectorManager) handlePodUpdateEvent(oldObj, newObj interface{}) {
 		}
 
 		// Timer is not running, add finalizer
-		podState.FinalizationTimer = cm.startFinalizationTimer(newPod)
-	} else if !newPodReady && oldPodReady {
-		cm.stopTimer(&newPod.ObjectMeta)
+		podState.FinalizationTimer = cm.startFinalizationTimer(pod)
+	} else {
+		cm.stopTimer(&pod.ObjectMeta)
 	}
 }
 
 // Timer function
 func (cm *CollectorManager) startFinalizationTimer(pod *v1.Pod) *time.Timer {
-	jitter := time.Duration(rand.Intn(int(cm.config.FinalizeJitter))) * time.Second
-	finalizationTimer := time.NewTimer(time.Duration(cm.config.FinalizeTime+uint64(jitter)) * time.Second)
+	jitter := uint64(rand.Intn(int(cm.config.FinalizeJitter)))
+	finalizationTimer := time.NewTimer(time.Duration(cm.config.FinalizeTime+jitter) * time.Second)
 
 	// This goroutine waits for the timer to finish.
 	go func() {
@@ -303,7 +283,11 @@ func (cm *CollectorManager) MarkPodNotRecording(pod, namespace string) {
 }
 
 func (cm *CollectorManager) StopFinalizerWatcher() {
-	close(cm.podFinalizerControl)
+	if cm.podFinalizerWatcher != nil {
+		cm.podFinalizerWatcher.Stop()
+	} else {
+		log.Printf("Pod finalizer watcher not started")
+	}
 }
 
 func ConvertInterfaceToPod(obj interface{}) (*v1.Pod, error) {