feat(node-observer): wait for topograph health in-process

charts/topograph/charts/node-observer/values.yaml

@@ -66,15 +66,6 @@ resources:
     cpu: 250m
     memory: 256Mi
 
-livenessProbe:
-  httpGet:
-    path: /
-    port: http
-readinessProbe:
-  httpGet:
-    path: /
-    port: http
-
 nodeSelector: {}
 
 tolerations: []

charts/topograph/tests/subcharts_test.yaml

@@ -26,6 +26,10 @@ tests:
       - notExists:
           path: spec.template.spec.initContainers
         template: charts/node-observer/templates/deployment.yaml
+      - equal:
+          path: spec.template.spec.containers[0].command[0]
+          value: /usr/local/bin/node-observer
+        template: charts/node-observer/templates/deployment.yaml
       - matchRegex:
           path: data["node-observer-config.yaml"]
           pattern: "apiServer:"

pkg/node_observer/controller.go

@@ -33,6 +33,7 @@ type Controller struct {
 	ctx            context.Context
 	client         kubernetes.Interface
 	statusInformer *StatusInformer
+	healthURL      string
 }
 
 func NewController(ctx context.Context, client kubernetes.Interface, cfg *Config) (*Controller, error) {
@@ -43,6 +44,11 @@ func NewController(ctx context.Context, client kubernetes.Interface, cfg *Config
 		return nil, fmt.Errorf("failed to marshal payload: %v", err)
 	}
 
+	healthURL, err := healthCheckURL(cfg.GenerateTopologyURL)
+	if err != nil {
+		return nil, err
+	}
+
 	f := httpreq.GetRequestFunc(ctx, http.MethodPost, headers, nil, data, cfg.GenerateTopologyURL)
 	statusInformer, err := NewStatusInformer(ctx, client, &cfg.Trigger, &cfg.APIServer, cfg.RetryDelay.Duration, f)
 	if err != nil {
@@ -52,10 +58,16 @@ func NewController(ctx context.Context, client kubernetes.Interface, cfg *Config
 		ctx:            ctx,
 		client:         client,
 		statusInformer: statusInformer,
+		healthURL:      healthURL,
 	}, nil
 }
 
 func (c *Controller) Start() error {
+	klog.Infof("Waiting for topograph API to become ready")
+	if err := waitForTopograph(c.ctx, c.healthURL, healthCheckInterval, healthCheckTimeout); err != nil {
+		return err
+	}
+
 	klog.Infof("Starting state observer")
 	return c.statusInformer.Start()
 }

pkg/node_observer/health.go

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2026 NVIDIA CORPORATION
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package node_observer
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+
+	"k8s.io/klog/v2"
+
+	"github.com/NVIDIA/topograph/internal/httpreq"
+)
+
+// healthCheckInterval is how long to wait between topograph health probes
+// while the API is not yet ready.
+const healthCheckInterval = 2 * time.Second
+
+// healthCheckTimeout bounds how long to wait for the topograph API to become
+// ready before giving up. When exceeded, waitForTopograph returns an error so
+// the process exits non-zero and the pod restarts.
+const healthCheckTimeout = 1 * time.Minute
+
+// healthCheckURL derives the topograph health endpoint from the generate
+// topology URL by replacing its path with /healthz.
+func healthCheckURL(generateTopologyURL string) (string, error) {
+	u, err := url.Parse(generateTopologyURL)
+	if err != nil {
+		return "", fmt.Errorf("failed to parse generateTopologyUrl %q: %w", generateTopologyURL, err)
+	}
+	u.Path = "/healthz"
+	u.RawQuery = ""
+	u.Fragment = ""
+	return u.String(), nil
+}
+
+// waitForTopograph blocks until the topograph API health endpoint responds
+// successfully, the context is cancelled, or timeout elapses. It replaces the
+// chart's former `wait` init container. On timeout it returns an error so the
+// caller can exit non-zero.
+func waitForTopograph(ctx context.Context, healthURL string, interval, timeout time.Duration) error {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	start := time.Now()
+	f := httpreq.GetRequestFunc(ctx, http.MethodGet, nil, nil, nil, healthURL)
+	timer := time.NewTimer(interval)
+	if !timer.Stop() {
+		<-timer.C
+	}
+	defer timer.Stop()
+
+	for {
+		_, _, err := httpreq.DoRequest(f, false)
+		if err == nil {
+			klog.Infof("Topograph API is ready at %s", healthURL)
+			return nil
+		}
+		klog.Infof("Waiting for topograph to start at %s: %v", healthURL, err)
+
+		timer.Reset(interval)
+		select {
+		case <-ctx.Done():
+			// Report the actual elapsed time rather than the nominal timeout:
+			// context.WithTimeout honours whichever deadline (this timeout or the
+			// parent context's) fires first, so the two can differ.
+			return fmt.Errorf("topograph API not ready at %s after %s: %w", healthURL, time.Since(start).Round(time.Second), ctx.Err())
+		case <-timer.C:
+		}
+	}
+}

pkg/node_observer/health_test.go

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2026 NVIDIA CORPORATION
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package node_observer
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHealthCheckURL(t *testing.T) {
+	testCases := []struct {
+		name        string
+		generateURL string
+		expected    string
+		err         string
+	}{
+		{
+			name:        "in-cluster service URL",
+			generateURL: "http://topograph.topograph.svc.cluster.local:49021/v1/generate",
+			expected:    "http://topograph.topograph.svc.cluster.local:49021/healthz",
+		},
+		{
+			name:        "strips query and fragment",
+			generateURL: "https://host:8443/v1/generate?foo=bar#frag",
+			expected:    "https://host:8443/healthz",
+		},
+		{
+			name:        "no path",
+			generateURL: "http://host:49021",
+			expected:    "http://host:49021/healthz",
+		},
+		{
+			name:        "invalid URL",
+			generateURL: "http://[::1]:namedport/v1/generate",
+			err:         "failed to parse generateTopologyUrl",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := healthCheckURL(tc.generateURL)
+			if len(tc.err) != 0 {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.expected, got)
+		})
+	}
+}
+
+func TestWaitForTopographReady(t *testing.T) {
+	var hits int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Fail the first probe, then succeed, to exercise the retry loop.
+		if atomic.AddInt32(&hits, 1) < 2 {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	err := waitForTopograph(context.Background(), srv.URL+"/healthz", time.Millisecond, time.Minute)
+	require.NoError(t, err)
+	require.GreaterOrEqual(t, atomic.LoadInt32(&hits), int32(2))
+}
+
+func TestWaitForTopographTimeout(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+
+	err := waitForTopograph(context.Background(), srv.URL+"/healthz", time.Millisecond, 50*time.Millisecond)
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+}
+
+func TestWaitForTopographContextCancelled(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := waitForTopograph(ctx, srv.URL+"/healthz", time.Hour, time.Hour)
+	require.ErrorIs(t, err, context.Canceled)
+}