fix broken things

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

Author: dims

health-events-analyzer/pkg/reconciler/reconciler.go

@@ -32,13 +32,9 @@ import (
 	"github.com/nvidia/nvsentinel/store-client/pkg/client"
 	"github.com/nvidia/nvsentinel/store-client/pkg/datastore"
 	"github.com/nvidia/nvsentinel/store-client/pkg/helper"
-	"google.golang.org/protobuf/proto"
 )
 
-const (
-	maxRetries int           = 5
-	delay      time.Duration = 10 * time.Second
-)
+// No retry constants needed - EventProcessor no longer retries internally
 
 type HealthEventsAnalyzerReconcilerConfig struct {
 	DataStoreConfig           *datastore.DataStoreConfig
@@ -74,11 +70,12 @@ func (r *Reconciler) Start(ctx context.Context) error {
 	r.databaseClient = bundle.DatabaseClient
 
 	// Create and configure the unified EventProcessor
+	// Note: EventProcessor no longer retries internally to prevent blocking the event stream
+	// Failed events will be retried on next pod restart (via resume token)
 	processorConfig := client.EventProcessorConfig{
-		MaxRetries:    maxRetries,
-		RetryDelay:    delay,
-		EnableMetrics: true,
-		MetricsLabels: map[string]string{"module": "health-events-analyzer"},
+		EnableMetrics:        true,
+		MetricsLabels:        map[string]string{"module": "health-events-analyzer"},
+		MarkProcessedOnError: false, // IMPORTANT: Don't mark failed events as processed
 	}
 
 	r.eventProcessor = client.NewEventProcessor(bundle.ChangeStreamWatcher, bundle.DatabaseClient, processorConfig)
@@ -115,8 +112,10 @@ func (r *Reconciler) processHealthEvent(ctx context.Context, event *datamodels.H
 	// Process the event using existing business logic
 	publishedNewEvent, err := r.handleEvent(ctx, event)
 	if err != nil {
-		// Log error but let the EventProcessor handle retry logic
+		// Return error - EventProcessor will NOT mark as processed
+		// Event will be retried on next pod restart
 		totalEventProcessingError.WithLabelValues("handle_event_error").Inc()
+		slog.Error("Failed to process health event", "error", err, "nodeName", labelValue)
 		return fmt.Errorf("failed to handle event: %w", err)
 	}
 
@@ -195,13 +194,9 @@ func (r *Reconciler) publishMatchedEvent(ctx context.Context,
 
 	actionVal := r.getRecommendedActionValue(rule.RecommendedAction, rule.Name)
 
-	// CRITICAL FIX: Clone the HealthEvent BEFORE passing to publisher
-	// The event pointer comes from EventProcessor's reusable variable and may be mutated
-	// by subsequent change stream events before the async gRPC call completes.
-	// This prevents data corruption where published events have wrong field values.
-	clonedHealthEvent := proto.Clone(event.HealthEvent).(*protos.HealthEvent)
-
-	err := r.config.Publisher.Publish(ctx, clonedHealthEvent, protos.RecommendedAction(actionVal), rule.Name)
+	// No need to clone here - Publisher.Publish already clones the event
+	// The EventProcessor creates a fresh stack variable for each event, so no mutation risk
+	err := r.config.Publisher.Publish(ctx, event.HealthEvent, protos.RecommendedAction(actionVal), rule.Name)
 	if err != nil {
 		slog.Error("Error in publishing the new fatal event", "error", err)
 		return fmt.Errorf("error in publishing the new fatal event: %w", err)

store-client/pkg/client/event_processor.go

@@ -25,14 +25,18 @@ import (
 
 // EventProcessorConfig holds configuration for the unified event processor
 type EventProcessorConfig struct {
-	// MaxRetries for event processing (0 = no retries, -1 = infinite retries)
+	// MaxRetries for event processing (used by QueueEventProcessor for retry limit)
+	// DefaultEventProcessor ignores this and does not retry
+	// Set to 0 for no retries, -1 for unlimited retries
 	MaxRetries int
-	// RetryDelay between attempts
-	RetryDelay time.Duration
 	// EnableMetrics controls whether to track processing metrics
 	EnableMetrics bool
 	// MetricsLabels for module-specific metrics categorization
 	MetricsLabels map[string]string
+	// MarkProcessedOnError determines whether to mark events as processed even if handler returns error
+	// Set to false (default) to preserve event for retry on next restart
+	// Set to true to skip failed events and continue (use with caution - may lose events)
+	MarkProcessedOnError bool
 }
 
 // EventProcessor provides a unified interface for processing change stream events
@@ -71,14 +75,6 @@ type DefaultEventProcessor struct {
 func NewEventProcessor(
 	watcher ChangeStreamWatcher, dbClient DatabaseClient, config EventProcessorConfig,
 ) EventProcessor {
-	if config.MaxRetries == 0 {
-		config.MaxRetries = 3 // Default retry attempts
-	}
-
-	if config.RetryDelay == 0 {
-		config.RetryDelay = time.Second * 2 // Default retry delay
-	}
-
 	return &DefaultEventProcessor{
 		changeStreamWatcher: watcher,
 		databaseClient:      dbClient,
@@ -143,73 +139,63 @@ func (p *DefaultEventProcessor) processEvents(ctx context.Context) error {
 	}
 }
 
-// handleSingleEvent processes a single event with retry logic
+// handleSingleEvent processes a single event
+// IMPORTANT: Does NOT retry internally - handler is responsible for its own retries if needed
+// This prevents retry-induced blocking of the event stream
 func (p *DefaultEventProcessor) handleSingleEvent(ctx context.Context, event Event) error {
 	startTime := time.Now()
 
 	// Unmarshal the event
 	var healthEventWithStatus model.HealthEventWithStatus
 	if err := event.UnmarshalDocument(&healthEventWithStatus); err != nil {
 		p.updateMetrics("unmarshal_error", "", time.Since(startTime), false)
+		// Unmarshal errors are non-recoverable, mark as processed to skip bad data
+		if markErr := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); markErr != nil {
+			slog.Error("Failed to mark processed after unmarshal error", "error", markErr)
+		}
 		return fmt.Errorf("failed to unmarshal event: %w", err)
 	}
 
 	eventID, err := event.GetDocumentID()
 	if err != nil {
 		p.updateMetrics("document_id_error", "", time.Since(startTime), false)
+		// Document ID errors are non-recoverable, mark as processed
+		if markErr := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); markErr != nil {
+			slog.Error("Failed to mark processed after document ID error", "error", markErr)
+		}
 		return fmt.Errorf("failed to get document ID: %w", err)
 	}
 
-	// DEBUG: Log RAW event immediately after unmarshal, before any business logic
-	if healthEventWithStatus.HealthEvent != nil {
-		slog.Info("EventProcessor - RAW event received from change stream",
-			"eventID", eventID,
-			"agent", healthEventWithStatus.HealthEvent.Agent,
-			"checkName", healthEventWithStatus.HealthEvent.CheckName,
-			"isFatal", healthEventWithStatus.HealthEvent.IsFatal,
-			"isHealthy", healthEventWithStatus.HealthEvent.IsHealthy,
-			"recommendedAction", healthEventWithStatus.HealthEvent.RecommendedAction,
-			"errorCode", healthEventWithStatus.HealthEvent.ErrorCode,
-			"nodeName", healthEventWithStatus.HealthEvent.NodeName,
-			"message", healthEventWithStatus.HealthEvent.Message,
-			"createdAt", healthEventWithStatus.CreatedAt)
-	}
-
 	slog.Debug("Processing event", "eventID", eventID, "event", healthEventWithStatus)
 
-	// Process with retry logic
-	var processErr error
-
-	maxRetries := p.config.MaxRetries
-	if maxRetries < 0 {
-		maxRetries = 1000000 // Treat -1 as infinite retries (large number)
-	}
-
-	for attempt := 1; attempt <= maxRetries; attempt++ {
-		slog.Debug("Processing event", "attempt", attempt, "eventID", eventID, "maxRetries", p.config.MaxRetries)
-
-		processErr = p.eventHandler.ProcessEvent(ctx, &healthEventWithStatus)
-		if processErr == nil {
-			// Success
-			p.updateMetrics("processing_success", eventID, time.Since(startTime), true)
-			break
-		}
-
-		slog.Error("Event processing failed", "attempt", attempt, "eventID", eventID, "error", processErr)
-
-		// Don't sleep on the last attempt
-		if attempt < maxRetries {
-			time.Sleep(p.config.RetryDelay)
-		}
-	}
+	// Process event - NO internal retries
+	// Handler is responsible for any retries it needs
+	processErr := p.eventHandler.ProcessEvent(ctx, &healthEventWithStatus)
 
 	if processErr != nil {
 		p.updateMetrics("processing_failed", eventID, time.Since(startTime), false)
-		slog.Error("Max retries reached, event processing failed", "eventID", eventID, "error", processErr)
+		slog.Error("Event processing failed", "eventID", eventID, "error", processErr)
+
+		// CRITICAL FIX: Only mark as processed if configured to do so OR if handler succeeded
+		if p.config.MarkProcessedOnError {
+			slog.Warn("Marking failed event as processed due to MarkProcessedOnError=true",
+				"eventID", eventID, "error", processErr)
+			if markErr := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); markErr != nil {
+				slog.Error("Failed to mark processed after error", "error", markErr)
+				return fmt.Errorf("failed to mark event as processed: %w", markErr)
+			}
+		} else {
+			// Do NOT mark as processed - event will be retried on next restart
+			slog.Error("Event processing failed, NOT marking as processed - will retry on restart",
+				"eventID", eventID, "error", processErr)
+			return processErr
+		}
+
+		return processErr
 	}
 
-	// Always mark as processed to advance the resume token
-	// Note: Passing empty token allows MongoDB watcher to use its internal cursor state
+	// Success - mark as processed
+	p.updateMetrics("processing_success", eventID, time.Since(startTime), true)
 	if err := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); err != nil {
 		p.updateMetrics("mark_processed_error", eventID, time.Since(startTime), false)
 		return fmt.Errorf("failed to mark event as processed: %w", err)

store-client/pkg/client/queue_event_processor.go

@@ -233,25 +233,38 @@ func (p *QueueEventProcessor) processQueuedEvent(ctx context.Context, queuedEven
 			// Re-queue for retry with exponential backoff
 			p.queue.AddRateLimited(queuedEvent)
 			p.updateMetrics("processing_retry", queuedEvent.EventID, time.Since(startTime), false)
-
+			// Do NOT mark as processed yet - still retrying
 			return
+		}
+
+		// Max retries reached - final failure
+		slog.Error("Max retries reached for event",
+			"eventID", queuedEvent.EventID,
+			"attempts", queuedEvent.Attempt)
+		p.updateMetrics("processing_failed", queuedEvent.EventID, time.Since(startTime), false)
+
+		// CRITICAL FIX: Only mark as processed if configured to do so
+		if p.config.MarkProcessedOnError {
+			slog.Warn("Marking failed event as processed due to MarkProcessedOnError=true",
+				"eventID", queuedEvent.EventID, "attempts", queuedEvent.Attempt)
+			if markErr := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); markErr != nil {
+				slog.Error("Failed to mark event as processed", "error", markErr)
+			}
 		} else {
-			// Max retries reached
-			slog.Error("Max retries reached for event",
-				"eventID", queuedEvent.EventID,
-				"attempts", queuedEvent.Attempt)
-			p.updateMetrics("processing_failed", queuedEvent.EventID, time.Since(startTime), false)
+			// Do NOT mark as processed - event will be retried on next restart
+			slog.Error("Event processing failed after max retries, NOT marking as processed - will retry on restart",
+				"eventID", queuedEvent.EventID, "attempts", queuedEvent.Attempt)
 		}
-	} else {
-		// Success
-		slog.Debug("Event processed successfully",
-			"workerID", workerID,
-			"eventID", queuedEvent.EventID)
-		p.updateMetrics("processing_success", queuedEvent.EventID, time.Since(startTime), true)
+
+		return
 	}
 
-	// Always mark as processed in the change stream to advance resume token
-	// This prevents the change stream from getting stuck on failed events
+	// Success - mark as processed
+	slog.Debug("Event processed successfully",
+		"workerID", workerID,
+		"eventID", queuedEvent.EventID)
+	p.updateMetrics("processing_success", queuedEvent.EventID, time.Since(startTime), true)
+
 	if markErr := p.changeStreamWatcher.MarkProcessed(ctx, []byte{}); markErr != nil {
 		slog.Error("Failed to mark event as processed in change stream",
 			"eventID", queuedEvent.EventID,