fix problems 4

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

Author: dims

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

@@ -17,12 +17,14 @@ package parser
 import (
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"reflect"
 	"regexp"
 	"strconv"
 	"strings"
 
 	datamodels "github.com/nvidia/nvsentinel/data-models/pkg/model"
+	"github.com/nvidia/nvsentinel/store-client/pkg/utils"
 )
 
 // ParseSequenceStage parses a JSON stage string and replaces "this." references with actual event values
@@ -56,7 +58,17 @@ func processValue(value interface{}, event datamodels.HealthEventWithStatus) (in
 				return nil, fmt.Errorf("error in getting value from path '%s': %w", fieldPath, err)
 			}
 
-			return resolvedValue, nil
+			// CRITICAL: Normalize field names for protobuf values embedded in MongoDB aggregation pipelines
+			// This converts camelCase JSON field names (entityType, entityValue) to lowercase BSON field names
+			// (entitytype, entityvalue) so they match MongoDB's storage and filter expectations
+			normalized := utils.NormalizeFieldNamesForMongoDB(resolvedValue)
+			
+			slog.Debug("Resolved this. reference",
+				"path", fieldPath,
+				"original_value", resolvedValue,
+				"normalized_value", normalized)
+			
+			return normalized, nil
 		}
 
 		return v, nil

health-events-analyzer/pkg/parser/parser_test.go

@@ -253,6 +253,33 @@ func TestParseSequenceStage(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name:  "stage with full entitiesimpacted array (MongoDB filter context)",
+			stage: `{"$match": {"$filter": {"input": "this.healthevent.entitiesimpacted", "cond": {"$eq": ["$$this.entitytype", "GPU"]}}}}`,
+			event: datamodels.HealthEventWithStatus{
+				HealthEvent: &protos.HealthEvent{
+					EntitiesImpacted: []*protos.Entity{
+						{EntityType: "GPU", EntityValue: "0"},
+						{EntityType: "CPU", EntityValue: "1"},
+					},
+				},
+			},
+			// After normalization, protobuf array should have lowercase field names
+			want: map[string]interface{}{
+				"$match": map[string]interface{}{
+					"$filter": map[string]interface{}{
+						"input": []interface{}{
+							map[string]interface{}{"entitytype": "GPU", "entityvalue": "0"},
+							map[string]interface{}{"entitytype": "CPU", "entityvalue": "1"},
+						},
+						"cond": map[string]interface{}{
+							"$eq": []interface{}{"$$this.entitytype", "GPU"},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
 		{
 			name:    "invalid JSON stage",
 			stage:   `{invalid json}`,

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

@@ -223,7 +223,11 @@ func (r *Reconciler) getRecommendedActionValue(recommendedAction, ruleName strin
 
 func (r *Reconciler) validateAllSequenceCriteria(ctx context.Context, rule config.HealthEventsAnalyzerRule,
 	healthEventWithStatus datamodels.HealthEventWithStatus) (bool, error) {
-	slog.Debug("Evaluating rule for event", "rule_name", rule.Name, "event", healthEventWithStatus)
+	slog.Info("→ Evaluating rule for event", 
+		"rule_name", rule.Name, 
+		"node", healthEventWithStatus.HealthEvent.NodeName,
+		"error_code", healthEventWithStatus.HealthEvent.ErrorCode,
+		"agent", healthEventWithStatus.HealthEvent.Agent)
 
 	// Build aggregation pipeline from stages
 	pipelineStages, err := r.getPipelineStages(rule, healthEventWithStatus)
@@ -234,12 +238,14 @@ func (r *Reconciler) validateAllSequenceCriteria(ctx context.Context, rule confi
 		return false, fmt.Errorf("failed to build pipeline stages: %w", err)
 	}
 
+	slog.Debug("Generated pipeline", "rule_name", rule.Name, "pipeline_stages", pipelineStages)
+
 	var result []map[string]interface{}
 
 	// Execute aggregation using store-client abstraction
 	cursor, err := r.databaseClient.Aggregate(ctx, pipelineStages)
 	if err != nil {
-		slog.Error("Failed to execute aggregation pipeline", "error", err)
+		slog.Error("Failed to execute aggregation pipeline", "error", err, "rule_name", rule.Name)
 		totalEventProcessingError.WithLabelValues("execute_pipeline_error").Inc()
 
 		return false, fmt.Errorf("failed to execute aggregation pipeline: %w", err)
@@ -248,33 +254,45 @@ func (r *Reconciler) validateAllSequenceCriteria(ctx context.Context, rule confi
 	defer cursor.Close(ctx)
 
 	if err = cursor.All(ctx, &result); err != nil {
-		slog.Error("Failed to decode cursor", "error", err)
+		slog.Error("Failed to decode cursor", "error", err, "rule_name", rule.Name)
 		totalEventProcessingError.WithLabelValues("decode_cursor_error").Inc()
 
 		return false, fmt.Errorf("failed to decode cursor: %w", err)
 	}
 
+	slog.Debug("Aggregation results", "rule_name", rule.Name, "result_count", len(result), "results", result)
+
 	// Check if we have results (rule matched)
 	if len(result) > 0 {
 		// Check for explicit ruleMatched field (used in tests and by SequenceFacet pipelines)
 		if matched, ok := result[0]["ruleMatched"].(bool); ok {
 			if matched {
-				slog.Debug("Rule matched (via ruleMatched field)", "rule_name", rule.Name)
+				slog.Info("✓ Rule matched via ruleMatched field", 
+					"rule_name", rule.Name,
+					"node", healthEventWithStatus.HealthEvent.NodeName)
 				return true, nil
 			}
 
-			slog.Debug("Rule did not match (ruleMatched=false)", "rule_name", rule.Name)
+			slog.Info("✗ Rule did not match (ruleMatched=false)", 
+				"rule_name", rule.Name,
+				"node", healthEventWithStatus.HealthEvent.NodeName,
+				"result", result[0])
 
 			return false, nil
 		}
 
 		// For Stage-based pipelines, presence of results indicates a match
-		slog.Debug("Rule matched (results exist)", "rule_name", rule.Name, "result_count", len(result))
+		slog.Info("✓ Rule matched via results existence", 
+			"rule_name", rule.Name,
+			"node", healthEventWithStatus.HealthEvent.NodeName,
+			"result_count", len(result))
 
 		return true, nil
 	}
 
-	slog.Debug("Rule did not match (no results)", "rule_name", rule.Name)
+	slog.Info("✗ Rule did not match (no results)", 
+		"rule_name", rule.Name,
+		"node", healthEventWithStatus.HealthEvent.NodeName)
 
 	return false, nil
 }
@@ -297,14 +315,18 @@ func (r *Reconciler) getPipelineStages(
 
 	for i, stageStr := range rule.Stage {
 		// Parse the stage and resolve "this." references
+		slog.Debug("Parsing stage", "stage_index", i, "stage_string", stageStr)
+		
 		stageMap, err := parser.ParseSequenceStage(stageStr, healthEventWithStatus)
 		if err != nil {
-			slog.Error("Failed to parse stage", "stage_index", i, "error", err)
+			slog.Error("Failed to parse stage", "stage_index", i, "error", err, "stage_string", stageStr)
 			totalEventProcessingError.WithLabelValues("parse_stage_error").Inc()
 
 			return nil, fmt.Errorf("failed to parse stage %d: %w", i, err)
 		}
 
+		slog.Debug("Parsed stage successfully", "stage_index", i, "parsed_stage", stageMap)
+		
 		pipeline = append(pipeline, stageMap)
 	}