Merge fix/duplicate-status-processing: Fix duplicate status processing and rate limiting

Author: mattmattox

pkg/detector/detector.go

@@ -96,11 +96,9 @@ type ProblemDetector struct {
 	configFilePath    string
 	monitorFactory    MonitorFactory
 	configChangeCh    <-chan struct{}
-	reloadMutex       sync.Mutex // Protects reload operations
-	started           bool
-	seenProblems      map[string]*types.Problem // For deduplication
-	problemsMutex     sync.RWMutex              // Protects seenProblems
-	passedMonitors    []types.Monitor           // Monitors passed directly to constructor
+	reloadMutex    sync.Mutex      // Protects reload operations
+	started        bool
+	passedMonitors []types.Monitor // Monitors passed directly to constructor
 }
 
 // MonitorFactory interface for creating monitor instances during hot reload
@@ -150,7 +148,6 @@ func NewProblemDetector(config *types.NodeDoctorConfig, monitors []types.Monitor
 		configWatcher:  configWatcher,
 		configFilePath: configFilePath,
 		monitorFactory: monitorFactory,
-		seenProblems:   make(map[string]*types.Problem),
 		passedMonitors: monitors, // Store passed monitors to start in Start()
 	}
 
@@ -345,19 +342,9 @@ func (pd *ProblemDetector) processStatus(status *types.Status) {
 	// Update statistics
 	pd.stats.IncrementStatusesReceived()
 
-	// Convert status to problems for testing
-	problems := pd.statusToProblems(status)
-	pd.stats.AddProblemsDetected(len(problems))
-
-	// Apply deduplication
-	newProblems := pd.deduplicateProblems(problems)
-	pd.stats.AddProblemsDeduplicated(len(newProblems))
-
-	for _, problem := range newProblems {
-		pd.processProblem(problem)
-	}
-
-	// Export to all exporters
+	// Export to all exporters (single path - Status contains all data)
+	// Note: Previously this also called ExportProblem() for converted problems,
+	// causing duplicate Kubernetes resources. See GitHub issue #7.
 	for _, exporter := range pd.exporters {
 		if err := exporter.ExportStatus(pd.ctx, status); err != nil {
 			log.Printf("[WARN] Failed to export status to exporter: %v", err)
@@ -368,98 +355,6 @@ func (pd *ProblemDetector) processStatus(status *types.Status) {
 	}
 }
 
-// processProblem processes a problem for testing purposes
-func (pd *ProblemDetector) processProblem(problem *types.Problem) {
-	// Export to all exporters
-	for _, exporter := range pd.exporters {
-		if err := exporter.ExportProblem(pd.ctx, problem); err != nil {
-			log.Printf("[WARN] Failed to export problem to exporter: %v", err)
-			pd.stats.IncrementExportsFailed()
-		} else {
-			pd.stats.IncrementExportsSucceeded()
-		}
-	}
-}
-
-// statusToProblems converts a status update to problems for testing
-func (pd *ProblemDetector) statusToProblems(status *types.Status) []*types.Problem {
-	var problems []*types.Problem
-
-	// Convert conditions to problems
-	for _, condition := range status.Conditions {
-		if condition.Status == types.ConditionFalse {
-			problems = append(problems, &types.Problem{
-				Type:       "condition-" + condition.Type,
-				Resource:   status.Source,
-				Message:    condition.Message,
-				Severity:   types.ProblemCritical, // False conditions are critical problems
-				DetectedAt: condition.Transition,
-				Metadata: map[string]string{
-					"condition": condition.Type,
-				},
-			})
-		}
-	}
-
-	// Convert high severity events to problems
-	for _, event := range status.Events {
-		if event.Severity == types.EventError || event.Severity == types.EventWarning {
-			severity := pd.eventSeverityToProblemSeverity(event.Severity)
-			problems = append(problems, &types.Problem{
-				Type:       "event-" + event.Reason,
-				Resource:   status.Source,
-				Message:    event.Message,
-				Severity:   severity,
-				DetectedAt: event.Timestamp,
-				Metadata: map[string]string{
-					"event": event.Reason,
-				},
-			})
-		}
-	}
-
-	return problems
-}
-
-// deduplicateProblems filters out problems that have already been seen with the same severity
-func (pd *ProblemDetector) deduplicateProblems(problems []*types.Problem) []*types.Problem {
-	pd.problemsMutex.Lock()
-	defer pd.problemsMutex.Unlock()
-
-	var newProblems []*types.Problem
-
-	for _, problem := range problems {
-		// Create a unique key for this problem
-		key := fmt.Sprintf("%s:%s:%s", problem.Type, problem.Resource, problem.Severity)
-
-		// Check if we've seen this exact problem before
-		if existingProblem, exists := pd.seenProblems[key]; exists {
-			// If severity is the same, skip (it's a duplicate)
-			if existingProblem.Severity == problem.Severity {
-				continue
-			}
-		}
-
-		// This is a new problem or severity has changed
-		pd.seenProblems[key] = problem
-		newProblems = append(newProblems, problem)
-	}
-
-	return newProblems
-}
-
-// eventSeverityToProblemSeverity converts EventSeverity to ProblemSeverity
-func (pd *ProblemDetector) eventSeverityToProblemSeverity(severity types.EventSeverity) types.ProblemSeverity {
-	switch severity {
-	case types.EventError:
-		return types.ProblemCritical
-	case types.EventWarning:
-		return types.ProblemWarning
-	default:
-		return types.ProblemInfo
-	}
-}
-
 // fanInFromMonitor reads statuses from a monitor and forwards them to the main status channel
 func (pd *ProblemDetector) fanInFromMonitor(ctx context.Context, statusCh <-chan *types.Status, monitorName string) {
 	log.Printf("[DEBUG] Starting fan-in for monitor %s", monitorName)

pkg/detector/detector_test.go

@@ -202,145 +202,6 @@ func TestProblemDetector_StatusProcessing(t *testing.T) {
 	}
 }
 
-func TestStatusToProblems_Events(t *testing.T) {
-	helper := NewTestHelper()
-	config := helper.CreateTestConfig()
-	factory := NewMockMonitorFactory()
-	detector, _ := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{NewMockExporter("test")}, "/tmp/test-config.yaml", factory)
-
-	tests := []struct {
-		name               string
-		status             *types.Status
-		expectedCount      int
-		containsTypes      []string // Check for presence, not order
-		containsSeverities []types.ProblemSeverity
-	}{
-		{
-			name: "error event",
-			status: func() *types.Status {
-				status := types.NewStatus("test-source")
-				status.AddEvent(types.NewEvent(types.EventError, "CriticalError", "Critical error occurred"))
-				status.AddCondition(types.NewCondition("SystemHealth", types.ConditionFalse, "HealthCheck", "System not healthy"))
-				return status
-			}(),
-			expectedCount:      2, // 1 error event + 1 false condition
-			containsTypes:      []string{"event-CriticalError", "condition-SystemHealth"},
-			containsSeverities: []types.ProblemSeverity{types.ProblemCritical}, // Both error events and false conditions are Critical
-		},
-		{
-			name: "warning event",
-			status: func() *types.Status {
-				status := types.NewStatus("test-source")
-				status.AddEvent(types.NewEvent(types.EventWarning, "PerformanceWarning", "Warning occurred"))
-				return status
-			}(),
-			expectedCount:      1,
-			containsTypes:      []string{"event-PerformanceWarning"},
-			containsSeverities: []types.ProblemSeverity{types.ProblemWarning},
-		},
-		{
-			name: "healthy status",
-			status: func() *types.Status {
-				status := types.NewStatus("test-source")
-				status.AddCondition(types.NewCondition("Ready", types.ConditionTrue, "Ready", "Ready"))
-				return status
-			}(),
-			expectedCount: 0, // Info events and true conditions are ignored
-		},
-		{
-			name:          "empty status",
-			status:        types.NewStatus("test-source"),
-			expectedCount: 0,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			problems := detector.statusToProblems(tt.status)
-
-			if len(problems) != tt.expectedCount {
-				t.Errorf("statusToProblems() problem count = %d, want %d", len(problems), tt.expectedCount)
-			}
-
-			// Check that all expected types are present
-			if tt.expectedCount > 0 {
-				problemTypes := make(map[string]bool)
-				problemSeverities := make(map[types.ProblemSeverity]bool)
-
-				for _, problem := range problems {
-					problemTypes[problem.Type] = true
-					problemSeverities[problem.Severity] = true
-				}
-
-				for _, expectedType := range tt.containsTypes {
-					if !problemTypes[expectedType] {
-						t.Errorf("statusToProblems() missing expected problem type: %s", expectedType)
-					}
-				}
-
-				for _, expectedSeverity := range tt.containsSeverities {
-					if !problemSeverities[expectedSeverity] {
-						t.Errorf("statusToProblems() missing expected problem severity: %s", expectedSeverity)
-					}
-				}
-			}
-		})
-	}
-}
-
-func TestStatusToProblems_Conditions(t *testing.T) {
-	helper := NewTestHelper()
-	config := helper.CreateTestConfig()
-	factory := NewMockMonitorFactory()
-	detector, _ := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{NewMockExporter("test")}, "/tmp/test-config.yaml", factory)
-
-	status := types.NewStatus("test-source")
-	status.AddCondition(types.NewCondition("DiskPressure", types.ConditionFalse, "DiskFull", "Disk is full"))
-	status.AddCondition(types.NewCondition("NetworkReady", types.ConditionTrue, "NetworkOK", "Network is ready"))
-	status.AddCondition(types.NewCondition("UnknownCondition", types.ConditionUnknown, "Unknown", "Status unknown"))
-
-	problems := detector.statusToProblems(status)
-
-	// Should only create problem for False condition
-	if len(problems) != 1 {
-		t.Errorf("statusToProblems() problem count = %d, want 1", len(problems))
-	}
-
-	if problems[0].Type != "condition-DiskPressure" {
-		t.Errorf("statusToProblems() problem type = %s, want condition-DiskPressure", problems[0].Type)
-	}
-}
-
-func TestDeduplicateProblems(t *testing.T) {
-	helper := NewTestHelper()
-	config := helper.CreateTestConfig()
-	factory := NewMockMonitorFactory()
-	detector, _ := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{NewMockExporter("test")}, "/tmp/test-config.yaml", factory)
-
-	// First set of problems
-	problem1 := types.NewProblem("disk-full", "node1", types.ProblemCritical, "Disk is full")
-	problem2 := types.NewProblem("memory-pressure", "node1", types.ProblemWarning, "Memory pressure detected")
-
-	newProblems := detector.deduplicateProblems([]*types.Problem{problem1, problem2})
-	if len(newProblems) != 2 {
-		t.Errorf("First deduplication expected 2 new problems, got %d", len(newProblems))
-	}
-
-	// Same problems again (should be deduplicated)
-	problem1Dup := types.NewProblem("disk-full", "node1", types.ProblemCritical, "Disk is full")
-	newProblems = detector.deduplicateProblems([]*types.Problem{problem1Dup})
-	if len(newProblems) != 0 {
-		t.Errorf("Second deduplication expected 0 new problems, got %d", len(newProblems))
-	}
-
-	// Same type/resource but different severity (should be reported)
-	problem1Updated := types.NewProblem("disk-full", "node1", types.ProblemWarning, "Disk pressure reduced")
-	newProblems = detector.deduplicateProblems([]*types.Problem{problem1Updated})
-	if len(newProblems) != 1 {
-		t.Errorf("Third deduplication expected 1 new problem, got %d", len(newProblems))
-	}
-}
-
 func TestExportDistribution(t *testing.T) {
 	helper := NewTestHelper()
 	config := helper.CreateTestConfig()
@@ -556,17 +417,17 @@ func TestExportFailureIsolation(t *testing.T) {
 	config := helper.CreateTestConfig()
 
 	// Create exporters with different failure behaviors
+	// Note: We only test status export failures since ExportProblem is no longer called (issue #7 fix)
 	successExporter := NewMockExporter("success-exporter")
 	failStatusExporter := NewMockExporter("fail-status").SetStatusExportError(fmt.Errorf("status export failed"))
-	failProblemExporter := NewMockExporter("fail-problem").SetProblemExportError(fmt.Errorf("problem export failed"))
 
 	factory := NewMockMonitorFactory().SetCreateFunc(func(config types.MonitorConfig) (types.Monitor, error) {
 		status := types.NewStatus(config.Name)
 		status.AddEvent(types.NewEvent(types.EventError, "CriticalError", "Critical error occurred"))
 		return NewMockMonitor(config.Name).AddStatusUpdate(status), nil
 	})
 
-	detector, err := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{successExporter, failStatusExporter, failProblemExporter}, "/tmp/test-config.yaml", factory)
+	detector, err := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{successExporter, failStatusExporter}, "/tmp/test-config.yaml", factory)
 	if err != nil {
 		t.Fatalf("NewProblemDetector() error = %v", err)
 	}
