diff --git a/charts/topograph/charts/node-observer/values.yaml b/charts/topograph/charts/node-observer/values.yaml index eea4f483..4360f8de 100644 --- a/charts/topograph/charts/node-observer/values.yaml +++ b/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: [] diff --git a/charts/topograph/tests/subcharts_test.yaml b/charts/topograph/tests/subcharts_test.yaml index 3364c37c..08c18ec6 100644 --- a/charts/topograph/tests/subcharts_test.yaml +++ b/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:" diff --git a/pkg/node_observer/controller.go b/pkg/node_observer/controller.go index 64dbb26a..f30fa774 100644 --- a/pkg/node_observer/controller.go +++ b/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() } diff --git a/pkg/node_observer/health.go b/pkg/node_observer/health.go new file mode 100644 index 00000000..feddadea --- /dev/null +++ b/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: + } + } +} diff --git a/pkg/node_observer/health_test.go b/pkg/node_observer/health_test.go new file mode 100644 index 00000000..a39782a1 --- /dev/null +++ b/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) +}