Calculate validation timeout from matrix plan

README.md

@@ -440,6 +440,13 @@ Validation modes control cross-rail coverage and gating:
 `--rdma-ib-write-size <BYTES>`, and
 `--rdma-ib-write-min-bandwidth-gbps <GBPS>` (`0` disables bandwidth gating).
 
+The connectivity timeout is automatic by default. After the test workload is
+ready and the matrix has been planned, validate derives a total budget from
+the selected checks, their per-command limits, ordered pod-pair batches, a
+bounded setup allowance, cleanup, and a safety margin. The calculated total
+is printed before test execution. Use `--connectivity-timeout <DURATION>` to
+replace it with an explicit end-to-end setup and execution deadline.
+
 A self-contained HTML report lands at `<deployment-files>/k8s-launch-kit-validation-report.html`
 by default (override with `--report-path`, disable with `--report-path=-`).
 The report has: header (l8k version, kubeconfig context, API-server version),

docs/reference/cli.md

@@ -150,7 +150,7 @@ full deletion boundary.
 | `--connectivity` | Enable or disable data-plane connectivity checks. |
 | `--validation-mode` | `quick`, `full`, or `strict`. |
 | `--validation-checks` | Comma-separated list of `icmp`, `rping`, and `ib_write_bw`. Enabled GPUDirect DMA-BUF validation follows the `ib_write_bw` selection. |
-| `--connectivity-timeout` | Wall-clock budget for connectivity workload rollout and test execution. |
+| `--connectivity-timeout` | Maximum connectivity workload setup and execution duration. `0` (default) calculates the total budget from the generated matrix plan; a positive duration is an explicit hard deadline. |
 | `--rdma-rping-iterations` | Override `validation.rdma.rpingIterations`. |
 | `--rdma-ib-write-size` | Override `validation.rdma.ibWriteSize`. |
 | `--rdma-ib-write-min-bandwidth-gbps` | Minimum `ib_write_bw` peak bandwidth. |

docs/user/validation.md

@@ -104,6 +104,31 @@ l8k validate \
   --rdma-ib-write-min-bandwidth-gbps 100
 ```
 
+## Connectivity Timeout
+
+By default, `--connectivity-timeout=0` selects an automatic timeout. Launch Kit
+first applies and discovers the generated validation workload under a bounded
+setup allowance. Once the matrix is known, it calculates the total budget from
+the selected tests, their individual command limits, ordered pod-pair batches,
+cleanup allowances, and a safety margin. The log reports the calculated total
+before connectivity test execution starts:
+
+```text
+Connectivity timeout automatically calculated from 144 planned tests: 2h10m24s total budget
+```
+
+Set a positive duration to replace the automatic budget with an explicit hard
+deadline for connectivity workload setup and execution:
+
+```bash
+l8k validate --connectivity-timeout 45m
+```
+
+The explicit deadline is useful for fitting validation into an external
+maintenance or CI window. Test DaemonSet and RDMA-process cleanup use short,
+independent best-effort contexts so cleanup is still attempted after either an
+automatic or user-supplied deadline expires.
+
 Disable only the connectivity stage:
 
 ```bash

pkg/cmd/validate.go

