diff --git a/internal/scheduling/reservations/commitments/api/change_commitments.go b/internal/scheduling/reservations/commitments/api/change_commitments.go index f181a0fbe..0aec12167 100644 --- a/internal/scheduling/reservations/commitments/api/change_commitments.go +++ b/internal/scheduling/reservations/commitments/api/change_commitments.go @@ -274,7 +274,8 @@ ProcessLoop: } if !handlesCommitments { failedReason = fmt.Sprintf("flavor group %q is not configured to handle %s commitments", flavorGroupName, resourceType) - sanitisedReason = "internal error processing request" + sanitisedReason = failedReason + isBadRequest = true rollback = true break ProcessLoop } diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller.go b/internal/scheduling/reservations/commitments/committed_resource_controller.go index 6d6d4b443..f8703d734 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller.go +++ b/internal/scheduling/reservations/commitments/committed_resource_controller.go @@ -7,6 +7,7 @@ import ( "context" "errors" "fmt" + "strconv" "time" "k8s.io/apimachinery/pkg/api/equality" @@ -56,7 +57,8 @@ func (r *CommittedResourceController) Reconcile(ctx context.Context, req ctrl.Re "committedResource", req.Name, ) - // Debug: timing anchor for CR reconcile relative to Phase 6 and Reservation echo. + // Timing anchor: shows which generation the cache served when this reconcile fired. + // Missing for a specific generation means the CR controller never dequeued that gen. logger.V(1).Info("reconcile entry", "rv", cr.ResourceVersion, "gen", cr.Generation) if !cr.DeletionTimestamp.IsZero() { @@ -124,7 +126,12 @@ func (r *CommittedResourceController) reconcilePending(ctx context.Context, logg // error would cause controller-runtime to ignore RequeueAfter and use exponential backoff. return ctrl.Result{RequeueAfter: delay}, nil } - allReady, anyFailed, readySlots, failReason, firstSlotReason, failedSlots, err := r.checkChildReservationStatus(ctx, cr, result.TotalSlots) + // alreadyReserving is true when the CR condition already shows Reserving for this generation, + // meaning at least one reconcile cycle (~1s) has elapsed since we first entered the waiting state. + // We only nudge stale Reservation slots after that delay to give the normal echo path a chance + // to complete first, keeping logs quiet in the happy path. + alreadyReserving := isAlreadyReserving(cr) + allReady, anyFailed, readySlots, failReason, firstSlotReason, failedSlots, err := r.checkChildReservationStatus(ctx, cr, result.TotalSlots, alreadyReserving) if err != nil { return ctrl.Result{}, err } @@ -151,10 +158,16 @@ func (r *CommittedResourceController) reconcilePending(ctx context.Context, logg // gate should not fire while we wait for slots to become ready. // Fallback requeue: in case the watch event is missed, // re-check after 1s rather than waiting for the full watch timeout. - if err := r.patchNotReady(ctx, cr, v1alpha1.CommittedResourceReasonReserving, "waiting for reservation placement", true); err != nil { + // Use "still waiting" message after the first cycle so the condition is a visible hint + // that a nudge has been (or is about to be) sent to stale slots. + msg := "waiting for reservation placement" + if alreadyReserving { + msg = "still waiting for reservation placement" + } + if err := r.patchNotReady(ctx, cr, v1alpha1.CommittedResourceReasonReserving, msg, true); err != nil { return ctrl.Result{}, err } - return ctrl.Result{RequeueAfter: time.Second}, nil + return ctrl.Result{RequeueAfter: 3 * time.Second}, nil } logger.Info("committed resource accepted", "generation", cr.Generation, "amount", cr.Spec.Amount.String()) return ctrl.Result{}, r.setAccepted(ctx, cr) @@ -183,7 +196,7 @@ func (r *CommittedResourceController) reconcileCommitted(ctx context.Context, lo if applyErr != nil { logger.V(1).Info("spurious-rejection recovery: apply failed, will retry", "err", applyErr) } else { - allReady, anyFailed, _, _, _, _, checkErr := r.checkChildReservationStatus(ctx, cr, result.TotalSlots) + allReady, anyFailed, _, _, _, _, checkErr := r.checkChildReservationStatus(ctx, cr, result.TotalSlots, false) if checkErr == nil && allReady && !anyFailed { logger.Info("recovered from spurious rejection: all slots ready", "generation", cr.Generation) return ctrl.Result{}, r.setAccepted(ctx, cr) @@ -222,7 +235,8 @@ func (r *CommittedResourceController) reconcileCommitted(ctx context.Context, lo // error would cause controller-runtime to ignore RequeueAfter and use exponential backoff. return ctrl.Result{RequeueAfter: delay}, nil } - allReady, anyFailed, readySlots, failReason, firstSlotReason, failedSlots, err := r.checkChildReservationStatus(ctx, cr, result.TotalSlots) + alreadyReserving := isAlreadyReserving(cr) + allReady, anyFailed, readySlots, failReason, firstSlotReason, failedSlots, err := r.checkChildReservationStatus(ctx, cr, result.TotalSlots, alreadyReserving) if err != nil { return ctrl.Result{}, err } @@ -247,13 +261,30 @@ func (r *CommittedResourceController) reconcileCommitted(ctx context.Context, lo // gate should not fire while we wait for slots to become ready. // Fallback requeue: in case the watch event is missed, // re-check after 1s rather than waiting for the full watch timeout. - if err := r.patchNotReady(ctx, cr, v1alpha1.CommittedResourceReasonReserving, "waiting for reservation placement", true); err != nil { + // Use "still waiting" message after the first cycle so the condition is a visible hint + // that a nudge has been (or is about to be) sent to stale slots. + msg := "waiting for reservation placement" + if alreadyReserving { + msg = "still waiting for reservation placement" + } + if err := r.patchNotReady(ctx, cr, v1alpha1.CommittedResourceReasonReserving, msg, true); err != nil { return ctrl.Result{}, err } - return ctrl.Result{RequeueAfter: time.Second}, nil + return ctrl.Result{RequeueAfter: 3 * time.Second}, nil } logger.Info("committed resource accepted", "generation", cr.Generation, "amount", cr.Spec.Amount.String()) - return ctrl.Result{}, r.setAccepted(ctx, cr) + if err := r.setAccepted(ctx, cr); err != nil { + return ctrl.Result{}, err + } + var latest v1alpha1.CommittedResource + if err := r.Get(ctx, client.ObjectKeyFromObject(cr), &latest); err == nil && latest.Generation > cr.Generation { + logger.Info("generation advanced during reconcile, requeueing", + "processedGen", cr.Generation, + "latestGen", latest.Generation, + ) + return ctrl.Result{RequeueAfter: 1 * time.Second}, nil + } + return ctrl.Result{}, nil } // reconcileCoresHeadroom handles acceptance of CPU core commitments. @@ -398,7 +429,7 @@ func (r *CommittedResourceController) applyReservationState(ctx context.Context, // Returns allReady=false, anyFailed=false when some children have no condition yet (placement pending). // failReason includes a slot count when more than one slot failed. // failedSlots lists the names of all failing reservation slots for log context. -func (r *CommittedResourceController) checkChildReservationStatus(ctx context.Context, cr *v1alpha1.CommittedResource, expectedSlots int) (allReady, anyFailed bool, readySlots int, failReason, firstFailMessage string, failedSlots []string, err error) { +func (r *CommittedResourceController) checkChildReservationStatus(ctx context.Context, cr *v1alpha1.CommittedResource, expectedSlots int, shouldNudge bool) (allReady, anyFailed bool, readySlots int, failReason, firstFailMessage string, failedSlots []string, err error) { var list v1alpha1.ReservationList if err := r.List(ctx, &list, client.MatchingLabels{v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource}, @@ -432,6 +463,34 @@ func (r *CommittedResourceController) checkChildReservationStatus(ctx context.Co ) if res.Status.CommittedResourceReservation == nil || res.Status.CommittedResourceReservation.ObservedParentGeneration != cr.Generation { + // parentGen has been written to the Reservation spec (syncReservationMetadata + // already ran for this generation) but the Reservation controller hasn't echoed + // it to status yet. Poke the Reservation to generate a fresh watch event so the + // Reservation controller wakes up and fires the echo. + // Use gen+ms as the annotation value so each nudge attempt is a distinct string: + // a same-value patch is a no-op on the API server (no watch event generated), and + // in a double-stale scenario the first nudge may itself be processed with a stale + // object, so periodic retries (every 3s via the fallback requeue) are needed. + // shouldNudge is false on the first cycle (CR condition not yet set for this + // generation) to give the normal echo path ~3s to complete before intervening. + if shouldNudge && + res.Spec.CommittedResourceReservation != nil && + res.Spec.CommittedResourceReservation.ParentGeneration == cr.Generation { + nudgeVal := strconv.FormatInt(cr.Generation, 10) + "-" + strconv.FormatInt(time.Now().UnixMilli(), 10) + ctrl.LoggerFrom(ctx).Info("nudging stale reservation: parentGen written to spec but echo pending", + "slot", res.Name, + "crGen", cr.Generation, + "obs", obs, + ) + old := res.DeepCopy() + if res.Annotations == nil { + res.Annotations = map[string]string{} + } + res.Annotations["cortex.cloud/echo-nudge"] = nudgeVal + if pErr := r.Patch(ctx, &res, client.MergeFrom(old)); client.IgnoreNotFound(pErr) != nil { + ctrl.LoggerFrom(ctx).V(1).Info("failed to nudge stale reservation echo", "slot", res.Name, "err", pErr) + } + } allReady = false continue } @@ -572,6 +631,18 @@ func isRejectedForGeneration(cr *v1alpha1.CommittedResource) bool { cond.ObservedGeneration == cr.Generation } +// isAlreadyReserving returns true when the CR condition already shows Reserving for this +// generation, meaning at least one reconcile cycle (~1s) has elapsed since we first entered +// the waiting-for-slots state. Used to gate the echo-nudge so the first reconcile cycle is +// always silent, giving the normal Reservation echo path a chance to complete first. +func isAlreadyReserving(cr *v1alpha1.CommittedResource) bool { + cond := meta.FindStatusCondition(cr.Status.Conditions, v1alpha1.CommittedResourceConditionReady) + return cond != nil && + cond.Status == metav1.ConditionFalse && + cond.Reason == v1alpha1.CommittedResourceReasonReserving && + cond.ObservedGeneration == cr.Generation +} + // retryDelay computes an exponential backoff interval for the AllowRejection=false retry paths. // The exponent is derived from the time already spent in Reserving state: each doubling period // advances one step, giving the same base→2*base→4*base→… sequence as a counter would, but diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller_test.go b/internal/scheduling/reservations/commitments/committed_resource_controller_test.go index 19730f432..d5732f0d0 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller_test.go +++ b/internal/scheduling/reservations/commitments/committed_resource_controller_test.go @@ -1157,6 +1157,74 @@ func TestCommittedResourceController_BadSpec(t *testing.T) { // Tests: checkChildReservationStatus generation guard // ============================================================================ +// TestCommittedResourceController_NudgeDelay verifies that the echo-nudge is not sent on +// the first reconcile cycle (when the condition is not yet Reserving for this generation), +// but is sent on the second cycle — and that the condition message reflects the cycle. +func TestCommittedResourceController_NudgeDelay(t *testing.T) { + scheme := newCRTestScheme(t) + cr := newTestCommittedResource("test-cr", v1alpha1.CommitmentStatusPending) + cr.Spec.AllowRejection = false + k8sClient := newCRTestClient(scheme, cr, newTestFlavorKnowledge()) + controller := &CommittedResourceController{Client: k8sClient, Scheme: scheme} + + getCR := func() v1alpha1.CommittedResource { + t.Helper() + var updated v1alpha1.CommittedResource + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: cr.Name}, &updated); err != nil { + t.Fatalf("get CR: %v", err) + } + return updated + } + getChildAnnotation := func() string { + t.Helper() + var list v1alpha1.ReservationList + if err := k8sClient.List(context.Background(), &list, client.MatchingLabels{ + v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource, + }); err != nil { + t.Fatalf("list reservations: %v", err) + } + for _, res := range list.Items { + if res.Spec.CommittedResourceReservation != nil && + res.Spec.CommittedResourceReservation.CommitmentUUID == cr.Spec.CommitmentUUID { + return res.Annotations["cortex.cloud/echo-nudge"] + } + } + return "" + } + + // Cycle 1: no Reserving condition exists yet → isAlreadyReserving=false → no nudge, + // condition message = "waiting for reservation placement". + if _, err := controller.Reconcile(context.Background(), reconcileReq(cr.Name)); err != nil { + t.Fatalf("reconcile 1: %v", err) + } + cond1 := meta.FindStatusCondition(getCR().Status.Conditions, v1alpha1.CommittedResourceConditionReady) + if cond1 == nil || cond1.Reason != v1alpha1.CommittedResourceReasonReserving { + t.Fatalf("cycle 1: expected Reserving condition, got %+v", cond1) + } + if cond1.Message != "waiting for reservation placement" { + t.Errorf("cycle 1: message want %q, got %q", "waiting for reservation placement", cond1.Message) + } + if got := getChildAnnotation(); got != "" { + t.Errorf("cycle 1: echo-nudge annotation should not be set, got %q", got) + } + + // Cycle 2: Reserving condition is now set for this generation → isAlreadyReserving=true → + // nudge fires, condition message = "still waiting for reservation placement". + if _, err := controller.Reconcile(context.Background(), reconcileReq(cr.Name)); err != nil { + t.Fatalf("reconcile 2: %v", err) + } + cond2 := meta.FindStatusCondition(getCR().Status.Conditions, v1alpha1.CommittedResourceConditionReady) + if cond2 == nil || cond2.Reason != v1alpha1.CommittedResourceReasonReserving { + t.Fatalf("cycle 2: expected Reserving condition, got %+v", cond2) + } + if cond2.Message != "still waiting for reservation placement" { + t.Errorf("cycle 2: message want %q, got %q", "still waiting for reservation placement", cond2.Message) + } + if got := getChildAnnotation(); got == "" { + t.Errorf("cycle 2: echo-nudge annotation should be set") + } +} + // TestCheckChildReservationStatus_GenerationGuard verifies the single-pass logic that // distinguishes a stale Ready=False (previous generation) from a current failure. func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { @@ -1165,26 +1233,42 @@ func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { obsGen int64 condStatus metav1.ConditionStatus // "" = no condition set condMessage string + shouldNudge bool wantAllReady bool wantAnyFailed bool wantReadySlots int wantReason string wantFailedSlots []string + wantNudge bool }{ { name: "Ready=False at stale generation: treated as pending", obsGen: 1, condStatus: metav1.ConditionFalse, condMessage: "no hosts available", + shouldNudge: true, wantAllReady: false, wantAnyFailed: false, wantReadySlots: 0, + wantNudge: true, + }, + { + name: "Ready=False at stale generation: no nudge on first cycle", + obsGen: 1, + condStatus: metav1.ConditionFalse, + condMessage: "no hosts available", + shouldNudge: false, + wantAllReady: false, + wantAnyFailed: false, + wantReadySlots: 0, + wantNudge: false, }, { name: "Ready=False at current generation: is a current failure", obsGen: 2, condStatus: metav1.ConditionFalse, condMessage: "no hosts available", + shouldNudge: true, wantAllReady: false, wantAnyFailed: true, wantReadySlots: 0, @@ -1195,6 +1279,7 @@ func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { name: "Ready=True at current generation: allReady", obsGen: 2, condStatus: metav1.ConditionTrue, + shouldNudge: true, wantAllReady: true, wantReadySlots: 1, }, @@ -1202,9 +1287,20 @@ func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { name: "no condition yet at current generation: still pending", obsGen: 2, condStatus: "", // no condition + shouldNudge: true, + wantAllReady: false, + wantAnyFailed: false, + wantReadySlots: 0, + }, + { + name: "parentGen written to spec but echo not yet in status: nudge annotation set", + obsGen: 1, // obs behind cr.Generation=2; spec.parentGeneration already set to cr.Generation + condStatus: metav1.ConditionTrue, + shouldNudge: true, wantAllReady: false, wantAnyFailed: false, wantReadySlots: 0, + wantNudge: true, }, } @@ -1248,7 +1344,7 @@ func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { } controller := &CommittedResourceController{Client: k8sClient, Scheme: scheme} - allReady, anyFailed, readySlots, reason, _, failedSlots, err := controller.checkChildReservationStatus(context.Background(), cr, 1) + allReady, anyFailed, readySlots, reason, _, failedSlots, err := controller.checkChildReservationStatus(context.Background(), cr, 1, tt.shouldNudge) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -1267,6 +1363,14 @@ func TestCheckChildReservationStatus_GenerationGuard(t *testing.T) { if !slices.Equal(failedSlots, tt.wantFailedSlots) { t.Errorf("failedSlots: want %v, got %v", tt.wantFailedSlots, failedSlots) } + var updated v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: child.Name}, &updated); err != nil { + t.Fatalf("get updated reservation: %v", err) + } + _, hasNudge := updated.Annotations["cortex.cloud/echo-nudge"] + if hasNudge != tt.wantNudge { + t.Errorf("echo-nudge annotation: want %v, got %v", tt.wantNudge, hasNudge) + } }) } } diff --git a/internal/scheduling/reservations/commitments/reservation_manager.go b/internal/scheduling/reservations/commitments/reservation_manager.go index 68eae90b7..cc83851e5 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager.go +++ b/internal/scheduling/reservations/commitments/reservation_manager.go @@ -403,7 +403,10 @@ func (m *ReservationManager) syncReservationMetadata( // Apply patch logger.V(1).Info("syncing reservation metadata", "reservation", reservation.Name, - "commitmentUUID", state.CommitmentUUID) + "commitmentUUID", state.CommitmentUUID, + "currentParentGen", reservation.Spec.CommittedResourceReservation.ParentGeneration, + "newParentGen", state.ParentGeneration, + ) patch := client.MergeFrom(reservation.DeepCopy())