multi-actor worker API - #1283
Conversation
|
x-ref: #1164 🙈 |
96e9c63 to
975778f
Compare
a7b750e to
0533447
Compare
Tim Hockin (thockin)
left a comment
There was a problem hiding this comment.
I focused on API stuff. The atomic update of an assignment and the worker seems like something that will hurt us later, but maybe workers just not updated often enough for that to matter?
It's tricky logic and could do with more explanation (and specifically the "not updated often enough to matter" rational).
I did not focus on validation yet
| // declared resource limits. An unset message, or a zero field within it, means | ||
| // "unknown/unset" for that dimension: treated as unconstrained so placement is | ||
| // not blocked (matching the pre-capacity behavior). | ||
| // WorkerCapacity is what a worker pod has to give the Actors it hosts, and also |
There was a problem hiding this comment.
This comment is also unclear. I think you should clarify a resource model but "give/take" wording is sort of clunky.
IIUC this is the total capacity of the worker, independent of what may currently be allocated to actors. Is that right? So "available" is "worker.capacity - worker.status.allocated".
How would we model overcommit? Worker lies about capacity? Maybe add an overcommit scalar (e.g. 1.5)?
There was a problem hiding this comment.
How would we model overcommit? Worker lies about capacity? Maybe add an overcommit scalar (e.g. 1.5)?
My gut take was worker lies about capacity.
We will have to filter this stuff up to autoscaling, that approach seems easy enough to reason about but perhaps too limiting.
There was a problem hiding this comment.
We plan to consolidate this with status.
FWIW I've pushed 28k claims/s with a test of the storage layer, it scales pretty well with cores.. For overcomitted toy actors, 97k in 60s ... mostly bound by dataplane. This is only going to hurt us if the rest of the system gets really fast or if we have reason to do a lot of other writes to workers unrelated to scheduling. We're by far dataplane bound even at fairly considerable scale with this. We could do something else like ... a distributed scheduling architecture, but I'm not sure that's necessary. I've realized some other defects in this as-is when I got some time to sit and think over the weekend. Working on it.
|
7e01bfb to
725a108
Compare
|
I will iterate on this a bit more, including some cleanup to the commits etc, and notify when it's ready. For now, logging off. |
4644d42 to
cd33f4d
Compare
cd33f4d to
4e67536
Compare
Eitan Yarmush (EItanya)
left a comment
There was a problem hiding this comment.
Overall this makes sense, I still need to mull the whole thing over, but I am really liking the direction.
Codex found a couple potential races that we can confirm:
Binding commits before the Actor is updated to reference the Worker. If ateapi crashes in
between, the Worker holds an assignment that the Actor does not know about.
Later, DeleteActor skips Worker cleanup when actor.status.worker_assignment is empty. There is
no assignment-to-Actor foreign key or reconciler, so the capacity can remain leaked until the
Worker disappears.
Smallest sound fix: deletion must look up and release an assignment by Actor UID even when the
Actor lacks a backlink. Add a test covering bind → missing Actor update → delete.
When an Actor is already assigned to the same Worker, BindActorToWorker subtracts the old
resources and adds the replacement, but does not call admit.
Because Actor.actor_template is mutable, a stale claim can be resumed with larger limits and
push the Worker over capacity.
Smallest fix: subtract the previous reservation, run admission against that effective Worker,
then add the replacement.
| // The Worker being reported on. atespace is always empty; Workers are | ||
| // global-scoped. | ||
| // +k8s:opaqueType | ||
| ObjectRef worker = 1; |
There was a problem hiding this comment.
Doesn't this need to be inferred from the certificate anyway, is this redundant?
There was a problem hiding this comment.
We don't currently dial directly from the worker, it comes from ateom => atelet.
I have confidence that will change in the future but ... not sure when. Probably after we actually have multi-actor worker ... bit chicken and egg.
| // +k8s:optional | ||
| // +k8s:minimum=1 | ||
| int64 cpu_milli = 1; | ||
| Resources resources = 1; |
There was a problem hiding this comment.
Feel free to throw this away. If we parse this into a map in go anyway, should it be a map in the proto as well?
There was a problem hiding this comment.
Tim Hockin (@thockin) had an earlier comment about reusing the Resources type. I'm on the fence. Later we probably need to model more things. I don't love the Resources type but I like that it matches what we have on ActorTemplate.
There was a problem hiding this comment.
So I lean towards keeping it for now. If we want to use cpu_milli / memory bytes consistently we should rewrite ActorTemplate too.
There was a problem hiding this comment.
I don't love Resources but it maps to k8s. There are ...issues... properly handling quantity strings, which I am chasing down on the k8s side. I think we should ultimately decide to either support quantity strings all over or enforce quantized int values (e.g. cpu_millicores, memory_mb) all over.
There was a problem hiding this comment.
I suggest we bulk edit that in a follow-up, it's somewhat orthogonal and I'm not totally sold either way yet.
We're also going to need more complex resource types O(soon) to express GPUs properly, a count really isn't sufficient.
4e67536 to
bfcd462
Compare
These are real and should be fixed. I need to do some other remaining cleanup. |
bfcd462 to
787f032
Compare
787f032 to
e56db3c
Compare
Tim Hockin (thockin)
left a comment
There was a problem hiding this comment.
Proto seems straight forward at this point
| // | ||
| // +k8s:optional | ||
| // +k8s:minimum=1 | ||
| int32 page_size = 2; |
There was a problem hiding this comment.
Julian Gutierrez Oschmann (@juli4n) Should we bundle these into a generic ListOptions ?
There was a problem hiding this comment.
Obviously for a followup PR
There was a problem hiding this comment.
I don't have a strong opinion. We can wrap it and reuse the validations. Each response still needs to define a next page token.
| // +k8s:optional | ||
| // +k8s:minimum=1 | ||
| int64 cpu_milli = 1; | ||
| Resources resources = 1; |
There was a problem hiding this comment.
I don't love Resources but it maps to k8s. There are ...issues... properly handling quantity strings, which I am chasing down on the k8s side. I think we should ultimately decide to either support quantity strings all over or enforce quantized int values (e.g. cpu_millicores, memory_mb) all over.
| // What a Worker supplies, named as an ActorTemplate names what it asks for, | ||
| // so the two are one vocabulary and subtract directly. A name the Worker does | ||
| // not report is unconstrained rather than absent, so a Worker that has said | ||
| // nothing is not unschedulable. |
There was a problem hiding this comment.
I don't think this semantic works -- if I ask for 128 cores, you can't put me somewhere unless you know I have them. If I ask for GPU, you can't put me on a worker that has no GPUs.
There was a problem hiding this comment.
Agree ...
Aside: this is also a meh claude comment for sure.
my main takeaway from this PR is to cut off claude for API changes and stick to boilerplate fixes. even writing out what you want winds up with ... ehh not there yet.
next time I will entirely hand-craft the API commit at minimum.
There was a problem hiding this comment.
I think what I want here is to require that these are set, and make ateom fully set them now.
There was a problem hiding this comment.
| // Actor is placed only where free covers its limits in every dimension. | ||
| // | ||
| // A Worker overcommits by reporting more than it has. | ||
| message WorkerCapacity { |
There was a problem hiding this comment.
Do we need this intermediate struct if we use Resources?
There was a problem hiding this comment.
I'm not sure if it will just be resources, and actor count is not a dimension we represent elsewhere.
There was a problem hiding this comment.
Elaborating: I think we're going to need a more complex model for devices than <name: count>. So we'd have to add a new type anyhow.
e56db3c to
15394e9
Compare
15394e9 to
42628d0
Compare
|
7d9ced4 is substantial new code hoisted to this branch to manage #1283 (comment) -- we have the ateoms fully self-report capacity now. For now it's just count = 1 and cpu/memory from downward API, to keep complexity down and behavior close to before (but still only after ateom boots). In the future we will report live capacity (read cgroups instead, consider IPPR), account for overcommit, etc. |
| // to the neutral badge, which is the right visual treatment. | ||
| func workerPhase(w *ateapipb.Worker) string { | ||
| if w.GetStatus().GetAllocated().GetActors() > 0 { | ||
| if w.GetStatus().GetAllocation().GetAllocated().GetActors() > 0 { |
There was a problem hiding this comment.
a little clunky but oh well
dbee797 to
2360665
Compare
|
I think all review comments are addressed. 💤 |
Eitan Yarmush (EItanya)
left a comment
There was a problem hiding this comment.
Mostly nits, but also a couple real potential races
| // mixed ateom versions. | ||
| // | ||
| // atelet calls this with its own client certificate, as it does for | ||
| // MintCert. Idempotent: re-sending the same capacity is not a write. |
There was a problem hiding this comment.
Idempotent: re-sending the same capacity is not a write.
Is this trying to signal that it won't kick off a watch event for that worker?
There was a problem hiding this comment.
Yes. reworded.
|
|
||
| // retryReport calls send until it succeeds or ctx ends, backing off between | ||
| // attempts. | ||
| func retryReport(ctx context.Context, send func() error, backoff time.Duration) error { |
There was a problem hiding this comment.
super nit: this function takes a callback to be generic, (and maybe for the unit tests), but makes reading it slightly less clear as it's only actually retrying the reportOnce. Consider just getting rid of the callback and call directly.
| // worker reports once, so an accepted call is the only thing that puts | ||
| // capacity on the Worker, and a Worker record the syncer has not created yet | ||
| // is the ordinary reason for a first attempt to fail. | ||
| func (s *workerCapacityService) SetWorkerCapacity(ctx context.Context, req *ateletpb.SetWorkerCapacityRequest) (*ateletpb.SetWorkerCapacityResponse, error) { |
There was a problem hiding this comment.
Are we doing this ateom -> atelet -> ateapi because that's what we already have, or because that's what we actually want?
There was a problem hiding this comment.
The former. But we could switch in the future with only authz changes. I'm not sure we want to introduce "ateom uses the network directly" in this PR.
| // writeAndAppendEventFor is writeAndAppendEvent for a write whose event | ||
| // payload is only known once the transaction has read what it is changing. | ||
| // fn returns the worker to announce, or nil for a write that changed nothing. | ||
| func (p *Persistence) writeAndAppendEventFor(ctx context.Context, eventType store.WorkerEventType, fn func(ctx context.Context, tx pgx.Tx) (*ateapipb.Worker, error)) error { |
There was a problem hiding this comment.
These helpers appear identical except that writeAndAppendEventFor discards the Worker result. The existing helper already supports discovering the event payload inside the transaction. Could Bind ignore its returned Worker and Release return it directly, then remove this duplicate?
| // belongs to the caller, which knows whether that is the state it wanted. | ||
| func (w *WorkerWorkflow) DeleteWorker(ctx context.Context, name string, pre store.DeletePreconditions) (*ateapipb.Worker, error) { | ||
| worker, err := w.loadWorkerForDelete(ctx, name) | ||
| if err != nil { |
There was a problem hiding this comment.
Codex pointed out something interesting here. Essentially Worker remains schedulable during this deletion run.
A Worker remains ACTIVE while this sweep runs, so a concurrent resume can bind a new Actor after
its page has been processed. finalizeDeleted then deletes the Worker and cascades that new
assignment without clearing the Actor's backlink. The worker syncer supplies no version
precondition, so the bind's Worker version bump does not prevent this. Could deletion first
persist a state that makes Applies reject new binds, then sweep assignments?
There was a problem hiding this comment.
should be fixed
status.assignment held a single Actor. It becomes repeated, reusing field 2 so an old record reads back as a set of one, and status.allocated carries the running total that placement reads for every Worker on every decision. WorkerCapacity now holds a Resources set, named the way an ActorTemplate names its limits so that the two subtract, alongside an actors ceiling for the costs that cpu and memory do not cover. ActorAssignment.resources records what an Actor was admitted for. Capacity can now change over a Worker's life, but it cannot be cleared. A Worker reports its own capacity through the new WorkerReporting service. That is separate from Control because it is served only to atelet, and only for the Workers on its own node.
Readers use resources.WorkerAssignmentFor, and binding and releasing update status.allocated alongside the assignment list. Accounting moves to named quantities in resources.Quantities, which emits them in sorted order. Resources is a repeated field, so two totals that differ only in order compare unequal and rewrite the record for no change. Placement compares what is free in each dimension, and eligibility is asked separately from room. A caller re-validating a Worker that already holds the Actor would otherwise be told there is no space and evict it. A Worker whose recorded capacity does not parse is treated as full. Clearing capacity is rejected by a hand-written check. Declarative validation can mark a field immutable, but it cannot allow a field to change while still requiring it to be set. ListWorkers omits the assignments and reports occupancy through allocated. kubectl-ate prints ASSIGNED(n/m).
atelet reports on behalf of the Workers on its node, authenticated by its own client certificate as it is for MintCert. A Worker on another node comes back as NOT_FOUND. The atelet authentication moves into a package both services share. Capacity is the Worker's to report rather than the control plane's to infer, so the syncer no longer derives cpu and memory from the pod. Until a report arrives, the ceiling CreateWorker reifies holds the Worker to one Actor. A reporter can leave out dimensions it does not know, and those keep whatever is already recorded. Re-sending an unchanged capacity does not write. Nothing calls this yet.
A Worker carried its assignments inside its own record, so the record, its change event, and every watcher's copy of it all grew with the number of Actors on it. Each assignment becomes a row of its own. status.allocated stays on the Worker, because placement reads it for every candidate. The store updates it in the same transaction as the assignment it counts. Binding does not first read the assignment it might be replacing. The insert landing is proof that nothing was there, and a read could not see a claim that commits after it.
Claiming a worker rewrote its whole record, so goroutines racing on the same worker refused one another: 21% of activations failed at 12 in flight, and 67% at 24. BindActorToWorker now takes an admit callback and asks inside the transaction that holds the worker's row lock, so the answer cannot go stale before the write, and a refusal rolls the speculative row back. That removes the version precondition, the fresh read, and the assignment lookup. The retry budget had to grow, because a claim refused for want of room re-runs scheduling.
Capacity is observed, not requested: the Worker reports it through SetWorkerCapacity and no client writes it. Move it from the Worker to WorkerStatus, beside the allocated total it is compared against. UpdateWorker restores the stored status, so a client can no longer change or clear capacity at all. The hand-written no-clearing check goes with it. Drops the reserved field numbers, which nothing shipped needs, and the opaqueType markers on SetWorkerCapacityRequest -- those cancel the validation of the type they annotate, which is the opposite of what this RPC wants.
Capacity read an absent dimension as unconstrained, so an Actor asking for a GPU could be placed on a Worker that never reported one. Absent now means none of it: a Worker reports everything it can supply, and a name missing from its capacity is one no Actor may ask it for. SetWorkerCapacity therefore replaces rather than merges. The merge existed to stop a reporter that knew only its actor ceiling from erasing compute it could not speak to, which a complete report removes -- and replacing is what the name says.
CreateWorker was fabricating a capacity of one actor for every Worker it stored. That contradicts the rule the API just adopted -- a Worker reports every dimension it has, so an absent one is none of it -- because the control plane was answering on the Worker's behalf before the worker had said anything. Nothing was calling SetWorkerCapacity, so add the reporter that makes it real: ateom answers a new GetCapacity, atelet sweeps the node's ateoms and forwards what they say, and a Worker holds no capacity until that lands. ateom learns its compute limits from the downward API, since it has no reader for its own cgroup and inventing one is a larger change. Left as a TODO: the environment is fixed at pod creation, so it goes stale under in-place pod resize.
Capacity and allocated are only ever read together -- placement subtracts one from the other -- so they sit under a single status.allocation rather than as two sibling fields. WorkerCapacity becomes WorkerResources: the same message carries both totals, so naming it after one of them was wrong.
Two problems Codex found in the assignment subresource. A rebind marshaled the metadata a first bind gets, so a retried claim replaced the recorded assignment with a fresh uid, create time and version 1, making an update look like a new subresource. It now carries the recorded identity forward. DeleteWorker swept a Worker that was still ACTIVE, so a resume could bind an Actor to a page already passed and the delete would cascade that assignment away, leaving the Actor pointing at a Worker that is gone. The delete drains first. The caller's guards are checked before that, against the Worker it observed, since the drain moves the version. Also drops writeAndAppendEventFor, which differed from writeAndAppendEvent only in discarding the returned Worker.
f5c0685 to
5886f4b
Compare
Tim Hockin (thockin)
left a comment
There was a problem hiding this comment.
Just the proto file again
| // | ||
| // +k8s:optional | ||
| // +k8s:minimum=1 | ||
| int32 page_size = 2; |
There was a problem hiding this comment.
Obviously for a followup PR
| // +k8s:optional | ||
| // +k8s:minimum=1 | ||
| int64 cpu_milli = 1; | ||
| Resources resources = 1; |
ReleaseActorFromWorker took an expectedVersion it never read, so callers passed a guard that did nothing. Honoring it would be wrong rather than a fix: a Worker's version moves whenever any of its Actors binds or releases, so on a Worker holding thousands the guard would refuse releases for changes that have nothing to do with the Actor being released, and the delete sweep would fail from its second page on. The assignment's own key and the Worker's row lock are what make the release exact.
The Go handler, its test and the apitool exemption follow the rename.
Tim Hockin (thockin)
left a comment
There was a problem hiding this comment.
This PR is big enough that there are no obvious bugs. I'm still worried about the multi-resource "dance" of creating the assignment. There are things I am not quite sure are right but are not obviously wrong either - for example a RESUMING actor bound to a DRAINING worker currently crashes but it's not clear to me if that is AI paranoia, if it is a legitimate "oh no" situation, or if it is actually an acceptable corner case.
Basically I'm trying to convince myself that the high-level flow is right. There are long sequences of tricky, conditional-rich code with no clear comments on assumed preconditions or asserted postconditions.
I'm still walking thru it and I will send you a diff in bit just for some comments I think could be better. I don't see a reason to block this going in at this point.
| // | ||
| // An absent name is none of that resource. A Worker reports every dimension it | ||
| // has, so a name missing from its capacity is one it cannot supply at all. | ||
| type Quantities map[string]resource.Quantity |
There was a problem hiding this comment.
OK for the moment, but we need to decide if we REALLY want to use Quantity the string format and if so if we really want to parse it into Quantity the type or into a math/big.Float or something
There was a problem hiding this comment.
Yeah. Do we already have a TODO for this? Should we file one?
Part of #1266
This is a draft of the core API + data store changes.
It's still a large PR, apologies.
The "as rows" commit could be split out, but this takes it to ~all of the breaking changes we can't hide behind updating internals.
Same for the claimlock, but in both cases it seems these are worth understanding when considering the API shape.
They're loadbearing for performance once we actually have multi-actor workers.