fix(validator): let the rollout allowance bound the probe, not just the sleeps

Regression from the previous commit, caught in review. Probing with the parent
context fixed the classification — a deadline landing mid-probe no longer
surfaced as a bare timeout — and silently removed the bound. Each probe makes
several sequential reads with their own timeouts, so a slow probe could cross
the allowance and still report success if the installation completed meanwhile.
On a loaded control plane that is an ordinary rollout, not a corner case.

The probe runs under pollCtx again, and both expiry paths route through one
classifier instead of the select owning the deadline alone. A canceled parent is
Timeout; a local expiry with the parent still live is NotFound with the last
known reason. Routing both through the same place is what keeps the verdict from
depending on whether the clock ran out during a probe or during a sleep — the
same cluster state must not produce two codes.

Adds a test that fails with the unbounded form and passes with the bound
restored, verified both ways.

Completes the diagnostic threading. An absent admission configuration had no
reason at all, a present configuration missing its webhook entry claimed the
configuration itself was gone, and the mutating-webhook path said 'validating'.
All three now name the object actually checked.

While splitting that branch I briefly turned 'err != nil || !found' into a
single not-found return, which would have reported a failed read as a missing
deployment — the exact misclassification this PR exists to prevent. The error
path is restored and propagates.

Refs #2297

Signed-off-by: Yuan Chen <yuanchen97@gmail.com>

Author: yuanchen8911

validators/performance/trainer_ensure_test.go

@@ -431,3 +431,41 @@ func TestEnsureTrainerInstalled_DeclaredRolloutDoesNotFallThrough(t *testing.T)
 			"does not cover the branch it claims to", probes)
 	}
 }
+
+// TestWaitForDeclaredTrainer_SlowProbeCannotOutrunTheDeadline pins the rollout
+// allowance as a bound on the whole wait, not just on the sleeps between probes.
+//
+// An earlier revision probed with the parent context so a deadline landing
+// mid-probe would not surface as a bare timeout. That fixed the classification and
+// removed the bound: each probe makes several sequential reads with their own
+// timeouts, so a slow probe could cross the allowance and still report success if
+// the installation happened to complete meanwhile. The probe runs under pollCtx
+// again, with the expiry classified rather than propagated blind.
+//
+// Here the first read sleeps past the allowance and the installation is complete
+// underneath, so a wait that respects its deadline must fail rather than succeed.
+func TestWaitForDeclaredTrainer_SlowProbeCannotOutrunTheDeadline(t *testing.T) {
+	oldTimeout, oldInterval := trainerInstallWaitTimeout, trainerInstallPollInterval
+	trainerInstallWaitTimeout = 20 * time.Millisecond
+	trainerInstallPollInterval = time.Millisecond
+	defer func() {
+		trainerInstallWaitTimeout, trainerInstallPollInterval = oldTimeout, oldInterval
+	}()
+
+	client := newTrainerFakeClient(completeTrainerInstall()...)
+	client.PrependReactor("get", "customresourcedefinitions",
+		func(k8stesting.Action) (bool, runtime.Object, error) {
+			time.Sleep(60 * time.Millisecond) // outlives the allowance
+			return false, nil, nil            // then let the complete install answer
+		})
+
+	_, err := waitForDeclaredTrainer(context.Background(), client)
+	if err == nil {
+		t.Fatal("wait returned success after its own deadline had passed; the rollout " +
+			"allowance must bound the probe, not only the sleeps between probes")
+	}
+	if !stderrors.Is(err, aicrErrors.New(aicrErrors.ErrCodeNotFound, "")) {
+		t.Errorf("error code = %v, want ErrCodeNotFound: the deadline expired locally "+
+			"with the parent still live, which is a deployment that never completed", err)
+	}
+}

validators/performance/trainer_lifecycle.go

