From 21ef0e71d8ae8c44e9d19fc9967a0e70313b7e97 Mon Sep 17 00:00:00 2001 From: Yuan Chen Date: Wed, 26 Aug 2026 16:14:46 -0700 Subject: [PATCH 1/4] fix(validators): retry transient reads while polling gang test pods A non-nil error from a wait.PollUntilContextCancel condition aborts the poll, so one throttled pod read failed the gang-scheduling check on a healthy cluster: client-go's own rate limiter returns "client rate limiter Wait returned an error" under load, and the next interval would have succeeded. Classify the read and retry the timeout forms, letting GangTestPodTimeout decide the verdict; genuine errors still fail closed. This is the defect #1513 fixed one step earlier in the same function, where an instantaneous deployment read became a bounded readiness wait. Signed-off-by: Yuan Chen --- validators/conformance/allocmode_bridge.go | 1 + .../conformance/gang_scheduling_check.go | 17 ++++- .../conformance/gang_scheduling_check_test.go | 70 +++++++++++++++++++ 3 files changed, 86 insertions(+), 2 deletions(-) diff --git a/validators/conformance/allocmode_bridge.go b/validators/conformance/allocmode_bridge.go index 6e5d3326c..f50b54720 100644 --- a/validators/conformance/allocmode_bridge.go +++ b/validators/conformance/allocmode_bridge.go @@ -32,5 +32,6 @@ var ( sortedNodeNames = allocmode.SortedNodeNames draAPIVersionPreference = allocmode.APIVersionPreference classifyK8sReadError = allocmode.ClassifyK8sReadError + isK8sTimeoutErr = allocmode.IsK8sTimeoutErr verifyGPUAllocationPolicy = allocmode.Verify ) diff --git a/validators/conformance/gang_scheduling_check.go b/validators/conformance/gang_scheduling_check.go index a9962eee4..8585618c6 100644 --- a/validators/conformance/gang_scheduling_check.go +++ b/validators/conformance/gang_scheduling_check.go @@ -19,6 +19,7 @@ import ( "crypto/rand" "encoding/hex" "fmt" + "log/slog" "strings" "time" @@ -341,8 +342,20 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru pod, err := clientset.CoreV1().Pods(run.namespace).Get( ctx, run.pods[i], metav1.GetOptions{}) if err != nil { - return false, errors.Wrap(errors.ErrCodeInternal, - fmt.Sprintf("failed to get gang test pod %s", run.pods[i]), err) + // A read that could not land is not a verdict. Returning a + // non-nil error here aborts the whole poll, so one throttled + // or timed-out call would fail a healthy cluster even though + // the next interval would have succeeded — the same defect + // #1513 fixed one step earlier in this function. Let the + // enclosing GangTestPodTimeout decide instead. + if isK8sTimeoutErr(err) { + slog.Debug("transient read while polling gang test pod; retrying", + "pod", run.pods[i], "error", err) + allDone = false + continue + } + return false, classifyK8sReadError(err, + fmt.Sprintf("gang test pod %s", run.pods[i])) } switch pod.Status.Phase { //nolint:exhaustive // only terminal states matter case corev1.PodSucceeded, corev1.PodFailed: diff --git a/validators/conformance/gang_scheduling_check_test.go b/validators/conformance/gang_scheduling_check_test.go index 8b33cbfc7..481b8189e 100644 --- a/validators/conformance/gang_scheduling_check_test.go +++ b/validators/conformance/gang_scheduling_check_test.go @@ -16,6 +16,8 @@ package main import ( "context" + "fmt" + "sync/atomic" "testing" corev1 "k8s.io/api/core/v1" @@ -25,6 +27,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/schema" dynamicfake "k8s.io/client-go/dynamic/fake" k8sfake "k8s.io/client-go/kubernetes/fake" + k8stesting "k8s.io/client-go/testing" ) // TestCleanupGangTestResourcesDeletesNamespace verifies the check tears down @@ -156,3 +159,70 @@ func TestCleanupGangTestResourcesPreservesConcurrentRun(t *testing.T) { } } } + +// TestWaitForGangTestPodsRetriesTransientReads pins the fix for #2406: a read +// that could not land must not decide the check. client-go's own rate limiter +// returns "client rate limiter Wait returned an error: context deadline +// exceeded" on a loaded cluster; before this fix that aborted the poll and +// failed a healthy cluster, even though the test pods had been created and the +// next interval would have succeeded. +// +// This mirrors the acceptance criteria of #1513, which fixed the same shape one +// step earlier in this function (instantaneous deployment read -> bounded wait). +func TestWaitForGangTestPodsRetriesTransientReads(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + + objs := make([]runtime.Object, 0, gangMinMembers) + for i := range gangMinMembers { + objs = append(objs, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: run.pods[i], Namespace: run.namespace}, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + }) + } + clientset := k8sfake.NewSimpleClientset(objs...) + + // Fail the first two reads the way a throttled client-go client does, then + // let the real objects through. + var reads atomic.Int32 + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) <= 2 { + return true, nil, fmt.Errorf( + "client rate limiter Wait returned an error: %w", context.DeadlineExceeded) + } + return false, nil, nil // fall through to the tracker + }) + + pods, err := waitForGangTestPods(context.Background(), clientset, run) + if err != nil { + t.Fatalf("waitForGangTestPods aborted on a transient read: %v", err) + } + for i := range gangMinMembers { + if pods[i] == nil { + t.Errorf("pod %d not collected after retry", i) + } + } + if got := reads.Load(); got < 3 { + t.Errorf("expected the poll to retry past the throttled reads, saw %d reads", got) + } +} + +// TestWaitForGangTestPodsFailsClosedOnTerminalRead is the other half: a genuine +// error (RBAC denial) must still abort rather than spin until the timeout. +func TestWaitForGangTestPodsFailsClosedOnTerminalRead(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + clientset := k8sfake.NewSimpleClientset() + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, k8serrors.NewForbidden( + schema.GroupResource{Resource: "pods"}, run.pods[0], fmt.Errorf("no access")) + }) + + if _, err := waitForGangTestPods(context.Background(), clientset, run); err == nil { + t.Fatal("expected a terminal read error to fail the check, got nil") + } +} From da26a02965c936462121e3310eb54983e364aca6 Mon Sep 17 00:00:00 2001 From: Yuan Chen Date: Wed, 26 Aug 2026 16:23:05 -0700 Subject: [PATCH 2/4] fix(validators): retry transient reads in the KAI deployment wait too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit waitForDeploymentAvailable aborts on any non-NotFound read error, and the gang-scheduling check calls it seconds before the pod poll under the same throttling — so fixing only the pod poll moved where the check breaks rather than making it robust. Apply the same guard, and keep the last transient read error so a sustained throttle is not reported as "never became ready" or "did not complete in time". Signed-off-by: Yuan Chen --- .../conformance/gang_scheduling_check.go | 8 ++ .../conformance/gang_scheduling_check_test.go | 75 ++++++++++++++++++- validators/conformance/helpers.go | 23 +++++- 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/validators/conformance/gang_scheduling_check.go b/validators/conformance/gang_scheduling_check.go index 8585618c6..e30372669 100644 --- a/validators/conformance/gang_scheduling_check.go +++ b/validators/conformance/gang_scheduling_check.go @@ -332,6 +332,7 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru waitCtx, cancel := context.WithTimeout(ctx, defaults.GangTestPodTimeout) defer cancel() + var lastReadErr error err := wait.PollUntilContextCancel(waitCtx, defaults.PodPollInterval, true, func(ctx context.Context) (bool, error) { allDone := true @@ -349,6 +350,7 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru // #1513 fixed one step earlier in this function. Let the // enclosing GangTestPodTimeout decide instead. if isK8sTimeoutErr(err) { + lastReadErr = err slog.Debug("transient read while polling gang test pod; retrying", "pod", run.pods[i], "error", err) allDone = false @@ -369,6 +371,12 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru ) if err != nil { if ctx.Err() != nil || waitCtx.Err() != nil { + // Preserve the last transient read error: a sustained throttle + // otherwise looks identical to pods that never completed. + if lastReadErr != nil { + return result, errors.Wrap(errors.ErrCodeTimeout, + "gang test pods unreadable (reads kept failing)", lastReadErr) + } return result, errors.Wrap(errors.ErrCodeTimeout, "gang test pods did not complete in time", err) } return result, errors.Wrap(errors.ErrCodeInternal, "gang test pod polling failed", err) diff --git a/validators/conformance/gang_scheduling_check_test.go b/validators/conformance/gang_scheduling_check_test.go index 481b8189e..8bce7fcd2 100644 --- a/validators/conformance/gang_scheduling_check_test.go +++ b/validators/conformance/gang_scheduling_check_test.go @@ -16,10 +16,16 @@ package main import ( "context" + stderrors "errors" "fmt" "sync/atomic" "testing" + "time" + "github.com/NVIDIA/aicr/pkg/errors" + "github.com/NVIDIA/aicr/validators" + + appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" k8serrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -222,7 +228,74 @@ func TestWaitForGangTestPodsFailsClosedOnTerminalRead(t *testing.T) { schema.GroupResource{Resource: "pods"}, run.pods[0], fmt.Errorf("no access")) }) - if _, err := waitForGangTestPods(context.Background(), clientset, run); err == nil { + _, err = waitForGangTestPods(context.Background(), clientset, run) + if err == nil { t.Fatal("expected a terminal read error to fail the check, got nil") } + // Assert the code, not just non-nil: a mis-coded Forbidden would otherwise + // slip through this guard. + if !stderrors.Is(err, errors.New(errors.ErrCodeInternal, "")) { + t.Errorf("Forbidden should classify as ErrCodeInternal, got %v", err) + } +} + +// TestWaitForGangTestPodsNotFoundIsTerminal pins the one error-mapping change +// this fix makes. The pods are created by deployGangTestResources immediately +// beforehand, and a Get with unset ResourceVersion is a quorum read, so +// NotFound means the pod genuinely went away — it must abort, not retry until +// the bound expires. +func TestWaitForGangTestPodsNotFoundIsTerminal(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + clientset := k8sfake.NewSimpleClientset() // no pods seeded + + start := time.Now() + _, err = waitForGangTestPods(context.Background(), clientset, run) + if err == nil { + t.Fatal("expected NotFound to fail the check, got nil") + } + if !stderrors.Is(err, errors.New(errors.ErrCodeNotFound, "")) { + t.Errorf("want ErrCodeNotFound, got %v", err) + } + // Terminal means immediate: it must not have burned the poll budget. + if elapsed := time.Since(start); elapsed > 30*time.Second { + t.Errorf("NotFound should abort immediately, took %s", elapsed) + } +} + +// TestWaitForDeploymentAvailableRetriesTransientReads covers the same defect at +// the site that runs FIRST in the gang check (step 1, via +// gang_scheduling_check.go's KAI deployment readiness loop). Fixing only the +// pod poll would have left the check flaking a few lines earlier under the same +// throttling. See #2406; this function is the one #1514 introduced for #1513. +func TestWaitForDeploymentAvailableRetriesTransientReads(t *testing.T) { + const ns, name = "kai-scheduler", "podgroup-controller" + ready := &appsv1.Deployment{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: ns}, + Status: appsv1.DeploymentStatus{AvailableReplicas: 1}, + } + clientset := k8sfake.NewSimpleClientset(ready) + + var reads atomic.Int32 + clientset.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) <= 2 { + return true, nil, fmt.Errorf( + "client rate limiter Wait returned an error: %w", context.DeadlineExceeded) + } + return false, nil, nil + }) + + vctx := &validators.Context{Ctx: context.Background(), Clientset: clientset} + deploy, err := waitForDeploymentAvailable(vctx, ns, name, 30*time.Second) + if err != nil { + t.Fatalf("aborted on a transient read: %v", err) + } + if deploy == nil || deploy.Status.AvailableReplicas != 1 { + t.Errorf("expected the ready deployment after retry, got %v", deploy) + } + if got := reads.Load(); got < 3 { + t.Errorf("expected retries past the throttled reads, saw %d", got) + } } diff --git a/validators/conformance/helpers.go b/validators/conformance/helpers.go index 75a421980..60964afb3 100644 --- a/validators/conformance/helpers.go +++ b/validators/conformance/helpers.go @@ -175,6 +175,7 @@ func waitForDeploymentAvailable(ctx *validators.Context, namespace, name string, defer cancel() var last *appsv1.Deployment + var lastReadErr error err := wait.PollUntilContextCancel(pollCtx, defaults.PodPollInterval, true, func(c context.Context) (bool, error) { deploy, getErr := ctx.Clientset.AppsV1().Deployments(namespace).Get(c, name, metav1.GetOptions{}) @@ -182,9 +183,20 @@ func waitForDeploymentAvailable(ctx *validators.Context, namespace, name string, if k8serrors.IsNotFound(getErr) { return false, nil // not created yet — keep waiting within the bound } - return false, errors.Wrap(errors.ErrCodeInternal, - fmt.Sprintf("failed to get deployment %s/%s", namespace, name), getErr) + // A read that could not land is not a verdict. Client-go's own + // rate limiter fails reads under load, and aborting here failed + // callers on a healthy cluster (#2406). Retry within the bound; + // genuine errors (RBAC, malformed) still abort immediately. + if isK8sTimeoutErr(getErr) { + lastReadErr = getErr + slog.Debug("transient read while waiting for deployment; retrying", + "namespace", namespace, "deployment", name, "error", getErr) + return false, nil + } + return false, classifyK8sReadError(getErr, + fmt.Sprintf("deployment %s/%s", namespace, name)) } + lastReadErr = nil last = deploy return deploy.Status.AvailableReplicas >= 1, nil }, @@ -214,6 +226,13 @@ func waitForDeploymentAvailable(ctx *validators.Context, namespace, name string, // available in time. Surface the NotFound-shaped not-available message the // caller wraps. A non-deadline error is a genuine API failure — propagate it. if pollCtx.Err() != nil { + // A sustained throttle would otherwise be indistinguishable from "never + // became ready" — keep the last read error so the operator sees why. + if lastReadErr != nil { + return last, errors.Wrap(errors.ErrCodeTimeout, + fmt.Sprintf("deployment %s/%s unreadable for %s (reads kept failing)", + namespace, name, timeout), lastReadErr) + } if last == nil { return nil, errors.New(errors.ErrCodeNotFound, fmt.Sprintf("deployment %s/%s not found after %s", namespace, name, timeout)) From bb711e8e63f80b99d4610fa57746747ccf30bb1e Mon Sep 17 00:00:00 2001 From: Yuan Chen Date: Wed, 26 Aug 2026 16:38:04 -0700 Subject: [PATCH 3/4] fix(validators): stop a recovered throttle from misdiagnosing the timeout Transient read state was sticky: the pod poll never cleared it, and the deployment wait's NotFound branch returned before clearing. One blip that fully recovered would then report a genuine "pods never completed" or "deployment missing" timeout as "reads kept failing", pointing the operator at a throttling ghost instead of the real cause. Track pod read errors per pod and clear each on a landed read; treat NotFound as the successful read it is. Report unreadable only when a still-pending object's most recent read failed. Test fixtures now use x/time/rate's sentinel-free string, so they exercise the plain-string branch of isK8sTimeoutErr rather than only its errors.Is path. Signed-off-by: Yuan Chen --- .../conformance/gang_scheduling_check.go | 23 ++++-- .../conformance/gang_scheduling_check_test.go | 76 +++++++++++++++++++ validators/conformance/helpers.go | 4 + 3 files changed, 97 insertions(+), 6 deletions(-) diff --git a/validators/conformance/gang_scheduling_check.go b/validators/conformance/gang_scheduling_check.go index e30372669..aca9d487d 100644 --- a/validators/conformance/gang_scheduling_check.go +++ b/validators/conformance/gang_scheduling_check.go @@ -332,7 +332,10 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru waitCtx, cancel := context.WithTimeout(ctx, defaults.GangTestPodTimeout) defer cancel() - var lastReadErr error + // Per-pod, not a single latch: a pod whose read recovers must clear its + // entry, otherwise one early blip would mislabel a genuine "never + // completed" timeout as "unreadable" long after reads recovered. + readErrs := make(map[string]error, gangMinMembers) err := wait.PollUntilContextCancel(waitCtx, defaults.PodPollInterval, true, func(ctx context.Context) (bool, error) { allDone := true @@ -350,7 +353,7 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru // #1513 fixed one step earlier in this function. Let the // enclosing GangTestPodTimeout decide instead. if isK8sTimeoutErr(err) { - lastReadErr = err + readErrs[run.pods[i]] = err slog.Debug("transient read while polling gang test pod; retrying", "pod", run.pods[i], "error", err) allDone = false @@ -359,7 +362,8 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru return false, classifyK8sReadError(err, fmt.Sprintf("gang test pod %s", run.pods[i])) } - switch pod.Status.Phase { //nolint:exhaustive // only terminal states matter + delete(readErrs, run.pods[i]) // this read landed + switch pod.Status.Phase { //nolint:exhaustive // only terminal states matter case corev1.PodSucceeded, corev1.PodFailed: result[i] = pod default: @@ -373,9 +377,16 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru if ctx.Err() != nil || waitCtx.Err() != nil { // Preserve the last transient read error: a sustained throttle // otherwise looks identical to pods that never completed. - if lastReadErr != nil { - return result, errors.Wrap(errors.ErrCodeTimeout, - "gang test pods unreadable (reads kept failing)", lastReadErr) + // Only if a still-pending pod's most recent read failed. + for i := range gangMinMembers { + if result[i] != nil { + continue + } + if readErr, ok := readErrs[run.pods[i]]; ok { + return result, errors.Wrap(errors.ErrCodeTimeout, + fmt.Sprintf("gang test pod %s unreadable (reads kept failing)", run.pods[i]), + readErr) + } } return result, errors.Wrap(errors.ErrCodeTimeout, "gang test pods did not complete in time", err) } diff --git a/validators/conformance/gang_scheduling_check_test.go b/validators/conformance/gang_scheduling_check_test.go index 8bce7fcd2..96c2954a2 100644 --- a/validators/conformance/gang_scheduling_check_test.go +++ b/validators/conformance/gang_scheduling_check_test.go @@ -18,6 +18,7 @@ import ( "context" stderrors "errors" "fmt" + "strings" "sync/atomic" "testing" "time" @@ -299,3 +300,78 @@ func TestWaitForDeploymentAvailableRetriesTransientReads(t *testing.T) { t.Errorf("expected retries past the throttled reads, saw %d", got) } } + +// rateLimitErr reproduces x/time/rate's raw form, which carries NO deadline +// sentinel — client-go surfaces it as a plain string. Fixtures that wrap +// context.DeadlineExceeded only exercise isK8sTimeoutErr's errors.Is path and +// would still pass if its plain-string branch broke. +func rateLimitErr() error { + //nolint:err113 // deliberately sentinel-free: mirrors x/time/rate's raw output + return stderrors.New("rate: Wait(n=1) would exceed context deadline") +} + +// TestWaitForDeploymentAvailableRecoveredReadNotReportedAsUnreadable pins that +// a transient throttle which RECOVERS does not poison the eventual timeout. A +// deployment that stays missing must be reported as not-found, not as "reads +// kept failing" — otherwise the operator chases a throttling ghost instead of +// the real cause. +func TestWaitForDeploymentAvailableRecoveredReadNotReportedAsUnreadable(t *testing.T) { + const ns, name = "kai-scheduler", "queue-controller" + clientset := k8sfake.NewSimpleClientset() // deployment never exists + + var reads atomic.Int32 + clientset.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) == 1 { + return true, nil, rateLimitErr() // one blip, then clean NotFound forever + } + return false, nil, nil + }) + + vctx := &validators.Context{Ctx: context.Background(), Clientset: clientset} + _, err := waitForDeploymentAvailable(vctx, ns, name, 2*time.Second) + if err == nil { + t.Fatal("expected a timeout error, got nil") + } + if !stderrors.Is(err, errors.New(errors.ErrCodeNotFound, "")) { + t.Errorf("want ErrCodeNotFound after a recovered blip, got %v", err) + } + if strings.Contains(err.Error(), "kept failing") { + t.Errorf("recovered throttle must not be reported as unreadable: %v", err) + } +} + +// TestWaitForGangTestPodsRecoveredReadNotReportedAsUnreadable is the same +// property for the pod poll: a blip that recovers, then pods that simply never +// reach a terminal phase, must time out as "did not complete", not "unreadable". +func TestWaitForGangTestPodsRecoveredReadNotReportedAsUnreadable(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + objs := make([]runtime.Object, 0, gangMinMembers) + for i := range gangMinMembers { + objs = append(objs, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: run.pods[i], Namespace: run.namespace}, + Status: corev1.PodStatus{Phase: corev1.PodRunning}, // never terminal + }) + } + clientset := k8sfake.NewSimpleClientset(objs...) + + var reads atomic.Int32 + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) == 1 { + return true, nil, rateLimitErr() + } + return false, nil, nil + }) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _, err = waitForGangTestPods(ctx, clientset, run) + if err == nil { + t.Fatal("expected a timeout error, got nil") + } + if strings.Contains(err.Error(), "kept failing") { + t.Errorf("recovered throttle must not be reported as unreadable: %v", err) + } +} diff --git a/validators/conformance/helpers.go b/validators/conformance/helpers.go index 60964afb3..9f6dc2087 100644 --- a/validators/conformance/helpers.go +++ b/validators/conformance/helpers.go @@ -181,6 +181,10 @@ func waitForDeploymentAvailable(ctx *validators.Context, namespace, name string, deploy, getErr := ctx.Clientset.AppsV1().Deployments(namespace).Get(c, name, metav1.GetOptions{}) if getErr != nil { if k8serrors.IsNotFound(getErr) { + // A NotFound is a successful read: the API answered. Clear + // any earlier transient error so a recovered throttle does + // not mislabel a genuinely-missing deployment at expiry. + lastReadErr = nil return false, nil // not created yet — keep waiting within the bound } // A read that could not land is not a verdict. Client-go's own From cba192d765abc2a21b65de25ed25bdf60b9ade08 Mon Sep 17 00:00:00 2001 From: Yuan Chen Date: Fri, 28 Aug 2026 09:25:05 -0700 Subject: [PATCH 4/4] fix(validators): preserve gang poll error diagnostics Do not retain a final read error caused by the poll context itself ending, and distinguish caller cancellation from the gang timeout. Preserve structured terminal error codes through the outer polling return. Strengthen the gang-path tests with plain-string rate limiting, ServerTimeout retry, ServiceUnavailable fail-closed behavior, parent cancellation, deadline-during-read diagnostics, and top-level code assertions. Signed-off-by: Yuan Chen --- .../conformance/gang_scheduling_check.go | 14 +- .../conformance/gang_scheduling_check_test.go | 256 ++++++++++++++---- validators/conformance/helpers.go | 14 + 3 files changed, 229 insertions(+), 55 deletions(-) diff --git a/validators/conformance/gang_scheduling_check.go b/validators/conformance/gang_scheduling_check.go index aca9d487d..c0ef36907 100644 --- a/validators/conformance/gang_scheduling_check.go +++ b/validators/conformance/gang_scheduling_check.go @@ -353,6 +353,12 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru // #1513 fixed one step earlier in this function. Let the // enclosing GangTestPodTimeout decide instead. if isK8sTimeoutErr(err) { + // The wait context ending during this Get is the poll's + // terminal signal, not evidence of sustained read failures. + if readFailedBecauseContextEnded(ctx, err) { + allDone = false + continue + } readErrs[run.pods[i]] = err slog.Debug("transient read while polling gang test pod; retrying", "pod", run.pods[i], "error", err) @@ -374,7 +380,11 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru }, ) if err != nil { - if ctx.Err() != nil || waitCtx.Err() != nil { + // Caller cancellation is an external abort, not the gang timing out. + if ctx.Err() != nil { + return result, errors.Wrap(errors.ErrCodeTimeout, "waiting for gang test pods canceled", ctx.Err()) + } + if waitCtx.Err() != nil { // Preserve the last transient read error: a sustained throttle // otherwise looks identical to pods that never completed. // Only if a still-pending pod's most recent read failed. @@ -390,7 +400,7 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru } return result, errors.Wrap(errors.ErrCodeTimeout, "gang test pods did not complete in time", err) } - return result, errors.Wrap(errors.ErrCodeInternal, "gang test pod polling failed", err) + return result, errors.PropagateOrWrap(err, errors.ErrCodeInternal, "gang test pod polling failed") } return result, nil diff --git a/validators/conformance/gang_scheduling_check_test.go b/validators/conformance/gang_scheduling_check_test.go index 96c2954a2..f3927ab6c 100644 --- a/validators/conformance/gang_scheduling_check_test.go +++ b/validators/conformance/gang_scheduling_check_test.go @@ -19,6 +19,7 @@ import ( stderrors "errors" "fmt" "strings" + "sync" "sync/atomic" "testing" "time" @@ -177,66 +178,105 @@ func TestCleanupGangTestResourcesPreservesConcurrentRun(t *testing.T) { // This mirrors the acceptance criteria of #1513, which fixed the same shape one // step earlier in this function (instantaneous deployment read -> bounded wait). func TestWaitForGangTestPodsRetriesTransientReads(t *testing.T) { - run, err := newGangTestRun() - if err != nil { - t.Fatalf("newGangTestRun: %v", err) - } + tests := []struct { + name string + readErr error + }{ + { + name: "client-go plain-string rate limit", + readErr: rateLimitErr(), + }, + { + name: "apiserver ServerTimeout", + readErr: k8serrors.NewServerTimeout( + schema.GroupResource{Resource: "pods"}, "get", 1), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + + objs := make([]runtime.Object, 0, gangMinMembers) + for i := range gangMinMembers { + objs = append(objs, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: run.pods[i], Namespace: run.namespace}, + Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + }) + } + clientset := k8sfake.NewSimpleClientset(objs...) + + // Fail the first two reads, then let the real objects through. + var reads atomic.Int32 + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + if reads.Add(1) <= 2 { + return true, nil, tt.readErr + } + return false, nil, nil // fall through to the tracker + }) - objs := make([]runtime.Object, 0, gangMinMembers) - for i := range gangMinMembers { - objs = append(objs, &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{Name: run.pods[i], Namespace: run.namespace}, - Status: corev1.PodStatus{Phase: corev1.PodSucceeded}, + pods, err := waitForGangTestPods(context.Background(), clientset, run) + if err != nil { + t.Fatalf("waitForGangTestPods aborted on a transient read: %v", err) + } + for i := range gangMinMembers { + if pods[i] == nil { + t.Errorf("pod %d not collected after retry", i) + } + } + if got := reads.Load(); got < 3 { + t.Errorf("expected the poll to retry past the transient reads, saw %d reads", got) + } }) } - clientset := k8sfake.NewSimpleClientset(objs...) - - // Fail the first two reads the way a throttled client-go client does, then - // let the real objects through. - var reads atomic.Int32 - clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { - if reads.Add(1) <= 2 { - return true, nil, fmt.Errorf( - "client rate limiter Wait returned an error: %w", context.DeadlineExceeded) - } - return false, nil, nil // fall through to the tracker - }) - - pods, err := waitForGangTestPods(context.Background(), clientset, run) - if err != nil { - t.Fatalf("waitForGangTestPods aborted on a transient read: %v", err) - } - for i := range gangMinMembers { - if pods[i] == nil { - t.Errorf("pod %d not collected after retry", i) - } - } - if got := reads.Load(); got < 3 { - t.Errorf("expected the poll to retry past the throttled reads, saw %d reads", got) - } } // TestWaitForGangTestPodsFailsClosedOnTerminalRead is the other half: a genuine // error (RBAC denial) must still abort rather than spin until the timeout. func TestWaitForGangTestPodsFailsClosedOnTerminalRead(t *testing.T) { - run, err := newGangTestRun() - if err != nil { - t.Fatalf("newGangTestRun: %v", err) - } - clientset := k8sfake.NewSimpleClientset() - clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { - return true, nil, k8serrors.NewForbidden( - schema.GroupResource{Resource: "pods"}, run.pods[0], fmt.Errorf("no access")) - }) + tests := []struct { + name string + readErr func(*gangTestRun) error + }{ + { + name: "Forbidden", + readErr: func(run *gangTestRun) error { + return k8serrors.NewForbidden( + schema.GroupResource{Resource: "pods"}, run.pods[0], fmt.Errorf("no access")) + }, + }, + { + name: "ServiceUnavailable", + readErr: func(*gangTestRun) error { + return k8serrors.NewServiceUnavailable("apiserver unavailable") + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + clientset := k8sfake.NewSimpleClientset() + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + return true, nil, tt.readErr(run) + }) - _, err = waitForGangTestPods(context.Background(), clientset, run) - if err == nil { - t.Fatal("expected a terminal read error to fail the check, got nil") - } - // Assert the code, not just non-nil: a mis-coded Forbidden would otherwise - // slip through this guard. - if !stderrors.Is(err, errors.New(errors.ErrCodeInternal, "")) { - t.Errorf("Forbidden should classify as ErrCodeInternal, got %v", err) + _, err = waitForGangTestPods(context.Background(), clientset, run) + if err == nil { + t.Fatal("expected a terminal read error to fail the check, got nil") + } + // Assert the code, not just non-nil: a terminal error accidentally + // routed through the retry path would otherwise burn the poll budget. + if !stderrors.Is(err, errors.New(errors.ErrCodeInternal, "")) { + t.Errorf("%s should classify as ErrCodeInternal, got %v", tt.name, err) + } + }) } } @@ -260,6 +300,13 @@ func TestWaitForGangTestPodsNotFoundIsTerminal(t *testing.T) { if !stderrors.Is(err, errors.New(errors.ErrCodeNotFound, "")) { t.Errorf("want ErrCodeNotFound, got %v", err) } + var structuredErr *errors.StructuredError + if !stderrors.As(err, &structuredErr) { + t.Fatalf("want top-level StructuredError, got %T", err) + } + if structuredErr.Code != errors.ErrCodeNotFound { + t.Errorf("want top-level ErrCodeNotFound, got %s", structuredErr.Code) + } // Terminal means immediate: it must not have burned the poll budget. if elapsed := time.Since(start); elapsed > 30*time.Second { t.Errorf("NotFound should abort immediately, took %s", elapsed) @@ -282,8 +329,7 @@ func TestWaitForDeploymentAvailableRetriesTransientReads(t *testing.T) { var reads atomic.Int32 clientset.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) { if reads.Add(1) <= 2 { - return true, nil, fmt.Errorf( - "client rate limiter Wait returned an error: %w", context.DeadlineExceeded) + return true, nil, rateLimitErr() } return false, nil, nil }) @@ -371,7 +417,111 @@ func TestWaitForGangTestPodsRecoveredReadNotReportedAsUnreadable(t *testing.T) { if err == nil { t.Fatal("expected a timeout error, got nil") } + if !stderrors.Is(err, errors.New(errors.ErrCodeTimeout, "")) { + t.Errorf("want ErrCodeTimeout after a recovered blip, got %v", err) + } if strings.Contains(err.Error(), "kept failing") { t.Errorf("recovered throttle must not be reported as unreadable: %v", err) } } + +func TestReadFailedBecauseContextEnded(t *testing.T) { + canceledCtx, cancel := context.WithCancel(context.Background()) + cancel() + expiredCtx, expiredCancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second)) + defer expiredCancel() + + tests := []struct { + name string + ctx context.Context + err error + want bool + }{ + { + name: "live context does not own deadline error", + ctx: context.Background(), + err: context.DeadlineExceeded, + want: false, + }, + { + name: "expired context owns deadline error", + ctx: expiredCtx, + err: context.DeadlineExceeded, + want: true, + }, + { + name: "canceled context owns cancellation error", + ctx: canceledCtx, + err: context.Canceled, + want: true, + }, + { + name: "expired context does not own plain rate-limit error", + ctx: expiredCtx, + err: rateLimitErr(), + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := readFailedBecauseContextEnded(tt.ctx, tt.err); got != tt.want { + t.Errorf("readFailedBecauseContextEnded() = %t, want %t", got, tt.want) + } + }) + } +} + +func TestWaitForGangTestPodsParentCanceledDuringRead(t *testing.T) { + run, err := newGangTestRun() + if err != nil { + t.Fatalf("newGangTestRun: %v", err) + } + clientset := k8sfake.NewSimpleClientset() + parent, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + readStarted := make(chan struct{}) + var signalOnce sync.Once + clientset.PrependReactor("get", "pods", func(k8stesting.Action) (bool, runtime.Object, error) { + signalOnce.Do(func() { close(readStarted) }) + <-parent.Done() + return true, nil, parent.Err() + }) + go func() { + <-readStarted + cancel() + }() + + _, err = waitForGangTestPods(parent, clientset, run) + if err == nil { + t.Fatal("expected cancellation error, got nil") + } + if !stderrors.Is(err, errors.New(errors.ErrCodeTimeout, "")) { + t.Errorf("want ErrCodeTimeout for caller cancellation, got %v", err) + } + if !strings.Contains(err.Error(), "canceled") { + t.Errorf("want caller-canceled diagnostic, got %v", err) + } + if strings.Contains(err.Error(), "unreadable") { + t.Errorf("caller cancellation must not be reported as unreadable: %v", err) + } +} + +func TestWaitForDeploymentAvailableDeadlineDuringReadNotReportedAsUnreadable(t *testing.T) { + const pollTimeout = 50 * time.Millisecond + clientset := k8sfake.NewSimpleClientset() + clientset.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) { + time.Sleep(2 * pollTimeout) + return true, nil, context.DeadlineExceeded + }) + + vctx := &validators.Context{Ctx: context.Background(), Clientset: clientset} + _, err := waitForDeploymentAvailable(vctx, "kai-scheduler", "queue-controller", pollTimeout) + if err == nil { + t.Fatal("expected timeout result, got nil") + } + if strings.Contains(err.Error(), "unreadable") { + t.Errorf("the wait's own deadline must not be reported as failed reads: %v", err) + } +} diff --git a/validators/conformance/helpers.go b/validators/conformance/helpers.go index 9f6dc2087..8a03d3ca0 100644 --- a/validators/conformance/helpers.go +++ b/validators/conformance/helpers.go @@ -16,6 +16,7 @@ package main import ( "context" + stderrors "errors" "fmt" "io" "log/slog" @@ -36,6 +37,13 @@ import ( "sigs.k8s.io/yaml" ) +// readFailedBecauseContextEnded reports whether an I/O error came from the +// polling context itself ending. Such an error is the wait's terminal signal, +// not evidence that preceding object reads were persistently failing. +func readFailedBecauseContextEnded(ctx context.Context, err error) bool { + return ctx.Err() != nil && stderrors.Is(err, ctx.Err()) +} + // getDynamicClient returns the dynamic client from context, or creates one from RESTConfig. func getDynamicClient(ctx *validators.Context) (dynamic.Interface, error) { if ctx.DynamicClient != nil { @@ -192,6 +200,12 @@ func waitForDeploymentAvailable(ctx *validators.Context, namespace, name string, // callers on a healthy cluster (#2406). Retry within the bound; // genuine errors (RBAC, malformed) still abort immediately. if isK8sTimeoutErr(getErr) { + // A Get interrupted by this wait's own cancellation is not a + // transient read failure. Do not let the final in-flight read + // overwrite the diagnostic from the preceding poll history. + if readFailedBecauseContextEnded(c, getErr) { + return false, nil + } lastReadErr = getErr slog.Debug("transient read while waiting for deployment; retrying", "namespace", namespace, "deployment", name, "error", getErr)