From 71b0d497bbb80bc3daf15320640b609b15150470 Mon Sep 17 00:00:00 2001 From: Alexander Maslennikov Date: Fri, 14 Aug 2026 11:48:08 +0200 Subject: [PATCH] Calculate validation timeout from matrix plan Replace the fixed five-minute connectivity deadline with a budget derived from selected checks and ordered pod-pair batches. Preserve a positive --connectivity-timeout as an explicit hard override and log the selected total before execution. Signed-off-by: Alexander Maslennikov --- README.md | 7 + docs/reference/cli.md | 2 +- docs/user/validation.md | 25 ++++ pkg/cmd/validate.go | 2 +- pkg/cmd/validate_config_test.go | 12 ++ .../connectivity/connectivity.go | 77 ++++++---- .../connectivity/icmp.go | 3 +- .../connectivity/rdma.go | 14 +- .../connectivity/source.go | 4 +- .../connectivity/timeout.go | 118 ++++++++++++++++ .../connectivity/timeout_test.go | 133 ++++++++++++++++++ skills/k8s-launch-kit-validate/SKILL.md | 9 +- 12 files changed, 367 insertions(+), 39 deletions(-) create mode 100644 pkg/networkoperatorplugin/connectivity/timeout.go create mode 100644 pkg/networkoperatorplugin/connectivity/timeout_test.go diff --git a/README.md b/README.md index 1058aeb5..d218bb4c 100644 --- a/README.md +++ b/README.md @@ -440,6 +440,13 @@ Validation modes control cross-rail coverage and gating: `--rdma-ib-write-size `, and `--rdma-ib-write-min-bandwidth-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 ` to +replace it with an explicit end-to-end setup and execution deadline. + A self-contained HTML report lands at `/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), diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 7478c328..cd0d3682 100644 --- a/docs/reference/cli.md +++ b/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. | diff --git a/docs/user/validation.md b/docs/user/validation.md index b15458c7..866381c1 100644 --- a/docs/user/validation.md +++ b/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 diff --git a/pkg/cmd/validate.go b/pkg/cmd/validate.go index 410fdd8f..13597d7f 100644 --- a/pkg/cmd/validate.go +++ b/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.") diff --git a/pkg/cmd/validate_config_test.go b/pkg/cmd/validate_config_test.go index 2262059e..288a5560 100644 --- a/pkg/cmd/validate_config_test.go +++ b/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) } diff --git a/pkg/networkoperatorplugin/connectivity/connectivity.go b/pkg/networkoperatorplugin/connectivity/connectivity.go index 1059dbcb..a21cdb6b 100644 --- a/pkg/networkoperatorplugin/connectivity/connectivity.go +++ b/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 { diff --git a/pkg/networkoperatorplugin/connectivity/icmp.go b/pkg/networkoperatorplugin/connectivity/icmp.go index 3d679aeb..a61fc76c 100644 --- a/pkg/networkoperatorplugin/connectivity/icmp.go +++ b/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) diff --git a/pkg/networkoperatorplugin/connectivity/rdma.go b/pkg/networkoperatorplugin/connectivity/rdma.go index 1fbf2001..aca2fcc3 100644 --- a/pkg/networkoperatorplugin/connectivity/rdma.go +++ b/pkg/networkoperatorplugin/connectivity/rdma.go @@ -222,7 +222,7 @@ func RunRPing(ctx context.Context, restConfig *rest.Config, serverNamespace stri clientCmd := shellWithTimeout( fmt.Sprintf("rping -c -I %s -a %s -p 9999 -C %d -v", shellArg(test.SrcIP), shellArg(test.DstIP), iterations), - commandTimeoutFor(test, 30*time.Second)) + commandTimeoutFor(test, rpingCommandTimeout)) cliRes, cliErr := kubeclient.ExecInPod(tctx, restConfig, clientNamespace, clientPod, clientContainer, []string{"/bin/sh", "-c", clientCmd}) r.Stdout, r.Stderr = cliRes.Stdout, cliRes.Stderr finalizeExpectedResult(&r, cliErr == nil, cliErr) @@ -361,7 +361,7 @@ func rpingBatchClientCommand(tests []PingTest, iterations int) string { var b strings.Builder b.WriteString(`run_with_timeout() { seconds="$1"; shift; if command -v timeout >/dev/null 2>&1; then timeout --kill-after=2s "${seconds}s" "$@"; else "$@" & pid=$!; (sleep "$seconds"; kill -TERM "$pid" 2>/dev/null; sleep 2; kill -KILL "$pid" 2>/dev/null) & watchdog=$!; wait "$pid"; rc=$?; kill "$watchdog" 2>/dev/null; wait "$watchdog" 2>/dev/null; return "$rc"; fi; }; `) for i, test := range tests { - timeoutSeconds := int(commandTimeoutFor(test, 30*time.Second).Seconds()) + timeoutSeconds := int(commandTimeoutFor(test, rpingCommandTimeout).Seconds()) if timeoutSeconds <= 0 { timeoutSeconds = 1 } @@ -400,7 +400,7 @@ func parseRPingBatchResults(stdout string) map[int]int { func rdmaBatchTimeoutFor(tests []PingTest) time.Duration { timeout := 15 * time.Second for _, test := range tests { - timeout += commandTimeoutFor(test, 30*time.Second) + rdmaBatchSettleDelayFor([]PingTest{test}) + timeout += commandTimeoutFor(test, rpingCommandTimeout) + rdmaBatchSettleDelayFor([]PingTest{test}) } return timeout } @@ -476,7 +476,7 @@ func RunIbWriteBw(ctx context.Context, restConfig *rest.Config, serverNamespace clientCmd := shellWithTimeout( fmt.Sprintf("ib_write_bw -d %s -R -s %d --report_gbits -p %d --bind_source_ip %s %s", shellArg(test.SrcRDMADev), size, port, shellArg(test.SrcIP), shellArg(test.DstIP)), - commandTimeoutFor(test, 45*time.Second)) + commandTimeoutFor(test, ibWriteBwCommandTimeout)) cliRes, cliErr := kubeclient.ExecInPod(tctx, restConfig, clientNamespace, clientPod, clientContainer, []string{"/bin/sh", "-c", clientCmd}) r.Stdout, r.Stderr = cliRes.Stdout, cliRes.Stderr @@ -645,7 +645,7 @@ func ibWriteBwBatchClientCommandMode(tests []PingTest, size int, gpuDirect bool) var b strings.Builder b.WriteString(`run_with_timeout() { seconds="$1"; shift; if command -v timeout >/dev/null 2>&1; then timeout --kill-after=2s "${seconds}s" "$@"; else "$@" & pid=$!; (sleep "$seconds"; kill -TERM "$pid" 2>/dev/null; sleep 2; kill -KILL "$pid" 2>/dev/null) & watchdog=$!; wait "$pid"; rc=$?; kill "$watchdog" 2>/dev/null; wait "$watchdog" 2>/dev/null; return "$rc"; fi; }; `) for i, test := range tests { - timeoutSeconds := int(commandTimeoutFor(test, 45*time.Second).Seconds()) + timeoutSeconds := int(commandTimeoutFor(test, ibWriteBwCommandTimeout).Seconds()) if timeoutSeconds <= 0 { timeoutSeconds = 1 } @@ -733,7 +733,7 @@ func ibWriteBwBatchPort(index int) int { func ibWriteBwBatchTimeoutFor(tests []PingTest) time.Duration { timeout := 20 * time.Second for _, test := range tests { - timeout += commandTimeoutFor(test, 45*time.Second) + rdmaBatchSettleDelayFor([]PingTest{test}) + timeout += commandTimeoutFor(test, ibWriteBwCommandTimeout) + rdmaBatchSettleDelayFor([]PingTest{test}) } return timeout } @@ -757,7 +757,7 @@ func bandwidthVerdict(bw, minBandwidthGbps float64) (bool, error) { func killRDMAServer(restConfig *rest.Config, namespace, pod, container, prog, pid string) { // Use a fresh background context so the cleanup runs even if // the test's deadline already fired. - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), rdmaServerCleanupTimeout) defer cancel() cmd := fmt.Sprintf("kill %s 2>/dev/null; pkill %s 2>/dev/null; true", pid, shellArg(prog)) _, _ = kubeclient.ExecInPod(ctx, restConfig, namespace, pod, container, []string{"/bin/sh", "-c", cmd}) diff --git a/pkg/networkoperatorplugin/connectivity/source.go b/pkg/networkoperatorplugin/connectivity/source.go index 91c7dffb..69f7421a 100644 --- a/pkg/networkoperatorplugin/connectivity/source.go +++ b/pkg/networkoperatorplugin/connectivity/source.go @@ -75,7 +75,7 @@ func shellWithTimeout(command string, timeout time.Duration) string { func commandTimeoutFor(test PingTest, defaultTimeout time.Duration) time.Duration { if test.Expectation == ExpectForbidden || test.Expectation == ExpectObserve { - return 5 * time.Second + return nonRequiredCommandTimeout } return defaultTimeout } @@ -100,7 +100,7 @@ func checkRoute(ctx context.Context, restConfig *rest.Config, namespace, pod, co out.Err = "missing source or destination IP" return out } - tctx, cancel := context.WithTimeout(ctx, 10*time.Second) + tctx, cancel := context.WithTimeout(ctx, routeCheckTimeout) defer cancel() res, err := kubeclient.ExecInPod(tctx, restConfig, namespace, pod, container, []string{"/bin/sh", "-c", cmd}) out.Output = strings.TrimSpace(strings.Join([]string{res.Stdout, res.Stderr}, "\n")) diff --git a/pkg/networkoperatorplugin/connectivity/timeout.go b/pkg/networkoperatorplugin/connectivity/timeout.go new file mode 100644 index 00000000..1de6dc39 --- /dev/null +++ b/pkg/networkoperatorplugin/connectivity/timeout.go @@ -0,0 +1,118 @@ +// Copyright 2026 NVIDIA CORPORATION & AFFILIATES +// +// 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. +// +// SPDX-License-Identifier: Apache-2.0 + +package connectivity + +import "time" + +const ( + automaticSetupTimeout = 10 * time.Minute + automaticSafetyMinimum = 30 * time.Second + matrixCleanupTimeout = 30 * time.Second + rdmaServerCleanupTimeout = 3 * time.Second + routeCheckTimeout = 10 * time.Second + nonRequiredCommandTimeout = 5 * time.Second + icmpCommandTimeout = 5 * time.Second + rpingCommandTimeout = 30 * time.Second + ibWriteBwCommandTimeout = 45 * time.Second + automaticSafetyMarginDivisor = 10 +) + +type timeoutBudget struct { + Setup time.Duration + Tests time.Duration + SafetyMargin time.Duration + Cleanup time.Duration + Total time.Duration + PlannedTests int +} + +func (b timeoutBudget) executionTimeout() time.Duration { + return b.Tests + b.SafetyMargin +} + +// automaticTimeoutBudget mirrors the serial execution graph in RunMatrix. +// Route probes run before each selected RDMA family, RDMA commands are grouped +// by ordered pod pair, and ICMP tests run one-by-one. Keeping the calculation +// beside the command timeout constants makes the default deadline grow with +// the actual plan instead of relying on a fixed cluster-size assumption. +func automaticTimeoutBudget(plan MatrixPlan, checks []Check) timeoutBudget { + checks = normalizeChecks(checks) + budget := timeoutBudget{ + Setup: automaticSetupTimeout, + Cleanup: matrixCleanupTimeout, + } + + for _, check := range checks { + tests := testsForCheck(plan, check) + budget.PlannedTests += len(tests) + if len(tests) == 0 { + continue + } + + switch check { + case CheckICMP: + budget.Tests += requiredRouteBudget(tests) + for _, test := range tests { + budget.Tests += commandTimeoutFor(test, icmpCommandTimeout) + } + case CheckRPing: + budget.Tests += requiredRouteBudget(tests) + for _, batch := range groupRPingTests(tests) { + budget.Tests += rdmaBatchTimeoutFor(batch) + rdmaServerCleanupTimeout + } + case CheckIBWriteBW, CheckGPUDirectDMABuf: + budget.Tests += requiredRouteBudget(tests) + for _, batch := range groupRPingTests(tests) { + budget.Tests += ibWriteBwBatchTimeoutFor(batch) + rdmaServerCleanupTimeout + } + } + } + + if budget.Tests > 0 { + budget.SafetyMargin = budget.Tests / automaticSafetyMarginDivisor + if budget.SafetyMargin < automaticSafetyMinimum { + budget.SafetyMargin = automaticSafetyMinimum + } + } + budget.Total = budget.Setup + budget.executionTimeout() + budget.Cleanup + return budget +} + +func testsForCheck(plan MatrixPlan, check Check) []PingTest { + switch check { + case CheckICMP: + return append(append([]PingTest{}, plan.ICMPSameRail...), plan.ICMPCrossRail...) + case CheckRPing: + return append(append([]PingTest{}, plan.RDMASameRail...), plan.RDMACrossRail...) + case CheckIBWriteBW: + return append(append([]PingTest{}, plan.RDMABwSameRail...), plan.RDMABwCrossRail...) + case CheckGPUDirectDMABuf: + return append(append([]PingTest{}, plan.GPUDirectDMABufSameRail...), plan.GPUDirectDMABufCrossRail...) + default: + return nil + } +} + +func requiredRouteBudget(tests []PingTest) time.Duration { + var budget time.Duration + for _, test := range tests { + if test.Expectation == "" || test.Expectation == ExpectRequired { + budget += routeCheckTimeout + } + } + return budget +} diff --git a/pkg/networkoperatorplugin/connectivity/timeout_test.go b/pkg/networkoperatorplugin/connectivity/timeout_test.go new file mode 100644 index 00000000..1256a39a --- /dev/null +++ b/pkg/networkoperatorplugin/connectivity/timeout_test.go @@ -0,0 +1,133 @@ +// Copyright 2026 NVIDIA CORPORATION & AFFILIATES +// +// 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. +// +// SPDX-License-Identifier: Apache-2.0 + +package connectivity + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestAutomaticTimeoutBudgetMirrorsSelectedExecutionPlan(t *testing.T) { + tests := []PingTest{ + {SrcPod: "pod-a", DstPod: "pod-b", Expectation: ExpectRequired}, + {SrcPod: "pod-a", DstPod: "pod-b", Expectation: ExpectObserve}, + {SrcPod: "pod-b", DstPod: "pod-a", Expectation: ExpectForbidden}, + } + plan := MatrixPlan{ + ICMPSameRail: append([]PingTest(nil), tests...), + RDMASameRail: append([]PingTest(nil), tests...), + RDMABwSameRail: append([]PingTest(nil), tests...), + GPUDirectDMABufSameRail: append([]PingTest(nil), tests...), + } + + budget := automaticTimeoutBudget(plan, []Check{ + CheckRPing, + CheckIBWriteBW, + CheckGPUDirectDMABuf, + CheckICMP, + }) + + assert.Equal(t, 12, budget.PlannedTests) + assert.Equal(t, automaticSetupTimeout, budget.Setup) + assert.Equal(t, 5*time.Minute+51*time.Second, budget.Tests) + assert.Equal(t, 35*time.Second+100*time.Millisecond, budget.SafetyMargin) + assert.Equal(t, 6*time.Minute+26*time.Second+100*time.Millisecond, budget.executionTimeout()) + assert.Equal(t, matrixCleanupTimeout, budget.Cleanup) + assert.Equal(t, 16*time.Minute+56*time.Second+100*time.Millisecond, budget.Total) +} + +func TestAutomaticTimeoutBudgetIncludesOnlySelectedChecks(t *testing.T) { + plan := MatrixPlan{ + ICMPSameRail: []PingTest{ + {SrcPod: "pod-a", DstPod: "pod-b", Expectation: ExpectRequired}, + {SrcPod: "pod-b", DstPod: "pod-a", Expectation: ExpectObserve}, + }, + RDMASameRail: []PingTest{ + {SrcPod: "pod-a", DstPod: "pod-b", Expectation: ExpectRequired}, + }, + } + + budget := automaticTimeoutBudget(plan, []Check{CheckICMP}) + + assert.Equal(t, 2, budget.PlannedTests) + assert.Equal(t, 20*time.Second, budget.Tests) + assert.Equal(t, automaticSafetyMinimum, budget.SafetyMargin) + assert.Equal(t, automaticSetupTimeout+50*time.Second+matrixCleanupTimeout, budget.Total) +} + +func TestAutomaticTimeoutBudgetForThreeNodeFourRailQuickMatrix(t *testing.T) { + pairs := [][2]string{ + {"pod-a", "pod-b"}, + {"pod-a", "pod-c"}, + {"pod-b", "pod-a"}, + {"pod-b", "pod-c"}, + {"pod-c", "pod-a"}, + {"pod-c", "pod-b"}, + } + tests := make([]PingTest, 0, 36) + for _, pair := range pairs { + for range 4 { + tests = append(tests, PingTest{ + SrcPod: pair[0], DstPod: pair[1], Expectation: ExpectRequired, + }) + } + } + for range 12 { + tests = append(tests, PingTest{ + SrcPod: "pod-a", DstPod: "pod-b", Expectation: ExpectRequired, + }) + } + plan := MatrixPlan{ + ICMPSameRail: append([]PingTest(nil), tests...), + RDMASameRail: append([]PingTest(nil), tests...), + RDMABwSameRail: append([]PingTest(nil), tests...), + GPUDirectDMABufSameRail: append([]PingTest(nil), tests...), + } + + budget := automaticTimeoutBudget(plan, []Check{ + CheckRPing, + CheckIBWriteBW, + CheckGPUDirectDMABuf, + CheckICMP, + }) + + assert.Equal(t, 144, budget.PlannedTests) + assert.Equal(t, 2*time.Hour+10*time.Minute+24*time.Second, budget.Total) +} + +func TestAutomaticTimeoutBudgetUsesDefaultChecksAndNoSafetyForEmptyPlan(t *testing.T) { + budget := automaticTimeoutBudget(MatrixPlan{}, nil) + + assert.Zero(t, budget.PlannedTests) + assert.Zero(t, budget.Tests) + assert.Zero(t, budget.SafetyMargin) + assert.Zero(t, budget.executionTimeout()) + assert.Equal(t, automaticSetupTimeout+matrixCleanupTimeout, budget.Total) +} + +func TestRequiredRouteBudgetTreatsEmptyExpectationAsRequired(t *testing.T) { + tests := []PingTest{ + {Expectation: ""}, + {Expectation: ExpectRequired}, + {Expectation: ExpectObserve}, + {Expectation: ExpectForbidden}, + } + + assert.Equal(t, 2*routeCheckTimeout, requiredRouteBudget(tests)) +} diff --git a/skills/k8s-launch-kit-validate/SKILL.md b/skills/k8s-launch-kit-validate/SKILL.md index 8311783a..907364f1 100644 --- a/skills/k8s-launch-kit-validate/SKILL.md +++ b/skills/k8s-launch-kit-validate/SKILL.md @@ -1,6 +1,6 @@ --- name: k8s-launch-kit-validate -version: 1.0.3 +version: 1.0.4 description: "Use this skill when the user wants to verify that an NVIDIA networking deployment matches the configuration that produced it. Activate for: 'is my deployment correct', 'are all the manifests applied', 'does the network operator version match', 'verify deployment', 'check cluster state against config', or any question about whether the cluster reflects what l8k generated. Wraps the `l8k validate` subcommand." metadata: requires: @@ -75,6 +75,7 @@ l8k validate [--user-config ] [--deployment-files ] [--kubeconfig