Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 59 additions & 18 deletions pkg/worker/controllers/worker/model_deployment_elastic_reconcile.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,18 +31,25 @@ const (
modelDeploymentElasticWithdrawnAnnotation = "modeldeployment.gpustack.ai/elastic-withdrawn"
elasticObservationLabel = "gpustack.ai/elastic-observation"

// ModelDeploymentConditionElasticResize reports whether an elastic scale command stands
// ModelDeploymentConditionElasticResize reports whether an elastic scale stands blocked or
// terminally refused. It is declared beside the convergence that decides it, like every other
// condition whose vocabulary sits with the code that can observe it. It is False with
// ScaleRefused only while the durable record says the retry bound is spent, and True with
// NoRefusal the rest of the time: a refusal still inside its bound is not terminal and must
// not read as one.
// ScaleRefused only while the durable record says the retry bound is spent, Unknown with
// WorldUnprovable while no record exists and the live width disagrees with the spec without
// proving itself, and True with NoRefusal the rest of the time: a refusal still inside its
// bound is not terminal and must not read as one, and an unprovable world is not a refusal.
ModelDeploymentConditionElasticResize kubeapistatus.ConditionType = "ElasticResize"

// modelDeploymentReasonScaleRefused is ElasticResize's False reason: the record is abandoned
// and its message carries the engine's own last refusal.
modelDeploymentReasonScaleRefused = "ScaleRefused"

// modelDeploymentReasonWorldUnprovable is ElasticResize's Unknown reason: no record exists,
// the live width disagrees with the spec or cannot be named, and the corrective scale is
// blocked until a pass proves the world it would have to act on. Its message carries the
// desired width, the observed width and the proof that failed.
modelDeploymentReasonWorldUnprovable = "WorldUnprovable"

// modelDeploymentReasonNoScaleRefusal is ElasticResize's True reason: no record stands
// refused, whether none exists, one is in flight, or one has just completed.
modelDeploymentReasonNoScaleRefusal = "NoRefusal"
Expand Down Expand Up @@ -412,27 +419,31 @@ func elasticOperationMembersMatch(op *elasticOperation, pods []core.Pod) bool {
// otherwise — including when no record exists, so a refusal never outlives the record it came
// from. The pass hands over the record as it stands after its own writes; a caller whose record
// was just deleted hands over nil.
func (r *ModelDeploymentReconciler) recordElasticResizeCondition(
ctx context.Context, md *workercore.ModelDeployment, op *elasticOperation,
) error {
target, reason, message := string(meta.ConditionTrue),
modelDeploymentReasonNoScaleRefusal, "no elastic scale command stands refused"
if op != nil && op.State == elasticStateAbandoned {
target, reason, message = string(meta.ConditionFalse),
modelDeploymentReasonScaleRefused, elasticScaleRefusalMessage(op)
}
return r.writeElasticResizeCondition(ctx, md, target, reason, message)
}

// writeElasticResizeCondition patches the deployment's ElasticResize condition to one outcome.
//
// IT IS ITS OWN WRITE rather than a field set before the pass's status sync, for the same reason
// the retirement reservation is: the sync compares the derived status with the observed one and
// would find an assigned field equal to itself. The patch is optimistic and asserts only this
// condition, and the in-memory status takes the server's response so the same pass's later status
// sync rebuilds from what the server now holds. An unchanged condition writes nothing.
func (r *ModelDeploymentReconciler) recordElasticResizeCondition(
ctx context.Context, md *workercore.ModelDeployment, op *elasticOperation,
func (r *ModelDeploymentReconciler) writeElasticResizeCondition(
ctx context.Context, md *workercore.ModelDeployment, status, reason, message string,
) error {
refused := op != nil && op.State == elasticStateAbandoned
reason, message := modelDeploymentReasonNoScaleRefusal, "no elastic scale command stands refused"
if refused {
reason, message = modelDeploymentReasonScaleRefused, elasticScaleRefusalMessage(op)
}
base := md.DeepCopy()
candidate := base.DeepCopy()
if refused {
ModelDeploymentConditionElasticResize.False(candidate, reason, message)
} else {
ModelDeploymentConditionElasticResize.True(candidate, reason, message)
}
ModelDeploymentConditionElasticResize.Status(candidate, status, reason, message)
if kubemeta.DeepEqual(base.Status, candidate.Status) {
return nil
}
Expand All @@ -445,6 +456,36 @@ func (r *ModelDeploymentReconciler) recordElasticResizeCondition(
return nil
}

// finishElasticWorldUnprovable parks an unprovable world loudly. With no record, a world the pass
// cannot prove blocks the corrective scale — acting on unproven ranks is unsafe by design — and a
// live width that disagrees with the spec, or cannot be named at all, must not read as health
// while the engine and the pod set hold the wider quota. The observation keeps the proof's own
// failure, and the condition turns Unknown with the block and a message carrying the desired
// width, the observed width and the failed proof. A live width that agrees with the spec owes no
// corrective scale, so its park stays as quiet as before. Recovery is ordinary: a later pass that
// proves the world reaches the corrective block and levels the condition through the
// record-driven writer.
func (r *ModelDeploymentReconciler) finishElasticWorldUnprovable(
ctx context.Context, md *workercore.ModelDeployment, status *modelDeploymentElasticStatus,
finish func(string) error, desired int,
) error {
if status.Ray.Known && status.Ray.Value == desired {
return finish(status.Effective.Reason)
}
status.Reason = status.Effective.Reason
if err := r.persistElasticObservation(ctx, md, *status); err != nil {
return err
}
ray := strconv.Itoa(status.Ray.Value)
if !status.Ray.Known {
ray = "unknown"
}
return r.writeElasticResizeCondition(ctx, md, string(meta.ConditionUnknown),
modelDeploymentReasonWorldUnprovable,
fmt.Sprintf("the corrective scale from the live width %s to the desired width %d is blocked: the live world cannot be proven (%s)",
ray, desired, status.Effective.Reason))
}

// elasticScaleRetryDue says how long the next refused-command attempt still owes, from the bound
// the record states: attempt n+1 waits min(15s*2^(n-1), 4m) after failure n, where n counts the
// definite refusals AND the re-issued commands whose answer never arrived — an unanswered
Expand Down Expand Up @@ -612,12 +653,12 @@ func (r *ModelDeploymentReconciler) reconcileModelDeploymentElasticResize(ctx co
status.Ray = unknownLayer(ray.RegisteredGPUUnknownReason)
}
if !proven {
return finish(status.Effective.Reason)
return r.finishElasticWorldUnprovable(ctx, md, &status, finish, int(role.ElasticEP.Width))
}
width = live
}
if !proven {
return finish(status.Effective.Reason)
return r.finishElasticWorldUnprovable(ctx, md, &status, finish, int(role.ElasticEP.Width))
}
status.StableWidth = width
if width == int(role.ElasticEP.Width) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1188,6 +1188,106 @@ func TestAnUnreadableRayWorldStillCreatesNoOperation(t *testing.T) {
require.Equal(t, "native rank identities are unknown", f.observation(t).Reason)
}

// TestAnUnprovableWorldSurfacesTheBlockedDrive pins the loud park: with no operation record and a
// live world this pass cannot prove, the corrective scale is blocked — no pass may act without the
// rank proof — and the deployment's condition must say so instead of reading healthy while the
// engine and the pod set hold more quota than the spec asks for. The observation keeps the proof's
// own failure, and the park synthesizes no operation, issues no command and retires no member.
func TestAnUnprovableWorldSurfacesTheBlockedDrive(t *testing.T) {
for _, tc := range []struct {
name string
liveWidth string
blockProof func(f *elasticConvergenceFixture)
}{
{
name: "the live width is unknown", liveWidth: "unknown",
blockProof: func(f *elasticConvergenceFixture) { f.rayMalformed.Store(true) },
},
{
name: "the live width fails its forward proof", liveWidth: "4",
blockProof: func(f *elasticConvergenceFixture) { f.forwardsFail.Store(true) },
},
} {
t.Run(tc.name, func(t *testing.T) {
ctx := context.Background()
f := newElasticConvergenceFixture(t)
f.stageLostUpscaleWorld(t)
tc.blockProof(f)
scalesBefore := f.scaleCalls.Load()
members := func() map[string]types.UID {
pods := new(core.PodList)
require.NoError(t, f.reconciler.Client.List(ctx, pods))
live := map[string]types.UID{}
for i := range pods.Items {
live[pods.Items[i].Name] = pods.Items[i].UID
}
return live
}
before := members()

require.NoError(t, f.reconciler.reconcileModelDeploymentElasticResize(ctx, f.md))

_, readErr := newElasticOperationStore(f.reconciler.Client).Read(
ctx, f.md.Namespace, f.md.Name, f.md.UID)
require.True(t, apierrors.IsNotFound(readErr),
"an unprovable world synthesizes no operation")
require.Equal(t, scalesBefore, f.scaleCalls.Load(),
"an unprovable world issues no corrective command")
require.Equal(t, before, members(), "an unprovable world retires no member")
observed := f.observation(t)
elasticCondition := kubeapistatus.ConditionType("ElasticResize")
live := new(workercore.ModelDeployment)
require.NoError(t, f.reconciler.Client.Get(ctx, ctrlcli.ObjectKeyFromObject(f.md), live))
require.Equal(t, "Unknown", elasticCondition.GetStatus(live),
"an unprovable world must not read as a healthy one: %s", observed.Reason)
require.Equal(t, "WorldUnprovable", elasticCondition.GetReason(live))
message := elasticCondition.GetMessage(live)
require.Contains(t, message, "desired width 2", message)
require.Contains(t, message, "live width "+tc.liveWidth, message)
require.Contains(t, message, observed.Reason, message)
})
}
}

// TestTheBlockedDriveLevelsBackOnceTheWorldProves pins the recovery: the blocked surfacing is a
// state of the world, not a verdict — once a pass proves the live width again, the ordinary
// corrective block creates the downscale and the condition levels back to True/NoRefusal.
func TestTheBlockedDriveLevelsBackOnceTheWorldProves(t *testing.T) {
ctx := context.Background()
f := newElasticConvergenceFixture(t)
f.stageLostUpscaleWorld(t)
f.rayMalformed.Store(true)
scalesBefore := f.scaleCalls.Load()
elasticCondition := kubeapistatus.ConditionType("ElasticResize")
liveCondition := func(t *testing.T) (status, reason string) {
t.Helper()
live := new(workercore.ModelDeployment)
require.NoError(t, f.reconciler.Client.Get(ctx, ctrlcli.ObjectKeyFromObject(f.md), live))
return elasticCondition.GetStatus(live), elasticCondition.GetReason(live)
}

require.NoError(t, f.reconciler.reconcileModelDeploymentElasticResize(ctx, f.md))
status, reason := liveCondition(t)
require.Equal(t, "Unknown", status, "the unprovable world must surface the blocked drive")
require.Equal(t, "WorldUnprovable", reason)

f.rayMalformed.Store(false)
require.NoError(t, f.reconciler.reconcileModelDeploymentElasticResize(ctx, f.md))

op, readErr := newElasticOperationStore(f.reconciler.Client).Read(
ctx, f.md.Namespace, f.md.Name, f.md.UID)
require.NoError(t, readErr, "a proven world reaches the ordinary corrective block: %s",
f.observation(t).Reason)
require.Equal(t, elasticStateRecorded, op.State)
require.Equal(t, elasticWidth{Old: 4, Target: 2}, op.Width)
require.Equal(t, 4, f.observation(t).StableWidth)
status, reason = liveCondition(t)
require.Equal(t, "True", status, "a proven world leaves no block standing")
require.Equal(t, "NoRefusal", reason)
require.Equal(t, scalesBefore, f.scaleCalls.Load(),
"the corrective operation is recorded; the ordinary flow sends it on a later pass")
}

// elasticWatchedReader records the options every List reached the API server with and can be made
// to fail one, so a test can see what a reconcile-hot-path read asked for instead of inferring it
// from the objects it happened to return.
Expand Down
Loading