@@ -145,7 +145,7 @@ func init() {
 	validateCmd.Flags().StringVar(&networkOperatorNamespace, "network-operator-namespace", "", "Override the network operator namespace from cluster-config.yaml")
 	validateCmd.Flags().BoolVar(&validateConnectivity, "connectivity", true, "Run a source-bound connectivity matrix (icmp + rping + ib_write_bw) between pods of the example DaemonSet. Default true. Pass --connectivity=false to skip when only the static manifest checks are wanted.")
 	validateCmd.Flags().BoolVar(&validateKeep, "keep", false, "Leave the example DaemonSet running after --connectivity completes (useful for debugging).")
-	validateCmd.Flags().DurationVar(&validateConnectivityTimeout, "connectivity-timeout", 5*time.Minute, "Wall-clock budget for the connectivity matrix (DaemonSet rollout + icmp + rping + ib_write_bw execs).")
+	validateCmd.Flags().DurationVar(&validateConnectivityTimeout, "connectivity-timeout", 0, "Maximum wall-clock budget for connectivity workload setup and test execution. 0 (default) calculates the budget from the generated matrix plan.")
 	validateCmd.Flags().StringVar(&validateMode, "validation-mode", "", "Connectivity validation mode: quick, full, or strict. Overrides validation.mode from cluster-config.yaml.")
 	validateCmd.Flags().StringSliceVar(&validateChecks, "validation-checks", nil, "Comma-separated checks to run during connectivity validation. Supported: icmp, rping, ib_write_bw. Overrides validation.checks from cluster-config.yaml.")
 	validateCmd.Flags().IntVar(&validateRDMAIterations, "rdma-rping-iterations", 0, "Number of rping client iterations. Overrides validation.rdma.rpingIterations from cluster-config.yaml.")

pkg/cmd/validate_config_test.go

@@ -18,6 +18,7 @@ package cmd
 
 import (
 	"testing"
+	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/assert"
@@ -32,6 +33,7 @@ func newValidateRequestTestCommand() *cobra.Command {
 	cmd.Flags().IntVar(&validateRDMAIterations, "rdma-rping-iterations", 0, "")
 	cmd.Flags().IntVar(&validateRDMAIBWriteSize, "rdma-ib-write-size", 0, "")
 	cmd.Flags().Float64Var(&validateRDMAIBWriteMinGbps, "rdma-ib-write-min-bandwidth-gbps", 0, "")
+	cmd.Flags().DurationVar(&validateConnectivityTimeout, "connectivity-timeout", 0, "")
 	return cmd
 }
 
@@ -43,6 +45,7 @@ func resetValidateRequestGlobals(t *testing.T) {
 	validateRDMAIterations = 0
 	validateRDMAIBWriteSize = 0
 	validateRDMAIBWriteMinGbps = 0
+	validateConnectivityTimeout = 0
 }
 
 func TestNewHostValidateRequestCapturesExplicitValues(t *testing.T) {
@@ -56,6 +59,7 @@ func TestNewHostValidateRequestCapturesExplicitValues(t *testing.T) {
 	require.NoError(t, cmd.Flags().Set("rdma-rping-iterations", "11"))
 	require.NoError(t, cmd.Flags().Set("rdma-ib-write-size", "8192"))
 	require.NoError(t, cmd.Flags().Set("rdma-ib-write-min-bandwidth-gbps", "0"))
+	require.NoError(t, cmd.Flags().Set("connectivity-timeout", "42m"))
 
 	request := newHostValidateRequest(cmd)
 	assert.True(t, request.Connectivity.Set)
@@ -70,6 +74,7 @@ func TestNewHostValidateRequestCapturesExplicitValues(t *testing.T) {
 	assert.True(t, request.RDMAIBWriteSize.Set)
 	assert.Zero(t, request.RDMAMinBandwidth.Value)
 	assert.True(t, request.RDMAMinBandwidth.Set)
+	assert.Equal(t, 42*time.Minute, request.ConnectivityTime)
 }
 
 func TestNewHostValidateRequestPreservesOmission(t *testing.T) {
@@ -83,4 +88,11 @@ func TestNewHostValidateRequestPreservesOmission(t *testing.T) {
 	assert.False(t, request.RDMAPIterations.Set)
 	assert.False(t, request.RDMAIBWriteSize.Set)
 	assert.False(t, request.RDMAMinBandwidth.Set)
+	assert.Zero(t, request.ConnectivityTime)
+}
+
+func TestValidateConnectivityTimeoutDefaultsToAutomatic(t *testing.T) {
+	flag := validateCmd.Flags().Lookup("connectivity-timeout")
+	require.NotNil(t, flag)
+	assert.Equal(t, "0s", flag.DefValue)
 }

pkg/networkoperatorplugin/connectivity/connectivity.go

@@ -48,8 +48,9 @@ type Options struct {
 	// manifests live under (validate uses the same dir it just
 	// validated).
 	ManifestDir string
-	// Timeout caps the whole connectivity phase: apply + DS rollout
-	// + connectivity test execs + cleanup. 0 falls back to 5 minutes.
+	// Timeout caps connectivity workload setup and test execution when
+	// positive. Zero selects an automatic budget derived from the generated
+	// matrix plan. Cleanup uses its own short context in either mode.
 	Timeout time.Duration
 	// Keep leaves the test DaemonSets running after the matrix
 	// completes, for follow-up debugging. Default is to delete.
@@ -130,8 +131,8 @@ type DaemonSetReport struct {
 // All UI output flows through `uiOutput` (caller passes
 // ui.FromContext(ctx)). Logs go to controller-runtime's logr.
 func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, uiOutput ui.Output, opts Options) (*MatrixResult, error) {
-	if opts.Timeout <= 0 {
-		opts.Timeout = 5 * time.Minute
+	if opts.Timeout < 0 {
+		return nil, fmt.Errorf("connectivity timeout must be greater than or equal to zero")
 	}
 	opts.Checks = normalizeChecks(opts.Checks)
 	if opts.Mode == "" {
@@ -144,15 +145,31 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		opts.IBWriteSize = 65536
 	}
 
-	ctx, cancel := context.WithTimeout(ctx, opts.Timeout)
-	defer cancel()
-
 	uiOutput.Section("Connectivity matrix")
 	if len(opts.Checks) == 0 {
 		return &MatrixResult{
 			Skipped: &MatrixSkip{Reason: "all connectivity checks are disabled"},
 		}, nil
 	}
+
+	automaticTimeout := opts.Timeout == 0
+	runCtx := ctx
+	runCancel := func() {}
+	setupTimeout := automaticSetupTimeout
+	if !automaticTimeout {
+		runCtx, runCancel = context.WithTimeout(ctx, opts.Timeout)
+		setupTimeout = opts.Timeout
+		uiOutput.Info("Connectivity timeout set by user: %s", opts.Timeout)
+	}
+	defer runCancel()
+
+	setupCtx := runCtx
+	setupCancel := func() {}
+	if automaticTimeout {
+		setupCtx, setupCancel = context.WithTimeout(runCtx, setupTimeout)
+	}
+	defer setupCancel()
+
 	uiOutput.Info("Loading example DaemonSet manifests from %s", opts.ManifestDir)
 
 	hasICMP := checksContain(opts.Checks, CheckICMP)
@@ -185,7 +202,7 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		}
 		// Use a fresh context for cleanup; the main ctx may have
 		// already been cancelled / timed out.
-		cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 30*time.Second)
+		cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), matrixCleanupTimeout)
 		defer cleanupCancel()
 		for _, ref := range refs {
 			if err := DeleteDaemonSet(cleanupCtx, c, ref); err != nil {
@@ -195,11 +212,11 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		}
 	}()
 
-	applyAndCollectPods := func(objs []*unstructured.Unstructured, refs []DaemonSetRef) ([]testPodWithDS, error) {
+	applyAndCollectPods := func(runCtx context.Context, objs []*unstructured.Unstructured, refs []DaemonSetRef) ([]testPodWithDS, error) {
 		for i, obj := range objs {
 			ref := refs[i]
 			uiOutput.Info("Applying %s/%s (from %s)", ref.Namespace, ref.Name, ref.SourceFile)
-			if err := ApplyDaemonSet(ctx, c, obj); err != nil {
+			if err := ApplyDaemonSet(runCtx, c, obj); err != nil {
 				return nil, fmt.Errorf("apply daemonset %s/%s: %w", ref.Namespace, ref.Name, err)
 			}
 		}
@@ -208,12 +225,12 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		out := make([]testPodWithDS, 0)
 		for _, ref := range refs {
 			uiOutput.Info("Waiting for DaemonSet %s/%s to roll out", ref.Namespace, ref.Name)
-			rollout, err := WaitForRollout(ctx, c, ref, opts.Timeout)
+			rollout, err := WaitForRollout(runCtx, c, ref, setupTimeout)
 			if err != nil {
 				return out, err
 			}
 			uiOutput.Success("DaemonSet %s/%s ready (%d/%d pods)", ref.Namespace, ref.Name, rollout.Ready, rollout.Desired)
-			pods, err := ListPods(ctx, c, ref)
+			pods, err := ListPods(runCtx, c, ref)
 			if err != nil {
 				return out, err
 			}
@@ -237,7 +254,7 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		return out, nil
 	}
 
-	buildMatrixPods := func(allPods []testPodWithDS, discoverRDMA bool) []TestPod {
+	buildMatrixPods := func(runCtx context.Context, allPods []testPodWithDS, discoverRDMA bool) []TestPod {
 		// Parse multus annotations and build the TestPod list. Each
 		// pod's RDMA-device-by-rail map is filled in from a single
 		// in-pod shell exec that reads
@@ -256,7 +273,7 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 				continue
 			}
 			if discoverRDMA {
-				tp.RDMADevsByRail = DiscoverRDMADevices(ctx, restConfig, p.pod.Namespace, p.pod.Name, p.ref.RDMAContainer, ifaceByRail)
+				tp.RDMADevsByRail = DiscoverRDMADevices(runCtx, restConfig, p.pod.Namespace, p.pod.Name, p.ref.RDMAContainer, ifaceByRail)
 				log.Log.V(1).Info("RDMA device discovery",
 					"pod", p.pod.Name, "rails", tp.RailOrder,
 					"rdmaDevsByRail", tp.RDMADevsByRail)
@@ -280,11 +297,11 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		return rdmaContainerByPod, icmpContainerByPod, namespaceByPod
 	}
 
-	allPods, err := applyAndCollectPods(objs, refs)
+	allPods, err := applyAndCollectPods(setupCtx, objs, refs)
 	if err != nil {
 		return result, err
 	}
-	testPods := buildMatrixPods(allPods, hasRDMA)
+	testPods := buildMatrixPods(setupCtx, allPods, hasRDMA)
 
 	plan := PlanWithOptions(testPods, opts.Mode, opts.Routing)
 	if plan.Skip != nil {
@@ -293,6 +310,16 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		return result, nil
 	}
 
+	executionCtx := runCtx
+	executionCancel := func() {}
+	if automaticTimeout {
+		budget := automaticTimeoutBudget(plan, opts.Checks)
+		uiOutput.Info("Connectivity timeout automatically calculated from %d planned tests: %s total budget", budget.PlannedTests, budget.Total)
+		setupCancel()
+		executionCtx, executionCancel = context.WithTimeout(ctx, budget.executionTimeout())
+	}
+	defer executionCancel()
+
 	totalSameRail := len(plan.RDMASameRail)
 	totalCrossRail := len(plan.RDMACrossRail)
 	plannedTests := 0
@@ -352,24 +379,24 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		{
 			check: CheckRPing,
 			label: "RDMA ping (rping)",
-			tests: append(append([]PingTest{}, plan.RDMASameRail...), plan.RDMACrossRail...),
+			tests: testsForCheck(plan, CheckRPing),
 		},
 		{
 			check: CheckIBWriteBW,
 			label: "RDMA bandwidth (ib_write_bw)",
-			tests: append(append([]PingTest{}, plan.RDMABwSameRail...), plan.RDMABwCrossRail...),
+			tests: testsForCheck(plan, CheckIBWriteBW),
 		},
 		{
 			check: CheckGPUDirectDMABuf,
 			label: "GPUDirect RDMA bandwidth (DMA-BUF)",
-			tests: append(append([]PingTest{}, plan.GPUDirectDMABufSameRail...), plan.GPUDirectDMABufCrossRail...),
+			tests: testsForCheck(plan, CheckGPUDirectDMABuf),
 		},
 		{
 			check: CheckICMP,
 			label: "Layer 3 ping (ICMP)",
-			tests: append(append([]PingTest{}, plan.ICMPSameRail...), plan.ICMPCrossRail...),
+			tests: testsForCheck(plan, CheckICMP),
 			run: func(t PingTest) PingResult {
-				return RunICMP(ctx, restConfig, namespaceByPod[t.SrcPod],
+				return RunICMP(executionCtx, restConfig, namespaceByPod[t.SrcPod],
 					t.SrcPod, icmpContainerByPod[t.SrcPod], t)
 			},
 		},
@@ -384,21 +411,21 @@ func RunMatrix(ctx context.Context, c client.Client, restConfig *rest.Config, ui
 		}
 		uiOutput.Info("Stage: %s — %d test(s)", stage.label, len(tests))
 		if stage.check != CheckICMP {
-			tests = checkSourceRoutes(ctx, restConfig, namespaceByPod, icmpContainerByPod, tests)
+			tests = checkSourceRoutes(executionCtx, restConfig, namespaceByPod, icmpContainerByPod, tests)
 		}
 		if stage.check == CheckRPing {
 			result.PingResults = append(result.PingResults,
-				RunRPingBatches(ctx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.RPingIterations)...)
+				RunRPingBatches(executionCtx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.RPingIterations)...)
 			continue
 		}
 		if stage.check == CheckIBWriteBW {
 			result.PingResults = append(result.PingResults,
-				RunIbWriteBwBatches(ctx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.IBWriteSize, opts.IBWriteMinBandwidthGbps)...)
+				RunIbWriteBwBatches(executionCtx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.IBWriteSize, opts.IBWriteMinBandwidthGbps)...)
 			continue
 		}
 		if stage.check == CheckGPUDirectDMABuf {
 			result.PingResults = append(result.PingResults,
-				RunGPUDirectDMABufBatches(ctx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.IBWriteSize, opts.IBWriteMinBandwidthGbps)...)
+				RunGPUDirectDMABufBatches(executionCtx, restConfig, namespaceByPod, rdmaContainerByPod, tests, opts.IBWriteSize, opts.IBWriteMinBandwidthGbps)...)
 			continue
 		}
 		for _, t := range tests {

pkg/networkoperatorplugin/connectivity/icmp.go

@@ -19,7 +19,6 @@ package connectivity
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/nvidia/k8s-launch-kit/pkg/kubeclient"
 	"k8s.io/client-go/rest"
@@ -42,7 +41,7 @@ func RunICMP(ctx context.Context, restConfig *rest.Config, namespace, pod, conta
 		}
 	}
 	cmd := shellWithTimeout(fmt.Sprintf("ping -c 1 -W 1 -I %s %s", shellArg(test.SrcIP), shellArg(test.DstIP)),
-		commandTimeoutFor(test, 5*time.Second))
+		commandTimeoutFor(test, icmpCommandTimeout))
 	res, err := kubeclient.ExecInPod(ctx, restConfig, namespace, pod, container, []string{"/bin/sh", "-c", cmd})
 	r.Stdout, r.Stderr = res.Stdout, res.Stderr
 	finalizeExpectedResult(&r, err == nil, err)