feat(node-drainer): replace event informer in node-drainer with eventrecorder

Signed-off-by: Ajay Mishra <ajmishra@nvidia.com>

Author: XRFXLP

distros/kubernetes/nvsentinel/charts/node-drainer/templates/clusterrole.yaml

@@ -53,10 +53,8 @@ rules:
   resources:
   - events
   verbs:
-  - list
   - create
-  - update
-  - watch
+  - patch
 - apiGroups:
   - ""
   resources:

docs/configuration/node-drainer.md

@@ -254,7 +254,7 @@ kubectl get events -n default \
   --field-selector "involvedObject.kind=Node,involvedObject.name=$NODE"
 ```
 
-Node Drainer events are created in the **`default`** namespace (`Type=NodeDraining`,
+Node Drainer events are created in the **`default`** namespace (`Type=Normal`,
 `Source=nvsentinel-node-drainer`). Example output for a pod `training-job-0` in namespace
 `batch-jobs` on node `gpu-node-42`:
 
@@ -283,7 +283,7 @@ database    postgres-0        1/1     Running   0          2h
 
 # kubectl get events -n default --field-selector "involvedObject.name=gpu-node-42"
 LAST SEEN   TYPE          REASON                  OBJECT           MESSAGE
-2m          NodeDraining  AwaitingPodCompletion   Node/gpu-node-42 Waiting for following pods to finish: [database/postgres-0]
+2m          Normal        AwaitingPodCompletion   Node/gpu-node-42 Waiting for following pods to finish: [database/postgres-0]
 ```
 
 #### `DeleteAfterTimeout`
@@ -295,7 +295,7 @@ batch-jobs  training-job-0    1/1     Running   0          2h
 
 # kubectl get events -n default --field-selector "involvedObject.name=gpu-node-42"
 LAST SEEN   TYPE          REASON                    OBJECT           MESSAGE
