fix node informer restart

Signed-off-by: Davanum Srinivas <[email protected]>

Author: dims

fault-quarantine/main.go

@@ -28,6 +28,8 @@ import (
 
 	"github.com/nvidia/nvsentinel/commons/pkg/logger"
 	"github.com/nvidia/nvsentinel/fault-quarantine/pkg/config"
+	"github.com/nvidia/nvsentinel/fault-quarantine/pkg/informer"
+	"github.com/nvidia/nvsentinel/fault-quarantine/pkg/nodeinfo"
 	"github.com/nvidia/nvsentinel/fault-quarantine/pkg/reconciler"
 	sdkconfig "github.com/nvidia/nvsentinel/store-client/pkg/config"
 	"github.com/nvidia/nvsentinel/store-client/pkg/datastore"
@@ -125,6 +127,7 @@ func startMetricsServer(metricsPort string) {
 	slog.Info("Metrics server goroutine started")
 }
 
+//nolint:cyclop // Complexity slightly increased by NodeInformer initialization (was 10, now 12)
 func run() error {
 	// Create a context that gets cancelled on OS interrupt signals
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -187,11 +190,36 @@ func run() error {
 
 	slog.Info("Successfully initialized k8sclient")
 
+	// Create the work signal channel (buffered channel acting as semaphore)
+	workSignal := make(chan struct{}, 1) // Buffer size 1 is usually sufficient
+
+	// Create NodeInfo for tracking node quarantine status
+	nodeInfo := nodeinfo.NewNodeInfo(workSignal)
+
+	// Create and start NodeInformer before reconciler initialization
+	// This matches the pre-postgres pattern where NodeInformer runs independently
+	slog.Info("Creating NodeInformer")
+
+	nodeInformer, err := informer.NewNodeInformer(k8sClient.GetK8sClient(),
+		30*time.Minute, workSignal, nodeInfo)
+	if err != nil {
+		return fmt.Errorf("failed to create node informer: %w", err)
+	}
+
+	slog.Info("Starting NodeInformer")
+
+	if err := nodeInformer.Run(ctx.Done()); err != nil {
+		return fmt.Errorf("failed to start node informer: %w", err)
+	}
+
+	slog.Info("NodeInformer started and cache synced")
+
 	reconcilerCfg := reconciler.ReconcilerConfig{
 		TomlConfig:            *tomlCfg,
 		DataStore:             dataStore, // Use new datastore abstraction
 		Pipeline:              pipeline,  // Use new pipeline types
 		K8sClient:             k8sClient,
+		NodeInformer:          nodeInformer, // Pass the NodeInformer to reconciler
 		DryRun:                *dryRun,
 		CircuitBreakerEnabled: *circuitBreakerEnabled,
 		UnprocessedEventsMetricUpdateInterval: time.Duration(unprocessedEventsMetricUpdateIntervalSeconds) *
@@ -204,9 +232,6 @@ func run() error {
 		},
 	}
 
-	// Create the work signal channel (buffered channel acting as semaphore)
-	workSignal := make(chan struct{}, 1) // Buffer size 1 is usually sufficient
-
 	// Pass the workSignal channel to the Reconciler
 	rec := reconciler.NewReconciler(ctx, reconcilerCfg, workSignal)
 

fault-quarantine/pkg/informer/node_informer.go

@@ -133,57 +133,25 @@ func (ni *NodeInformer) Run(stopCh <-chan struct{}) error {
 
 	klog.Info("NodeInformer cache synced")
 
-	// Retry recalculateCounts if it returns 0 nodes (cache might still be populating)
-	// This handles the race between cache sync and cache population
-	maxRetries := 10
-	retryDelay := 100 * time.Millisecond
-
-	for attempt := 0; attempt < maxRetries; attempt++ {
-		changed, err := ni.recalculateCounts()
-		if err != nil {
-			klog.Errorf("Initial count calculation failed on attempt %d: %v", attempt+1, err)
-
-			if attempt < maxRetries-1 {
-				time.Sleep(retryDelay)
-				retryDelay *= 2 // Exponential backoff
-
-				if retryDelay > 2*time.Second {
-					retryDelay = 2 * time.Second
-				}
-
-				continue
-			}
-
-			break
-		}
+	// Perform initial count calculation without failing if 0 nodes
+	// Nodes may be added dynamically (e.g., in CI environments or with cluster autoscaling)
+	changed, err := ni.recalculateCounts()
+	if err != nil {
+		klog.Warningf("Initial count calculation failed: %v - will retry on node events", err)
+	} else {
+		_ = changed // Ignore changed status during initial population
 
-		// Check if we got nodes
 		ni.mutex.RLock()
 		totalNodes := ni.totalGpuNodes
 		ni.mutex.RUnlock()
 
 		if totalNodes > 0 {
 			klog.Infof("NodeInformer cache populated: %d GPU nodes found", totalNodes)
-			break
-		}
-
-		if attempt < maxRetries-1 {
-			klog.V(2).Infof("NodeInformer cache still empty after sync, retry %d/%d in %v",
-				attempt+1, maxRetries, retryDelay)
-			time.Sleep(retryDelay)
-			retryDelay *= 2 // Exponential backoff
-
-			if retryDelay > 2*time.Second {
-				retryDelay = 2 * time.Second
-			}
 		} else {
-			klog.Fatalf("NodeInformer cache still reports 0 nodes after %d retries and ~%v total wait time. "+
-				"This indicates a critical initialization failure - either cluster has no GPU nodes "+
-				"or NodeInformer cannot populate its cache. Pod will restart.",
-				maxRetries, time.Duration(maxRetries)*retryDelay)
+			klog.Warningf("NodeInformer cache synced but found 0 GPU nodes. " +
+				"This is expected in CI environments where nodes are created asynchronously. " +
+				"Nodes will be discovered as they are added to the cluster.")
 		}
-
-		_ = changed // Ignore changed status during initial population
 	}
 
 	return nil

fault-quarantine/pkg/reconciler/reconciler.go

@@ -52,6 +52,7 @@ type ReconcilerConfig struct {
 	DataStore                             datastore.DataStore
 	Pipeline                              interface{} // Custom pipeline for change stream filtering (e.g., mongo.Pipeline)
 	K8sClient                             K8sClientInterface
+	NodeInformer                          *informer.NodeInformer // NodeInformer instance (created in main.go)
 	DryRun                                bool
 	CircuitBreakerEnabled                 bool
 	UnprocessedEventsMetricUpdateInterval time.Duration
@@ -153,10 +154,10 @@ func (r *Reconciler) SetLabelKeys(labelKeyPrefix string) {
 
 // nolint: cyclop, gocognit //fix this as part of NGCC-21793
 func (r *Reconciler) Start(ctx context.Context) {
-	nodeInformer, err := informer.NewNodeInformer(r.config.K8sClient.GetK8sClient(),
-		30*time.Minute, r.workSignal, r.nodeInfo)
-	if err != nil {
-		klog.Fatalf("failed to initialize node informer: %+v", err)
+	// Use the NodeInformer passed from main.go (already created and running)
+	nodeInformer := r.config.NodeInformer
+	if nodeInformer == nil {
+		klog.Fatalf("NodeInformer is nil - must be created and started before calling reconciler.Start()")
 	}
 
 	// Set the callback to decrement the metric when a quarantined node with annotations is deleted
@@ -238,12 +239,8 @@ func (r *Reconciler) Start(ctx context.Context) {
 		klog.Infof("Initial quarantinedNodesMap is: %+v, total of %d nodes", quarantinedNodesMap, len(quarantinedNodesMap))
 	}
 
-	err = nodeInformer.Run(ctx.Done())
-	if err != nil {
-		klog.Fatalf("failed to run node informer: %+v", err)
-	}
-
-	// Wait for NodeInformer cache to sync before processing any events
+	// NodeInformer.Run() is already called in main.go before reconciler.Start()
+	// Just wait for it to sync before processing events
 	klog.Info("Waiting for NodeInformer cache to sync before starting event processing...")
 
 	for !nodeInformer.HasSynced() {

platform-connectors/pkg/connectors/kubernetes/k8s_connector.go

@@ -86,6 +86,7 @@ func (r *K8sConnector) FetchAndProcessHealthMetric(ctx context.Context) {
 			if healthEvents == nil {
 				continue
 			}
+
 			if err := r.processHealthEvents(ctx, healthEvents); err != nil {
 				slog.Error("Not able to process healthEvent", "error", err)
 				r.ringBuffer.HealthMetricEleProcessingFailed(healthEvents)