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..c0ef36907 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" @@ -331,6 +332,10 @@ func waitForGangTestPods(ctx context.Context, clientset kubernetes.Interface, ru waitCtx, cancel := context.WithTimeout(ctx, defaults.GangTestPodTimeout) defer cancel() + // 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 @@ -341,10 +346,30 @@ 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) { + // 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) + 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 + 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: @@ -355,10 +380,27 @@ 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. + 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) } - 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 8b33cbfc7..f3927ab6c 100644 --- a/validators/conformance/gang_scheduling_check_test.go +++ b/validators/conformance/gang_scheduling_check_test.go @@ -16,8 +16,18 @@ package main import ( "context" + stderrors "errors" + "fmt" + "strings" + "sync" + "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" @@ -25,6 +35,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 +167,361 @@ 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) { + 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 + }) + + 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) + } + }) + } +} + +// TestWaitForGangTestPodsFailsClosedOnTerminalRead is the other half: a genuine +// error (RBAC denial) must still abort rather than spin until the timeout. +func TestWaitForGangTestPodsFailsClosedOnTerminalRead(t *testing.T) { + 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 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) + } + }) + } +} + +// 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) + } + 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) + } +} + +// 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, rateLimitErr() + } + 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) + } +} + +// 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 !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 75a421980..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 { @@ -175,16 +183,38 @@ 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{}) 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 } - 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) { + // 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) + 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 +244,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))