@@ -269,7 +269,8 @@ func isTrainerInstalled(ctx context.Context, dynamicClient dynamic.Interface) (t
 	if !ok {
 		slog.Info("Kubeflow Trainer incomplete: admission webhook missing",
 			"configuration", trainerMutatingWebhookConfig, "webhook", trainerMutatingWebhookName)
-		return trainerInstall{Incomplete: "the validating admission webhook configuration is missing"}, false, nil
+		return trainerInstall{Incomplete: fmt.Sprintf(
+			"admission configuration %q is missing", trainerMutatingWebhookConfig)}, false, nil
 	}
 
 	// The controller Deployment is found by label: its name is release-derived on
@@ -313,13 +314,18 @@ func discoverTrainerInstall(ctx context.Context, dynamicClient dynamic.Interface
 	gvr schema.GroupVersionResource, configName, webhookName string) (trainerInstall, bool, error) {
 
 	obj, found, err := getTrainerObject(ctx, dynamicClient, gvr, "", configName)
-	if err != nil || !found {
-		if err == nil {
-			slog.Info("Kubeflow Trainer incomplete: admission configuration missing",
-				"configuration", configName)
-		}
+	if err != nil {
+		// A failed read is not evidence of absence. Propagate it so the caller
+		// classifies it as transport or timeout rather than as a deployment that
+		// never happened.
 		return trainerInstall{}, false, err
 	}
+	if !found {
+		slog.Info("Kubeflow Trainer incomplete: admission configuration missing",
+			"configuration", configName)
+		return trainerInstall{Incomplete: fmt.Sprintf(
+			"admission configuration %q is missing", configName)}, false, nil
+	}
 
 	entries, _, err := unstructured.NestedSlice(obj.Object, "webhooks")
 	if err != nil {
@@ -346,7 +352,9 @@ func discoverTrainerInstall(ctx context.Context, dynamicClient dynamic.Interface
 
 	slog.Info("Kubeflow Trainer incomplete: admission webhook missing",
 		"configuration", configName, "webhook", webhookName)
-	return trainerInstall{Incomplete: "the validating admission webhook configuration is missing"}, false, nil
+	return trainerInstall{Incomplete: fmt.Sprintf(
+		"admission configuration %q exists but does not contain the %q webhook",
+		configName, webhookName)}, false, nil
 }
 
 // getTrainerObject fetches one object. NotFound reports found=false with no
@@ -537,6 +545,42 @@ func ensureTrainerInstalled(ctx context.Context, dynamicClient dynamic.Interface
 	return nil, awaitTrainerController(ctx, dynamicClient, install)
 }
 
+// classifyPollExpiry turns an expired poll context into the right verdict, or nil
+// when the poll context is still live and the caller should handle its own error.
+//
+// The two expiries mean opposite things. A canceled parent means the run was
+// aborted — catalog timeout, canceled phase, killed Job — which is not a customer
+// deployment defect. A local deadline means the delivered Trainer never became
+// complete, which is.
+//
+// Both paths route through here so the verdict does not depend on whether the clock
+// ran out during a probe or during a sleep: the same cluster state must not produce
+// two different codes.
+func classifyPollExpiry(ctx, pollCtx context.Context, last trainerInstall) error {
+	if ctx.Err() != nil {
+		return aicrErrors.Wrap(aicrErrors.ErrCodeTimeout,
+			"canceled while waiting for the recipe-declared Kubeflow Trainer", ctx.Err())
+	}
+	if pollCtx.Err() == nil {
+		return nil
+	}
+
+	// ErrCodeNotFound, not Unavailable: the read succeeded and the answer was "not
+	// deployed". Unavailable is this package's code for a transport failure — see
+	// the decision table on validators.Require — and using it here would file a
+	// product defect alongside apiserver hiccups, telling whoever triages it to
+	// re-run rather than to fix their deployment.
+	reason := last.Incomplete
+	if reason == "" {
+		reason = "no complete installation was found"
+	}
+	return aicrErrors.New(aicrErrors.ErrCodeNotFound, fmt.Sprintf(
+		"the recipe declares the %s component but its Kubeflow Trainer installation "+
+			"did not become complete within %s: %s. The benchmark will not self-install "+
+			"over a delivered component that failed to deploy",
+		kubeflowTrainerComponent, trainerInstallWaitTimeout, reason))
+}
+
 // awaitTrainerController waits for a Trainer the benchmark does not own to finish
 // starting. The probe confirms every object exists; the controller may still be
 // rolling, and waiting is the alternative to reinstalling over a healthy Trainer.
@@ -573,13 +617,20 @@ func waitForDeclaredTrainer(ctx context.Context, dynamicClient dynamic.Interface
 	defer cancel()
 
 	for {
-		// Probe with the parent context, not pollCtx: getTrainerObject checks
-		// ctx.Err() at the top of every read and returns Timeout, so a deadline
-		// landing mid-probe would surface as a bare timeout instead of the
-		// NotFound-plus-diagnosis this function exists to produce. Letting the
-		// select below own the deadline keeps the two conditions separable.
-		install, ok, err := isTrainerInstalled(ctx, dynamicClient)
+		// Probe under pollCtx so the rollout allowance actually bounds it. Each probe
+		// makes several sequential reads, each with its own DiagnosticTimeout, so a
+		// probe running on the parent context could cross the deadline and still
+		// report success — the allowance would bound only the sleeps.
+		//
+		// The cost is that getTrainerObject checks ctx.Err() at the top of every read
+		// and returns Timeout, so an expiring deadline surfaces here as a probe error
+		// rather than through the select. Classify it below instead of propagating it
+		// blind, so the verdict does not depend on where in the loop the clock ran out.
+		install, ok, err := isTrainerInstalled(pollCtx, dynamicClient)
 		if err != nil {
+			if verdict := classifyPollExpiry(ctx, pollCtx, last); verdict != nil {
+				return trainerInstall{}, verdict
+			}
 			return trainerInstall{}, aicrErrors.PropagateOrWrap(err, aicrErrors.ErrCodeInternal,
 				"failed to check Kubeflow Trainer installation")
 		}
@@ -590,30 +641,11 @@ func waitForDeclaredTrainer(ctx context.Context, dynamicClient dynamic.Interface
 
 		select {
 		case <-pollCtx.Done():
-			// pollCtx expires for two different reasons and they mean opposite
-			// things. A canceled parent means the run was aborted — catalog timeout,
-			// phase cancellation, the Job killed — and reporting that as a customer
-			// deployment defect is the same misclassification the NotFound code
-			// above exists to avoid, one level down.
-			if ctx.Err() != nil {
-				return trainerInstall{}, aicrErrors.Wrap(aicrErrors.ErrCodeTimeout,
-					"canceled while waiting for the recipe-declared Kubeflow Trainer", ctx.Err())
-			}
-
-			// ErrCodeNotFound, not Unavailable: the read succeeded and the answer was
-			// "not deployed". Unavailable is this package's code for a transport
-			// failure — see the decision table on validators.Require — and using it
-			// here would file a product defect alongside apiserver hiccups, telling
-			// whoever triages it to re-run rather than to fix their deployment.
-			reason := last.Incomplete
-			if reason == "" {
-				reason = "no complete installation was found"
+			if verdict := classifyPollExpiry(ctx, pollCtx, last); verdict != nil {
+				return trainerInstall{}, verdict
 			}
-			return trainerInstall{}, aicrErrors.New(aicrErrors.ErrCodeNotFound, fmt.Sprintf(
-				"the recipe declares the %s component but its Kubeflow Trainer installation "+
-					"did not become complete within %s: %s. The benchmark will not self-install "+
-					"over a delivered component that failed to deploy",
-				kubeflowTrainerComponent, trainerInstallWaitTimeout, reason))
+			return trainerInstall{}, aicrErrors.New(aicrErrors.ErrCodeInternal,
+				"Kubeflow Trainer wait ended without a verdict")
 		case <-time.After(trainerInstallPollInterval):
 		}
 	}