chore: address review coments

Signed-off-by: Ajay Mishra <[email protected]>

Author: XRFXLP

health-monitors/kubernetes-object-monitor/go.mod

@@ -6,7 +6,6 @@ require (
 	github.com/google/cel-go v0.26.1
 	github.com/nvidia/nvsentinel/commons v0.0.0
 	github.com/nvidia/nvsentinel/data-models v0.0.0
-	github.com/pelletier/go-toml/v2 v2.2.4
 	github.com/prometheus/client_golang v1.23.2
 	github.com/prometheus/client_model v0.6.2
 	github.com/stretchr/testify v1.11.1
@@ -19,6 +18,7 @@ require (
 
 require (
 	cel.dev/expr v0.25.1 // indirect
+	github.com/BurntSushi/toml v1.5.0 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect

health-monitors/kubernetes-object-monitor/go.sum

@@ -1,5 +1,7 @@
 cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4=
 cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4=
+github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg=
+github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
 github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ=
 github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -102,8 +104,6 @@ github.com/onsi/ginkgo/v2 v2.22.0 h1:Yed107/8DjTr0lKCNt7Dn8yQ6ybuDRQoMGrNFKzMfHg
 github.com/onsi/ginkgo/v2 v2.22.0/go.mod h1:7Du3c42kxCUegi0IImZ1wUQzMBVecgIHjR1C+NkhLQo=
 github.com/onsi/gomega v1.36.1 h1:bJDPBO7ibjxcbHMgSCoo4Yj18UWbKDlLwX1x9sybDcw=
 github.com/onsi/gomega v1.36.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog=
-github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
-github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

health-monitors/kubernetes-object-monitor/main.go

@@ -60,7 +60,7 @@ var (
 	)
 	maxConcurrentReconciles = flag.Int(
 		"max-concurrent-reconciles",
-		10,
+		1,
 		"Maximum number of resources to reconcile concurrently",
 	)
 	platformConnectorSocket = flag.String(

health-monitors/kubernetes-object-monitor/pkg/cel/environment.go

@@ -141,11 +141,11 @@ func (e *Environment) lookup(args ...ref.Val) ref.Val {
 	}
 
 	if err := e.client.Get(ctx, key, obj); err != nil {
-		slog.Error("Failed to get object from informer cache", "error", err)
+		slog.Error("Failed to get object using cached client", "error", err)
 		return types.NullValue
 	}
 
-	slog.Info("Successfully got object from informer cache", "object", obj.Object)
+	slog.Info("Successfully got object using cached client", "object", obj.Object)
 
 	return types.DefaultTypeAdapter.NativeToValue(obj.Object)
 }

health-monitors/kubernetes-object-monitor/pkg/config/loader.go

@@ -15,20 +15,14 @@ package config
 
 import (
 	"fmt"
-	"os"
 
-	"github.com/pelletier/go-toml/v2"
+	"github.com/nvidia/nvsentinel/commons/pkg/configmanager"
 )
 
 func Load(path string) (*Config, error) {
-	data, err := os.ReadFile(path)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read config file: %w", err)
-	}
-
 	var cfg Config
-	if err := toml.Unmarshal(data, &cfg); err != nil {
-		return nil, fmt.Errorf("failed to parse TOML config: %w", err)
+	if err := configmanager.LoadTOMLConfig(path, &cfg); err != nil {
+		return nil, err
 	}
 
 	if err := validate(&cfg); err != nil {

health-monitors/kubernetes-object-monitor/pkg/config/types.go

@@ -41,10 +41,11 @@ type AssociationSpec struct {
 }
 
 type HealthEventSpec struct {
-	ComponentClass    string `toml:"componentClass"`
-	IsFatal           bool   `toml:"isFatal"`
-	Message           string `toml:"message"`
-	RecommendedAction string `toml:"recommendedAction"`
+	ComponentClass    string   `toml:"componentClass"`
+	IsFatal           bool     `toml:"isFatal"`
+	Message           string   `toml:"message"`
+	RecommendedAction string   `toml:"recommendedAction"`
+	ErrorCode         []string `toml:"errorCode"`
 }
 
 func (r *ResourceSpec) GVK() string {

health-monitors/kubernetes-object-monitor/pkg/controller/reconciler.go

@@ -64,7 +64,7 @@ func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	obj.SetGroupVersionKind(r.gvk)
 
 	if err := r.Get(ctx, req.NamespacedName, obj); err != nil {
-		return r.handleGetError(err, req)
+		return r.handleGetError(ctx, err, req)
 	}
 
 	for _, p := range r.policies {
@@ -81,21 +81,18 @@ func (r *ResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	return ctrl.Result{}, nil
 }
 
-func (r *ResourceReconciler) handleGetError(err error, req ctrl.Request) (ctrl.Result, error) {
+func (r *ResourceReconciler) handleGetError(ctx context.Context, err error, req ctrl.Request) (ctrl.Result, error) {
 	if client.IgnoreNotFound(err) == nil {
-		r.cleanupDeletedResource(req)
+		r.cleanupDeletedResource(ctx, req)
 		return ctrl.Result{}, nil
 	}
 
-	metrics.ReconciliationErrors.WithLabelValues(r.gvk.Kind, "get_error").Inc()
+	metrics.ReconciliationErrors.WithLabelValues(r.gvk.Kind, "get_resource_error").Inc()
 
-	return ctrl.Result{}, err
+	return ctrl.Result{}, fmt.Errorf("failed to get resource: %w", err)
 }
 
-func (r *ResourceReconciler) cleanupDeletedResource(req ctrl.Request) {
-	r.matchStatesMu.Lock()
-	defer r.matchStatesMu.Unlock()
-
+func (r *ResourceReconciler) cleanupDeletedResource(ctx context.Context, req ctrl.Request) {
 	for _, p := range r.policies {
 		if !p.Enabled {
 			continue
@@ -107,8 +104,25 @@ func (r *ResourceReconciler) cleanupDeletedResource(req ctrl.Request) {
 		obj.SetName(req.Name)
 
 		stateKey := r.getStateKey(&p, obj)
-		if r.matchStates[stateKey] {
+
+		r.matchStatesMu.RLock()
+		wasMatched := r.matchStates[stateKey]
+		r.matchStatesMu.RUnlock()
+
+		if wasMatched {
+			nodeName := obj.GetName()
+
+			if err := r.publisher.PublishHealthEvent(ctx, &p, nodeName, true); err != nil {
+				slog.Error("Failed to publish healthy event for deleted resource",
+					"policy", p.Name,
+					"resource", req.NamespacedName,
+					"error", err)
+				metrics.HealthEventsPublishErrors.WithLabelValues(p.Name, "grpc_error").Inc()
+			}
+
+			r.matchStatesMu.Lock()
 			delete(r.matchStates, stateKey)
+			r.matchStatesMu.Unlock()
 		}
 	}
 }

health-monitors/kubernetes-object-monitor/pkg/controller/reconciler_test.go

@@ -16,6 +16,7 @@ package controller_test
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -217,6 +218,103 @@ func TestReconciler_DisabledPolicy(t *testing.T) {
 	}, 500*time.Millisecond, 50*time.Millisecond)
 }
 
+func TestReconciler_ErrorCodePropagation(t *testing.T) {
+	setup := setupTest(t)
+	nodeName := "test-node-error-code"
+
+	createNode(t, setup, nodeName, v1.ConditionFalse)
+
+	result, err := setup.reconciler.Reconcile(setup.ctx, ctrl.Request{
+		NamespacedName: types.NamespacedName{Name: nodeName},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, ctrl.Result{}, result)
+
+	require.Eventually(t, func() bool {
+		if len(setup.publisher.publishedEvents) != 1 {
+			return false
+		}
+		event := setup.publisher.publishedEvents[0]
+		return len(event.policy.HealthEvent.ErrorCode) == 2 &&
+			event.policy.HealthEvent.ErrorCode[0] == "NODE_NOT_READY" &&
+			event.policy.HealthEvent.ErrorCode[1] == "CONDITION_FAILED"
+	}, time.Second, 50*time.Millisecond)
+}
+
+func TestReconciler_ColdStart(t *testing.T) {
+	tests := []struct {
+		name              string
+		postRestartAction func(*testing.T, *testSetup, string, *v1.Node)
+		expectEvent       bool
+		expectHealthy     bool
+	}{
+		{
+			name: "unhealthy resource",
+			postRestartAction: func(t *testing.T, s *testSetup, nodeName string, _ *v1.Node) {
+				updateNodeStatus(t, s, nodeName, v1.ConditionFalse)
+			},
+			expectEvent:   true,
+			expectHealthy: false,
+		},
+		{
+			name: "healthy resource",
+			postRestartAction: func(t *testing.T, s *testSetup, nodeName string, _ *v1.Node) {
+				updateNodeStatus(t, s, nodeName, v1.ConditionTrue)
+			},
+			expectEvent: false,
+		},
+		{
+			name: "deleted resource",
+			postRestartAction: func(t *testing.T, s *testSetup, _ string, node *v1.Node) {
+				require.NoError(t, s.k8sClient.Delete(s.ctx, node))
+			},
+			expectEvent: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			setup := setupTest(t)
+			nodeName := "cold-start-" + strings.ReplaceAll(tt.name, " ", "-")
+
+			node := createNode(t, setup, nodeName, v1.ConditionFalse)
+
+			result, err := setup.reconciler.Reconcile(setup.ctx, ctrl.Request{
+				NamespacedName: types.NamespacedName{Name: nodeName},
+			})
+			assert.NoError(t, err)
+			assert.Equal(t, ctrl.Result{}, result)
+
+			require.Eventually(t, func() bool {
+				return len(setup.publisher.publishedEvents) == 1
+			}, time.Second, 50*time.Millisecond)
+
+			coldStartSetup := restartReconciler(t, setup)
+
+			tt.postRestartAction(t, coldStartSetup, nodeName, node)
+
+			result, err = coldStartSetup.reconciler.Reconcile(coldStartSetup.ctx, ctrl.Request{
+				NamespacedName: types.NamespacedName{Name: nodeName},
+			})
+			assert.NoError(t, err)
+			assert.Equal(t, ctrl.Result{}, result)
+
+			if tt.expectEvent {
+				require.Eventually(t, func() bool {
+					if len(coldStartSetup.publisher.publishedEvents) != 1 {
+						return false
+					}
+					return coldStartSetup.publisher.publishedEvents[0].isHealthy == tt.expectHealthy
+				}, time.Second, 50*time.Millisecond)
+			} else {
+				require.Never(t, func() bool {
+					return len(coldStartSetup.publisher.publishedEvents) > 0
+				}, 500*time.Millisecond, 50*time.Millisecond)
+			}
+		})
+	}
+}
+
 type testSetup struct {
 	ctx        context.Context
 	k8sClient  client.Client
@@ -226,31 +324,31 @@ type testSetup struct {
 	testEnv    *envtest.Environment
 }
 
-func setupTest(t *testing.T) *testSetup {
-	t.Helper()
-
-	policies := []config.Policy{
-		{
-			Name:    "node-not-ready",
-			Enabled: true,
-			Resource: config.ResourceSpec{
-				Group:   "",
-				Version: "v1",
-				Kind:    "Node",
-			},
-			Predicate: config.PredicateSpec{
-				Expression: `resource.status.conditions.filter(c, c.type == "Ready" && c.status == "False").size() > 0`,
-			},
-			HealthEvent: config.HealthEventSpec{
-				ComponentClass:    "Node",
-				IsFatal:           true,
-				Message:           "Node is not ready",
-				RecommendedAction: "CONTACT_SUPPORT",
-			},
+func defaultNodeNotReadyPolicy() config.Policy {
+	return config.Policy{
+		Name:    "node-not-ready",
+		Enabled: true,
+		Resource: config.ResourceSpec{
+			Group:   "",
+			Version: "v1",
+			Kind:    "Node",
+		},
+		Predicate: config.PredicateSpec{
+			Expression: `resource.status.conditions.filter(c, c.type == "Ready" && c.status == "False").size() > 0`,
+		},
+		HealthEvent: config.HealthEventSpec{
+			ComponentClass:    "Node",
+			IsFatal:           true,
+			Message:           "Node is not ready",
+			RecommendedAction: "CONTACT_SUPPORT",
+			ErrorCode:         []string{"NODE_NOT_READY", "CONDITION_FAILED"},
 		},
 	}
+}
 
-	return setupTestWithPolicies(t, policies)
+func setupTest(t *testing.T) *testSetup {
+	t.Helper()
+	return setupTestWithPolicies(t, []config.Policy{defaultNodeNotReadyPolicy()})
 }
 
 func setupTestWithPolicies(t *testing.T, policies []config.Policy) *testSetup {
@@ -375,6 +473,39 @@ func updateNodeStatus(t *testing.T, setup *testSetup, name string, readyStatus v
 	}, time.Second, 50*time.Millisecond)
 }
 
+func restartReconciler(t *testing.T, setup *testSetup) *testSetup {
+	t.Helper()
+
+	gvk := schema.GroupVersionKind{
+		Group:   "",
+		Version: "v1",
+		Kind:    "Node",
+	}
+
+	mockPub := &mockPublisher{
+		publishedEvents: []mockPublishedEvent{},
+	}
+
+	policies := []config.Policy{defaultNodeNotReadyPolicy()}
+
+	reconciler := controller.NewResourceReconciler(
+		setup.k8sClient,
+		setup.evaluator,
+		mockPub,
+		policies,
+		gvk,
+	)
+
+	return &testSetup{
+		ctx:        setup.ctx,
+		k8sClient:  setup.k8sClient,
+		reconciler: reconciler,
+		publisher:  mockPub,
+		evaluator:  setup.evaluator,
+		testEnv:    setup.testEnv,
+	}
+}
+
 func getCounterVecValue(t *testing.T, counterVec *prometheus.CounterVec, labelValues ...string) float64 {
 	t.Helper()
 	counter, err := counterVec.GetMetricWithLabelValues(labelValues...)

health-monitors/kubernetes-object-monitor/pkg/publisher/publisher.go

@@ -55,6 +55,7 @@ func (p *Publisher) PublishHealthEvent(ctx context.Context,
 		IsHealthy:          isHealthy,
 		NodeName:           nodeName,
 		RecommendedAction:  mapRecommendedAction(policy.HealthEvent.RecommendedAction),
+		ErrorCode:          policy.HealthEvent.ErrorCode,
 	}
 
 	healthEvents := &pb.HealthEvents{