fix: in the status observer, trigger topology discovery when pods are ready (#183)

Signed-off-by: Dmitry Shmulevich <[email protected]>

Author: dmitsh

internal/k8s/utils.go

@@ -33,6 +33,15 @@ func GetPodsByLabels(ctx context.Context, client *kubernetes.Clientset, namespac
 	return client.CoreV1().Pods(namespace).List(ctx, opt)
 }
 
+func IsPodReady(pod *corev1.Pod) bool {
+	for _, cond := range pod.Status.Conditions {
+		if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 func GetDaemonSetPods(ctx context.Context, client *kubernetes.Clientset, name, namespace, nodename string) (*corev1.PodList, error) {
 	ds, err := client.AppsV1().DaemonSets(namespace).Get(ctx, name, metav1.GetOptions{})
 	if err != nil {

internal/k8s/utils_test.go

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2025 NVIDIA CORPORATION
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package k8s
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func TestIsPodReady(t *testing.T) {
+	testCases := []struct {
+		name  string
+		pod   *corev1.Pod
+		ready bool
+	}{
+		{
+			name: "Case 1: ready",
+			pod: &corev1.Pod{
+				Status: corev1.PodStatus{
+					Conditions: []corev1.PodCondition{
+						{
+							Type:   corev1.DisruptionTarget,
+							Status: corev1.ConditionUnknown,
+						},
+						{
+							Type:   corev1.PodReady,
+							Status: corev1.ConditionTrue,
+						},
+					},
+				},
+			},
+			ready: true,
+		},
+		{
+			name: "Case 2: implicit not ready",
+			pod: &corev1.Pod{
+				Status: corev1.PodStatus{
+					Conditions: []corev1.PodCondition{
+						{
+							Type:   corev1.DisruptionTarget,
+							Status: corev1.ConditionUnknown,
+						},
+						{
+							Type:   corev1.ContainersReady,
+							Status: corev1.ConditionTrue,
+						},
+					},
+				},
+			},
+			ready: false,
+		},
+		{
+			name: "Case 3: explicit not ready",
+			pod: &corev1.Pod{
+				Status: corev1.PodStatus{
+					Conditions: []corev1.PodCondition{
+						{
+							Type:   corev1.DisruptionTarget,
+							Status: corev1.ConditionUnknown,
+						},
+						{
+							Type:   corev1.PodReady,
+							Status: corev1.ConditionFalse,
+						},
+					},
+				},
+			},
+			ready: false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.ready, IsPodReady(tc.pod))
+		})
+	}
+}

pkg/engines/slinky/engine.go

@@ -133,6 +133,9 @@ func (eng *SlinkyEngine) GetComputeInstances(ctx context.Context, _ engines.Envi
 	// map k8s host name to SLURM host name
 	nodeMap := make(map[string]string)
 	for _, pod := range pods.Items {
+		if !k8s.IsPodReady(&pod) {
+			continue
+		}
 		host, ok := pod.Labels["slurm.node.name"]
 		if !ok {
 			host = pod.Spec.Hostname

pkg/node_observer/controller.go

@@ -31,9 +31,9 @@ import (
 )
 
 type Controller struct {
-	ctx          context.Context
-	client       kubernetes.Interface
-	nodeInformer *NodeInformer
+	ctx            context.Context
+	client         kubernetes.Interface
+	statusInformer *StatusInformer
 }
 
 func NewController(ctx context.Context, client kubernetes.Interface, cfg *Config) (*Controller, error) {
@@ -50,23 +50,23 @@ func NewController(ctx context.Context, client kubernetes.Interface, cfg *Config
 		req.Header.Set("Content-Type", "application/json")
 		return req, nil
 	}
-	nodeInformer, err := NewNodeInformer(ctx, client, &cfg.Trigger, f)
+	statusInformer, err := NewStatusInformer(ctx, client, &cfg.Trigger, f)
 	if err != nil {
 		return nil, err
 	}
 	return &Controller{
-		ctx:          ctx,
-		client:       client,
-		nodeInformer: nodeInformer,
+		ctx:            ctx,
+		client:         client,
+		statusInformer: statusInformer,
 	}, nil
 }
 
 func (c *Controller) Start() error {
 	klog.Infof("Starting state observer")
-	return c.nodeInformer.Start()
+	return c.statusInformer.Start()
 }
 
 func (c *Controller) Stop(err error) {
 	klog.Infof("Stopping state observer")
-	c.nodeInformer.Stop(err)
+	c.statusInformer.Stop(err)
 }