diff --git a/docs/cluster.md b/docs/cluster.md index f6cc8fc3d..3215577b6 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -186,7 +186,8 @@ it while it sleeps write it, so the claim path is unchanged. not hold the claim reads the row; when the token hash matches, it takes the row with one compare-and-set on the epoch. The same statement requires the owner's [node lease](#node-lease) to be expired, unless the owner is this - node (with its record gone). The node records the claim as archived and + node (with its record gone) or the row has no owner, since its owner + released it in an [evacuation](#evacuating-a-node). The node records the claim as archived and answers as the owner. Two nodes adopting at once leave one owner; the other answers 404. A tombstone is never adopted. The next call that reaches the guest wakes the claim from the store as usual. An owner with a live lease answers for @@ -262,6 +263,49 @@ never returns leaves behind only the tombstones and checkpoints of the claims it was waking or releasing when it died, plus what a replicated claim leaves (see [running claims](#running-claims-across-node-loss)). +### Evacuating a node + +[`POST /v1/drain?evacuate=1`](sandboxd-api.md#post-v1drain) empties a node for +maintenance or retirement without losing its claims. The node cordons as a +drain does, then moves each claim with bounded concurrency, on the maintenance +budget, archived claims first and running ones last: it hibernates a running claim, archives a hibernated one through the +archive path above, releases the row, and asks a peer to adopt it. A release is +one compare-and-set: the row this node owns at the epoch it holds moves to no +owner at a new epoch. A row with no owner has no lease to wait out, so a peer +takes it with the same epoch compare-and-set as any adoption, and two peers +adopting at once still leave one owner. This node keeps the claim listed until +a peer answers `GET /v1/sandboxes/{id}/owner`, so a root listing finds the claim +on at least one node at every moment, then forgets it under the claim's transition lock +and answers 404 for it; the checkpoint stays, since it belongs to the adopter. +With no adopter (no peer, every peer draining or unreachable) the node takes the +row back and the claim stays here. Until the hand-off, a wake of a claim the +evacuation still has to move answers 503 with `Retry-After`, so a client call +cannot wake it back onto the node. A running claim keeps serving until its turn. +The run takes the budget one batch at a time, so crash recovery and reaps of the +claims that stay keep their turn while it runs. + +The release keeps the fencing intact. Every later write of the old owner is +refused: a put is owner-guarded, and a replace, consume or second release +needs the epoch the release moved on. A release that finds its row moved +reads it: a row with no owner is one an outage cut short or a later owner +released, so the node asks a peer to adopt it and never takes it back; a row another node took means +the claim moved already, and the node drops it; a row this node still owns at +another epoch keeps the claim, which is skipped. A crash between the release and a peer's answer leaves a row no node +lists: a restart drops the claim, and the next `Lookup` with its token adopts it, +so a holder that recovers the token from a root listing loses that claim. A +released row past its retention deadline is purged by any +node's hourly sweep, like the row of a departed owner. + +An archive the sweeps already run for a claim is waited out and the claim +retried every 5 seconds. Claims that cannot move are skipped and keep serving +here; once `evacuating` is gone `claimed` counts them, and the log names each +one with its reason. A client that holds a handle gets 404 from this node after +the move, the same answer as for a released claim, and only a `Lookup` tells +the two apart: the SDKs' `Close` does this itself. A reader-served connection +gets the retry as an error frame, not a 503, and a preview URL minted before +the move stops working until it is minted again. `DELETE /v1/drain` stops the evacuation: claims not yet started +stay as they are, and one mid-archive finishes its archive but keeps its row. + ### Node lease With `archive_failover` every node renews one row of `sandboxd_nodes (cell, diff --git a/docs/deploy.md b/docs/deploy.md index e33970185..76ff3f2b8 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -157,7 +157,7 @@ the intersection of the two. | `disk_reserve` | unset | `{"free": "100G", "dirs": ["/srv/sbx/run"]}`: while the filesystem of any listed directory (default `data_dir`) has less than `free` available, checked every 2 s, the node refuses new claims, forks, checkpoint claims, wakes of hibernated or archived sandboxes, hibernates, checkpoint captures and promotes with 429, and parks refill; the idle sweep and a lease-end archive leave their sandboxes running and retry on the next tick; `/v1/info` reports it as at capacity and the mesh sees no warm VMs on it. Running sandboxes keep running. On an XFS project quota the free space is the project's. List cocoon's `run_dir` when it sits on another filesystem, since the copy-on-write disks grow there. Size `free` to at least two sandboxes' worst-case growth (copy-on-write disk plus memory file each) and keep the refusal above your disk alert's warning level, so the warning comes first: on a 640G project with 40G disks and 8G memory files, `100G` refuses at 84% used | | `idle_hibernate_seconds` | 0 (off) | node-wide idle policy for unpooled claims (template/checkpoint claims): a none-lane claim is hibernated once it has had no open data-plane connection (relay, buffered exec, preview) and no egress request in flight for this long; the clock restarts when the last connection closes, so a long command is never cut short. The next call that reaches the guest wakes it transparently. Per-pool `idle_hibernate_seconds` does the same for that pool's claims; pooled keys ignore the node-wide value, and egress pools reject it because they cannot resume safely. Opt in deliberately: a wake costs latency and the snapshot, so callers with their own idle logic must not pay twice | | `archive_after_seconds` | 0 (off) | tier below hibernation: a hibernated claim idle this long is checkpointed to the store and its local VM dropped, freeing the node entirely; the next call that reaches the guest restores it transparently (a checkpoint restore's latency) with a fresh lease of the length the claim was granted (the server default of 5m when it asked for none); a lease moves only when a client calls [`/renew`](sandboxd-api.md#post-v1sandboxesidrenew). Keep `idle_hibernate_seconds` under the leases your claims use, or a woken sandbox that idles again is destroyed by the reap before it can hibernate, its archive already consumed by the wake; one still in use when its lease ends is destroyed the same way, unless it claimed `on_expire: archive`, which archives at the lease end whatever this setting. Requires `idle_hibernate_seconds > 0` and must exceed it. Node-wide for unpooled keys; per-pool overrides for that pool | -| `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived; the claim is then gone for good. On S3, chunks and manifests remain until [offline reclamation](#reclaiming-s3-storage). On archive this retention window replaces the live claim deadline; 0 clears the deadline so the archive is kept forever. Same node-wide/per-pool split | +| `archive_delete_after_seconds` | 0 (keep) | purge an archived claim's store checkpoint this long after it was archived; the claim is then gone for good. On S3, chunks and manifests remain until [offline reclamation](#reclaiming-s3-storage). On archive this retention window replaces the live claim deadline; 0 clears the deadline so the archive is kept forever. An [evacuation](cluster.md#evacuating-a-node) keeps a longer live deadline, and a claim of a pool without an archive tier, whose `on_expire` is not `archive`, keeps its deadline. Same node-wide/per-pool split | | `archive_failover` | false | let another node adopt this node's archived claims once its node lease expires, and the running claims of pools with `replicate_seconds`: each archive and replication round also writes a row (token SHA-256, owner, epoch, checkpoint id, retention deadline, and the claim record with its env, `guest: false` values included, in plaintext) to the shared PostgreSQL, and a `Lookup` that reaches any node takes the claim over there (see [archived claims across node loss](cluster.md#archived-claims-across-node-loss)). Requires `meta_store` with a `cell` (rows are scoped by it, so meshes sharing one database never adopt from each other), a `checkpoint_store` of kind `s3` and a `mesh`, all checked at config load. The node also renews a lease row every 5 s (see [node lease](cluster.md#node-lease)). With the meta store down, archives, wakes, releases and env writes of archived claims, and env writes and renews of hibernated replicated claims, fail and retry, and replicated claims stop being served once the lease is overdue (see [running claims](cluster.md#running-claims-across-node-loss)); other running claims never touch it. Restart only | | `mesh` | unset | join a cluster ([Clusters](cluster.md)); unset = single node | | `pools[]` | — | warm pools, keyed by `(template, net, size)`. `warm` defaults to 4 only when both `warm` and `warm_max` are omitted; an explicit `warm: 0` is kept, so `warm: 0` under a `warm_max` starts the pool empty and grows it on demand, and `warm: 0` alone registers the key with no warm VMs (`PUT /v1/pools` never defaults `warm`); `net` is `none` or `egress`; `size` is a tier, below. Retune online without a restart via [`PUT /v1/pools`](sandboxd-api.md#put-v1pools) — omitted pools drain, and their goldens are deleted once they are idle (`warm: 0` keeps the golden). This is the **first-boot seed**: once a node takes a `PUT /v1/pools`, the applied set persists to `/pools.json` and overrides this section on every later boot (a startup log notes it); delete `pools.json` to return to config-owned pools. With a `meta_store` `cell`, the cell's set overrides it instead (see [shared pool set](#shared-pool-set)). Egress stays config-owned either way. A template re-pulled or re-imported under the same ref is picked up within a minute: the pool rebuilds its golden and replaces its warm VMs. See [state ownership](cluster.md#state-ownership) | @@ -713,7 +713,11 @@ To empty a node for maintenance, cordon it first: [`POST /v1/drain`](sandboxd-api.md#post-v1drain) (root) stops new claims and drains the warm pools; poll `GET /v1/info` until `claimed` reaches zero (or let the leases expire), then stop sandboxd. `DELETE /v1/drain` uncordons. -The drain is not persisted — a restarted node serves again. +The drain is not persisted — a restarted node serves again. With +`archive_failover`, `POST /v1/drain?evacuate=1` also moves the claims to the +peers with their state, and a peer holds each one before this node drops it; poll `GET /v1/info` until `evacuating` is gone, and +`claimed` then counts the claims it skipped (see +[evacuating a node](cluster.md#evacuating-a-node)). ### Disabling archive failover diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index e00de3fea..cadaa0d16 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -518,6 +518,7 @@ success, 404 unknown id or wrong token, 409 a failed egress-lane or volume sandbox, or one whose VM record is gone, 403 a removed tenant's archive, 429 while the disk reserve is low or the node is at [`max_running`](deploy.md#configuration), 500 when the cold boot or a restore fails, 503 `lease lost; retry` for a replicated claim while its node's lease is overdue, +503 `sandbox moving to another node; retry` while an [evacuation](#post-v1drain) has yet to move it, 503 `archive registry down; retry` for an archived claim under `archive_failover` while the meta store is down. ## POST /v1/sandboxes/{id}/renew @@ -784,10 +785,35 @@ or TTL. Pool ownership is untouched — no pools.json write, no config change. Answers the fresh `GET /v1/info` payload; poll `claimed` to zero to know the node is empty. Deliberately not persisted: a restarted node serves again. +`POST /v1/drain?evacuate=1` cordons the same way, then moves the claims to the +peers with their state, so the node empties without waiting for the leases to +run out (see [evacuating a node](cluster.md#evacuating-a-node)). It needs +[`archive_failover`](cluster.md#archived-claims-across-node-loss). In the +background, with bounded concurrency, each claim is hibernated if it runs, +archived to the checkpoint store if it is not archived, its registry row is +released, and a peer is asked to adopt it through `GET /v1/sandboxes/{id}/owner`, +so the root listing on that peer shows it with its token. Open streams to a +claim end when it hibernates. Until a peer holds it, a call that would wake the +claim answers 503 `sandbox moving to another node; retry` with `Retry-After`; +after that this node answers 404 and the client looks the claim up again. With +no peer to adopt it, the node takes the row back and the claim stays. A claim +that cannot move stays here +and keeps serving: an egress-lane claim, one with volumes, a failed one, one +whose tenant was removed, one archived before `archive_failover` was on, one +whose step failed, and every claim of a node without `archive_failover`. The +answer and `GET /v1/info` count the claims left in `evacuating`; once it is +gone, `claimed` counts the ones that stayed, and the log names each and why. A +claim of a pool without an archive tier, whose `on_expire` is not `archive`, +keeps its lease deadline across the move; any other keeps the later of its +lease and `archive_delete_after_seconds`, or no deadline when that is 0. A call +while an evacuation runs changes nothing; a call after one finished starts a +new one over the claims still here. + ## DELETE /v1/drain -Auth: root only (tenant tokens get 403). Lifts the drain and kicks an immediate refill. Answers the -fresh info payload. +Auth: root only (tenant tokens get 403). Lifts the drain, stops an evacuation, and kicks an immediate refill. Answers the +fresh info payload. A claim an evacuation is archiving finishes its archive and +stays here archived; the next call wakes it. ## POST /v1/sandboxes/{id}/preview @@ -1010,7 +1036,7 @@ verb never wakes one) or its image serves no instance metadata. Auth: root only (tenant tokens get 403). Prometheus text format, hand-rendered: pool warm/target gauges, claimed/running/max_running/hibernated/archived/failed/draining -gauges, a per-tenant live-claim gauge (`sandboxd_tenant_claims{tenant="…"}`, +gauges, `sandboxd_evacuating` (the claims an evacuation has yet to move), a per-tenant live-claim gauge (`sandboxd_tenant_claims{tenant="…"}`, tenants holding live claims only, so the label set is bounded by the node's claims), `sandboxd_config_digest_mismatch` on a mesh, claims by tier (warm/clone/cold), wake/hibernate/fork/checkpoint/promote/release/reap counters plus @@ -1224,8 +1250,9 @@ Auth: the sandbox's own token. Answers `{"owner_addr": "host:port"}` when this node owns the sandbox, 404 otherwise. With [`archive_failover`](cluster.md#archived-claims-across-node-loss), a node that does not hold the sandbox adopts it when its row names an owner whose node -lease expired, and answers 200; 503 with `Retry-After` when the meta store is -down. A row whose owner's lease is live is never adopted: 404. A draining or fenced +lease expired, or whose owner released it in an +[evacuation](#post-v1drain), and answers 200; 503 with `Retry-After` when the meta store is +down. A row whose owner's lease is live is never adopted unless the owner released it: 404. A draining or fenced node, or one that lifted its fence less than 8 seconds ago, does not adopt: 404. A node whose lease is overdue answers 404 for the replicated claims it holds, so the `Lookup` scatter moves on. With `client_advertise` configured, @@ -1267,7 +1294,8 @@ and was not restarted (omitted at zero); all are included in `claimed`. `running` counts claims that hold a VM, with a wake in flight counted, and `max_running` is the [cap](deploy.md#configuration) on it (omitted when off). A node cordoned via [`POST /v1/drain`](#post-v1drain) additionally reports -`"draining": true`. When refill is parked because the node cannot start +`"draining": true`, and during an evacuation `evacuating` (claims left to +move, omitted at zero). When refill is parked because the node cannot start another VM, `"at_capacity": true` distinguishes a full node from one still filling its warm target, and `at_capacity_reason` carries the engine's capacity reason. Both capacity fields are omitted while refill is not capacity-blocked. Each VM holds a console pty, so a node stops at `kernel.pty.max` (4096 by default) with `no space left on device`; raise it past the VM count you plan to run. diff --git a/docs/sdk.md b/docs/sdk.md index 752393d3f..462972c0a 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -227,8 +227,11 @@ are exported. `Sandbox.TemplateDigest` is the exact content identity when the claim cloned a promoted template; it is empty for other sources. `Owner()` names the owning node, and `Token()` returns the per-sandbox bearer to persist with `ID` for a -later `Lookup`; `Close()` releases the sandbox (releasing one already gone is -not an error, and `Close` is bounded internally so it stays defer-friendly). +later `Lookup`; `Close()` releases the sandbox (it follows a sandbox that +moved to another node with a `Lookup`, releasing one already gone is not an +error, and `Close` is bounded internally so it stays defer-friendly). Any other +call on a handle answers 404 once its sandbox has moved; run `Lookup` to tell +that from a released sandbox. Volume sandboxes cannot hibernate, fork, checkpoint, or promote. Egress-lane sandboxes cannot either, nor take `WithArchiveOnExpire`; each answers 409. Passing `WithVolumes` or `WithClaimRef` to `Checkpoint.New` returns a local error: diff --git a/sandboxd/archives/archives.go b/sandboxd/archives/archives.go index 606208a69..f81f1dc63 100644 --- a/sandboxd/archives/archives.go +++ b/sandboxd/archives/archives.go @@ -1,4 +1,4 @@ -// Package archives keeps one cluster-wide row per archived or replicated claim, so another node can adopt the claim once its owner's lease expires. +// Package archives keeps one cluster-wide row per archived or replicated claim, so another node can adopt the claim once its owner's lease expires or its owner releases the row. package archives import ( @@ -37,6 +37,8 @@ type Registry interface { Replace(ctx context.Context, rec Record, expect int64) (int64, error) // Take moves the live row at epoch expect to owner at a new epoch and returns it; another node's row moves only once its owner's lease expired. Take(ctx context.Context, id string, expect int64, owner string) (int64, error) + // Release moves the live row owner holds at epoch expect to no owner at a new epoch and returns it, so any node can take the row at once. + Release(ctx context.Context, id string, expect int64, owner string) (int64, error) // TakeOwned moves each live row owner holds at its epoch in expect to a new epoch in one call and returns the new epochs of the rows it moved. TakeOwned(ctx context.Context, owner string, expect map[string]int64) (map[string]int64, error) // Consume turns the live row at epoch expect into a tombstone at a new epoch and returns it. diff --git a/sandboxd/archives/archivestest/archivestest.go b/sandboxd/archives/archivestest/archivestest.go index f66030b46..21fa63370 100644 --- a/sandboxd/archives/archivestest/archivestest.go +++ b/sandboxd/archives/archivestest/archivestest.go @@ -83,6 +83,16 @@ func (m *Mem) Take(_ context.Context, id string, expect int64, owner string) (in }) } +func (m *Mem) Release(_ context.Context, id string, expect int64, owner string) (int64, error) { + return m.advance(id, expect, func(rec *archives.Record) bool { + if rec.Owner != owner { + return false + } + rec.Owner = "" + return true + }) +} + func (m *Mem) TakeOwned(_ context.Context, owner string, expect map[string]int64) (map[string]int64, error) { taken := map[string]int64{} for id, epoch := range expect { @@ -228,6 +238,7 @@ func Run(t *testing.T, open Opener) { t.Run("rows lists the named ids", func(t *testing.T) { testRows(t, open(t)) }) t.Run("replace is a compare-and-set", func(t *testing.T) { testReplace(t, open(t)) }) t.Run("take waits out a live lease", func(t *testing.T) { testTakeLease(t, open(t)) }) + t.Run("release lets another node take the row at once", func(t *testing.T) { testRelease(t, open(t)) }) t.Run("sweepable spares a live owner's expired row", func(t *testing.T) { testSweepableLease(t, open(t)) }) t.Run("renew moves a lease forward", func(t *testing.T) { testRenewExtends(t, open(t)) }) } @@ -533,6 +544,54 @@ func testTakeLease(t *testing.T, r archives.Registry) { } } +func testRelease(t *testing.T, r archives.Registry) { + ctx := t.Context() + first := mustPut(t, r, record("sb_1", "a", time.Time{})) + if err := r.Renew(ctx, "a", time.Hour); err != nil { + t.Fatalf("renew: %v", err) + } + if _, err := r.Release(ctx, "sb_1", first, "b"); !errors.Is(err, archives.ErrLost) { + t.Errorf("release by a node that does not own the row: %v, want ErrLost", err) + } + if _, err := r.Release(ctx, "sb_1", first-1, "a"); !errors.Is(err, archives.ErrLost) { + t.Errorf("release at a stale epoch: %v, want ErrLost", err) + } + moved, err := r.Release(ctx, "sb_1", first, "a") + if err != nil { + t.Fatalf("release: %v", err) + } + released, _, _ := r.Get(ctx, "sb_1") + if released.Owner != "" || released.Epoch != moved || released.Epoch <= first || released.Consumed { + t.Fatalf("row after release: %+v, want a live row with no owner above epoch %d", released, first) + } + if _, err = r.Put(ctx, record("sb_1", "a", time.Time{})); !errors.Is(err, archives.ErrLost) { + t.Errorf("the old owner's put after release: %v, want ErrLost", err) + } + if _, err = r.Replace(ctx, record("sb_1", "a", time.Time{}), first); !errors.Is(err, archives.ErrLost) { + t.Errorf("the old owner's replace after release: %v, want ErrLost", err) + } + if _, err = r.Consume(ctx, "sb_1", first); !errors.Is(err, archives.ErrLost) { + t.Errorf("a consume at the epoch before release: %v, want ErrLost", err) + } + if _, err = r.Release(ctx, "sb_1", released.Epoch, "a"); !errors.Is(err, archives.ErrLost) { + t.Errorf("a second release by the old owner: %v, want ErrLost", err) + } + epoch, err := r.Take(ctx, "sb_1", released.Epoch, "b") + if err != nil || epoch <= released.Epoch { + t.Fatalf("take of a released row while the old owner's lease is live: %d %v, want a new epoch", epoch, err) + } + if got, _, _ := r.Get(ctx, "sb_1"); got.Owner != "b" || got.Epoch != epoch { + t.Errorf("row after the take: %s@%d, want b@%d", got.Owner, got.Epoch, epoch) + } + spent, err := r.Consume(ctx, "sb_1", epoch) + if err != nil { + t.Fatalf("consume: %v", err) + } + if _, err = r.Release(ctx, "sb_1", spent, "b"); !errors.Is(err, archives.ErrLost) { + t.Errorf("release of a tombstone: %v, want ErrLost", err) + } +} + func testSweepableLease(t *testing.T, r archives.Registry) { ctx := t.Context() past := time.Now().Add(-time.Minute) diff --git a/sandboxd/archives/pg/pg.go b/sandboxd/archives/pg/pg.go index 1c0668561..b8c3fa258 100644 --- a/sandboxd/archives/pg/pg.go +++ b/sandboxd/archives/pg/pg.go @@ -78,6 +78,12 @@ func (r *Registry) Take(ctx context.Context, id string, expect int64, owner stri RETURNING epoch`, r.cell, id, expect, owner) } +func (r *Registry) Release(ctx context.Context, id string, expect int64, owner string) (int64, error) { + return r.advance(ctx, `UPDATE sandboxd_archives SET owner = '', epoch = nextval('sandboxd_archives_epoch'), updated_at = now() + WHERE cell = $1 AND id = $2 AND epoch = $3 AND owner = $4 AND NOT consumed + RETURNING epoch`, r.cell, id, expect, owner) +} + func (r *Registry) TakeOwned(ctx context.Context, owner string, expect map[string]int64) (map[string]int64, error) { ids, epochs := make([]string, 0, len(expect)), make([]int64, 0, len(expect)) for id, epoch := range expect { diff --git a/sandboxd/main.go b/sandboxd/main.go index c2bd40963..d8c1dd6b7 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -135,6 +135,8 @@ func main() { }, &peer.HTTPPuller{Token: cfg.APIToken}) broadcaster := &peer.Broadcaster{Peers: msh.PeerAddrs, Token: cfg.APIToken} mgr.WithPeerDelete(broadcaster.Delete) + adopter := &peer.Adopter{Peers: msh.PeerAddrs} + mgr.WithPeerAdopt(adopter.Adopt) go gossipNodeState(ctx, msh, mgr) logger.Infof(ctx, "mesh %s joined (%d seeds)", cfg.Mesh.ID(), len(cfg.Mesh.Join)) } diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index f997a5cbe..6289afd94 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -43,6 +43,16 @@ func (m *Manager) retentionDeadline(key types.PoolKey) time.Time { return time.Time{} } +// archiveDeadline is the retention deadline unless the claim's lease must survive the archive. +func (m *Manager) archiveDeadline(sb *types.Sandbox) time.Time { + retention := m.retentionDeadline(sb.Key) + optedIn := sb.OnExpire == types.ExpireArchive || m.archiveAfterFor(sb.Key) > 0 + if !optedIn || m.evacuatingLocked(sb.ID) && !retention.IsZero() && sb.Deadline.After(retention) { + return sb.Deadline + } + return retention +} + func (m *Manager) archiveOnce(ctx context.Context) { if !m.archiveSweep.CompareAndSwap(false, true) { return // the previous sweep's archives are still draining @@ -146,7 +156,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { snap, vmName, sock, prevDeadline, prevPending, replica := sb.HibernateSnap, sb.VMName, sb.VsockSocket, sb.Deadline, sb.ArchivePending, sb.ReplicaCk sb.ArchiveCk, sb.ArchiveEpoch, sb.ArchivePending, sb.ReplicaCk = ck.ID, epoch, "", "" sb.HibernateSnap, sb.VsockSocket, sb.VMName = "", "", "" - sb.Deadline = m.retentionDeadline(sb.Key) + sb.Deadline = m.archiveDeadline(sb) js := m.store.set(sb) m.mu.Unlock() if saveErr := m.store.commit(js); saveErr != nil { diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 96a8cade3..887422dad 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -439,6 +439,9 @@ func (m *Manager) beginWake(sb *types.Sandbox) error { } m.mu.Lock() defer m.mu.Unlock() + if sb.Failed == "" && m.evacuatingLocked(sb.ID) { + return ErrEvacuating + } if err := m.runningErr(1); err != nil { return err } @@ -699,7 +702,7 @@ func (m *Manager) reapOnce(ctx context.Context) { case reapArchive: logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) case reapPause: - logSweepResult(ctx, logger, m.pauseExpired(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) + logSweepResult(ctx, logger, m.pauseArchive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) default: td := m.quiesceVolumes(ctx, v.sb) td.holds = v.holds @@ -716,7 +719,7 @@ func (m *Manager) archivesAtDeadline(sb *types.Sandbox) bool { return sb.Failed == "" && !m.tenantGone(sb.Tenant) && (sb.OnExpire == types.ExpireArchive || sb.HibernateSnap != "" && m.archiveAfterFor(sb.Key) > 0) } -func (m *Manager) pauseExpired(ctx context.Context, sb *types.Sandbox) error { +func (m *Manager) pauseArchive(ctx context.Context, sb *types.Sandbox) error { sb.Transition.Lock() m.mu.Lock() live := m.claimed[sb.ID] == sb && sb.ArchiveCk == "" diff --git a/sandboxd/pool/drain.go b/sandboxd/pool/drain.go index e78912097..1363c08d9 100644 --- a/sandboxd/pool/drain.go +++ b/sandboxd/pool/drain.go @@ -1,8 +1,26 @@ package pool -import "context" +import ( + "cmp" + "context" + "errors" + "fmt" + "maps" + "slices" + "time" -// Drain cordons the node: claims answer 429 and warm VMs are destroyed. Not persisted. + "github.com/projecteru2/core/log" + + "github.com/cocoonstack/sandbox/sandboxd/archives" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +type evacuation struct { + pending map[string]*types.Sandbox // guarded by m.mu + stop context.CancelFunc +} + +// Drain cordons the node: claims answer 429 and warm VMs are destroyed. func (m *Manager) Drain(ctx context.Context) { var trim []string m.mu.Lock() @@ -14,10 +32,171 @@ func (m *Manager) Drain(ctx context.Context) { m.destroyAll(context.WithoutCancel(ctx), trim).Wait() } -// Uncordon lifts a drain and kicks an immediate refill. +// Evacuate drains the node, then archives each claim it can and releases its row, so a peer adopts it at once. +func (m *Manager) Evacuate(ctx context.Context) { + m.Drain(ctx) + m.mu.Lock() + defer m.mu.Unlock() + if m.evac != nil && len(m.evac.pending) > 0 { + return + } + ctx, stop := context.WithCancel(context.WithoutCancel(ctx)) + m.evac = &evacuation{pending: maps.Clone(m.claimed), stop: stop} + go m.runEvacuation(ctx, m.evac) +} + +// Uncordon lifts a drain, stops an evacuation, and kicks an immediate refill. func (m *Manager) Uncordon(ctx context.Context) { m.mu.Lock() m.draining = false + e := m.evac + m.evac = nil m.mu.Unlock() + if e != nil { + e.stop() + } m.refillOnce(context.WithoutCancel(ctx)) } + +// runEvacuation repeats its pass while a claim waits for an archive the sweeps run. +func (m *Manager) runEvacuation(ctx context.Context, e *evacuation) { + defer e.stop() + for { + m.mu.Lock() + batch := slices.SortedFunc(maps.Values(e.pending), func(a, b *types.Sandbox) int { return cmp.Compare(moveCost(a), moveCost(b)) }) + m.mu.Unlock() + // One budget's worth at a time leaves crash recovery and reaps their FIFO turn. + for chunk := range slices.Chunk(batch, cap(m.maintSem)) { + m.runBounded(ctx, len(chunk), func(ctx context.Context, i int) { m.evacuateOne(ctx, e, chunk[i]) }).Wait() + } + m.mu.Lock() + left := len(e.pending) + m.mu.Unlock() + if left == 0 { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(reapInterval): + } + } +} + +func (m *Manager) evacuateOne(ctx context.Context, e *evacuation, sb *types.Sandbox) { + err := m.evacuateClaim(ctx, sb) + if errors.Is(err, errArchiveBusy) { + return + } + m.mu.Lock() + skip := err != nil && m.claimed[sb.ID] == sb + delete(e.pending, sb.ID) + m.mu.Unlock() + if skip { + log.WithFunc("pool.evacuateOne").Warnf(ctx, "evacuation skips %s: %v", sb.ID, err) + } +} + +func (m *Manager) evacuateClaim(ctx context.Context, sb *types.Sandbox) error { + if m.archives == nil { + return errNoFailover + } + m.mu.Lock() + _, busy := m.archiving[sb.ID] + gone := m.claimed[sb.ID] != sb || !m.evacuatingLocked(sb.ID) + archived, tenantGone := sb.ArchiveCk != "", m.tenantGone(sb.Tenant) + pause := !gone && !tenantGone && !busy && !archived + if pause { + m.archiving[sb.ID] = struct{}{} + } + m.mu.Unlock() + switch { + case gone: + return nil + case tenantGone: + return ErrTenantRemoved + case busy: + return errArchiveBusy + case pause: + if err := m.pauseArchive(ctx, sb); err != nil { + return err + } + } + sb.Transition.Lock() + defer sb.Transition.Unlock() + return m.handOff(ctx, sb) +} + +// Caller holds Transition. +func (m *Manager) handOff(ctx context.Context, sb *types.Sandbox) error { + m.mu.Lock() + live, archived := m.claimed[sb.ID] == sb && m.evacuatingLocked(sb.ID), sb.ArchiveCk != "" + epoch, token := sb.ArchiveEpoch, sb.Token + m.mu.Unlock() + switch { + case !live: + return nil + case !archived: + return errArchiveBusy + case epoch == 0: + return errNoArchiveRow + } + ctx = context.WithoutCancel(ctx) + released, err := m.archives.Release(ctx, sb.ID, epoch, m.nodeID) + if errors.Is(err, archives.ErrLost) { + return m.resumeHandOff(ctx, sb, token) + } + if err != nil { + return fmt.Errorf("hand off %s: %w", sb.ID, err) + } + if !m.peerAdopt(ctx, sb.ID, token) { + return m.takeBack(ctx, sb, released) + } + m.dropAdopted(ctx, sb) + return nil +} + +// A row with no owner may hold a later owner's writes, so it outranks this claim and only a peer may take it. +func (m *Manager) resumeHandOff(ctx context.Context, sb *types.Sandbox, token string) error { + rec, ok, err := m.archives.Get(ctx, sb.ID) + switch { + case err != nil: + return fmt.Errorf("hand off %s: %w", sb.ID, err) + case !ok || rec.Owner != "": + if err = m.refusedReplace(ctx, sb); err == nil { + err = archives.ErrLost + } + return fmt.Errorf("hand off %s: %w", sb.ID, err) + case !m.peerAdopt(ctx, sb.ID, token): + return errNoAdopter + } + m.dropAdopted(ctx, sb) + return nil +} + +func (m *Manager) takeBack(ctx context.Context, sb *types.Sandbox, released int64) error { + epoch, err := m.archives.Take(ctx, sb.ID, released, m.nodeID) + if errors.Is(err, archives.ErrLost) { + return m.refusedReplace(ctx, sb) + } + if err != nil { + return fmt.Errorf("take back %s: %w", sb.ID, err) + } + _, _ = m.commitIfLive(ctx, sb, func() { sb.ArchiveEpoch = epoch }) + return errNoAdopter +} + +func (m *Manager) evacuatingLocked(id string) bool { + return m.evac != nil && m.evac.pending[id] != nil +} + +// A parked claim answers 503 until its turn, so it moves first. +func moveCost(sb *types.Sandbox) int { + switch { + case sb.ArchiveCk != "": + return 0 + case sb.HibernateSnap != "": + return 1 + } + return 2 +} diff --git a/sandboxd/pool/drain_test.go b/sandboxd/pool/drain_test.go index 908e2cceb..26bd96aba 100644 --- a/sandboxd/pool/drain_test.go +++ b/sandboxd/pool/drain_test.go @@ -1,11 +1,17 @@ package pool import ( + "context" "errors" + "slices" "testing" "testing/synctest" + "time" + "github.com/cocoonstack/sandbox/sandboxd/archives" "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/tenants" + "github.com/cocoonstack/sandbox/sandboxd/types" ) func TestDrainRefusesClaimsTrimsWarmAndUncordonRefills(t *testing.T) { @@ -51,3 +57,685 @@ func TestDrainRefusesClaimsTrimsWarmAndUncordonRefills(t *testing.T) { } }) } + +func TestEvacuateHandsClaimsToAPeerWhileItsLeaseIsLive(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + running, asleep, archived := mustClaim(t, a, testKey), mustClaim(t, a, testKey), mustClaim(t, a, testKey) + hibernateClaim(t, a, asleep) + mustArchive(t, a, archived) + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 0) + for _, sb := range []*types.Sandbox{running, asleep, archived} { + if _, held := a.byID(sb.ID); held { + t.Errorf("a kept %s", sb.ID) + } + rec, ok, _ := c.reg.Get(t.Context(), sb.ID) + if !ok || rec.Owner != "b" || rec.Consumed || !ckExists(t, b, rec.Checkpoint) { + t.Fatalf("row of %s %+v, want a live row b owns on a stored checkpoint", sb.ID, rec) + } + if _, err := a.putArchive(t.Context(), sb, rec.Checkpoint); !errors.Is(err, archives.ErrLost) { + t.Errorf("a's late put over the row of %s: %v, want ErrLost", sb.ID, err) + } + if row, held := b.Sandbox(sb.ID); !held || row.Token != sb.Token { + t.Errorf("root read of %s on b: held %v, want it with its token", sb.ID, held) + } + if _, _, _, err := b.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Errorf("adopter's wake of %s: %v", sb.ID, err) + } + } + }) +} + +func TestEvacuateHandOffRefusedByAConcurrentOwnerWriteKeepsTheClaim(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + mustArchive(t, a, sb) + a.archives = &hookRegistry{Registry: c.reg, beforeRelease: func() { + rec, _, _ := c.reg.Get(t.Context(), sb.ID) + if _, err := c.reg.Replace(t.Context(), rec, rec.Epoch); err != nil { + t.Errorf("owner write: %v", err) + } + }} + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 1) + if _, held := a.byID(sb.ID); !held { + t.Fatal("a refused hand-off dropped the claim") + } + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Owner != "a" { + t.Errorf("row owner %q after a refused hand-off, want a", rec.Owner) + } + assertNotAdopted(t, b, sb, "adopt a row the hand-off did not release") + + a.Evacuate(t.Context()) + synctest.Wait() + assertEvacuation(t, a, 0, 0) + mustAdopt(t, b, sb, "adopt after the retried hand-off") + }) +} + +func TestEvacuateHandOffOfARowAPeerTookDropsTheClaim(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + mustArchive(t, a, sb) + a.archives = &hookRegistry{Registry: c.reg, beforeRelease: func() { + c.expire("a") + mustAdopt(t, b, sb, "adopt once a's lease lapsed") + c.renew(t, "a") + }} + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 0) + if _, held := a.byID(sb.ID); held { + t.Error("a kept a claim b adopted") + } + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Owner != "b" { + t.Errorf("row owner %q, want b", rec.Owner) + } + }) +} + +func TestEvacuateKeepsClaimsThatCannotMove(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + c.node(t, "b", t.TempDir(), keepAnHour) + failed, withVolume, moved := mustClaim(t, a, testKey), mustClaim(t, a, testKey), mustClaim(t, a, testKey) + a.mu.Lock() + failed.Failed = "vmm down" + withVolume.Volumes = []types.Volume{{Name: "data"}} + a.store.set(failed, withVolume) + a.mu.Unlock() + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 2) + for _, sb := range []*types.Sandbox{failed, withVolume} { + if held, ok := a.byID(sb.ID); !ok || held.HibernateSnap != "" || held.ArchiveCk != "" { + t.Errorf("skipped %s: held %v, want it kept as it was", sb.ID, ok) + } + } + if _, held := a.byID(moved.ID); held { + t.Error("a kept the claim it could move") + } + if _, _, _, err := a.WakeAgentSocket(t.Context(), withVolume.ID, withVolume.Token); err != nil { + t.Errorf("a skipped claim's data plane: %v", err) + } + }) +} + +func TestEvacuateWithoutArchiveFailoverSkipsEveryClaim(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(keepAnHour)) + sb := mustClaim(t, m, testKey) + + m.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, m, 0, 1) + if held, ok := m.byID(sb.ID); !ok || held.HibernateSnap != "" || len(eng.hibernates) != 0 { + t.Errorf("held %v hibernates %v, want the claim left running", ok, eng.hibernates) + } + }) +} + +func TestEvacuateWaitsOutAnArchiveInFlight(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + mustArchive(t, a, sb) + a.mu.Lock() + a.archiving[sb.ID] = struct{}{} + a.mu.Unlock() + + a.Evacuate(t.Context()) + synctest.Wait() + assertEvacuation(t, a, 1, 0) + + a.untrack(a.archiving, sb.ID) + time.Sleep(reapInterval) + synctest.Wait() + assertEvacuation(t, a, 0, 0) + mustAdopt(t, b, sb, "adopt once the archive in flight ended") + }) +} + +func TestUncordonStopsAnEvacuationBeforeItsHandOff(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + stall := stallExports(a) + sb := mustClaim(t, a, testKey) + + a.Evacuate(t.Context()) + synctest.Wait() + assertEvacuation(t, a, 1, 0) + + a.Uncordon(t.Context()) + close(stall) + synctest.Wait() + + assertEvacuation(t, a, 0, 0) + if _, g := a.Info(); g.Draining { + t.Error("still draining after uncordon") + } + held, ok := a.byID(sb.ID) + if !ok || held.ArchiveCk == "" { + t.Fatalf("held %v, want the claim kept archived on a", ok) + } + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Owner != "a" { + t.Errorf("row owner %q after uncordon, want a", rec.Owner) + } + assertNotAdopted(t, b, sb, "adopt a claim whose evacuation stopped") + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Errorf("wake after uncordon: %v", err) + } + }) +} + +func TestEvacuateHoldsTheWakeOfAQueuedClaimUntilUncordon(t *testing.T) { + tests := []struct { + name string + park func(*testing.T, *Manager, *types.Sandbox) + }{ + {"hibernated", hibernateClaim}, + {"archived", mustArchive}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + tt.park(t, a, sb) + snap, ck := sb.HibernateSnap, sb.ArchiveCk + release := holdMaint(a) + + a.Evacuate(t.Context()) + synctest.Wait() + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); !errors.Is(err, ErrEvacuating) { + t.Fatalf("wake of a queued claim: %v, want ErrEvacuating", err) + } + assertEvacuation(t, a, 1, 0) + + a.Uncordon(t.Context()) + release() + synctest.Wait() + + if held, ok := a.byID(sb.ID); !ok || held.HibernateSnap != snap || held.ArchiveCk != ck { + t.Fatalf("held %v, want the claim left as it was parked", ok) + } + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Errorf("wake after uncordon: %v", err) + } + }) + }) + } +} + +func TestAQueuedFailedClaimStillRestarts(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + tuneView(a, func(v *configView) { v.vmmRestart = types.VMMRestartNone }) + sb := mustClaim(t, a, testKey) + killVMM(fakeOf(a), sb.VMName) + a.recoverVMM(t.Context(), sb) + release := holdMaint(a) + + a.Evacuate(t.Context()) + synctest.Wait() + if err := a.Wake(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Errorf("wake of a queued failed claim: %v", err) + } + + a.Uncordon(t.Context()) + release() + synctest.Wait() + }) +} + +func TestEvacuateRequeuesAClaimAWakeUnarchived(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + release := holdMaint(a) + + a.Evacuate(t.Context()) + synctest.Wait() + sb.Transition.Lock() + err := a.handOff(t.Context(), sb) + sb.Transition.Unlock() + if !errors.Is(err, errArchiveBusy) { + t.Errorf("hand-off of a running claim: %v, want errArchiveBusy", err) + } + + a.Uncordon(t.Context()) + release() + synctest.Wait() + }) +} + +func TestEvacuateKeepsAClaimOfARemovedTenant(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := tenantManager(t, t.TempDir(), tenants.TenantSpec{Name: "acme", Token: "acme-tok"}) + m.archives = newFailoverCluster(t).reg + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{TTL: time.Hour, Tenant: "acme", OnExpire: types.ExpireArchive}) + if err != nil { + t.Fatalf("claim: %v", err) + } + if err = m.DeleteTenant(t.Context(), "acme"); err != nil { + t.Fatalf("DeleteTenant: %v", err) + } + + m.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, m, 0, 1) + if held, ok := m.byID(sb.ID); !ok || held.HibernateSnap != "" || held.ArchiveCk != "" { + t.Errorf("held %v, want the claim left running", ok) + } + }) +} + +func TestAnEnvWriteThatLostTheClaimAnswersUnknown(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + mustArchive(t, a, sb) + sb.Transition.Lock() + a.dropAdopted(t.Context(), sb) + sb.Transition.Unlock() + + if err := a.rewriteArchiveRow(t.Context(), sb, false); !errors.Is(err, ErrUnknownSandbox) { + t.Errorf("row rewrite of a dropped claim: %v, want ErrUnknownSandbox", err) + } + }) +} + +func TestAnEnvWriteDuringTheArchiveReachesItsRow(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + hibernateClaim(t, a, sb) + written := make(chan error, 1) + a.archives = &hookRegistry{Registry: c.reg, beforePut: func() { + go func() { written <- a.SetEnv(t.Context(), sb.ID, types.Env{"H": {Value: "new", Guest: hostOnly}}, "") }() + waitFor(t, func() bool { + env, err := a.Env(sb.ID, "") + _, stored := env["H"] + return err == nil && stored + }) + }} + + if err := a.archive(t.Context(), sb); err != nil { + t.Fatalf("archive: %v", err) + } + if err := <-written; err != nil { + t.Fatalf("env write: %v", err) + } + rec, _, _ := c.reg.Get(t.Context(), sb.ID) + if _, err := c.reg.Release(t.Context(), sb.ID, rec.Epoch, "a"); err != nil { + t.Fatalf("release: %v", err) + } + mustAdopt(t, b, sb, "adopt the released row") + if held, ok := b.byID(sb.ID); !ok || held.Env["H"].Value != "new" { + t.Errorf("adopter env %v, want the write the archive raced", held.Env) + } +} + +func TestASecondEvacuateWhileOneRunsChangesNothing(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + release := holdMaint(a) + + a.Evacuate(t.Context()) + synctest.Wait() + first := a.evac + a.Evacuate(t.Context()) + if a.evac != first { + t.Error("a second evacuate replaced the run") + } + + a.Uncordon(t.Context()) + release() + synctest.Wait() + if held, ok := a.byID(sb.ID); !ok || held.HibernateSnap != "" || held.ArchiveCk != "" { + t.Errorf("held %v, want the claim left running after uncordon", ok) + } + }) +} + +func TestAStoppedEvacuationDoesNotPauseAClaimItHadQueued(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + release := holdMaint(a) + + a.Evacuate(t.Context()) + synctest.Wait() + a.Uncordon(t.Context()) + if err := a.evacuateClaim(t.Context(), sb); err != nil { + t.Errorf("evacuateClaim after uncordon: %v, want nil", err) + } + if held, ok := a.byID(sb.ID); !ok || held.HibernateSnap != "" || held.ArchiveCk != "" { + t.Errorf("held %v, want the claim left running", ok) + } + + release() + synctest.Wait() + }) +} + +func TestEvacuateKeepsAClaimArchivedWithoutARow(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + mustArchive(t, a, sb) + a.mu.Lock() + epoch := sb.ArchiveEpoch + sb.ArchiveEpoch = 0 + a.mu.Unlock() + if err := c.reg.Delete(t.Context(), sb.ID, epoch); err != nil { + t.Fatalf("delete row: %v", err) + } + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 1) + if held, ok := a.byID(sb.ID); !ok || held.ArchiveCk == "" { + t.Errorf("held %v, want the rowless archive kept", ok) + } + }) +} + +func TestEvacuateMovesTheCheapestClaimsFirst(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + running, hibernated, archived := mustClaim(t, a, testKey), mustClaim(t, a, testKey), mustClaim(t, a, testKey) + hibernateClaim(t, a, hibernated) + mustArchive(t, a, archived) + a.maintSem = make(chan struct{}, 1) + var moved []string + a.WithPeerAdopt(func(ctx context.Context, id, token string) bool { + moved = append(moved, id) + adopted, err := b.AdoptArchived(ctx, id, token) + return err == nil && adopted + }) + + a.Evacuate(t.Context()) + synctest.Wait() + + if want := []string{archived.ID, hibernated.ID, running.ID}; !slices.Equal(moved, want) { + t.Errorf("moved %v, want %v", moved, want) + } + }) +} + +func TestAnEvacuationLeavesOtherMaintenanceItsTurn(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + c.node(t, "b", t.TempDir(), keepAnHour) + stall := stallExports(a) + for range 3 { + mustClaim(t, a, testKey) + } + a.maintSem = make(chan struct{}, 1) + + a.Evacuate(t.Context()) + synctest.Wait() + left := -1 + a.runBounded(t.Context(), 1, func(context.Context, int) { + _, g := a.Info() + left = g.Evacuating + }) + synctest.Wait() + close(stall) + synctest.Wait() + + if left != 2 { + t.Errorf("maintenance ran with %d claims left, want its turn after the first of 3", left) + } + }) +} + +func TestEvacuateWithNoPeerToAdoptKeepsTheClaim(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 1) + rec, _, _ := c.reg.Get(t.Context(), sb.ID) + held, ok := a.byID(sb.ID) + if !ok || rec.Owner != "a" || held.ArchiveEpoch != rec.Epoch { + t.Fatalf("held %v row %s@%d, want the claim kept on the row it took back", ok, rec.Owner, rec.Epoch) + } + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Errorf("wake of the kept claim: %v", err) + } + }) +} + +func TestEvacuateConvergesWhenAPeersAnswerIsLost(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + a.WithPeerAdopt(func(ctx context.Context, id, token string) bool { + if _, err := b.AdoptArchived(ctx, id, token); err != nil { + t.Errorf("adopt: %v", err) + } + return false + }) + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 0) + if _, held := a.byID(sb.ID); held { + t.Error("a kept a claim b adopted") + } + if _, held := b.byID(sb.ID); !held { + t.Error("b lost the claim") + } + }) +} + +func TestEvacuateConvergesWhenTheReleaseReplyIsLost(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + a.archives = &hookRegistry{Registry: c.reg, loseReleaseReply: true} + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 1) + if err := a.checkArchiveRow(t.Context(), sb.ID); err != nil { + t.Errorf("a released row read as lost: %v", err) + } + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); !errors.Is(err, ErrUnknownSandbox) { + t.Errorf("old owner's wake: %v, want ErrUnknownSandbox", err) + } + mustAdopt(t, b, sb, "adopt the released row") + }) +} + +func TestARepeatEvacuateResumesAHandOffAnOutageCutShort(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + a.archives = &hookRegistry{Registry: c.reg, loseReleaseReply: true} + + a.Evacuate(t.Context()) + synctest.Wait() + assertEvacuation(t, a, 0, 1) + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 0) + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Owner != "b" { + t.Errorf("row owner %q, want the peer the hand-off resumed at", rec.Owner) + } + if _, held := b.byID(sb.ID); !held { + t.Error("b does not list the claim") + } + }) +} + +func TestARepeatEvacuateWithNoAdopterLeavesAReleasedRowReleased(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + sb := mustClaim(t, a, testKey) + a.archives = &hookRegistry{Registry: c.reg, loseReleaseReply: true} + a.Evacuate(t.Context()) + synctest.Wait() + a.WithPeerAdopt(func(context.Context, string, string) bool { return false }) + + a.Evacuate(t.Context()) + synctest.Wait() + + assertEvacuation(t, a, 0, 1) + if rec, _, _ := c.reg.Get(t.Context(), sb.ID); rec.Owner != "" { + t.Errorf("row owner %q, want it left released", rec.Owner) + } + if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); !errors.Is(err, ErrUnknownSandbox) { + t.Errorf("wake of the stale claim: %v, want ErrUnknownSandbox", err) + } + mustAdopt(t, b, sb, "adopt from the row") + }) +} + +func TestACrashBeforeThePeersAnswerLeavesARowAnyNodeSettles(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) + adopted, purged := mustClaim(t, a, testKey), mustClaim(t, a, testKey) + for _, sb := range []*types.Sandbox{adopted, purged} { + mustArchive(t, a, sb) + rec, _, _ := c.reg.Get(t.Context(), sb.ID) + if _, err := c.reg.Release(t.Context(), sb.ID, rec.Epoch, "a"); err != nil { + t.Fatalf("release: %v", err) + } + } + + restarted := c.restart(t, a) + + if _, held := restarted.byID(adopted.ID); held { + t.Error("the restarted node kept a released claim") + } + mustAdopt(t, b, adopted, "adopt after the crash") + rec, _, _ := c.reg.Get(t.Context(), purged.ID) + time.Sleep(2 * time.Hour) + b.sweepOrphanArchives(t.Context()) + synctest.Wait() + if _, ok, _ := c.reg.Get(t.Context(), purged.ID); ok || ckExists(t, b, rec.Checkpoint) { + t.Errorf("row %v checkpoint %v after the retention, want both gone", ok, ckExists(t, b, rec.Checkpoint)) + } + }) +} + +func TestEvacuationDeadlines(t *testing.T) { + tests := []struct { + name string + key types.PoolKey + onExpire types.ExpireAction + ttl time.Duration + retention int + want time.Duration + }{ + {"a claim that never chose archiving keeps its lease", plainKey, "", time.Minute, keepAnHour, time.Minute}, + {"a claim that never chose archiving is not kept forever", plainKey, "", time.Minute, 0, time.Minute}, + {"a long lease outlives a short retention", testKey, types.ExpireArchive, 24 * time.Hour, keepAnHour, 24 * time.Hour}, + {"a long retention outlives a short lease", testKey, types.ExpireArchive, time.Minute, keepAnHour, time.Hour}, + {"no retention keeps a chosen archive forever", testKey, types.ExpireArchive, time.Minute, 0, 0}, + {"a pool archive tier counts as the choice", testKey, "", time.Minute, keepAnHour, time.Hour}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := newFailoverCluster(t) + a := c.node(t, "a", t.TempDir(), tt.retention) + sb, err := claimAny(t.Context(), a, tt.key, ClaimOptions{TTL: tt.ttl, OnExpire: tt.onExpire}) + if err != nil { + t.Fatalf("claim: %v", err) + } + now := time.Now() + + a.Evacuate(t.Context()) + synctest.Wait() + + rec, ok, _ := c.reg.Get(t.Context(), sb.ID) + if !ok { + t.Fatal("no row after the evacuation") + } + var want time.Time + if tt.want != 0 { + want = now.Add(tt.want) + } + if !rec.Deadline.Equal(want) { + t.Errorf("row deadline %v, want %v", rec.Deadline, want) + } + }) + }) + } +} + +func assertEvacuation(t *testing.T, m *Manager, left, kept int) { + t.Helper() + if _, g := m.Info(); g.Evacuating != left || g.Draining && g.Claimed-g.Evacuating != kept { + t.Errorf("evacuation left %d of %d claims, want %d left and %d kept", g.Evacuating, g.Claimed, left, kept) + } +} + +func holdMaint(m *Manager) (release func()) { + for range cap(m.maintSem) { + m.maintSem <- struct{}{} + } + return func() { + for range cap(m.maintSem) { + <-m.maintSem + } + } +} + +func stallExports(m *Manager) chan struct{} { + eng := fakeOf(m) + stall := make(chan struct{}) + eng.mu.Lock() + eng.exportStall = stall + eng.mu.Unlock() + return stall +} diff --git a/sandboxd/pool/env.go b/sandboxd/pool/env.go index 6314d4917..3ab5dc8af 100644 --- a/sandboxd/pool/env.go +++ b/sandboxd/pool/env.go @@ -81,7 +81,7 @@ func (m *Manager) writeEnv(ctx context.Context, id, tenant, verb string, next fu } return ErrPaused } - rowed, deliver := m.rowFollows(sb), !paused + rowed, deliver := m.rowFollows(sb) || sb.ArchivePending != "", !paused if env.Equal(prev) { m.mu.Unlock() } else { @@ -116,9 +116,12 @@ func (m *Manager) rewriteArchiveRow(ctx context.Context, sb *types.Sandbox, lock } for range 2 { m.mu.Lock() - ck, replica, rowed := sb.ArchiveCk, sb.ReplicaCk, m.claimed[sb.ID] == sb && m.rowFollows(sb) + ck, replica, live, rowed := sb.ArchiveCk, sb.ReplicaCk, m.claimed[sb.ID] == sb, m.rowFollows(sb) m.mu.Unlock() - if !rowed { + switch { + case !live: + return ErrUnknownSandbox + case !rowed: return nil } epoch, err := m.putRow(ctx, sb, cmp.Or(ck, replica), ck == "", true) diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index e5b71761e..104c76d2c 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -122,7 +122,7 @@ func (m *Manager) putRow(ctx context.Context, sb *types.Sandbox, ck string, repl case replica: deadline = time.Time{} case sb.ArchiveCk == "": - deadline = m.retentionDeadline(sb.Key) + deadline = m.archiveDeadline(sb) } m.mu.Unlock() dto.Token, dto.VMName, dto.VsockSocket, dto.HibernateSnap, dto.PendingSnap, dto.ArchivePending, dto.ReplicaCk, dto.ReplicaPending = "", "", "", "", "", "", "", "" @@ -161,7 +161,7 @@ func (m *Manager) checkArchiveRow(ctx context.Context, id string) error { return nil } rec, ok, err := m.archives.Get(ctx, id) - if err == nil && ok && rec.Owner != m.nodeID { + if err == nil && ok && rec.Owner != m.nodeID && rec.Owner != "" { return fmt.Errorf("%w: held by %s", archives.ErrLost, rec.Owner) } return err diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index f5d1e0c40..1e168a151 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -1862,6 +1862,7 @@ type failoverCluster struct { ckDir string replicate int volumes []config.VolumeSpec + nodes []*Manager } func newFailoverCluster(t *testing.T) *failoverCluster { @@ -1930,6 +1931,16 @@ func (c *failoverCluster) nodeOn(t *testing.T, eng *fakeEngine, name, dataDir st t.Fatalf("manager %s: %v", name, err) } _ = c.reg.Renew(t.Context(), name, time.Hour) + m.WithPeerAdopt(func(ctx context.Context, id, token string) bool { + return slices.ContainsFunc(c.nodes, func(p *Manager) bool { + if p.nodeID == name { + return false + } + adopted, err := p.AdoptArchived(ctx, id, token) + return err == nil && adopted + }) + }) + c.nodes = append(slices.DeleteFunc(c.nodes, func(p *Manager) bool { return p.nodeID == name }), m) t.Cleanup(func() { assertRunningCount(t, m) }) t.Cleanup(func() { settleGoldenBuilds(m) }) return m @@ -1940,10 +1951,12 @@ type hookRegistry struct { beforePut func() beforeTake func() beforeConsume func() + beforeRelease func() beforeDelete func() afterRows func() loseConsumeReply bool loseTakeReply bool + loseReleaseReply bool losePutReply bool failPut bool replaces int @@ -1995,6 +2008,19 @@ func (r *hookRegistry) Take(ctx context.Context, id string, expect int64, owner return epoch, err } +func (r *hookRegistry) Release(ctx context.Context, id string, expect int64, owner string) (int64, error) { + if hook := r.beforeRelease; hook != nil { + r.beforeRelease = nil + hook() + } + released, err := r.Registry.Release(ctx, id, expect, owner) + if r.loseReleaseReply && err == nil { + r.loseReleaseReply = false + return 0, archives.ErrUnavailable + } + return released, err +} + func (r *hookRegistry) Delete(ctx context.Context, id string, epoch int64) error { if hook := r.beforeDelete; hook != nil { r.beforeDelete = nil diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index f02b5af28..cbd225191 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -114,6 +114,7 @@ var ( ErrArchived = errors.New("sandbox is archived; a wake or an exec wakes it first") ErrLeaseLost = errors.New("node lease lost; retry") ErrPaused = errors.New("sandbox is paused; an operator call does not wake it") + ErrEvacuating = errors.New("sandbox is moving to another node; retry") ErrFailed = errors.New("sandbox failed: its VMM is down; release it, or wake it to cold-boot its disk") ErrNoInstanceMetadata = engine.ErrNoInstanceMetadata ErrQuota = errors.New("node claim quota reached") @@ -123,6 +124,10 @@ var ( errNotSleeping = errors.New("sandbox is no longer hibernated or archived") errWokeMeanwhile = errors.New("woke between sweep and hibernate") errNoEgressTap = errors.New("egress-lane claim has no lockable tap") + errNoFailover = errors.New("archive_failover is off, so no peer can adopt the claim") + errNoArchiveRow = errors.New("archived before archive_failover was on, so no registry row names it") + errArchiveBusy = errors.New("an archive of the claim is in flight") + errNoAdopter = errors.New("no peer adopted the claim") ) // Engine is the slice of the cocoon driver the manager consumes. @@ -213,6 +218,8 @@ type Gauges struct { Archived int Failed int Draining bool + // Evacuating counts the claims an evacuation has yet to move. + Evacuating int // AtCapacity marks refill parked because the node refused another VM. AtCapacity bool AtCapacityReason string @@ -230,6 +237,9 @@ type Backends struct { // PeerDeleteFunc broadcasts a checkpoint delete to every peer. type PeerDeleteFunc func(ctx context.Context, id string) +// PeerAdoptFunc asks the peers to adopt a released claim and reports whether one holds it now. +type PeerAdoptFunc func(ctx context.Context, id, token string) bool + type pendingRemoval struct { sandboxID string tap string @@ -392,6 +402,7 @@ type Manager struct { reloadMu sync.Mutex draining bool // guarded by m.mu; deliberately not persisted + evac *evacuation tenants tenants.Source rootSum string @@ -411,6 +422,7 @@ type Manager struct { healPending map[string]struct{} healAbort map[string]struct{} peerDelete PeerDeleteFunc + peerAdopt PeerAdoptFunc tpls store.Store ckptSweeping atomic.Bool genSweeping atomic.Bool @@ -653,6 +665,9 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) { }) } g := Gauges{Claimed: len(m.claimed), Running: m.store.runningCount(), MaxRunning: m.view.Load().maxRunning, Draining: m.draining} + if m.evac != nil { + g.Evacuating = len(m.evac.pending) + } g.AtCapacityReason = m.capacityReasonLocked(now) g.AtCapacity = g.AtCapacityReason != "" for _, sb := range m.claimed { @@ -717,6 +732,11 @@ func (m *Manager) WithPeerDelete(fn PeerDeleteFunc) { m.peerDelete = fn } +// WithPeerAdopt wires the push an evacuation makes after it releases a row. +func (m *Manager) WithPeerAdopt(fn PeerAdoptFunc) { + m.peerAdopt = fn +} + func (m *Manager) sweepStoreGenerations(ctx context.Context) { if !m.genSweeping.CompareAndSwap(false, true) { return diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index 3f026a13b..debce3cea 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -596,7 +596,7 @@ func TestTheSweepsLeaveAReplicatedClaimAloneWhileFenced(t *testing.T) { a.mu.Lock() a.archiving[sb.ID] = struct{}{} a.mu.Unlock() - if err := a.pauseExpired(t.Context(), sb); !errors.Is(err, ErrLeaseLost) || !benignSweepErr(err) { + if err := a.pauseArchive(t.Context(), sb); !errors.Is(err, ErrLeaseLost) || !benignSweepErr(err) { t.Errorf("pause of an expired claim while fenced: %v, want a benign ErrLeaseLost", err) } a.mu.Lock() diff --git a/sandboxd/server/metrics.go b/sandboxd/server/metrics.go index a070ca850..ce1f33c04 100644 --- a/sandboxd/server/metrics.go +++ b/sandboxd/server/metrics.go @@ -34,6 +34,7 @@ func (s *Server) handleMetrics(w http.ResponseWriter, _ *http.Request) { scalar("archived", "claims archived to the checkpoint store", g.Archived) scalar("failed", "claims whose VMM is down and not restarted", g.Failed) scalar("draining", "1 while the node is cordoned for maintenance", draining) + scalar("evacuating", "claims an evacuation has yet to move to a peer", g.Evacuating) if tenants := s.mgr.TenantClaims(); len(tenants) > 0 { metric("tenant_claims", "gauge", "live claims per configured tenant") diff --git a/sandboxd/server/preview.go b/sandboxd/server/preview.go index 0d35cac16..8b57f99f6 100644 --- a/sandboxd/server/preview.go +++ b/sandboxd/server/preview.go @@ -132,9 +132,9 @@ func (p *PreviewServer) proxyLocal(w http.ResponseWriter, r *http.Request, claim }, Transport: p.transport, ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { - if errors.Is(err, pool.ErrLeaseLost) { - w.Header().Set("Retry-After", retryAfterSeconds) - http.Error(w, "lease lost; retry", http.StatusServiceUnavailable) + if errors.Is(err, pool.ErrLeaseLost) || errors.Is(err, pool.ErrEvacuating) { + code, msg, _ := mapPoolErr(w, err) + http.Error(w, msg, code) return } if errors.Is(err, pool.ErrQuota) { diff --git a/sandboxd/server/preview_test.go b/sandboxd/server/preview_test.go index 73ec2753c..aecbf0335 100644 --- a/sandboxd/server/preview_test.go +++ b/sandboxd/server/preview_test.go @@ -97,21 +97,25 @@ func TestPreviewRevokedWhenDialFails(t *testing.T) { } } -func TestPreviewAnswers503WhileTheLeaseIsLost(t *testing.T) { - ps := NewPreviewServer("secret", "node:9000", "node:7777", &fakePreviewMgr{ - dial: func(string, uint16) (net.Conn, error) { return nil, pool.ErrLeaseLost }, - }) - ts := httptest.NewServer(ps.Handler()) - t.Cleanup(ts.Close) - token := mintToken(ps, "sb_1", 3000, time.Hour) +func TestPreviewAnswers503WhileAClaimCannotServe(t *testing.T) { + for _, dialErr := range []error{pool.ErrLeaseLost, pool.ErrEvacuating} { + t.Run(dialErr.Error(), func(t *testing.T) { + ps := NewPreviewServer("secret", "node:9000", "node:7777", &fakePreviewMgr{ + dial: func(string, uint16) (net.Conn, error) { return nil, dialErr }, + }) + ts := httptest.NewServer(ps.Handler()) + t.Cleanup(ts.Close) + token := mintToken(ps, "sb_1", 3000, time.Hour) - resp, err := http.Get(ts.URL + "/p/" + token + "/") - if err != nil { - t.Fatalf("get: %v", err) - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") == "" { - t.Errorf("status %d retry-after %q, want 503 with a retry hint", resp.StatusCode, resp.Header.Get("Retry-After")) + resp, err := http.Get(ts.URL + "/p/" + token + "/") + if err != nil { + t.Fatalf("get: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") == "" { + t.Errorf("status %d retry-after %q, want 503 with a retry hint", resp.StatusCode, resp.Header.Get("Retry-After")) + } + }) } } diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 8769801bf..ebb3e0358 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -60,6 +60,7 @@ var poolErrHTTP = []struct { {pool.ErrArchiveStoreDown, http.StatusServiceUnavailable, "archive registry down; retry"}, {pool.ErrVolumeCatalogUnavailable, http.StatusServiceUnavailable, "volume catalog unavailable; retry"}, {pool.ErrLeaseLost, http.StatusServiceUnavailable, "lease lost; retry"}, + {pool.ErrEvacuating, http.StatusServiceUnavailable, "sandbox moving to another node; retry"}, {pool.ErrNoEgress, http.StatusConflict, ""}, {pool.ErrNoEgressHibernate, http.StatusConflict, ""}, {pool.ErrNoEgressFork, http.StatusConflict, ""}, @@ -138,6 +139,7 @@ type Manager interface { PutTenant(ctx context.Context, spec tenants.TenantSpec) error DeleteTenant(ctx context.Context, name string) error Drain(ctx context.Context) + Evacuate(ctx context.Context) Uncordon(ctx context.Context) Info() ([]pool.PoolInfo, pool.Gauges) Templates() []pool.TemplateInfo @@ -187,6 +189,7 @@ type InfoResponse struct { Archived int `json:"archived"` Failed int `json:"failed,omitzero"` Draining bool `json:"draining,omitzero"` + Evacuating int `json:"evacuating,omitzero"` AdvertiseAddr string `json:"advertise_addr,omitempty"` Peers []string `json:"peers,omitempty"` AtCapacity bool `json:"at_capacity,omitzero"` @@ -820,7 +823,11 @@ func (s *Server) handleDeleteTenant(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleDrain(w http.ResponseWriter, r *http.Request) { - s.mgr.Drain(r.Context()) + if r.URL.Query().Get("evacuate") != "" { + s.mgr.Evacuate(r.Context()) + } else { + s.mgr.Drain(r.Context()) + } s.handleInfo(w, r) } @@ -841,7 +848,7 @@ func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) { pools, g := s.mgr.Info() resp := InfoResponse{ Pools: pools, Templates: s.mgr.Templates(), Claimed: g.Claimed, Running: g.Running, MaxRunning: g.MaxRunning, Hibernated: g.Hibernated, Archived: g.Archived, Failed: g.Failed, - Draining: g.Draining, AtCapacity: g.AtCapacity, AtCapacityReason: g.AtCapacityReason, + Draining: g.Draining, Evacuating: g.Evacuating, AtCapacity: g.AtCapacity, AtCapacityReason: g.AtCapacityReason, AdvertiseAddr: s.advertise, } if s.placer != nil { diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index 781a9c149..95530336b 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -717,6 +717,35 @@ func TestDrainEndpoints(t *testing.T) { } } +func TestDrainEvacuateStartsAnEvacuationAndReportsIt(t *testing.T) { + mgr := &fakeManager{evacuateLeft: 2} + ts := newTestServer(t, "", mgr, nil) + statusOf(t, http.MethodPost, ts.URL+"/v1/drain", "", "") + if mgr.evacuated { + t.Fatal("a plain drain started an evacuation") + } + resp := doReq(t, http.MethodPost, ts.URL+"/v1/drain?evacuate=1", "", "") + info := decodeJSON[InfoResponse](t, resp.Body) + resp.Body.Close() + if resp.StatusCode != http.StatusOK || !mgr.evacuated || !info.Draining || info.Evacuating != 2 { + t.Fatalf("status=%d evacuated=%v info=%+v, want 200, an evacuation and 2 left", resp.StatusCode, mgr.evacuated, info) + } + body := scrapeMetrics(t, ts.URL) + if want := "sandboxd_evacuating 2\n"; !strings.Contains(body, want) { + t.Errorf("metrics miss %q:\n%s", want, body) + } +} + +func TestAnEvacuatingSandboxAnswersRetry(t *testing.T) { + mgr := &fakeManager{socket: func(string, string) (string, error) { return "", pool.ErrEvacuating }} + ts := newTestServer(t, "", mgr, nil) + resp := doReq(t, http.MethodPost, ts.URL+"/v1/sandboxes/sb_1/exec", "tok", `{"argv":["true"]}`) + resp.Body.Close() + if resp.StatusCode != http.StatusServiceUnavailable || resp.Header.Get("Retry-After") == "" { + t.Errorf("exec on an evacuating sandbox: status %d Retry-After %q, want 503 with Retry-After", resp.StatusCode, resp.Header.Get("Retry-After")) + } +} + func TestPutPoolsUpdatesTargets(t *testing.T) { var got []config.PoolSpec mgr := &fakeManager{ @@ -2227,6 +2256,8 @@ type fakeManager struct { sandboxIndex func(tenant, claimRef string) []pool.SandboxSummary sandboxByID func(id string) (pool.SandboxSummary, bool) draining bool + evacuated bool + evacuateLeft int reader *fakeReader } @@ -2550,7 +2581,7 @@ func (f *fakeManager) SetPools(_ context.Context, pools []config.PoolSpec) error } func (f *fakeManager) Info() ([]pool.PoolInfo, pool.Gauges) { - return f.infoPools, pool.Gauges{Draining: f.draining} + return f.infoPools, pool.Gauges{Draining: f.draining, Evacuating: f.evacuateLeft} } func (f *fakeManager) Templates() []pool.TemplateInfo { return f.infoTemplates } @@ -2580,6 +2611,8 @@ func (f *fakeManager) Env(_, tenant string) (types.Env, error) { func (f *fakeManager) Drain(context.Context) { f.draining = true } +func (f *fakeManager) Evacuate(context.Context) { f.draining, f.evacuated = true, true } + func (f *fakeManager) Uncordon(context.Context) { f.draining = false } func (f *fakeManager) FetchCheckpoint(_ context.Context, _ string) (string, []byte, func(), error) { diff --git a/sandboxd/store/peer/adopt.go b/sandboxd/store/peer/adopt.go new file mode 100644 index 000000000..5d1dea0c9 --- /dev/null +++ b/sandboxd/store/peer/adopt.go @@ -0,0 +1,27 @@ +package peer + +import ( + "context" + "net/http" + "net/url" + + "github.com/projecteru2/core/log" +) + +// Adopter asks peers to take over a claim their owner released. +type Adopter struct { + Peers func() []string +} + +// Adopt tries the peers until one holds the claim and reports whether one does. +func (a *Adopter) Adopt(ctx context.Context, id, token string) bool { + logger := log.WithFunc("peer.Adopter.Adopt") + for _, addr := range a.Peers() { + err := peerCall(ctx, http.MethodGet, peerURL(addr, "/v1/sandboxes/"+url.PathEscape(id)+"/owner"), token, http.StatusOK) + if err == nil { + return true + } + logger.Warnf(ctx, "adopt %s on %s: %v", id, addr, err) + } + return false +} diff --git a/sandboxd/store/peer/adopt_test.go b/sandboxd/store/peer/adopt_test.go new file mode 100644 index 000000000..f000dc6ce --- /dev/null +++ b/sandboxd/store/peer/adopt_test.go @@ -0,0 +1,50 @@ +package peer + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestAdoptAsksPeersUntilOneHoldsTheClaim(t *testing.T) { + tests := []struct { + name string + statuses []int + want bool + }{ + {"a peer adopts", []int{http.StatusOK}, true}, + {"a draining peer refuses and another adopts", []int{http.StatusNotFound, http.StatusOK}, true}, + {"every peer refuses", []int{http.StatusNotFound, http.StatusServiceUnavailable}, false}, + {"no peer", nil, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + addrs := make([]string, 0, len(tt.statuses)) + for _, status := range tt.statuses { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/v1/sandboxes/sb_1/owner" || r.Header.Get("Authorization") != "Bearer tok" { + t.Errorf("request %s %s auth %q", r.Method, r.URL.Path, r.Header.Get("Authorization")) + } + w.WriteHeader(status) + })) + t.Cleanup(ts.Close) + addrs = append(addrs, strings.TrimPrefix(ts.URL, "http://")) + } + a := &Adopter{Peers: func() []string { return addrs }} + if got := a.Adopt(t.Context(), "sb_1", "tok"); got != tt.want { + t.Errorf("Adopt = %v, want %v", got, tt.want) + } + }) + } +} + +func TestAdoptSkipsAnUnreachablePeer(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + addr := strings.TrimPrefix(ts.URL, "http://") + ts.Close() + a := &Adopter{Peers: func() []string { return []string{addr} }} + if a.Adopt(t.Context(), "sb_1", "tok") { + t.Error("Adopt = true with the only peer down") + } +} diff --git a/sandboxd/store/peer/broadcast.go b/sandboxd/store/peer/broadcast.go index 082cb1ea8..7b5978aee 100644 --- a/sandboxd/store/peer/broadcast.go +++ b/sandboxd/store/peer/broadcast.go @@ -12,7 +12,7 @@ import ( "github.com/projecteru2/core/log" ) -const deleteTimeout = 5 * time.Second +const peerCallTimeout = 5 * time.Second // Broadcaster fans a checkpoint delete out to every peer so a healed copy does not outlive it. type Broadcaster struct { @@ -30,7 +30,8 @@ func (b *Broadcaster) Delete(ctx context.Context, id string) { var wg sync.WaitGroup for _, addr := range addrs { wg.Go(func() { - if err := deleteOn(ctx, addr, id, b.Token); err != nil { + // 404 means the peer never healed a copy — not a failure worth a retry. + if err := peerCall(ctx, http.MethodDelete, checkpointURL(addr, id, "?no_forward=1"), b.Token, http.StatusNoContent, http.StatusNotFound); err != nil { logger.Warnf(ctx, "broadcast delete %s to %s: %v", id, addr, err) } }) @@ -38,11 +39,12 @@ func (b *Broadcaster) Delete(ctx context.Context, id string) { wg.Wait() } -func deleteOn(ctx context.Context, addr, id, token string) error { - ctx, cancel := context.WithTimeout(ctx, deleteTimeout) +// peerCall bounds one bodyless request to a peer; any status outside ok is an error. +func peerCall(ctx context.Context, method, target, token string, ok ...int) error { + ctx, cancel := context.WithTimeout(ctx, peerCallTimeout) defer cancel() - req, err := http.NewRequestWithContext(ctx, http.MethodDelete, checkpointURL(addr, id, "?no_forward=1"), nil) + req, err := http.NewRequestWithContext(ctx, method, target, nil) if err != nil { return err } @@ -57,8 +59,7 @@ func deleteOn(ctx context.Context, addr, id, token string) error { _, _ = io.Copy(io.Discard, resp.Body) _ = resp.Body.Close() }() - // 404 means the peer never healed a copy — not a failure worth a retry. - if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusNotFound { + if !slices.Contains(ok, resp.StatusCode) { return fmt.Errorf("unexpected status %d", resp.StatusCode) } return nil diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index e5511fda4..b2e911d73 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -217,10 +217,14 @@ func writeFile(target string, r io.Reader, mode os.FileMode) error { } func checkpointURL(addr, id, suffix string) string { + return peerURL(addr, "/v1/checkpoints/"+url.PathEscape(id)+suffix) +} + +func peerURL(addr, path string) string { if !strings.Contains(addr, "://") { addr = "http://" + addr } - return addr + "/v1/checkpoints/" + url.PathEscape(id) + suffix + return addr + path } // safeJoin resolves name under root, rejecting absolute paths and names that escape root. diff --git a/sdk/go/client_test.go b/sdk/go/client_test.go index ffb71173e..c836fd02b 100644 --- a/sdk/go/client_test.go +++ b/sdk/go/client_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "reflect" "strings" + "sync/atomic" "testing" "time" ) @@ -106,7 +107,9 @@ func TestCloseReleases(t *testing.T) { t.Run(tt.name, func(t *testing.T) { var gotPath, gotAuth string ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - gotPath, gotAuth = r.URL.Path, r.Header.Get("Authorization") + if strings.HasSuffix(r.URL.Path, "/release") { + gotPath, gotAuth = r.URL.Path, r.Header.Get("Authorization") + } w.WriteHeader(tt.status) })) t.Cleanup(ts.Close) @@ -124,6 +127,38 @@ func TestCloseReleases(t *testing.T) { } } +func TestCloseReleasesASandboxThatMoved(t *testing.T) { + var released atomic.Bool + var adopter *httptest.Server + adopter = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/sandboxes/sb_9/owner": + _ = json.MarshalWrite(w, map[string]string{"owner_addr": strings.TrimPrefix(adopter.URL, "http://")}) + case "/v1/sandboxes/sb_9/release": + released.Store(true) + w.WriteHeader(http.StatusNoContent) + } + })) + t.Cleanup(adopter.Close) + old := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/peers" { + _ = json.MarshalWrite(w, map[string]any{"peers": []string{strings.TrimPrefix(adopter.URL, "http://")}}) + return + } + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(old.Close) + + c := testClient(t, old) + sb := &Sandbox{ID: "sb_9", c: c, token: "tok9", owner: c.addr} + if err := sb.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if !released.Load() { + t.Error("Close reported success without releasing at the adopter") + } +} + func TestNewFollowsRedirect(t *testing.T) { var nodeB *httptest.Server nodeB = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { diff --git a/sdk/go/info.go b/sdk/go/info.go index 184928598..d0f17ebfc 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -20,6 +20,7 @@ type NodeInfo struct { Archived int `json:"archived"` Failed int `json:"failed,omitzero"` Draining bool `json:"draining,omitzero"` + Evacuating int `json:"evacuating,omitzero"` AdvertiseAddr string `json:"advertise_addr,omitempty"` Peers []string `json:"peers,omitempty"` AtCapacity bool `json:"at_capacity,omitzero"` diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index cf28fd302..4a180d8c0 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -224,15 +224,31 @@ func (s *Sandbox) Close() error { s.pool.drain() ctx, cancel := context.WithTimeout(context.Background(), releaseTimeout) defer cancel() - resp, err := s.c.roundTrip(ctx, http.MethodPost, s.owner, "/v1/sandboxes/"+s.ID+"/release", nil, s.token) - if err != nil { + found, err := s.releaseAt(ctx, s.owner) + if err != nil || found { return err } - defer func() { _ = resp.Body.Close() }() - if resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusNotFound { + moved, err := s.c.Lookup(ctx, s.ID, s.token) + if err != nil { return nil } - return apiError("release", resp) + _, err = s.releaseAt(ctx, moved.owner) + return err +} + +func (s *Sandbox) releaseAt(ctx context.Context, addr string) (found bool, err error) { + resp, err := s.c.roundTrip(ctx, http.MethodPost, addr, "/v1/sandboxes/"+s.ID+"/release", nil, s.token) + if err != nil { + return false, err + } + defer func() { _ = resp.Body.Close() }() + switch resp.StatusCode { + case http.StatusNoContent: + return true, nil + case http.StatusNotFound: + return false, nil + } + return false, apiError("release", resp) } // call sends req on a relay connection and hands back the lease that ends the RPC; ctx cancellation closes the connection under it. diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 347bd3745..82933f8a5 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -410,12 +410,20 @@ def close(self) -> None: """Releases the sandbox; its VM is destroyed.""" self._pool.drain() try: - self._client._request( - self.owner, "POST", f"/v1/sandboxes/{self.id}/release", None, "release", bearer=self.token - ) + self._release_at(self.owner) + return except APIError as exc: if exc.status != 404: raise + try: + moved = self._client.lookup(self.id, self.token) + self._release_at(moved.owner) + except APIError as exc: + if exc.status != 404: + raise + + def _release_at(self, addr: str) -> None: + self._client._request(addr, "POST", f"/v1/sandboxes/{self.id}/release", None, "release", bearer=self.token) def _volumes_request( self, method: str, body: dict[str, Any] | None, verb: str, timeout: float | None diff --git a/sdk/python/tests/test_fault_matrix.py b/sdk/python/tests/test_fault_matrix.py index b16b17e17..cd0c56f5c 100644 --- a/sdk/python/tests/test_fault_matrix.py +++ b/sdk/python/tests/test_fault_matrix.py @@ -209,3 +209,38 @@ def test_lookup_all_miss(spawn_node, dead_addr): with pytest.raises(APIError) as exc: Client(entry).lookup("sb_1", "tok") assert exc.value.status == 404 and "no owner found" in exc.value.message + + +def test_close_releases_a_sandbox_that_moved(spawn_node): + released = [] + + def release(body, path): + released.append(path) + return (204, {}) + + adopter = spawn_node( + { + ("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (200, {}), + ("POST", "/v1/sandboxes/sb_1/release"): release, + } + ) + old = spawn_node( + { + ("POST", "/v1/sandboxes/sb_1/release"): lambda body, path: (404, {"error": "unknown sandbox"}), + ("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (404, {"error": "not here"}), + ("GET", "/v1/peers"): lambda body, path: (200, {"peers": [adopter]}), + } + ) + Client(old).attach(old, "sb_1", "tok").close() + assert released + + +def test_close_of_a_sandbox_no_node_holds_is_not_an_error(spawn_node): + gone = spawn_node( + { + ("POST", "/v1/sandboxes/sb_1/release"): lambda body, path: (404, {"error": "unknown sandbox"}), + ("GET", "/v1/sandboxes/sb_1/owner"): lambda body, path: (404, {"error": "not here"}), + ("GET", "/v1/peers"): lambda body, path: (200, {"peers": []}), + } + ) + Client(gone).attach(gone, "sb_1", "tok").close()