fix(validator): surface + re-read NCCL launcher log on bandwidth parse miss (NVIDIA#1691)

Signed-off-by: Nathan Hensley <nhensley@nvidia.com>

Author: njhensley

pkg/defaults/timeouts.go

@@ -923,6 +923,17 @@ const (
 	// attempts while the Kubeflow Trainer validating webhook's informer cache
 	// catches up to a freshly-created TrainingRuntime.
 	TrainJobAdmissionRetryInterval = 500 * time.Millisecond
+
+	// NCCLLauncherLogReadInterval is the backoff between re-reads of a succeeded
+	// NCCL launcher pod's log while waiting for the results table to be fully
+	// captured. A pod that has just reached Succeeded can briefly serve an empty
+	// or truncated log if its container is being torn down mid-read.
+	NCCLLauncherLogReadInterval = 2 * time.Second
+
+	// NCCLLauncherLogReadAttempts bounds how many times the succeeded launcher
+	// pod's log is re-read before giving up and returning the last read for
+	// diagnosis (the parser then fails and the log is surfaced).
+	NCCLLauncherLogReadAttempts = 5
 )
 
 // Termination and truncation limits for validator output.

validators/performance/nccl_all_reduce_bw_constraint.go

@@ -328,6 +328,16 @@ func validateNcclAllReduceBw(ctx *validators.Context, constraint recipe.Constrai
 	// Parse bandwidth from logs (shared across all service types).
 	bandwidth, err := parseBandwidthFromLogs(logs)
 	if err != nil {
+		// The launcher pod succeeded but its log yielded no parseable bandwidth
+		// row. Surface the retrieved log into report.json the way the pod-failed
+		// path does via emitDiagnosticBlock — without it, a succeeded-but-
+		// unparseable run is a dead end: we cannot tell an empty/truncated log
+		// capture from a benchmark that exited 0 without emitting the results
+		// table. (The caller discards the returned logs string on error, so
+		// logging is the only way this reaches the check's captured stdout.)
+		slog.Error("NCCL launcher succeeded but bandwidth could not be parsed; dumping launcher log",
+			"logBytes", len(logs))
+		emitDiagnosticBlock("launcher log (bandwidth parse failed)", tailLines(strings.TrimSpace(logs), maxDiagLogLines))
 		return logs, false, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to parse bandwidth from logs", err)
 	}
 
@@ -1315,16 +1325,78 @@ func waitForLauncherPodAndGetLogs(ctx *validators.Context, podHelper *helper.Pod
 		return logs, aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "pod failed to complete successfully", err)
 	}
 
-	// Get logs from completed pod using helper method
+	// Get logs from the completed pod. A pod that has just reached Succeeded can
+	// briefly serve an empty or truncated log if its container is being torn down
+	// mid-read, and the NCCL results table (which parseBandwidthFromLogs keys on)
+	// prints last — so re-read until the results are present before returning.
 	slog.Info("Retrieving logs from successful pod...")
-	logs, err := podHelper.GetPodLogs(ctx.Ctx, launcherPod)
+	logs, err := getCompleteLauncherLogs(ctx.Ctx, podHelper, launcherPod)
 	if err != nil {
 		return "", aicrErrors.Wrap(aicrErrors.ErrCodeInternal, "failed to get pod logs", err)
 	}
 
 	return logs, nil
 }
 