@@ -650,7 +511,7 @@ func TestStatisticsTracking(t *testing.T) {
 	})
 
 	successExporter := NewMockExporter("success-exp")
-	failExporter := NewMockExporter("fail-exp").SetProblemExportError(fmt.Errorf("export failed"))
+	failExporter := NewMockExporter("fail-exp").SetStatusExportError(fmt.Errorf("export failed"))
 
 	detector, err := NewProblemDetector(config, []types.Monitor{}, []types.Exporter{successExporter, failExporter}, "/tmp/test-config.yaml", factory)
 	if err != nil {
@@ -679,9 +540,6 @@ func TestStatisticsTracking(t *testing.T) {
 	if stats.GetStatusesReceived() == 0 {
 		t.Errorf("Expected statuses to be received")
 	}
-	if stats.GetProblemsDetected() == 0 {
-		t.Errorf("Expected problems to be detected")
-	}
 
 	// Verify export statistics
 	if stats.GetExportsSucceeded() == 0 {

pkg/exporters/kubernetes/event_manager.go

@@ -103,21 +103,21 @@ func (em *EventManager) CreateEvent(ctx context.Context, event corev1.Event) err
 		namespace = "default"
 	}
 
-	// Check rate limiting first
-	if !em.checkRateLimit() {
-		log.Printf("[WARN] Event rate limit exceeded (%d/min), dropping event: %s",
-			em.maxEventsPerMin, event.Reason)
-		return fmt.Errorf("event rate limit exceeded")
-	}
-
-	// Check deduplication
+	// Check deduplication FIRST - duplicates should not consume rate limit quota
 	signature := CreateEventSignature(event)
 	if em.isDuplicate(signature) {
 		log.Printf("[DEBUG] Duplicate event suppressed: %s (signature: %s)",
 			event.Reason, signature.Hash())
 		return nil
 	}
 
+	// Check rate limiting only for non-duplicate events
+	if !em.checkRateLimit() {
+		log.Printf("[WARN] Event rate limit exceeded (%d/min), dropping event: %s",
+			em.maxEventsPerMin, event.Reason)
+		return fmt.Errorf("event rate limit exceeded")
+	}
+
 	// Create the event
 	err := em.client.CreateEvent(ctx, event, namespace)
 	if err != nil {

pkg/exporters/kubernetes/event_manager_test.go

@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -163,19 +164,19 @@ func TestEventRateLimiting(t *testing.T) {
 	manager := createTestEventManager(2, time.Minute) // Limit to 2 events per minute
 	ctx := context.Background()
 
-	// Create events up to the limit
+	// Create unique events up to the limit (different Reasons to avoid deduplication)
 	for i := 0; i < 2; i++ {
 		event := corev1.Event{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      "rate-limit-event-" + string(rune('1'+i)),
+				Name:      fmt.Sprintf("rate-limit-event-%d", i+1),
 				Namespace: "test-namespace",
 			},
 			InvolvedObject: corev1.ObjectReference{
 				Kind: "Node",
 				Name: "test-node",
 			},
-			Reason:         "RateLimitTest",
-			Message:        "Rate limit message",
+			Reason:         fmt.Sprintf("RateLimitTest%d", i+1), // Unique reason to avoid deduplication
+			Message:        fmt.Sprintf("Rate limit message %d", i+1),
 			Type:           corev1.EventTypeNormal,
 			FirstTimestamp: metav1.NewTime(time.Now()),
 			LastTimestamp:  metav1.NewTime(time.Now()),