-1m          NodeDraining  WaitingBeforeForceDelete  Node/gpu-node-42 Waiting for following pods to finish: [training-job-0] in namespace: [batch-jobs] or they will be force deleted on: 2026-07-01 18:30:00 +0000 UTC
+1m          Normal        WaitingBeforeForceDelete  Node/gpu-node-42 Waiting for following pods to finish: [training-job-0] in namespace: [batch-jobs] or they will be force deleted on: 2026-07-01 18:30:00 +0000 UTC
 ```
 
 After the deadline, the pod is force-deleted and `kubectl get pods` shows it gone.

node-drainer/pkg/informers/informers.go

@@ -34,7 +34,10 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
 
 	"github.com/nvidia/nvsentinel/data-models/pkg/model"
@@ -44,20 +47,19 @@ import (
 )
 
 const (
-	NodeIndex            = "node"
-	NamespaceNodeIndex   = "namespace-node"
-	NodeEventReasonIndex = "node-event-reason"
+	NodeIndex          = "node"
+	NamespaceNodeIndex = "namespace-node"
 )
 
 type Informers struct {
 	podInformer            cache.SharedIndexInformer
-	eventInformer          cache.SharedIndexInformer
 	nodeInformer           cache.SharedIndexInformer
+	eventBroadcaster       record.EventBroadcaster
+	eventRecorder          record.EventRecorder
 	clientset              kubernetes.Interface
 	notReadyTimeoutMinutes *int
 	drainGPUPods           bool
 	dryRunMode             []string
-	namespace              string
 }
 
 func NewInformers(clientset kubernetes.Interface, resyncPeriod time.Duration,
@@ -87,41 +89,30 @@ func NewInformers(clientset kubernetes.Interface, resyncPeriod time.Duration,
 		return nil, fmt.Errorf("failed to add indexer: %w", err)
 	}
 
-	eventInformerFactory := informers.NewSharedInformerFactoryWithOptions(
-		clientset,
-		resyncPeriod,
-		informers.WithNamespace(metav1.NamespaceDefault),
-	)
-
-	eventInformer := eventInformerFactory.Core().V1().Events().Informer()
-
-	err = eventInformer.GetIndexer().AddIndexers(
-		cache.Indexers{
-			NodeEventReasonIndex: NodeEventReasonIndexFunc,
-		})
-	if err != nil {
-		return nil, fmt.Errorf("failed to add event indexer: %w", err)
-	}
-
 	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
 	if err := nodeInformer.SetTransform(nodeTransform); err != nil {
 		return nil, fmt.Errorf("failed to set node informer transform: %w", err)
 	}
 
+	eventBroadcaster := record.NewBroadcaster()
+
 	dryRunMode := []string{}
 	if dryRun {
 		dryRunMode = []string{metav1.DryRunAll}
 	}
 
 	return &Informers{
-		clientset:              clientset,
-		podInformer:            podInformer,
-		eventInformer:          eventInformer,
-		nodeInformer:           nodeInformer,
+		clientset:        clientset,
+		podInformer:      podInformer,
+		nodeInformer:     nodeInformer,
+		eventBroadcaster: eventBroadcaster,
+		eventRecorder: eventBroadcaster.NewRecorder(
+			scheme.Scheme,
+			v1.EventSource{Component: "nvsentinel-node-drainer"},
+		),
 		notReadyTimeoutMinutes: notReadyTimeoutMinutes,
 		drainGPUPods:           drainGPUPods,
 		dryRunMode:             dryRunMode,
-		namespace:              metav1.NamespaceDefault,
 	}, nil
 }
 
@@ -273,7 +264,7 @@ func isDaemonSetOwned(ownerReferences []metav1.OwnerReference) bool {
 }
 
 func (i *Informers) HasSynced() bool {
-	return i.podInformer.HasSynced() && i.eventInformer.HasSynced() && i.nodeInformer.HasSynced()
+	return i.podInformer.HasSynced() && i.nodeInformer.HasSynced()
 }
 
 func NodeIndexFunc(obj any) ([]string, error) {
@@ -304,25 +295,17 @@ func NamespaceNodeIndexFunc(obj any) ([]string, error) {
 	return []string{compositeKey}, nil
 }
 
-func NodeEventReasonIndexFunc(obj any) ([]string, error) {
-	event, ok := obj.(*v1.Event)
-	if !ok {
-		return []string{}, nil
-	}
-
-	if event.InvolvedObject.Kind != "Node" {
-		return []string{}, nil
-	}
-
-	compositeKey := fmt.Sprintf("%s/%s", event.InvolvedObject.Name, event.Reason)
-
-	return []string{compositeKey}, nil
-}
-
 func (i *Informers) Run(ctx context.Context) error {
+	i.eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{
+		Interface: i.clientset.CoreV1().Events(metav1.NamespaceDefault),
+	})
+
 	go i.podInformer.Run(ctx.Done())
-	go i.eventInformer.Run(ctx.Done())
 	go i.nodeInformer.Run(ctx.Done())
+	go func() {
+		<-ctx.Done()
+		i.eventBroadcaster.Shutdown()
+	}()
 
 	if ok := cache.WaitForCacheSync(ctx.Done(),
 		i.HasSynced); !ok {
@@ -714,71 +697,13 @@ func (i *Informers) sendEvictionRequestForPod(ctx context.Context, namespace str
 	return nil
 }
 
-func (i *Informers) UpdateNodeEvent(ctx context.Context, nodeName string, reason string, message string) error {
-	compositeKey := fmt.Sprintf("%s/%s", nodeName, reason)
-
-	cachedEvents, err := i.eventInformer.GetIndexer().ByIndex(NodeEventReasonIndex, compositeKey)
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to query event cache", "error", err)
-		return fmt.Errorf("error querying event cache: %w", err)
-	}
-
-	now := metav1.NewTime(time.Now())
-	eventsClient := i.clientset.CoreV1().Events(i.namespace)
-
-	for _, obj := range cachedEvents {
-		existingEvent, ok := obj.(*v1.Event)
-		if !ok {
-			continue
-		}
-
-		if existingEvent.Message == message {
-			eventCopy := existingEvent.DeepCopy()
-			eventCopy.Count++
-			eventCopy.LastTimestamp = now
-
-			_, err = eventsClient.Update(ctx, eventCopy, metav1.UpdateOptions{})
-			if err != nil {
-				slog.ErrorContext(ctx, "Failed to update event occurrence count", "error", err)
-				return fmt.Errorf("error in updating event occurrence count: %w", err)
-			}
-
-			return nil
-		}
-	}
-
-	// Get node from informer cache to retrieve its UID for proper event association
+func (i *Informers) UpdateNodeEvent(_ context.Context, nodeName string, reason string, message string) error {
 	node, err := i.GetNode(nodeName)
 	if err != nil {
 		return fmt.Errorf("error getting node from cache %s: %w", nodeName, err)
 	}
 
-	newEvent := &v1.Event{
-		ObjectMeta: metav1.ObjectMeta{
-			GenerateName: nodeName + "-",
-			Namespace:    i.namespace,
-		},
-		InvolvedObject: v1.ObjectReference{
-			Kind:       "Node",
-			Name:       nodeName,
-			UID:        node.UID,
-			APIVersion: "v1",
-		},
-		Reason:         reason,
-		Message:        message,
-		Type:           "NodeDraining",
-		Source:         v1.EventSource{Component: "nvsentinel-node-drainer"},
-		FirstTimestamp: now,
-		LastTimestamp:  now,
-		Count:          1,
-	}
-
-	_, err = eventsClient.Create(ctx, newEvent, metav1.CreateOptions{})
-	if err != nil {
-		slog.ErrorContext(ctx, "Failed to create event", "error", err, "node", nodeName, "reason", reason)
-		return fmt.Errorf("error in creating event: %w", err)
-	}
-
+	i.eventRecorder.Event(node, v1.EventTypeNormal, reason, message)
 	return nil
 }
 

node-drainer/pkg/informers/informers_transform_test.go

@@ -245,10 +245,19 @@ func TestInformerTransformsIntegrateWithIndexesAndNodeEvents(t *testing.T) {
 	assert.Equal(t, node.Annotations, cachedNode.Annotations)
 
 	require.NoError(t, informers.UpdateNodeEvent(ctx, node.Name, "AwaitingPodCompletion", "waiting"))
-	events, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{})
-	require.NoError(t, err)
-	require.Len(t, events.Items, 1)
-	assert.Equal(t, node.UID, events.Items[0].InvolvedObject.UID)
+	require.NoError(t, informers.UpdateNodeEvent(ctx, node.Name, "AwaitingPodCompletion", "waiting"))
+	require.EventuallyWithT(t, func(collect *assert.CollectT) {
+		events, listErr := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{})
+		if !assert.NoError(collect, listErr) || !assert.Len(collect, events.Items, 1) {
+			return
+		}
+
+		event := events.Items[0]
+		assert.Equal(collect, node.UID, event.InvolvedObject.UID)
+		assert.Equal(collect, v1.EventTypeNormal, event.Type)
+		assert.Equal(collect, "nvsentinel-node-drainer", event.Source.Component)
+		assert.Equal(collect, int32(2), event.Count)
+	}, 5*time.Second, 50*time.Millisecond)
 }
 
 func richDrainEligiblePod(namespace, name, nodeName string) *v1.Pod {

node-drainer/pkg/reconciler/reconciler_integration_test.go

@@ -504,9 +504,16 @@ func TestReconciler_ProcessEvent(t *testing.T) {
 				assert.Error(t, err)
 				assert.Contains(t, err.Error(), "waiting for pods to complete")
 
-				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)})
+				eventListOptions := metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)}
+				require.Eventually(t, func() bool {
+					nodeEvents, listErr := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
+					return listErr == nil && len(nodeEvents.Items) == 1
+				}, 5*time.Second, 50*time.Millisecond, "node event should be recorded asynchronously")
+
+				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
 				require.NoError(t, err)
 				require.Len(t, nodeEvents.Items, 1, "only one event should be created despite multiple reconciliations")
+				require.Equal(t, v1.EventTypeNormal, nodeEvents.Items[0].Type)
 				require.Equal(t, nodeEvents.Items[0].Reason, "AwaitingPodCompletion")
 				expectedMessage := "Waiting for following pods to finish: [completion-test/running-pod-1]"
 				require.Equal(t, expectedMessage, nodeEvents.Items[0].Message, "only expected pods should be drained")
@@ -661,9 +668,16 @@ func TestReconciler_ProcessEvent(t *testing.T) {
 				assert.Error(t, err)
 				assert.Contains(t, err.Error(), "failed timeout eviction for node test-node: waiting for 1 pods to complete or timeout")
 
-				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)})
+				eventListOptions := metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)}
+				require.Eventually(t, func() bool {
+					nodeEvents, listErr := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
+					return listErr == nil && len(nodeEvents.Items) == 1
+				}, 5*time.Second, 50*time.Millisecond, "node event should be recorded asynchronously")
+
+				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
 				require.NoError(t, err)
 				require.Len(t, nodeEvents.Items, 1, "only one event should be created despite multiple reconciliations")
+				require.Equal(t, v1.EventTypeNormal, nodeEvents.Items[0].Type)
 				require.Equal(t, nodeEvents.Items[0].Reason, "WaitingBeforeForceDelete")
 				expectedMessage := "Waiting for following pods to finish: [pod-1] in namespace: [timeout-test] or they will be force deleted on:"
 				require.Contains(t, nodeEvents.Items[0].Message, expectedMessage, "only expected pods should be drained")
@@ -892,9 +906,16 @@ func TestReconciler_ProcessEvent(t *testing.T) {
 				assert.Error(t, err)
 				assert.Contains(t, err.Error(), "waiting for pods to complete")
 
-				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)})
+				eventListOptions := metav1.ListOptions{FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Node", nodeName)}
+				require.Eventually(t, func() bool {
+					nodeEvents, listErr := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
+					return listErr == nil && len(nodeEvents.Items) == 1
+				}, 5*time.Second, 50*time.Millisecond, "node event should be recorded asynchronously")
+
+				nodeEvents, err := client.CoreV1().Events(metav1.NamespaceDefault).List(ctx, eventListOptions)
 				require.NoError(t, err)
 				require.Len(t, nodeEvents.Items, 1, "only one event should be created despite multiple reconciliations")
+				require.Equal(t, v1.EventTypeNormal, nodeEvents.Items[0].Type)
 				require.Equal(t, nodeEvents.Items[0].Reason, "AwaitingPodCompletion")
 				expectedMessage := "Waiting for following pods to finish: [completion-test/running-pod-1 completion-test/running-pod-2 completion-test/running-pod-3]"
 				require.Equal(t, expectedMessage, nodeEvents.Items[0].Message, "pod list should be in sorted order")

tests/fault_management_test.go

@@ -346,7 +346,7 @@ func TestNodeRecoveryDuringDrain(t *testing.T) {
 		require.NoError(t, err)
 
 		require.Eventually(t, func() bool {
-			found, _ := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, "NodeDraining", "WaitingBeforeForceDelete", time.Time{})
+			found, _ := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, v1.EventTypeNormal, "WaitingBeforeForceDelete", time.Time{})
 			return found
 		}, helpers.EventuallyWaitTimeout, helpers.WaitInterval, "WaitingBeforeForceDelete event should be created")
 
@@ -446,7 +446,7 @@ func TestManualUncordonPropagation(t *testing.T) {
 		require.NoError(t, err)
 
 		require.Eventually(t, func() bool {
-			found, _ := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, "NodeDraining", "WaitingBeforeForceDelete", time.Time{})
+			found, _ := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, v1.EventTypeNormal, "WaitingBeforeForceDelete", time.Time{})
 			return found
 		}, helpers.EventuallyWaitTimeout, helpers.WaitInterval, "WaitingBeforeForceDelete event should be created")
 

tests/node_drainer_test.go

@@ -169,7 +169,7 @@ func TestNodeDrainerEvictionModes(t *testing.T) {
 		require.NoError(t, err)
 
 		require.Eventually(t, func() bool {
-			found, event := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, "NodeDraining", "", restartTime)
+			found, event := helpers.CheckNodeEventExists(ctx, client, testCtx.NodeName, v1.EventTypeNormal, "", restartTime)
 			if found {
 				t.Logf("Found event after restart: %s", event.Reason)
 			}

tests/smoke_test.go

@@ -130,7 +130,7 @@ func TestFatalHealthEvent(t *testing.T) {
 		helpers.WaitForNodesWithLabel(ctx, t, client, []string{nodeName}, statemanager.NVSentinelStateLabelKey, string(statemanager.DrainingLabelValue))
 
 		expectedDrainingNodeEvent := v1.Event{
-			Type:   "NodeDraining",
+			Type:   v1.EventTypeNormal,
 			Reason: "AwaitingPodCompletion",
 		}
 		helpers.WaitForNodeEvent(ctx, t, client, nodeName, expectedDrainingNodeEvent)
@@ -321,7 +321,7 @@ func TestFatalUnsupportedHealthEvent(t *testing.T) {
 		helpers.WaitForNodesWithLabel(ctx, t, client, []string{nodeName}, statemanager.NVSentinelStateLabelKey, string(statemanager.DrainingLabelValue))
 
 		expectedDrainingNodeEvent := v1.Event{
-			Type:   "NodeDraining",
+			Type:   v1.EventTypeNormal,
 			Reason: "AwaitingPodCompletion",
 		}
 		helpers.WaitForNodeEvent(ctx, t, client, nodeName, expectedDrainingNodeEvent)