+// ncclLauncherLogComplete reports whether a launcher log contains the NCCL
+// results parseBandwidthFromLogs needs. all_reduce_perf prints its "Avg bus
+// bandwidth" summary line only after the full size sweep finishes, so its
+// presence guarantees the trailing largest-message-size row — the row the parser
+// keys on (last regexp match) — is already in the log. We deliberately do NOT
+// accept a bare data-row match here: an early row can appear while the log is
+// still streaming, and gating on it would let the retry loop short-circuit
+// before the largest row lands, defeating the purpose (parseBandwidthFromLogs
+// would then read a smaller-size row).
+func ncclLauncherLogComplete(logs string) bool {
+	return strings.Contains(logs, "Avg bus bandwidth")
+}
+
+// getCompleteLauncherLogs retrieves the launcher pod's logs, re-reading until the
+// NCCL results are present or the attempt budget is exhausted. A pod that has
+// just reached Succeeded can serve an empty or truncated log if its container is
+// torn down while we read; because the parser keys on the trailing
+// largest-message-size row, a truncated read loses exactly that row and yields
+// "could not find bandwidth value in logs".
+func getCompleteLauncherLogs(ctx context.Context, podHelper *helper.PodLifecycle, pod *v1.Pod) (string, error) {
+	return readLauncherLogsUntilComplete(ctx,
+		func(c context.Context) (string, error) { return podHelper.GetPodLogs(c, pod) },
+		defaults.NCCLLauncherLogReadAttempts, defaults.NCCLLauncherLogReadInterval)
+}
+
+// readLauncherLogsUntilComplete re-reads via fetch until ncclLauncherLogComplete
+// is satisfied or attempts is exhausted, sleeping interval between tries. It
+// returns the last read even when still incomplete, so the caller's parse-failure
+// path can surface it for diagnosis rather than discarding it. Split from
+// getCompleteLauncherLogs so the retry logic is unit-testable without a cluster.
+func readLauncherLogsUntilComplete(ctx context.Context, fetch func(context.Context) (string, error), attempts int, interval time.Duration) (string, error) {
+	var logs string
+	for attempt := 1; ; attempt++ {
+		var err error
+		logs, err = fetch(ctx)
+		if err != nil {
+			return "", err
+		}
+		if ncclLauncherLogComplete(logs) {
+			if attempt > 1 {
+				slog.Info("launcher log complete after re-read", "attempts", attempt, "logBytes", len(logs))
+			}
+			return logs, nil
+		}
+		if attempt >= attempts {
+			slog.Warn("launcher log still lacks NCCL results after re-reads; returning last read for diagnosis",
+				"attempts", attempt, "logBytes", len(logs))
+			return logs, nil
+		}
+		slog.Info("launcher log has no NCCL results yet; re-reading", "attempt", attempt, "logBytes", len(logs))
+		select {
+		case <-ctx.Done():
+			// Return what we have; the caller's parse path will surface it.
+			return logs, nil
+		case <-time.After(interval):
+		}
+	}
+}
+
 // maxDiagLogLines bounds how many trailing log lines are kept per worker
 // container in the failure diagnostics. The fatal error is almost always near
 // the end, so the tail is what matters; the cap keeps a verbose worker

validators/performance/nccl_launcher_logs_test.go

