sync changes

Author: lalitadithya

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

@@ -47,6 +47,7 @@ rules:
   - list
   - patch
   - update
+  - watch
 - apiGroups:
   - ""
   resources:

node-drainer-module/pkg/informers/informers.go

@@ -46,6 +46,7 @@ const (
 type Informers struct {
 	podInformer            cache.SharedIndexInformer
 	eventInformer          cache.SharedIndexInformer
+	nodeInformer           cache.SharedIndexInformer
 	clientset              kubernetes.Interface
 	notReadyTimeoutMinutes *int
 	dryRunMode             []string
@@ -86,6 +87,8 @@ func NewInformers(clientset kubernetes.Interface, resyncPeriod time.Duration,
 		return nil, fmt.Errorf("failed to add event indexer: %w", err)
 	}
 
+	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
+
 	dryRunMode := []string{}
 	if dryRun {
 		dryRunMode = []string{metav1.DryRunAll}
@@ -95,14 +98,15 @@ func NewInformers(clientset kubernetes.Interface, resyncPeriod time.Duration,
 		clientset:              clientset,
 		podInformer:            podInformer,
 		eventInformer:          eventInformer,
+		nodeInformer:           nodeInformer,
 		notReadyTimeoutMinutes: notReadyTimeoutMinutes,
 		dryRunMode:             dryRunMode,
 		namespace:              metav1.NamespaceDefault,
 	}, nil
 }
 
 func (i *Informers) HasSynced() bool {
-	return i.podInformer.HasSynced() && i.eventInformer.HasSynced()
+	return i.podInformer.HasSynced() && i.eventInformer.HasSynced() && i.nodeInformer.HasSynced()
 }
 
 func NodeIndexFunc(obj any) ([]string, error) {
@@ -151,6 +155,7 @@ func NodeEventReasonIndexFunc(obj any) ([]string, error) {
 func (i *Informers) Run(ctx context.Context) error {
 	go i.podInformer.Run(ctx.Done())
 	go i.eventInformer.Run(ctx.Done())
+	go i.nodeInformer.Run(ctx.Done())
 
 	if ok := cache.WaitForCacheSync(ctx.Done(),
 		i.HasSynced); !ok {
@@ -382,6 +387,23 @@ func (i *Informers) UpdateNodeEvent(ctx context.Context, nodeName string, reason
 		}
 	}
 
+	// Get node from informer cache to retrieve its UID for proper event association
+	nodeObj, exists, err := i.nodeInformer.GetIndexer().GetByKey(nodeName)
+	if err != nil {
+		klog.Errorf("Failed to get node %s from cache: %v", nodeName, err)
+		return fmt.Errorf("error getting node %s from cache: %w", nodeName, err)
+	}
+
+	if !exists {
+		klog.Errorf("Node %s not found in cache", nodeName)
+		return fmt.Errorf("node %s not found in cache", nodeName)
+	}
+
+	node, ok := nodeObj.(*v1.Node)
+	if !ok {
+		return fmt.Errorf("failed to cast node object for %s", nodeName)
+	}
+
 	newEvent := &v1.Event{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: nodeName + "-",
@@ -390,6 +412,7 @@ func (i *Informers) UpdateNodeEvent(ctx context.Context, nodeName string, reason
 		InvolvedObject: v1.ObjectReference{
 			Kind:       "Node",
 			Name:       nodeName,
+			UID:        node.UID,
 			APIVersion: "v1",
 		},
 		Reason:         reason,

node-drainer-module/pkg/queue/types.go

@@ -36,10 +36,6 @@ type MongoCollectionAPI interface {
 	Find(ctx context.Context, filter interface{}, opts ...*options.FindOptions) (*mongo.Cursor, error)
 }
 
-const (
-	maxRetries = 5
-)
-
 type EventQueueManager interface {
 	EnqueueEvent(ctx context.Context, nodeName string, event bson.M, collection MongoCollectionAPI) error
 	Start(ctx context.Context)

node-drainer-module/pkg/queue/worker.go

@@ -50,15 +50,9 @@ func (m *eventQueueManager) processNextWorkItem(ctx context.Context) bool {
 
 	err := m.processEvent(ctx, *nodeEvent.Event, nodeEvent.Collection, nodeEvent.NodeName)
 	if err != nil {
-		if m.queue.NumRequeues(nodeEvent) < maxRetries {
-			klog.Warningf("Error processing event for node %s (attempt %d/%d): %v",
-				nodeEvent.NodeName, m.queue.NumRequeues(nodeEvent)+1, maxRetries, err)
-			m.queue.AddRateLimited(nodeEvent)
-		} else {
-			klog.Errorf("Failed to process event for node %s after %d retries: %v",
-				nodeEvent.NodeName, maxRetries, err)
-			m.queue.Forget(nodeEvent)
-		}
+		klog.Warningf("Error processing event for node %s (attempt %d): %v (will retry)",
+			nodeEvent.NodeName, m.queue.NumRequeues(nodeEvent)+1, err)
+		m.queue.AddRateLimited(nodeEvent)
 	} else {
 		m.queue.Forget(nodeEvent)
 	}

node-drainer-module/pkg/reconciler/reconciler.go

@@ -162,15 +162,20 @@ func (r *Reconciler) executeImmediateEviction(ctx context.Context,
 		}
 	}
 
-	return nil
+	return fmt.Errorf("immediate eviction completed, requeuing for status verification")
 }
 
 func (r *Reconciler) executeTimeoutEviction(ctx context.Context,
 	action *evaluator.DrainActionResult, healthEvent storeconnector.HealthEventWithStatus) error {
 	nodeName := healthEvent.HealthEvent.NodeName
 	timeoutMinutes := int(action.Timeout.Minutes())
 
-	return r.informers.DeletePodsAfterTimeout(ctx, nodeName, action.Namespaces, timeoutMinutes, &healthEvent)
+	if err := r.informers.DeletePodsAfterTimeout(ctx,
+		nodeName, action.Namespaces, timeoutMinutes, &healthEvent); err != nil {
+		return err
+	}
+
+	return fmt.Errorf("timeout eviction initiated, requeuing for status verification")
 }
 
 func (r *Reconciler) executeCheckCompletion(ctx context.Context,
@@ -211,7 +216,7 @@ func (r *Reconciler) executeCheckCompletion(ctx context.Context,
 
 	klog.Infof("All pods completed on node %s", nodeName)
 
-	return nil
+	return fmt.Errorf("pod completion verified, requeuing for status update")
 }
 
 func (r *Reconciler) executeMarkAlreadyDrained(ctx context.Context,