fix postgres impl

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

Author: dims

fault-quarantine/pkg/informer/k8s_client.go

@@ -203,6 +203,10 @@ func (c *FaultQuarantineClient) QuarantineNodeAndSetAnnotations(
 	annotations map[string]string,
 	labels map[string]string,
 ) error {
+	slog.Info("DRY-RUN DEBUG: QuarantineNodeAndSetAnnotations called", "node", nodename,
+		"numAnnotations", len(annotations), "isCordon", isCordon, "numTaints", len(taints),
+		"dryRunMode", c.DryRunMode)
+
 	updateFn := func(node *v1.Node) error {
 		if len(taints) > 0 {
 			if err := c.applyTaints(node, taints, nodename); err != nil {
@@ -264,6 +268,10 @@ func (c *FaultQuarantineClient) applyTaints(node *v1.Node, taints []config.Taint
 
 func (c *FaultQuarantineClient) handleCordon(node *v1.Node, nodename string) bool {
 	_, exist := node.Annotations[common.QuarantineHealthEventAnnotationKey]
+
+	slog.Info("DRY-RUN DEBUG: handleCordon", "node", nodename, "isUnschedulable", node.Spec.Unschedulable,
+		"hasQuarantineAnnotation", exist, "dryRunMode", c.DryRunMode)
+
 	if node.Spec.Unschedulable {
 		if exist {
 			slog.Info("Node already cordoned by FQM; skipping taint/annotation updates", "node", nodename)
@@ -287,6 +295,16 @@ func (c *FaultQuarantineClient) applyAnnotations(node *v1.Node, annotations map[
 		node.Annotations = make(map[string]string)
 	}
 
+	slog.Info("DRY-RUN DEBUG: applyAnnotations called", "node", nodename, "numAnnotations", len(annotations),
+		"annotationKeys", func() []string {
+			keys := make([]string, 0, len(annotations))
+			for k := range annotations {
+				keys = append(keys, k)
+			}
+
+			return keys
+		}())
+
 	slog.Info("Setting annotations on node", "node", nodename, "annotations", annotations)
 
 	for annotationKey, annotationValue := range annotations {

fault-quarantine/pkg/reconciler/reconciler.go

@@ -436,7 +436,9 @@ func (r *Reconciler) handleEvent(
 	annotationsMap := r.prepareAnnotations(taintsToBeApplied, &labelsMap, &isCordoned)
 
 	isNodeQuarantined := len(taintsToBeApplied) > 0 || isCordoned.Load()
-	if !isNodeQuarantined {
+
+	// In dry-run mode, always apply annotations for observability even if no actions would be taken
+	if !isNodeQuarantined && !r.config.DryRun {
 		return nil
 	}
 
@@ -679,6 +681,8 @@ func (r *Reconciler) applyQuarantine(
 	labelsMap *sync.Map,
 	isCordoned *atomic.Bool,
 ) *model.Status {
+	slog.Info("DRY-RUN DEBUG: applyQuarantine called", "node", event.HealthEvent.NodeName,
+		"numAnnotations", len(annotationsMap), "dryRunMode", r.config.DryRun)
 	r.recordCordonEventInCircuitBreaker(event)
 
 	healthEvents := healthEventsAnnotation.NewHealthEventsAnnotationMap()
@@ -692,9 +696,21 @@ func (r *Reconciler) applyQuarantine(
 	}
 
 	if err := r.addHealthEventAnnotation(healthEvents, annotationsMap); err != nil {
+		slog.Error("DRY-RUN DEBUG: Failed to add health event annotation", "error", err, "node", event.HealthEvent.NodeName)
+
 		return nil
 	}
 
+	slog.Info("DRY-RUN DEBUG: Added health event annotation", "node", event.HealthEvent.NodeName,
+		"numAnnotations", len(annotationsMap), "annotationKeys", func() []string {
+			keys := make([]string, 0, len(annotationsMap))
+			for k := range annotationsMap {
+				keys = append(keys, k)
+			}
+
+			return keys
+		}())
+
 	// Remove manual uncordon annotation if present before applying new quarantine
 	r.cleanupManualUncordonAnnotation(ctx, event.HealthEvent.NodeName, annotations)
 

platform-connectors/pkg/connectors/store/store_connector.go

@@ -136,28 +136,46 @@ func (r *DatabaseStoreConnector) insertHealthEvents(
 	// Prepare all documents for batch insertion
 	healthEventWithStatusList := make([]interface{}, 0, len(healthEvents.GetEvents()))
 
-	for _, healthEvent := range healthEvents.GetEvents() {
+	slog.Info("=== [PLATFORM-CONNECTORS-DEBUG] insertHealthEvents ENTRY ===",
+		"eventCount", len(healthEvents.GetEvents()))
+
+	for i, healthEvent := range healthEvents.GetEvents() {
 		// CRITICAL FIX: Clone the HealthEvent to avoid pointer reuse issues with gRPC buffers
 		// Without this clone, the healthEvent pointer may point to reused gRPC buffer memory
 		// that gets overwritten by subsequent requests, causing data corruption in MongoDB.
 		// This manifests as events having wrong isfatal/ishealthy/message values.
 		clonedHealthEvent := proto.Clone(healthEvent).(*protos.HealthEvent)
 
+		slog.Info("[PLATFORM-CONNECTORS-DEBUG] Processing health event",
+			"index", i,
+			"nodeName", clonedHealthEvent.NodeName,
+			"checkName", clonedHealthEvent.CheckName,
+			"isFatal", clonedHealthEvent.IsFatal,
+			"componentClass", clonedHealthEvent.ComponentClass)
+
 		healthEventWithStatusObj := model.HealthEventWithStatus{
 			CreatedAt:   time.Now().UTC(),
 			HealthEvent: clonedHealthEvent,
 		}
 		healthEventWithStatusList = append(healthEventWithStatusList, healthEventWithStatusObj)
 	}
 
+	slog.Info("=== [PLATFORM-CONNECTORS-DEBUG] About to call databaseClient.InsertMany ===",
+		"documentCount", len(healthEventWithStatusList),
+		"documentType", fmt.Sprintf("%T", healthEventWithStatusList[0]))
+
 	// Insert all documents in a single batch operation
 	// This ensures MongoDB generates INSERT operations (not UPDATE) for change streams
 	// Note: InsertMany is already atomic - either all documents are inserted or none are
 	_, err := r.databaseClient.InsertMany(ctx, healthEventWithStatusList)
 	if err != nil {
+		slog.Error("=== [PLATFORM-CONNECTORS-DEBUG] InsertMany FAILED ===",
+			"error", err)
 		return fmt.Errorf("insertMany failed: %w", err)
 	}
 
+	slog.Info("=== [PLATFORM-CONNECTORS-DEBUG] InsertMany SUCCESS ===")
+
 	return nil
 }
 

store-client/pkg/client/postgresql_client.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"strings"
 
 	_ "github.com/lib/pq" // PostgreSQL driver
@@ -85,10 +86,23 @@ func (c *PostgreSQLClient) Close(ctx context.Context) error {
 
 // InsertMany inserts multiple documents
 func (c *PostgreSQLClient) InsertMany(ctx context.Context, documents []interface{}) (*InsertManyResult, error) {
+	slog.Info("=== [POSTGRESQL-CLIENT-DEBUG] PostgreSQLClient.InsertMany ENTRY ===",
+		"documentCount", len(documents),
+		"tableName", c.table)
+
 	if len(documents) == 0 {
+		slog.Info("[POSTGRESQL-CLIENT-DEBUG] Empty documents, returning early")
 		return &InsertManyResult{InsertedIDs: []interface{}{}}, nil
 	}
 
+	// Check type of first document to see if it's a health event
+	if len(documents) > 0 {
+		firstDocType := fmt.Sprintf("%T", documents[0])
+		slog.Info("=== [POSTGRESQL-CLIENT-DEBUG] Document type check ===",
+			"documentType", firstDocType,
+			"documentCount", len(documents))
+	}
+
 	// Build batch insert query
 	// INSERT INTO table (document) VALUES ($1), ($2), ... RETURNING id
 	placeholders := make([]string, len(documents))
@@ -98,13 +112,20 @@ func (c *PostgreSQLClient) InsertMany(ctx context.Context, documents []interface
 		// Marshal document to JSON
 		docJSON, err := json.Marshal(doc)
 		if err != nil {
+			slog.Error("[POSTGRESQL-CLIENT-DEBUG] Failed to marshal document",
+				"index", i,
+				"error", err)
 			return nil, datastore.NewSerializationError(
 				datastore.ProviderPostgreSQL,
 				fmt.Sprintf("failed to marshal document at index %d", i),
 				err,
 			)
 		}
 
+		slog.Info("[POSTGRESQL-CLIENT-DEBUG] Marshaled document",
+			"index", i,
+			"jsonLength", len(docJSON))
+
 		placeholders[i] = fmt.Sprintf("($%d)", i+1)
 		args[i] = docJSON
 	}
@@ -116,9 +137,16 @@ func (c *PostgreSQLClient) InsertMany(ctx context.Context, documents []interface
 		strings.Join(placeholders, ", "),
 	)
 
+	slog.Info("=== [POSTGRESQL-CLIENT-DEBUG] About to execute INSERT query ===",
+		"query", query,
+		"argCount", len(args))
+
 	// Execute query and collect returned IDs
 	rows, err := c.db.QueryContext(ctx, query, args...)
 	if err != nil {
+		slog.Error("=== [POSTGRESQL-CLIENT-DEBUG] INSERT query FAILED ===",
+			"error", err,
+			"query", query)
 		return nil, datastore.NewInsertError(
 			datastore.ProviderPostgreSQL,
 			"failed to insert documents",
@@ -127,6 +155,8 @@ func (c *PostgreSQLClient) InsertMany(ctx context.Context, documents []interface
 	}
 	defer rows.Close()
 
+	slog.Info("[POSTGRESQL-CLIENT-DEBUG] INSERT query succeeded, reading IDs")
+
 	var insertedIDs []interface{}
 
 	for rows.Next() {

store-client/pkg/datastore/providers/postgresql/changestream.go

@@ -451,8 +451,29 @@ func (e *PostgreSQLEventAdapter) GetResumeToken() []byte {
 func (e *PostgreSQLEventAdapter) UnmarshalDocument(v interface{}) error {
 	// The fullDocument contains the actual document data
 	if fullDoc, ok := e.eventData["fullDocument"]; ok {
+		// Convert to map for easier manipulation
+		docMap, ok := fullDoc.(map[string]interface{})
+		if !ok {
+			return fmt.Errorf("fullDocument is not a map[string]interface{}")
+		}
+
+		// The PostgreSQL provider stores the document in a nested "document" field
+		// Extract just the document field which contains the actual HealthEventWithStatus
+		var actualDoc map[string]interface{}
+		if nestedDoc, ok := docMap["document"].(map[string]interface{}); ok {
+			actualDoc = nestedDoc
+		} else {
+			// Fall back to using the whole docMap if there's no nested document
+			actualDoc = docMap
+		}
+
+		// Transform lowercase keys to match the struct field names
+		// This handles the case where PostgreSQL stores lowercase JSON field names
+		// but Go struct tags may expect different casing
+		transformedDoc := transformJSONKeys(actualDoc)
+
 		// Use JSON marshaling/unmarshaling for type conversion
-		jsonData, err := json.Marshal(fullDoc)
+		jsonData, err := json.Marshal(transformedDoc)
 		if err != nil {
 			return fmt.Errorf("failed to marshal event document: %w", err)
 		}
@@ -467,6 +488,45 @@ func (e *PostgreSQLEventAdapter) UnmarshalDocument(v interface{}) error {
 	return fmt.Errorf("fullDocument not found in event")
 }
 
+// transformJSONKeys transforms lowercase JSON keys to match Go struct field names
+// This is needed because PostgreSQL stores lowercase JSON field names from bson tags
+// but protobuf fields need specific casing for proper unmarshaling
+func transformJSONKeys(doc map[string]interface{}) map[string]interface{} {
+	result := make(map[string]interface{})
+
+	for key, value := range doc {
+		// Handle nested maps recursively
+		if nestedMap, ok := value.(map[string]interface{}); ok {
+			value = transformJSONKeys(nestedMap)
+		}
+
+		// Apply specific transformations for known fields
+		// Map lowercase keys from bson tags to proper JSON keys
+		switch key {
+		case "healthevent":
+			// Keep it as healthevent since that's what the bson tag expects
+			result["healthevent"] = value
+		case "healtheventstatus":
+			result["healtheventstatus"] = value
+		case "createdat":
+			result["createdAt"] = value
+		case "nodequarantined":
+			result["nodequarantined"] = value
+		case "userpodsevictionstatus":
+			result["userpodsevictionstatus"] = value
+		case "faultremediated":
+			result["faultremediated"] = value
+		case "lastremediationtimestamp":
+			result["lastremediationtimestamp"] = value
+		default:
+			// Keep other keys as-is
+			result[key] = value
+		}
+	}
+
+	return result
+}
+
 // Verify PostgreSQLEventAdapter implements client.Event interface at compile time
 var _ client.Event = (*PostgreSQLEventAdapter)(nil)
 

store-client/pkg/factory/client_factory.go

@@ -16,12 +16,15 @@ package factory
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"os"
 
+	_ "github.com/lib/pq" // PostgreSQL driver
 	"github.com/nvidia/nvsentinel/store-client/pkg/client"
 	"github.com/nvidia/nvsentinel/store-client/pkg/config"
 	"github.com/nvidia/nvsentinel/store-client/pkg/datastore"
+	providers_postgresql "github.com/nvidia/nvsentinel/store-client/pkg/datastore/providers/postgresql"
 )
 
 // ClientFactory provides a simple interface for creating database clients
@@ -72,8 +75,30 @@ func (f *ClientFactory) CreateDatabaseClient(ctx context.Context) (client.Databa
 
 	switch provider {
 	case string(datastore.ProviderPostgreSQL):
-		// PostgreSQL client (Phase 2-3: Basic skeleton with Ping/Close implemented)
-		return client.NewPostgreSQLClient(ctx, f.dbConfig)
+		// Create PostgreSQL connection
+		db, err := sql.Open("postgres", f.dbConfig.GetConnectionURI())
+		if err != nil {
+			return nil, datastore.NewConnectionError(
+				datastore.ProviderPostgreSQL,
+				"failed to open PostgreSQL connection",
+				err,
+			)
+		}
+
+		// Test the connection
+		if err := db.PingContext(ctx); err != nil {
+			db.Close()
+
+			return nil, datastore.NewConnectionError(
+				datastore.ProviderPostgreSQL,
+				"failed to connect to PostgreSQL",
+				err,
+			)
+		}
+
+		// Return the provider's PostgreSQL database client which has health event field extraction
+		tableName := f.dbConfig.GetCollectionName() // In PostgreSQL context, collection = table
+		return providers_postgresql.NewPostgreSQLDatabaseClient(db, tableName), nil
 
 	case string(datastore.ProviderMongoDB), "":
 		// Default to MongoDB for backward compatibility