@@ -0,0 +1,137 @@
+// Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.  All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	stderrors "errors"
+	"testing"
+	"time"
+)
+
+// A representative all_reduce_perf results block (one data row + summary trailer).
+const ncclCompleteLog = `#       size         count      type   redop    root     time   algbw   busbw #wrong
+#        (B)    (elements)                               (us)  (GB/s)  (GB/s)
+ 8589934592    2147483648     float     sum      -1   48298   177.85  333.47      0   48292   177.87  333.51      0
+# Out of bounds values : 0 OK
+# Avg bus bandwidth    : 333.49
+`
+
+func TestNcclLauncherLogComplete(t *testing.T) {
+	tests := []struct {
+		name string
+		logs string
+		want bool
+	}{
+		{"empty", "", false},
+		{"header only (no data rows, no trailer)", "#   size   count   type\n#   (B)\n", false},
+		{"init noise only", "NCCL INFO Bootstrap : Using eth0\nsome mpirun warmup line\n", false},
+		// An early data row without the trailer must NOT count as complete: the
+		// parser keys on the *last* row, so a log truncated before the largest
+		// size would otherwise short-circuit the retry loop (CodeRabbit #1691).
+		{"early data row but no trailer", "#       size ...\n         1024          256     float     sum      -1     12    0.10    0.09      0     12    0.10    0.09      0\n", false},
+		{"full sweep with trailer", ncclCompleteLog, true},
+		{"summary trailer only", "# Avg bus bandwidth    : 333.49\n", true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ncclLauncherLogComplete(tt.logs); got != tt.want {
+				t.Errorf("ncclLauncherLogComplete() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestReadLauncherLogsUntilComplete_ReturnsWhenComplete(t *testing.T) {
+	// First read is already complete — returns immediately, no retries.
+	calls := 0
+	fetch := func(context.Context) (string, error) { calls++; return ncclCompleteLog, nil }
+	logs, err := readLauncherLogsUntilComplete(context.Background(), fetch, 5, time.Millisecond)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !ncclLauncherLogComplete(logs) {
+		t.Errorf("expected complete logs, got %q", logs)
+	}
+	if calls != 1 {
+		t.Errorf("expected 1 fetch, got %d", calls)
+	}
+}
+
+func TestReadLauncherLogsUntilComplete_RetriesUntilComplete(t *testing.T) {
+	// Truncated (empty) reads twice, then the full results table lands.
+	calls := 0
+	fetch := func(context.Context) (string, error) {
+		calls++
+		if calls < 3 {
+			return "", nil
+		}
+		return ncclCompleteLog, nil
+	}
+	logs, err := readLauncherLogsUntilComplete(context.Background(), fetch, 5, time.Millisecond)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !ncclLauncherLogComplete(logs) {
+		t.Errorf("expected complete logs after retries, got %q", logs)
+	}
+	if calls != 3 {
+		t.Errorf("expected 3 fetches, got %d", calls)
+	}
+}
+
+func TestReadLauncherLogsUntilComplete_ReturnsLastReadWhenNeverComplete(t *testing.T) {
+	// Never completes: must return the last (incomplete) read after the budget,
+	// not an error — so the caller's parse-failure path can surface it.
+	calls := 0
+	fetch := func(context.Context) (string, error) { calls++; return "partial noise, no table", nil }
+	logs, err := readLauncherLogsUntilComplete(context.Background(), fetch, 3, time.Millisecond)
+	if err != nil {
+		t.Fatalf("expected no error (last read returned for diagnosis), got %v", err)
+	}
+	if ncclLauncherLogComplete(logs) {
+		t.Errorf("expected incomplete logs to be returned as-is, got complete %q", logs)
+	}
+	if calls != 3 {
+		t.Errorf("expected exactly the attempt budget (3) fetches, got %d", calls)
+	}
+}
+
+func TestReadLauncherLogsUntilComplete_PropagatesFetchError(t *testing.T) {
+	sentinel := stderrors.New("logs api unavailable")
+	fetch := func(context.Context) (string, error) { return "", sentinel }
+	_, err := readLauncherLogsUntilComplete(context.Background(), fetch, 5, time.Millisecond)
+	if !stderrors.Is(err, sentinel) {
+		t.Errorf("expected the fetch error to propagate, got %v", err)
+	}
+}
+
+func TestReadLauncherLogsUntilComplete_ReturnsOnContextCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // canceled before the retry sleep
+	calls := 0
+	fetch := func(context.Context) (string, error) { calls++; return "still no table", nil }
+	logs, err := readLauncherLogsUntilComplete(ctx, fetch, 5, time.Hour)
+	if err != nil {
+		t.Fatalf("expected no error on cancel (last read returned), got %v", err)
+	}
+	if ncclLauncherLogComplete(logs) {
+		t.Errorf("expected incomplete logs, got complete %q", logs)
+	}
+	// One fetch, then the canceled context short-circuits the sleep.
+	if calls != 1 {
+		t.Errorf("expected 1 fetch before cancel short-circuit, got %d", calls)
+	}
+}