diff --git a/docs/cluster.md b/docs/cluster.md index e3dba7f4..490c337e 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -332,8 +332,8 @@ exec and pty streams, port forwards and egress connections end; the client meets a 404 on the source and runs `Lookup`. **The record.** Each move has a record of its own in the meta store (table -`sandboxd_moves`), separate from the archive rows: `id`, `from`, `to`, `state` -and `epoch`, each change a compare-and-set at a new epoch. +`sandboxd_moves`), separate from the archive rows: `id`, `from_node`, +`to_node`, `state` and `epoch`, each change a compare-and-set at a new epoch. | From | To | Who | When | |---|---|---|---| @@ -379,21 +379,21 @@ never fit), the source pauses the guest for the final round and the target receives the series and starts the guest. A move whose deltas never fit answers 503 and the guest runs on; from the second delta on, one that shrinks by under 10% and is still too large answers at once. `downtime_ms` in the answer -is the pause the source measured, from the final capture to the target's -commit. Each round has one minute plus its size at 10 MB/s to arrive, so a +is the pause the source measured, from the final capture until the target +answers the commit, after it started the guest. Each round has one minute plus its size at 10 MB/s to arrive, so a target that stops reading fails the move. The final capture, its transfer and the target's commit have 2 minutes together; past that the source resolves the move through the record, and the guest resumes on the source unless the target committed. -**Counting.** The target admits the move like a new claim: node `max_claims`, -`max_running`, the tenant's `max_claims`, `memory_limit` and -`move_concurrency`, and holds a reservation from admission to its commit. The +**Counting.** The target admits the move like a new claim: drain, +`disk_reserve`, `memory_limit`, node `max_claims` and `max_running`, the +tenant's `max_claims` and `move_concurrency`, and holds a reservation from admission to its commit. The reservation counts in every cap but is no claim: listings, reaps and gossip never see it. At the commit the reservation becomes the claim under one lock, so the target never counts it twice. The source counts the claim until it drops it after the commit, so the cluster counts a moving claim twice, which can make a node refuse earlier but never over-admit. `memory_limit` is checked again -before the receive. A new admission from the source that holds the reservation +before the receive, and a target that started draining refuses the commit. A new admission from the source that holds the reservation replaces it and resets its staged files, so a move retried after a source crash does not wait out the idle drop; one from another source answers 409. A reservation with no round arriving and none finished or @@ -401,11 +401,11 @@ commit asked for 2 minutes is dropped with its staged files. **Same paths.** Both nodes need the same path strings in cocoon's `root_dir` and `run_dir`. The requirement is on the string, not the real directory: a symlink -of one name can lead to different disks on each node, and `data_dir` may differ. -The string has to be set before the VM starts, because the VM's record keeps it. -A move between nodes with different strings fails at the receive, after the -guest paused: cocoon refuses the series with `untrusted storage path in snapshot -metadata`, the call answers 502 and the guest resumes on the source. +of one name can lead to different disks on each node, and sandboxd's `data_dir` +may differ. Set the string before the VM starts; a VM started under another +string cannot move. A move between nodes with different strings fails at the +receive, after the guest paused: `cocoon vm migrate receive` refuses the series, +the call answers 502 and the guest resumes on the source. **Known limits.** diff --git a/docs/deploy.md b/docs/deploy.md index 0ab90fcb..4c246e7d 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -160,7 +160,7 @@ the intersection of the two. | `storage` (pool entry) | unset (cocoon's default, 10G) | size of the copy-on-write disk the pool's golden is cold-booted with, in cocoon's own `--storage` spelling (`"40G"`, `"40GiB"`, `"40Gi"`: Docker/Kubernetes units, all binary) and passed to it verbatim; every clone, fork, checkpoint branch and promoted template of the pool inherits it. At least cocoon's default of 10G. Equal sizes spelled differently build the same golden; a changed size rebuilds the golden on the next build, and claims already out keep their disk. Config-owned like `egress`: `PUT /v1/pools` rejects it | | `max_claims` | 0 (unlimited) | node-wide cap on live claims; claim/fork/branch requests beyond it answer 429 with the pool state unharmed (on a cluster a non-volume claim first tries a redirect to a warm peer, else to a peer that serves the pool; a volume claim answers the 429 with no redirect) | | `max_running` | 0 (off) | node-wide cap on claims that hold a VM: not hibernated, not archived, not failed, with a wake in flight counted; claim, fork, branch and wake beyond it answer 429 and leave the claim as it was; a claim takes the same redirect as `max_claims`. Hibernated and archived claims hold no slot, so `max_claims` can follow disk while this follows RAM and CPU. Warm and reader VMs are not counted: keep `max_running` × the largest size plus every pool's warm target × its size within the node's RAM. A wake cannot move to another node, so its 429 stays until a running claim sleeps or goes. Lowering the cap stops nothing that runs. A [reload](#reloading-the-config) applies a change | -| `move_concurrency` | 2 | node-wide cap on [live moves](cluster.md#moving-a-running-claim) in flight, counting moves this node sends and reservations it holds for moves to it; a move past it answers 429. Moves need `meta_store` with a `cell`, for the move record, and `mesh.cluster_key`, but not `archive_failover` or an S3 `checkpoint_store`: a node that sets this without them refuses to load, and one that leaves it unset without them cannot move (409). A move also needs a cocoon with live migration (v0.7.7 or a master build), the Cloud Hypervisor fork at `35b098c65260` or newer (a VM keeps the VMM it started with), a run dir that supports reflink such as XFS, nodes of the same CPU model, and the same path strings in cocoon's `root_dir` and `run_dir` on both nodes ([same paths](cluster.md#moving-a-running-claim)). Lowering the cap stops no move in flight. A [reload](#reloading-the-config) applies a change | +| `move_concurrency` | 2 | node-wide cap on [live moves](cluster.md#moving-a-running-claim) in flight, counting moves this node sends and reservations it holds for moves to it; a move past it answers 429. Moves need `meta_store` with a `cell`, for the move record, and `mesh.cluster_key`, but not `archive_failover` or an S3 `checkpoint_store`: a node that sets this without them refuses to load, and one that leaves it unset without them cannot move (409). A move also needs a cocoon with live migration (v0.7.7 or a master build), the Cloud Hypervisor fork at `35b098c65260` or newer (a VM started before the upgrade cannot move), a run dir that supports reflink such as XFS, nodes of the same CPU model, and the same path strings in cocoon's `root_dir` and `run_dir` on both nodes ([same paths](cluster.md#moving-a-running-claim)). Lowering the cap stops no move in flight. A [reload](#reloading-the-config) applies a change | | `audit_log` | false | append every relayed frame that opens an RPC — its op + addressing fields, never payloads — to `/audit.jsonl`; continuation `data`, `data_end`, `stdin`, and `stdin_close` frames pass through unrecorded. The file is size-rotated with one `.1` backup. Records are `{t, id, op}` plus whichever addressing fields the op carries (`argv`, `path`, `dest`, `from`, `to`, `url`, `session`, `port`), plus `method` (`GET`, `CONNECT`, `SOCKS5`, …), `decision` (`allow`, `deny`, `rate_limited`), `secret` (the ref name, never its value), `upstream` (the [upstream proxy](egress.md#upstream-proxies)'s `host:port`, when one carried it), `tenant`, and on an allowed connection `src_addr` and `dst_addr` (the node-side and destination `ip:port` of the upstream TCP connection, the tunnel target through an upstream proxy) on `egress` records, which a relayed frame cannot set; preview accesses record as op `preview`, one per request; a tenant change records as op `tenants` with the `added`, `changed` and `removed` names, never a token; a config reload records as op `config_reload` with `changed`, and a runtime volume change as op `volumes` with `added` or `removed`. An opening frame whose first line exceeds 4 KiB records as op `oversized` with no addressing fields | | `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 | | `memory_limit` | unset | `{"anon": "300G", "psi_some_avg10": 40}`, either key alone: while `AnonPages` in `/proc/meminfo` is above `anon`, or the `some avg10` of `/proc/pressure/memory` is above `psi_some_avg10`, checked every 2 s, the node refuses new claims, forks and checkpoint claims with 429 and parks refill; `/v1/info` reports it as at capacity and the mesh sees no pools on it, so a warm or pool redirect skips it; a template or volume held only there still redirects to it and answers 429. Wakes of its own claims, hibernates, checkpoints and archives go on, and running sandboxes keep running. Guest memory is a private file mapping, so page cache and clean guest pages are reclaimable and only anon is what a hibernate has to free. Both signals are node-wide; `AnonPages` leaves out swapped pages and explicit hugepages. A signal the node cannot read, such as PSI on a kernel without it, logs an error and is ignored. PSI also counts a task stalled by its own cgroup limit and weighs by busy CPUs, so on a mostly idle node a few throttled tasks in one small cgroup can cross it: 16 tasks under a 64M `MemoryHigh` reached 75 on an idle 384-CPU host with memory to spare. Treat `anon` as the main limit, and set `psi_some_avg10` only where no workload runs under a tight memory cgroup. Size `anon` so the running guests' further growth still fits under physical memory | @@ -677,20 +677,21 @@ Stopping sandboxd leaves VMs alive; the next start reconciles them. The unit's `KillMode=process` is what guarantees it: systemd signals sandboxd alone, so a VMM that cocoon could not move into its own scope outlives the stop instead of dying with the service's cgroup. sandboxd starts no new `cocoon` command once -stopped and waits for the running ones, since cocoon dies on its next log line -once sandboxd's end of its stderr closes, and a hibernate killed that way leaves -the guest paused. A save or hibernate can take 26 minutes, hence +stopped and waits for the running ones, since a `cocoon` command outlives a +stopped sandboxd only until its next log line, and a hibernate cut short that +way leaves the guest paused. A save or hibernate can take 26 minutes, hence `TimeoutStopSec=27min`; stops normally take seconds. If sandboxd dies abruptly during a capture instead, the next start holds that claim until the `cocoon` command has -exited, then settles it: a hibernate whose snapshot landed stays hibernated and wakes -from it, and any other capture resumes the guest, or marks the claim failed if it +exited, then settles it: with cocoon v0.7.7 or newer, a hibernate whose snapshot +landed stays hibernated and wakes from it, and any other capture resumes the guest, or marks the claim failed if it does not answer within 30 minutes. Claimed sandboxes are reaped when their TTL expires (default 5m, capped at 24h). -A snapshot save, hibernate, export, restore or clone has 15 minutes. Once canceled, a -save or hibernate has 11 more to exit, since cocoon v0.7.6 or newer waits up to 10 for -the VMM to resume the guest; an export, restore or clone has 3. At most 8 saves and -hibernates run at once on a node. A call that +A snapshot save, hibernate, live-move capture, export, restore, clone or +live-move receive has 15 minutes. Once canceled, a save, hibernate or live-move +capture has 11 more to exit, since cocoon v0.7.6 or newer can take up to 10 to +exit after a cancel; an export, restore, clone or receive has 3. At most 8 +saves, hibernates and live-move captures run at once on a node. A call that dials a guest gives up after 10 s, since a paused or wedged guest never answers, and gets `502`. @@ -788,6 +789,7 @@ through `/v1/tenants` on any node. `max_connections`. - Every other shared table a node opens adds a query pool of the same size: the pool set with a `cell`, the [archive registry](#shared-archive-registry) with `archive_failover`, + the [move registry](#shared-move-registry) with a `cell` and `mesh.cluster_key`, and the usage table with `usage`. Count each in `max_connections`. - LISTEN needs a session-mode connection, so a connection pooler in front of the database must not run in transaction mode for that DSN. @@ -979,6 +981,29 @@ With `archive_failover`, each node keeps its archive and replica rows in ); ``` +### Shared move registry + +With `meta_store.cell` and `mesh.cluster_key`, each node keeps its +[live-move](cluster.md#moving-a-running-claim) records in `sandboxd_moves`, +whether or not `move_concurrency` is set. + +- **Schema:** sandboxd creates the table at boot when it is missing; create it + ahead with the same DDL when the node's role has no `CREATE` right: + + ```sql + CREATE SEQUENCE IF NOT EXISTS sandboxd_moves_epoch; + CREATE TABLE IF NOT EXISTS sandboxd_moves ( + cell text NOT NULL, + id text NOT NULL, + from_node text NOT NULL, + to_node text NOT NULL, + state text NOT NULL CHECK (state IN ('pending', 'final', 'committed', 'started', 'aborted')), + epoch bigint NOT NULL CHECK (epoch > 0), + updated_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (cell, id) + ); + ``` + ### Reloading the config `kill -HUP ` or a root `POST /v1/config/reload` diff --git a/docs/index.md b/docs/index.md index af0ccd2c..774af3e0 100644 --- a/docs/index.md +++ b/docs/index.md @@ -22,13 +22,13 @@ vsock-only I/O (hardened default); `net=egress` attaches a bridge/CNI NIC. images, running as a service, the bare-metal e2e and acceptance drivers, the boot contract - [Clusters](cluster.md) — joining a mesh, querying members, how redirect - placement works, relocating handles, and surviving node loss (archived and - replicated claims) + placement works, relocating handles, draining and moving claims between + nodes, and surviving node loss (archived and replicated claims) - [sandboxd HTTP API](sandboxd-api.md) — every endpoint: claim/release, renew, hibernate, fork, promote, checkpoints, preview, the guest-port relay, claim env, runtime volume registration, live volume changes, - runtime tenants, config reload, online pool reconfigure, metrics, the usage - journal and its shared table + runtime tenants, config reload, online pool reconfigure, drain and + evacuation, live move, metrics, the usage journal and its shared table - [Replace a dataset in a running sandbox](live-volumes.md) — swapping a read-only catalog volume on a running claim with the volume API - [Hibernate with an external volume controller](volume-controller.md) — detach, diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 497c8664..64a74a6c 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -855,16 +855,19 @@ its guest's processes and memory (see and never this node (else 400), `max_downtime_ms` optional with a default of 2000 and a floor: 400 or less answers 400 at once, because the target's receive and start alone are allowed that long. Answers `{"owner_addr": "", "rounds": n, "downtime_ms": d}` once the -peer runs the guest and this node dropped the claim; `downtime_ms` is the pause -from the final capture to the peer's commit. While the move runs, release, fork, +peer took the claim and this node dropped it; a guest the peer cannot start is +failed there. `downtime_ms` is the pause from the final capture until the peer +answers the commit, after it started the guest. While the move runs, release, fork, checkpoint, promote, hibernate and volume changes of the claim answer 503 `sandbox moving to another node; retry`. Env writes, renew and data-plane calls keep working until the final round and answer the same 503 during it, because the peer would miss -them. Every error leaves the claim running here, except a 503 with the registry -down after the final round paused the guest: the guest stays paused until the -record answers, the guest then resumes or, when the record reads `committed`, the +them. Every error leaves the claim running here, except one after the final round +paused the guest whose abort did not land (a 503 with the registry down, or a +failed `cocoon vm migrate abort`): the guest stays paused until the abort is +retried through, the guest then resumes or, when the record reads `committed`, the node drops the claim (see [moving a running claim](cluster.md#moving-a-running-claim)): +- 403 the claim's tenant was removed - 404 unknown sandbox - 409 the claim cannot move (hibernated, archived, failed, waking, with volumes, egress lane, a move already running), cocoon or its run dir cannot @@ -873,7 +876,8 @@ node drops the claim (see [moving a running claim](cluster.md#moving-a-running-c - 429 `move_concurrency` reached here or on the target, or the target's admission refused (`target : `) - 502 the target is unreachable or failed -- 503 the move or archive registry is down, the node lease is overdue, or the +- 503 the move or archive registry is down, the node lease is overdue, the node + is evacuating the claim (`sandbox moving to another node; retry`), or the final round would not fit `max_downtime_ms`, after at most 4 deltas The target serves root-only peer routes for a move, `POST /v1/moves/{id}`, diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 9a5e0c5d..3c0ae416 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -26,13 +26,14 @@ import ( ) const ( - defaultListen = ":7777" - defaultDataDir = "/var/lib/sandboxd" - defaultCocoonBin = "cocoon" - defaultWarm = 4 DefaultMaxForkCount = 16 - refillFloor = 4 - refillCeiling = 256 + + defaultListen = ":7777" + defaultDataDir = "/var/lib/sandboxd" + defaultCocoonBin = "cocoon" + defaultWarm = 4 + refillFloor = 4 + refillCeiling = 256 // cocoon builds a VM with a 10G disk by default, so a pool never asks for less minStorageBytes = 10 << 30 ) @@ -185,10 +186,10 @@ type MemoryLimit struct { // MeshConfig configures cluster membership; every node shares one APIToken and tenant set. type MeshConfig struct { - NodeID string `json:"node_id"` // unique name; defaults to Bind - Bind string `json:"bind"` // memberlist host:port - Join []string `json:"join,omitempty"` // seed members; empty = mesh of one - ClusterKey string `json:"cluster_key,omitempty"` // base64 gossip-encryption key + NodeID string `json:"node_id"` + Bind string `json:"bind"` // memberlist host:port + Join []string `json:"join,omitempty"` // seed members; empty = mesh of one + ClusterKey string `json:"cluster_key,omitempty"` } // ID is the node's mesh name: NodeID, else Bind. diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index c779b717..e7b3f5af 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -13,7 +13,7 @@ import ( ) func TestLoadAppliesDefaults(t *testing.T) { - path := writeConfig(t, `{"pools":[{"template":"rt:24.04","net":"none","size":"small"}]}`) + path := writeConfig(t, `{"pools":[{"template":"rt:24.04"}]}`) cfg := mustLoad(t, path) if cfg.Listen == "" || cfg.DataDir == "" || cfg.CocoonBin == "" { t.Errorf("defaults not applied: %+v", cfg) @@ -111,7 +111,7 @@ func TestLoadRejectsInvalid(t *testing.T) { {"memory limit past a percentage", `{"memory_limit":{"psi_some_avg10":101},"pools":[]}`, "memory_limit.psi_some_avg10"}, {"negative max running", `{"max_running":-1,"pools":[]}`, "max_running"}, {"bad restore mode", `{"restore_mode":"Mmap","pools":[]}`, "restore_mode"}, - {"bad pool key", `{"pools":[{"template":"","net":"none","size":"small"}]}`, "pool"}, + {"bad pool key", `{"pools":[{"template":"","net":"none","size":"small"}]}`, `pool "": template must not be empty`}, {"egress without attachment", `{"pools":[{"template":"rt:24.04","net":"egress","size":"small"}]}`, "egress lane needs"}, {"egress idle hibernate", `{"bridges":["br0"],"pools":[{"template":"rt:24.04","net":"egress","size":"small","idle_hibernate_seconds":1}]}`, "not supported for egress"}, {"negative warm", `{"pools":[{"template":"rt:24.04","net":"none","size":"small","warm":-2}]}`, "negative"}, diff --git a/sandboxd/engine/cocoond.go b/sandboxd/engine/cocoond.go index 9bece4c4..9d195bab 100644 --- a/sandboxd/engine/cocoond.go +++ b/sandboxd/engine/cocoond.go @@ -11,11 +11,11 @@ import ( ) const ( - cocoondEventsURL = "http://cocoond/v1/events" - sseMaxLine = 1 << 20 - // VMDeleted is the change kind cocoon daemon sends when a VM record goes away. VMDeleted = "DELETED" + + cocoondEventsURL = "http://cocoond/v1/events" + sseMaxLine = 1 << 20 ) // VMStatus is cocoon daemon's view of one supervised VM. diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index bdfece4f..73d111b9 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -169,7 +169,7 @@ func (e *Engine) Remove(ctx context.Context, name string) error { return err } -// Start cold-boots a stopped VM from its own disk, or resumes the paused guest of a received VM. +// Start cold-boots a stopped VM from its own disk, or resumes a paused guest. func (e *Engine) Start(ctx context.Context, name string) error { _, err := e.run(ctx, "vm", "start", name) return err @@ -202,7 +202,7 @@ func (e *Engine) SnapshotSave(ctx context.Context, vmName, snapName string) erro return err } -// Hibernate atomically snapshots a running VM under snapName and stops it. +// Hibernate snapshots a running VM under snapName and stops it. func (e *Engine) Hibernate(ctx context.Context, vmName, snapName string) error { _, err := e.runDump(ctx, "vm", "hibernate", argName, snapName, vmName) return err diff --git a/sandboxd/engine/migrate.go b/sandboxd/engine/migrate.go index c07d453d..e87b502e 100644 --- a/sandboxd/engine/migrate.go +++ b/sandboxd/engine/migrate.go @@ -11,19 +11,10 @@ import ( const argSeries = "--series" -// MigrateUnsupported reports a runtime without cocoon's live migration. -type MigrateUnsupported struct { - Version string -} - -func (e *MigrateUnsupported) Error() string { - return fmt.Sprintf("live migration requires Cocoon >= %s; installed version is %s", MigrateCocoon, cmp.Or(e.Version, "unavailable")) -} - // CheckMigrateSupport rejects runtimes without live migration. func (e *Engine) CheckMigrateSupport(ctx context.Context) error { if version, misses := e.missesFloor(ctx, MigrateCocoon); misses { - return &MigrateUnsupported{Version: version} + return fmt.Errorf("live migration requires Cocoon >= %s; installed version is %s", MigrateCocoon, cmp.Or(version, "unavailable")) } return nil } diff --git a/sandboxd/engine/version_test.go b/sandboxd/engine/version_test.go index f07bb053..90d05f29 100644 --- a/sandboxd/engine/version_test.go +++ b/sandboxd/engine/version_test.go @@ -2,7 +2,6 @@ package engine import ( "context" - "errors" "fmt" "strings" "testing" @@ -73,9 +72,7 @@ printf 'Version: %s\n' "$COCOON_TEST_VERSION" } return } - _, isVolume := errors.AsType[*LiveVolumeUnsupported](err) - _, isMigrate := errors.AsType[*MigrateUnsupported](err) - if !isVolume && !isMigrate || !strings.Contains(err.Error(), tt.floor) || strings.Contains(err.Error(), "/private") { + if err == nil || !strings.Contains(err.Error(), tt.floor) || strings.Contains(err.Error(), "/private") { t.Fatalf("error %v", err) } }) diff --git a/sandboxd/main.go b/sandboxd/main.go index 370eafc5..e9792cd9 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -100,6 +100,9 @@ func main() { ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop() + if cfg.Mesh != nil { + mgr.WithMovePeer(&server.MoveClient{Token: cfg.APIToken}) + } if err := mgr.Reconcile(ctx); err != nil { logger.Fatalf(ctx, err, "reconcile") } @@ -134,7 +137,6 @@ func main() { mgr.WithPeerDelete(broadcaster.Delete) adopter := &peer.Adopter{Peers: msh.PeerAddrs} mgr.WithPeerAdopt(adopter.Adopt) - mgr.WithMovePeer(&server.MoveClient{Token: cfg.APIToken}) 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 d562431a..1b136d18 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -91,10 +91,14 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { } sb.Transition.Lock() err := m.resolveIntent(ctx, sb) + snap := sb.HibernateSnap sb.Transition.Unlock() if err != nil { return fmt.Errorf("archive %s: settle intent: %w", sb.ID, err) } + if snap == "" { + return errWokeMeanwhile + } if err = m.checkArchiveRow(ctx, sb.ID); err != nil { return fmt.Errorf("archive %s: registry: %w", sb.ID, err) } @@ -105,7 +109,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { m.mu.Unlock() defer m.untrack(m.pendingCks, ckID) // A hibernated source is copied from its wake image, so no VM starts. - ck, srcSnap, err := m.publishCheckpoint(ctx, sb, ckID, "archive", sb.Tenant, true) + ck, srcSnap, err := m.publishCheckpoint(ctx, sb, ckID, "archive", sb.Tenant, snap, true) if err != nil { return err } diff --git a/sandboxd/pool/archive_test.go b/sandboxd/pool/archive_test.go index 67c9f3dc..5dab0449 100644 --- a/sandboxd/pool/archive_test.go +++ b/sandboxd/pool/archive_test.go @@ -107,6 +107,29 @@ func TestArchiveWakeRehibernateWindowAborts(t *testing.T) { } } +func TestArchiveNeverCapturesAGuestThatWokeBeforeItsExport(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng, archivePool(3600)) + sb := mustClaim(t, m, testKey) + hibernateClaim(t, m, sb) + m.ckpts = &stagingStore{Store: m.ckpts, onStage: func() { + if _, _, release, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { + t.Errorf("wake before the export: %v", err) + } else { + release() + } + }} + if err := m.archive(t.Context(), sb); !errors.Is(err, errWokeMeanwhile) { + t.Fatalf("archive: %v, want errWokeMeanwhile", err) + } + eng.mu.Lock() + saves := eng.snapSavesEntered + eng.mu.Unlock() + if saves != 0 { + t.Errorf("the archive paused the woken guest for %d snapshot saves", saves) + } +} + func TestArchiveCkHiddenAcrossNodes(t *testing.T) { shared := t.TempDir() mkMgr := func() *Manager { @@ -914,6 +937,16 @@ func BenchmarkReapPurgesArchivedClaims(b *testing.B) { } } +type stagingStore struct { + store.Store + onStage func() +} + +func (s *stagingStore) Stage(id string) (string, error) { + s.onStage() + return s.Store.Stage(id) +} + type stallingStore struct { store.Store published chan string diff --git a/sandboxd/pool/capture.go b/sandboxd/pool/capture.go index a1cd3b56..57da7e01 100644 --- a/sandboxd/pool/capture.go +++ b/sandboxd/pool/capture.go @@ -58,10 +58,14 @@ func (m *Manager) captureAllowed(sb *types.Sandbox) error { } // exportSource captures a claimed sandbox into exportDir and returns its snapshot, VM name and guest-env presence. -func (m *Manager) exportSource(ctx context.Context, sb *types.Sandbox, exportDir string) (snap, vmName string, guestEnv bool, err error) { +func (m *Manager) exportSource(ctx context.Context, sb *types.Sandbox, exportDir, from string) (snap, vmName string, guestEnv bool, err error) { if lockErr := m.lockTransition(sb); lockErr != nil { return "", "", false, lockErr } + if from != "" && sb.HibernateSnap != from { + sb.Transition.Unlock() + return "", "", false, errWokeMeanwhile + } if captureErr := m.captureAllowed(sb); captureErr != nil { sb.Transition.Unlock() return "", "", false, captureErr diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index a89cb1e5..efa26a61 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -51,7 +51,7 @@ func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, te } // A started capture must finish even if the caller hangs up. ctx = context.WithoutCancel(ctx) - ckpt, _, err := m.publishCheckpoint(ctx, sb, store.CheckpointID(randHex(8)), name, tenant, false) + ckpt, _, err := m.publishCheckpoint(ctx, sb, store.CheckpointID(randHex(8)), name, tenant, "", false) if err != nil { return types.Checkpoint{}, err } @@ -167,7 +167,7 @@ func (m *Manager) FetchCheckpoint(ctx context.Context, ckptID string) (string, [ return dir, meta, unlock, nil } -func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID, name, tenant string, archive bool) (types.Checkpoint, string, error) { +func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID, name, tenant, from string, archive bool) (types.Checkpoint, string, error) { ckpt := types.Checkpoint{ ID: ckID, Name: name, @@ -187,7 +187,7 @@ func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID return types.Checkpoint{}, "", fmt.Errorf("stage checkpoint: %w", err) } defer func() { _ = os.RemoveAll(staging) }() - srcSnap, _, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) + srcSnap, _, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir), from) if err != nil { return types.Checkpoint{}, "", fmt.Errorf("checkpoint %s: %w", sb.ID, err) } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index 6314f327..c9c0feb4 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -41,7 +41,6 @@ func (m *Manager) ClaimWarm(ctx context.Context, key types.PoolKey, o ClaimOptio if err != nil { return nil, err } - // Holds belong to this path until finalize; past it the sandbox carries them. reserved := applied defer func() { m.unreserveVolumes(reserved) }() m.mu.Lock() @@ -286,6 +285,10 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand m.mu.Unlock() return ErrUnknownSandbox } + if m.midMoveLocked(sb) { + m.mu.Unlock() + return ErrEvacuating + } ck := sb.ArchiveCk if ck == "" || ck == marked { break @@ -408,10 +411,7 @@ func (m *Manager) quotaErr(extra int, tenant string) error { if limit <= 0 { return nil } - live := m.tenantLive[tenant] - if len(m.incoming) > 0 { - live += m.incomingOf(tenant) - } + live := m.tenantLive[tenant] + m.incomingOf(tenant) if live+extra > limit { return fmt.Errorf("%w: tenant %s at %d live claims, cap %d", ErrQuota, tenant, live, limit) } diff --git a/sandboxd/pool/env.go b/sandboxd/pool/env.go index 7e266d6c..26691629 100644 --- a/sandboxd/pool/env.go +++ b/sandboxd/pool/env.go @@ -34,7 +34,6 @@ func (m *Manager) Env(id, tenant string) (types.Env, error) { return sb.Env.Redacted(), nil } -// writeEnv stores next's env for claim id; an unchanged result reaches a running guest again, so a resend repairs a failed write. func (m *Manager) writeEnv(ctx context.Context, id, tenant, verb string, next func(types.Env) types.Env) error { sb, ok := m.byID(id) if !ok || !tenantOwns(tenant, sb.Tenant) { diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index 068b4910..e1b7056d 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -387,7 +387,10 @@ func (m *Manager) unrowExpired(ctx context.Context, sb *types.Sandbox) { sb.Transition.Lock() defer sb.Transition.Unlock() var err error - if m.replicaHeld(sb) { + m.mu.Lock() + replicaHeld := sb.HasReplica() + m.mu.Unlock() + if replicaHeld { err = m.dropReplica(ctx, sb) } else { _, err = m.consumeArchive(ctx, sb) diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index e18e4fba..e4c607fc 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -1951,6 +1951,7 @@ func (c *failoverCluster) nodeOn(t *testing.T, eng *fakeEngine, name, dataDir st if err != nil { t.Fatalf("manager %s: %v", name, err) } + t.Cleanup(func() { settleCaptures(m) }) _ = 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 { @@ -2129,7 +2130,7 @@ func assertReleasedForGood(t *testing.T, c *failoverCluster, sb *types.Sandbox, func interruptArchive(t *testing.T, m *Manager, sb *types.Sandbox, putCommitted bool) string { t.Helper() - ck, _, err := m.publishCheckpoint(t.Context(), sb, store.CheckpointID(randHex(8)), "archive", sb.Tenant, true) + ck, _, err := m.publishCheckpoint(t.Context(), sb, store.CheckpointID(randHex(8)), "archive", sb.Tenant, "", true) if err != nil { t.Fatalf("publish: %v", err) } diff --git a/sandboxd/pool/hibernate_test.go b/sandboxd/pool/hibernate_test.go index 7da12a3c..77ebe54e 100644 --- a/sandboxd/pool/hibernate_test.go +++ b/sandboxd/pool/hibernate_test.go @@ -223,8 +223,9 @@ func TestWakeResolvesDanglingIntent(t *testing.T) { sb := mustClaim(t, m, testKey) leaveDanglingIntent(t, m, eng, sb) - eng.hibernateErr = nil - eng.snapListErr = nil + eng.mu.Lock() + eng.hibernateErr, eng.snapListErr = nil, nil + eng.mu.Unlock() sock, _, _, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token) if err != nil { t.Fatalf("wake: %v", err) diff --git a/sandboxd/pool/interrupted.go b/sandboxd/pool/interrupted.go index 5b3c8e57..e6cb1fcf 100644 --- a/sandboxd/pool/interrupted.go +++ b/sandboxd/pool/interrupted.go @@ -23,12 +23,16 @@ func (m *Manager) settleInterrupted(ctx context.Context, sb *types.Sandbox) { defer sb.Transition.Unlock() logger := log.WithFunc("pool.settleInterrupted") m.mu.Lock() - sb.ReaderSafe = false snap := cmp.Or(sb.PendingSnap, sb.SaveSnap) + if snap == "" { + m.mu.Unlock() + return + } + sb.ReaderSafe = false m.mu.Unlock() logger.Warnf(ctx, "%s has an interrupted capture %s; holding it until the capture exits", sb.ID, snap) if err := m.settleCapture(ctx, sb); err != nil { - logger.Warnf(ctx, "%s keeps its capture intent for the next start: %v", sb.ID, err) + logger.Warnf(ctx, "%s did not settle its capture: %v", sb.ID, err) return } logger.Infof(ctx, "%s settled its interrupted capture", sb.ID) @@ -55,16 +59,19 @@ func (m *Manager) settleCapture(ctx context.Context, sb *types.Sandbox) (err err return err } listed := pending != "" && slices.Contains(snaps, pending) + var failed error if !listed && running { if err = m.resumeGuest(ctx, sb, sock); err != nil { if ctx.Err() != nil || errors.Is(err, engine.ErrStopping) { return err } - m.failLocked(ctx, sb, fmt.Sprintf("capture stopped short: guest did not resume: %v", err)) + reason := fmt.Sprintf("capture stopped short: guest did not resume: %v", err) + m.failLocked(ctx, sb, reason) + failed = fmt.Errorf("%w: %s", ErrFailed, reason) } } _, err = m.finishCapture(ctx, sb, running && listed, saved, snaps) - return err + return errors.Join(failed, err) } func (m *Manager) settleFailedCapture(ctx context.Context, sb *types.Sandbox) (adopted bool, err error) { diff --git a/sandboxd/pool/interrupted_test.go b/sandboxd/pool/interrupted_test.go index 631c67f5..219104ee 100644 --- a/sandboxd/pool/interrupted_test.go +++ b/sandboxd/pool/interrupted_test.go @@ -227,10 +227,10 @@ func TestInterruptedCaptureFailsTheClaim(t *testing.T) { func TestInterruptedCaptureKeepsItsIntentWhenStopping(t *testing.T) { for _, tt := range []struct { name string - run func(*testing.T, *fakeEngine, context.CancelFunc) + run func(*fakeEngine, context.CancelFunc) }{ - {"engine stopping", func(_ *testing.T, eng *fakeEngine, _ context.CancelFunc) { eng.startErr = engine.ErrStopping }}, - {"context canceled mid retry", func(_ *testing.T, eng *fakeEngine, cancel context.CancelFunc) { + {"engine stopping", func(eng *fakeEngine, _ context.CancelFunc) { eng.startErr = engine.ErrStopping }}, + {"context canceled mid retry", func(eng *fakeEngine, cancel context.CancelFunc) { eng.startFails = 1 << 30 eng.startHook = cancel }}, @@ -239,7 +239,7 @@ func TestInterruptedCaptureKeepsItsIntentWhenStopping(t *testing.T) { eng := newFakeEngine() ctx, cancel := context.WithCancel(t.Context()) defer cancel() - tt.run(t, eng, cancel) + tt.run(eng, cancel) m, sb := reconcileInterruptedCtx(ctx, t, eng, interruptedHib, "", nil) waitSettled(t, sb) if row := journaled(t, m)[sb.ID]; row.PendingSnap != interruptedHib || sb.Failed != "" { @@ -250,20 +250,28 @@ func TestInterruptedCaptureKeepsItsIntentWhenStopping(t *testing.T) { } func TestAFailedHibernateResumesTheGuest(t *testing.T) { - for _, listFails := range []int{0, 3} { - synctest.Test(t, func(t *testing.T) { - eng := newFakeEngine() - m := newTestManager(t, eng) - sb := mustClaim(t, m, testKey) - eng.hibernateErr, eng.listFails = errors.New("signal: killed"), listFails - if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { - t.Fatal("Hibernate reported success") - } - time.Sleep(time.Duration(listFails+1) * captureResumeRetry) - synctest.Wait() - if !slices.Equal(eng.starts, []string{sb.VMName}) { - t.Errorf("starts %q, want the guest resumed", eng.starts) - } + for _, tt := range []struct { + name string + listFails int + }{ + {"list answers", 0}, + {"list fails three times", 3}, + } { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + m := newTestManager(t, eng) + sb := mustClaim(t, m, testKey) + eng.hibernateErr, eng.listFails = errors.New("signal: killed"), tt.listFails + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { + t.Fatal("Hibernate reported success") + } + time.Sleep(time.Duration(tt.listFails+1) * captureResumeRetry) + synctest.Wait() + if !slices.Equal(eng.starts, []string{sb.VMName}) { + t.Errorf("starts %q, want the guest resumed", eng.starts) + } + }) }) } } @@ -329,18 +337,18 @@ func TestAFailedHibernateThatPersistedStopsTheVMMAndAdopts(t *testing.T) { func TestAFailedCaptureKeepsItsIntentWhileSandboxdStops(t *testing.T) { for _, tt := range []struct { name string - capture func(*Manager, *types.Sandbox, *fakeEngine) error + capture func(context.Context, *Manager, *types.Sandbox, *fakeEngine) error intent func(*types.Sandbox) string }{ - {"hibernate", func(m *Manager, sb *types.Sandbox, eng *fakeEngine) error { + {"hibernate", func(ctx context.Context, m *Manager, sb *types.Sandbox, eng *fakeEngine) error { eng.hibernateErr = errors.New("signal: killed") - return m.Hibernate(context.Background(), sb.ID, Cred{Token: sb.Token}) + return m.Hibernate(ctx, sb.ID, Cred{Token: sb.Token}) }, func(row *types.Sandbox) string { return row.PendingSnap }}, - {"save", func(m *Manager, sb *types.Sandbox, eng *fakeEngine) error { + {"save", func(ctx context.Context, m *Manager, sb *types.Sandbox, eng *fakeEngine) error { eng.snapSaveErr = errors.New("signal: killed") sb.Transition.Lock() defer sb.Transition.Unlock() - _, _, err := m.sourceSnap(context.Background(), sb) + _, _, err := m.sourceSnap(ctx, sb) return err }, func(row *types.Sandbox) string { return row.SaveSnap }}, } { @@ -349,7 +357,7 @@ func TestAFailedCaptureKeepsItsIntentWhileSandboxdStops(t *testing.T) { m := newTestManager(t, eng) sb := mustClaim(t, m, testKey) eng.startErr = engine.ErrStopping - if err := tt.capture(m, sb, eng); err == nil { + if err := tt.capture(t.Context(), m, sb, eng); err == nil { t.Fatal("the capture reported success") } waitStarted(t, eng, 1) @@ -400,6 +408,65 @@ func TestASaveJournalsItsIntentAndAFailureResumesTheGuest(t *testing.T) { waitFor(t, func() bool { return journaled(t, m)[sb.ID].SaveSnap == "" }) } +func TestAWakeWhoseGuestDoesNotResumeAnswersFailed(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.snapListErr = errors.New("cocoon unreachable") + m, sb := reconcileInterrupted(t, eng, interruptedHib, "", nil) + time.Sleep(captureResumeBudget + 2*captureResumeRetry) + synctest.Wait() + eng.mu.Lock() + eng.snapListErr, eng.probeErr = nil, errors.New("never ready") + eng.mu.Unlock() + if _, _, _, err := m.WakeAgentSocket(t.Context(), sb.ID, sb.Token); !errors.Is(err, ErrFailed) { + t.Fatalf("wake: %v, want ErrFailed", err) + } + if err := m.Hibernate(t.Context(), sb.ID, Cred{Token: sb.Token}); !errors.Is(err, ErrFailed) { + t.Errorf("hibernate: %v, want ErrFailed", err) + } + }) +} + +func TestASettleWithNoIntentLeftKeepsTheReaderSafeHibernate(t *testing.T) { + m, sb := reconcileInterrupted(t, newFakeEngine(), "", "", nil) + m.mu.Lock() + sb.ReaderSafe = true + m.mu.Unlock() + sb.Transition.Lock() + m.settleInterrupted(t.Context(), sb) + if !sb.ReaderSafe { + t.Error("a settle with nothing to settle dropped ReaderSafe") + } +} + +func TestReconcileWithoutASnapshotListSettlesAStoppedHibernate(t *testing.T) { + for _, tt := range []struct { + name string + listed []string + want string + }{ + {"snapshot persisted", []string{interruptedHib}, interruptedHib}, + {"snapshot never persisted", nil, ""}, + } { + t.Run(tt.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eng := newFakeEngine() + eng.vmStates = map[string]string{interruptedVM: "stopped"} + eng.snapListErr = errors.New("cocoon unreachable") + _, sb := reconcileInterrupted(t, eng, interruptedHib, "", tt.listed) + eng.mu.Lock() + eng.snapListErr = nil + eng.mu.Unlock() + time.Sleep(2 * captureResumeRetry) + synctest.Wait() + if sb.HibernateSnap != tt.want || sb.PendingSnap != "" { + t.Errorf("hibernate snap %q intent %q, want %q and no intent", sb.HibernateSnap, sb.PendingSnap, tt.want) + } + }) + }) + } +} + func waitStarted(t *testing.T, eng *fakeEngine, n int) { t.Helper() waitFor(t, func() bool { diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go index 0e3084e2..c5a7f09f 100644 --- a/sandboxd/pool/liveness.go +++ b/sandboxd/pool/liveness.go @@ -314,7 +314,7 @@ func (m *Manager) vmSettledLocked(sb *types.Sandbox) bool { return runningShaped(sb) && !archiving && !moving } -// runningShaped reports whether sb's VM should be up; callers hold m.mu. +// runningShaped reports whether sb's VM should be up; callers hold m.mu or sb.Transition. func runningShaped(sb *types.Sandbox) bool { return sb.HibernateSnap == "" && sb.PendingSnap == "" && sb.ArchiveCk == "" && sb.Failed == "" && !sb.MoveIn } diff --git a/sandboxd/pool/liveness_test.go b/sandboxd/pool/liveness_test.go index 01c40439..8ee93b9b 100644 --- a/sandboxd/pool/liveness_test.go +++ b/sandboxd/pool/liveness_test.go @@ -39,8 +39,12 @@ func TestLivenessRestartsExitedVMM(t *testing.T) { t.Errorf("events %s, want %s", got, want) } reloaded, err := newClaimStore(m.dataDir, false).load() - if err != nil || reloaded[sb.ID].Restarts != 1 { - t.Errorf("journal restarts %d (err %v), want 1", reloaded[sb.ID].Restarts, err) + if err != nil { + t.Fatalf("load journal: %v", err) + } + rec, ok := reloaded[sb.ID] + if !ok || rec.Restarts != 1 { + t.Errorf("journal record %+v (present %v), want one restart", rec, ok) } } @@ -201,7 +205,6 @@ func TestLivenessLeavesALapsedLeaseToReap(t *testing.T) { func TestWakeOfAFailedClaimRearmsEgressAfterARestart(t *testing.T) { eng := newFakeEngine() eng.sockRoot = outboundtest.SockRoot(t) - t.Setenv("GH_TOKEN", "s3cr3t") cfg := &config.Config{ DataDir: t.TempDir(), Bridges: []string{"sbxbr0"}, EgressCA: writeTestEgressCA(t), VMMRestart: types.VMMRestartNone, Pools: []config.PoolSpec{{PoolKey: testKey, Egress: &egress.Policy{Allow: []egress.Rule{{Host: "example.com"}}}}}, @@ -317,7 +320,7 @@ func TestDaemonExitEventRecoversWithoutAList(t *testing.T) { killVMM(eng, other.VMName) eng.vmEvents <- fakeVMEvent{change: engine.VMChange{Kind: "MODIFIED", VM: engine.VMStatus{Name: sb.VMName}}} - waitFor(t, func() bool { got, _ := m.Sandbox(sb.ID); return got.Restarts == 1 }) + waitFor(t, func() bool { got, _ := m.Sandbox(sb.ID); return got.Restarts == 1 && len(m.maintSem) == 0 }) if got, _ := m.Sandbox(other.ID); got.Restarts != 0 { t.Errorf("an event for %s restarted %s too", sb.VMName, other.VMName) } diff --git a/sandboxd/pool/move.go b/sandboxd/pool/move.go index 70f4ea99..3e0c6612 100644 --- a/sandboxd/pool/move.go +++ b/sandboxd/pool/move.go @@ -54,8 +54,6 @@ type MovePeer interface { // outMove is a source's move in flight; final is guarded by m.mu, the rest belongs to the goroutine driving the move. type outMove struct { types.MoveOut - addr string - to string replica string rounds int final bool @@ -93,7 +91,7 @@ func (m *Manager) moveDir(id string) string { return filepath.Join(m.dataDir, movesDir, id) } -// A target only marks a committed move started; only the source deletes an ended one. +// A target marks a committed move started and aborts a stale pending or final one; only the source deletes an ended one. func (m *Manager) sweepStaleMoves(ctx context.Context, now time.Time) { logger := log.WithFunc("pool.sweepStaleMoves") recs, err := m.moves.Stale(ctx, now.Add(-moveStaleAge)) diff --git a/sandboxd/pool/move_in.go b/sandboxd/pool/move_in.go index 80810810..0d447118 100644 --- a/sandboxd/pool/move_in.go +++ b/sandboxd/pool/move_in.go @@ -33,6 +33,9 @@ func (m *Manager) AdmitMove(ctx context.Context, id string, a MoveAdmission) (st if m.moves == nil { return "", errMovesOff } + if err := m.eng.CheckMigrateSupport(ctx); err != nil { + return "", fmt.Errorf("%w: %w", ErrNotMovable, err) + } if err := m.validate(a.Key); err != nil { return "", fmt.Errorf("%w: %w", ErrNotMovable, err) } @@ -103,7 +106,7 @@ func (m *Manager) CommitMove(ctx context.Context, id string, claim []byte, epoch sb.Transition.Lock() defer sb.Transition.Unlock() m.mu.Lock() - if m.incoming[id] != admitted || m.claimed[id] != nil { + if m.incoming[id] != admitted || m.claimed[id] != nil || m.draining { m.mu.Unlock() m.destroy(ctx, name) return fmt.Errorf("%w: the reservation for %s ended", ErrNotMovable, id) @@ -126,7 +129,7 @@ func (m *Manager) CommitMove(ctx context.Context, id string, claim []byte, epoch go m.settleMoveIn(ctx, sb) return fmt.Errorf("move %s: %w", id, err) } - m.startMoveIn(ctx, sb, admitted.from, committed) + m.startMoveIn(ctx, sb, admitted.from, committed, false) return nil } @@ -142,15 +145,37 @@ func (m *Manager) DropMove(id string) { } // A guest that cannot start fails the claim here, never back to the source; the caller holds sb.Transition. -func (m *Manager) startMoveIn(ctx context.Context, sb *types.Sandbox, from string, committed int64) { +func (m *Manager) startMoveIn(ctx context.Context, sb *types.Sandbox, from string, committed int64, cold bool) { logger := log.WithFunc("pool.startMoveIn") + if !cold && !m.startMovedGuest(ctx, sb, committed) { + return + } + live, err := m.commitIfLive(ctx, sb, func() { sb.MoveIn = false }) + if !live { + m.out.Disarm(sb.ID, true) + return + } + if err != nil { + logger.Warnf(ctx, "persist the start of moved claim %s: %v", sb.ID, err) + } + sb.Touch() + m.recordUsage(ctx, usage.Event{Event: "move_in", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Deadline: m.deadlineOf(sb)}) + _ = os.RemoveAll(m.moveDir(sb.ID)) + if committed != 0 && err == nil { + m.markMoveStarted(ctx, sb.ID, committed) + } + logger.Infof(ctx, "sandbox %s moved here from %s", sb.ID, from) +} + +func (m *Manager) startMovedGuest(ctx context.Context, sb *types.Sandbox, committed int64) bool { + logger := log.WithFunc("pool.startMovedGuest") var err error for range 2 { if err = m.eng.Start(ctx, sb.VMName); err == nil { break } if errors.Is(err, engine.ErrStopping) { - return + return false } logger.Warnf(ctx, "start moved claim %s: %v", sb.ID, err) } @@ -163,7 +188,7 @@ func (m *Manager) startMoveIn(ctx context.Context, sb *types.Sandbox, from strin if committed != 0 && persistErr == nil { m.markMoveStarted(ctx, sb.ID, committed) } - return + return false } if sock, probeErr := m.probeReady(ctx, sb.VMName, sb.VsockSocket, claimProbeTimeout); probeErr != nil { logger.Warnf(ctx, "probe moved claim %s: %v", sb.ID, probeErr) @@ -175,21 +200,7 @@ func (m *Manager) startMoveIn(ctx context.Context, sb *types.Sandbox, from strin if proxyErr := m.out.ArmProxy(ctx, sb); proxyErr != nil { logger.Errorf(ctx, proxyErr, "arm egress proxy %s", sb.ID) } - live, err := m.commitIfLive(ctx, sb, func() { sb.MoveIn = false }) - if !live { - m.out.Disarm(sb.ID, true) - return - } - if err != nil { - logger.Warnf(ctx, "persist the start of moved claim %s: %v", sb.ID, err) - } - sb.Touch() - m.recordUsage(ctx, usage.Event{Event: "move_in", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Deadline: m.deadlineOf(sb)}) - _ = os.RemoveAll(m.moveDir(sb.ID)) - if committed != 0 && err == nil { - m.markMoveStarted(ctx, sb.ID, committed) - } - logger.Infof(ctx, "sandbox %s moved here from %s", sb.ID, from) + return true } func (m *Manager) markMoveStarted(ctx context.Context, id string, epoch int64) { @@ -229,11 +240,17 @@ func (m *Manager) resolveMoveIn(ctx context.Context, sb *types.Sandbox) bool { } toHere := ok && rec.To == m.nodeID switch { - case toHere && rec.State == moves.Committed: - m.startMoveIn(ctx, sb, rec.From, rec.Epoch) - return true - case toHere && rec.State == moves.Started: - m.startMoveIn(ctx, sb, rec.From, 0) + case toHere && (rec.State == moves.Committed || rec.State == moves.Started): + committed := rec.Epoch + if rec.State == moves.Started { + committed = 0 + } + vm, found, inspectErr := m.eng.Inspect(ctx, sb.VMName) + if inspectErr != nil { + log.WithFunc("pool.resolveMoveIn").Warnf(ctx, "inspect moved claim %s, retrying: %v", sb.ID, inspectErr) + return false + } + m.startMoveIn(ctx, sb, rec.From, committed, found && vm.State != vmStateRunning) return true case toHere && (rec.State == moves.Pending || rec.State == moves.Final): if _, err = m.moves.Advance(ctx, sb.ID, rec.Epoch, rec.State, moves.Aborted); err != nil { diff --git a/sandboxd/pool/move_out.go b/sandboxd/pool/move_out.go index 3a3d4ccf..025b8f11 100644 --- a/sandboxd/pool/move_out.go +++ b/sandboxd/pool/move_out.go @@ -31,7 +31,7 @@ func (m *Manager) Move(ctx context.Context, id, addr string, maxDowntime time.Du if err := m.eng.CheckMigrateSupport(ctx); err != nil { return MoveResult{}, fmt.Errorf("%w: %w", ErrNotMovable, err) } - mv := &outMove{addr: addr} + mv := &outMove{Addr: addr} if err := m.markMoving(sb, mv); err != nil { return MoveResult{}, err } @@ -50,11 +50,11 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b m.mu.Lock() admission, vm := MoveAdmission{Tenant: sb.Tenant, Key: sb.Key, From: m.nodeID}, sb.VMName m.mu.Unlock() - to, err := m.movePeer.Admit(ctx, mv.addr, sb.ID, admission) + to, err := m.movePeer.Admit(ctx, mv.Addr, sb.ID, admission) if err != nil { return err } - mv.to = to + mv.To = to epoch, err := m.beginMove(ctx, sb, to) if errors.Is(err, moves.ErrLost) { return fmt.Errorf("%w: another move of %s is on record", ErrNotMovable, sb.ID) @@ -125,7 +125,7 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b if err != nil { return fmt.Errorf("encode claim: %w", err) } - return m.movePeer.Commit(ctx, mv.addr, sb.ID, claim, final) + return m.movePeer.Commit(ctx, mv.Addr, sb.ID, claim, final) } func (m *Manager) beginMove(ctx context.Context, sb *types.Sandbox, to string) (int64, error) { @@ -157,7 +157,7 @@ func (m *Manager) sendRound(ctx context.Context, id string, mv *outMove, round t } ctx, cancel := context.WithTimeout(ctx, roundTimeout(roundBytes(round))) defer cancel() - return m.movePeer.SendRound(ctx, mv.addr, id, mv.rounds, round.Dir, names) + return m.movePeer.SendRound(ctx, mv.Addr, id, mv.rounds, round.Dir, names) } func (m *Manager) markMoving(sb *types.Sandbox, mv *outMove) error { @@ -283,9 +283,9 @@ func (m *Manager) failMoveOut(ctx context.Context, sb *types.Sandbox, mv *outMov moved, err := m.settleMoveOut(ctx, sb, mv) switch { case err != nil: - log.WithFunc("pool.failMoveOut").Warnf(ctx, "move of %s failed (%v) and its abort waits for the registry: %v", sb.ID, cause, err) + log.WithFunc("pool.failMoveOut").Warnf(ctx, "move of %s failed (%v) and its abort is retrying: %v", sb.ID, cause, err) go m.retryMoveOut(ctx, sb, mv) - return MoveResult{}, fmt.Errorf("%w: %w", ErrMoveStoreDown, cause) + return MoveResult{}, errors.Join(cause, err) case moved: return result, nil } @@ -340,9 +340,9 @@ func (m *Manager) abortMoveOut(ctx context.Context, sb *types.Sandbox, mv *outMo return fmt.Errorf("abort series %s: %w", mv.Series, err) } } - if mv.addr != "" && mv.to != "" { - if err := m.movePeer.Drop(ctx, mv.addr, sb.ID); err != nil { - logger.Warnf(ctx, "drop the reservation for %s on %s: %v", sb.ID, mv.addr, err) + if mv.Addr != "" && mv.To != "" { + if err := m.movePeer.Drop(ctx, mv.Addr, sb.ID); err != nil { + logger.Warnf(ctx, "drop the reservation for %s on %s: %v", sb.ID, mv.Addr, err) } } _, journalErr := m.commitIfLive(ctx, sb, func() { sb.MoveOut = nil }) @@ -422,8 +422,5 @@ func roundTimeout(size int64) time.Duration { // finalPause estimates the final round's pause from the last delta's size at the rate the rounds so far reached. func finalPause(last, sent int64, sending time.Duration) time.Duration { - if sent <= 0 { - return MoveStartAllowance - } return MoveStartAllowance + time.Duration(float64(sending)*float64(last)/float64(sent)) } diff --git a/sandboxd/pool/move_test.go b/sandboxd/pool/move_test.go index 00711f0b..42e31730 100644 --- a/sandboxd/pool/move_test.go +++ b/sandboxd/pool/move_test.go @@ -151,7 +151,7 @@ func TestMoveFailingAtAnyStepKeepsOneOwner(t *testing.T) { "receive": {fail: func(mc *moveCluster) { fakeOf(mc.t).receiveErr = errors.New("restore failed") }}, "target memory": {fail: func(mc *moveCluster) { mc.p.commitHook = func(int) error { mc.t.memoryLow.Store(true); return nil } }, want: ErrQuota}, "source aborted first": {fail: func(mc *moveCluster) { - mc.p.commitHook = func(int) error { abortRecord(mc); return nil } + mc.p.commitHook = func(int) error { abortRecord(t, mc); return nil } }, want: ErrNotMovable}, } { t.Run(name, func(t *testing.T) { @@ -227,6 +227,47 @@ func TestMoveKeepsThePausedGuestUntilItsAbortLands(t *testing.T) { } } +func TestMoveWhoseAbortFailsBlamesTheAbortNotTheRegistry(t *testing.T) { + mc := newMoveCluster(t) + sb := claimTenant(t, mc.s, "acme") + eng := fakeOf(mc.s) + eng.abortErr = errors.New("vmm busy") + mc.p.commitHook = func(int) error { return ErrMoveTarget } + _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Minute) + if !errors.Is(err, ErrMoveTarget) || errors.Is(err, ErrMoveStoreDown) { + t.Fatalf("move whose abort failed: %v, want ErrMoveTarget and not ErrMoveStoreDown", err) + } + eng.mu.Lock() + eng.abortErr = nil + eng.mu.Unlock() + waitFor(t, func() bool { return !pausedOn(mc.s, sb.VMName) && !movingOn(mc.s, sb) }) + assertStillOnSource(t, mc, sb) +} + +func TestMoveRestartedSourceDropsTheTargetsReservation(t *testing.T) { + mc := newMoveCluster(t) + const id, vm, series = "sb_x", "sbx-x", "s1" + reserveMove(t, mc.t, id, "acme") + epoch := seedMoveRecord(t, mc.moveReg, id, "a", "b", moves.Pending) + dataDir := t.TempDir() + seedClaims(t, dataDir, map[string]*types.Sandbox{id: { + ID: id, VMName: vm, Key: testKey, Token: "tok", Tenant: "acme", Deadline: time.Now().Add(time.Hour), + MoveOut: &types.MoveOut{Epoch: epoch, Series: series, Addr: "b", To: "b"}, + }}) + eng := newFakeEngine() + eng.vms[vm], eng.migrateSeries[vm] = "/vsock/"+vm, series + restarted := mc.nodeOn(t, eng, "a", dataDir, keepAnHour) + restarted.liftFence() + restarted.WithMovePeer(mc.p) + if err := restarted.Reconcile(t.Context()); err != nil { + t.Fatalf("reconcile: %v", err) + } + waitFor(t, func() bool { return !restarted.moveHeld(id) }) + if mc.t.moveHeld(id) { + t.Error("the target still holds the reservation of a move the restarted source aborted") + } +} + func TestMoveTargetCountsAReservationLikeAClaim(t *testing.T) { mc := newMoveCluster(t) tuneView(mc.t, func(v *configView) { v.maxClaims = 1 }) @@ -602,48 +643,28 @@ func TestMoveRestartAfterTheConsumeKeepsTheClaim(t *testing.T) { } func TestMoveHoldsOffCapturesReleasesAndEvacuationWhileItRuns(t *testing.T) { + for name, archived := range map[string]bool{"archives": true, "no archives": false} { + t.Run(name, func(t *testing.T) { testMoveHoldsOffCallers(t, archived) }) + } +} + +func TestMoveToANodeThatStartedDrainingAborts(t *testing.T) { mc := newMoveCluster(t) sb := mustClaim(t, mc.s, testKey) - other := mustClaim(t, mc.s, testKey) - cred := Cred{Token: sb.Token} mc.p.roundHook = func(round int) error { - if round != 2 { - return nil - } - _, forkErr := mc.s.Fork(t.Context(), sb.ID, cred, 1, time.Minute, "", "") - _, ckErr := mc.s.Checkpoint(t.Context(), sb.ID, cred, "ck", "") - _, _, promoteErr := mc.s.Promote(t.Context(), sb.ID, cred, "tpl", "") - for op, err := range map[string]error{ - "fork": forkErr, "checkpoint": ckErr, "promote": promoteErr, - "hibernate": mc.s.Hibernate(t.Context(), sb.ID, cred), - "release": mc.s.Release(t.Context(), sb.ID, cred), - } { - if !errors.Is(err, ErrEvacuating) { - t.Errorf("%s during a move: %v, want ErrEvacuating", op, err) - } - } - mc.s.mu.Lock() - replicable := mc.s.replicable(sb) - mc.s.mu.Unlock() - if replicable { - t.Error("a replica round may capture the moving claim") - } - expire(mc.s, sb) - mc.s.reapOnce(t.Context()) - if _, held := mc.s.byID(sb.ID); !held { - t.Error("the reap took the moving claim") - } - tuneView(mc.s, func(v *configView) { v.moveConcurrency = 1 }) - if _, err := mc.s.Move(t.Context(), other.ID, "b", time.Minute); !errors.Is(err, ErrQuota) { - t.Errorf("a second move past a lowered move_concurrency: %v, want ErrQuota", err) + if round == 2 { + mc.t.Drain(t.Context()) } return nil } - if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Minute); err != nil { - t.Fatalf("the move the lowered budget did not stop: %v", err) + if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Minute); !errors.Is(err, ErrNotMovable) { + t.Fatalf("move to a draining node: %v, want ErrNotMovable", err) } - if _, held := mc.t.byID(sb.ID); !held { - t.Error("the target does not hold the moved claim") + if _, held := mc.t.byID(sb.ID); held { + t.Error("the draining node took the claim") + } + if _, held := mc.s.byID(sb.ID); !held { + t.Error("the source dropped the claim the target refused") } } @@ -898,6 +919,14 @@ func TestMoveRefusesClaimsThatCannotMove(t *testing.T) { if len(fakeOf(mc.s).captures) != 0 { t.Errorf("a refused move captured %v", fakeOf(mc.s).captures) } + fakeOf(mc.s).migrateSupportErr = nil + fakeOf(mc.t).migrateSupportErr = errors.New("cocoon too old") + if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Minute); !errors.Is(err, ErrNotMovable) { + t.Errorf("move to a cocoon without migration: %v, want ErrNotMovable", err) + } + if len(mc.p.roundBudgets) != 0 { + t.Errorf("a move the target refused sent %d rounds", len(mc.p.roundBudgets)) + } } func TestMoveRestartResolvesAJournaledMoveThroughItsRecord(t *testing.T) { @@ -917,31 +946,40 @@ func TestMoveRestartResolvesAJournaledMoveThroughItsRecord(t *testing.T) { t.Errorf("starts %v removes %v, want the received VM started", eng.starts, eng.removes) } } + coldRestarted := func(t *testing.T, eng *fakeEngine) { + if len(eng.starts) != 1 || slices.Contains(eng.removes, vm) { + t.Errorf("starts %v removes %v, want one cold boot by liveness", eng.starts, eng.removes) + } + } unstarted := func(t *testing.T, eng *fakeEngine) { if slices.Contains(eng.starts, vm) || !slices.Contains(eng.removes, vm) { t.Errorf("starts %v removes %v, want the received VM removed unstarted", eng.starts, eng.removes) } } for name, tt := range map[string]struct { - node string - from, to string - reach moves.State - wantClaim bool - wantRecord moves.State - check func(t *testing.T, eng *fakeEngine) + node string + from, to string + reach moves.State + wantClaim bool + wantRecord moves.State + vmmLost bool + inspectFails int + check func(t *testing.T, eng *fakeEngine) }{ - "source at pending": {node: "a", from: "a", to: "b", reach: moves.Pending, wantClaim: true, check: aborted}, - "source at final": {node: "a", from: "a", to: "b", reach: moves.Final, wantClaim: true, check: aborted}, - "source at aborted": {node: "a", from: "a", to: "b", reach: moves.Aborted, wantClaim: true, check: aborted}, - "source at committed": {node: "a", from: "a", to: "b", reach: moves.Committed, wantRecord: moves.Committed, check: movedOn}, - "source at started": {node: "a", from: "a", to: "b", reach: moves.Started, check: movedOn}, - "source, record missing": {node: "a", check: movedOn}, - "source, another node moved": {node: "a", from: "b", to: "c", reach: moves.Pending, wantRecord: moves.Pending, check: movedOn}, - "target at final": {node: "b", from: "a", to: "b", reach: moves.Final, wantRecord: moves.Aborted, check: unstarted}, - "target at committed": {node: "b", from: "a", to: "b", reach: moves.Committed, wantClaim: true, wantRecord: moves.Started, check: started}, - "target at started": {node: "b", from: "a", to: "b", reach: moves.Started, wantClaim: true, wantRecord: moves.Started, check: started}, - "target at aborted": {node: "b", from: "a", to: "b", reach: moves.Aborted, wantRecord: moves.Aborted, check: unstarted}, - "target, record missing": {node: "b", check: unstarted}, + "source at pending": {node: "a", from: "a", to: "b", reach: moves.Pending, wantClaim: true, check: aborted}, + "source at final": {node: "a", from: "a", to: "b", reach: moves.Final, wantClaim: true, check: aborted}, + "source at aborted": {node: "a", from: "a", to: "b", reach: moves.Aborted, wantClaim: true, check: aborted}, + "source at committed": {node: "a", from: "a", to: "b", reach: moves.Committed, wantRecord: moves.Committed, check: movedOn}, + "source at started": {node: "a", from: "a", to: "b", reach: moves.Started, check: movedOn}, + "source, record missing": {node: "a", check: movedOn}, + "source, another node moved": {node: "a", from: "b", to: "c", reach: moves.Pending, wantRecord: moves.Pending, check: movedOn}, + "target at final": {node: "b", from: "a", to: "b", reach: moves.Final, wantRecord: moves.Aborted, check: unstarted}, + "target at committed": {node: "b", from: "a", to: "b", reach: moves.Committed, wantClaim: true, wantRecord: moves.Started, check: started}, + "target at started": {node: "b", from: "a", to: "b", reach: moves.Started, wantClaim: true, wantRecord: moves.Started, check: started}, + "target at committed, VMM lost with the host": {node: "b", from: "a", to: "b", reach: moves.Committed, wantClaim: true, wantRecord: moves.Started, vmmLost: true, check: coldRestarted}, + "target at committed, VMM lost, first inspect fails": {node: "b", from: "a", to: "b", reach: moves.Committed, wantClaim: true, wantRecord: moves.Started, vmmLost: true, inspectFails: 1, check: coldRestarted}, + "target at aborted": {node: "b", from: "a", to: "b", reach: moves.Aborted, wantRecord: moves.Aborted, check: unstarted}, + "target, record missing": {node: "b", check: unstarted}, } { t.Run(name, func(t *testing.T) { c := newFailoverCluster(t) @@ -964,8 +1002,8 @@ func TestMoveRestartResolvesAJournaledMoveThroughItsRecord(t *testing.T) { t.Fatal(err) } eng := newFakeEngine() - eng.vms[vm], eng.stopped[vm] = "/vsock/"+vm, sb.MoveIn - eng.migrateSeries[vm] = series + eng.vms[vm], eng.paused[vm], eng.stopped[vm] = "/vsock/"+vm, sb.MoveIn, tt.vmmLost + eng.migrateSeries[vm], eng.inspectFails = series, tt.inspectFails m := c.nodeOn(t, eng, tt.node, dataDir, keepAnHour) m.liftFence() if err := m.Reconcile(ctx); err != nil { @@ -983,6 +1021,14 @@ func TestMoveRestartResolvesAJournaledMoveThroughItsRecord(t *testing.T) { if ok && (held.MoveOut != nil || held.MoveIn) { t.Errorf("claim journal out %+v in %+v, want both cleared", held.MoveOut, held.MoveIn) } + if tt.vmmLost { + m.livenessOnce(ctx) + waitFor(t, func() bool { + m.mu.Lock() + defer m.mu.Unlock() + return held.Restarts == 1 && !m.sweeps.liveness.Load() + }) + } eng.mu.Lock() defer eng.mu.Unlock() tt.check(t, eng) @@ -1392,6 +1438,73 @@ func BenchmarkQuotaErr(b *testing.B) { } } +func testMoveHoldsOffCallers(t *testing.T, archived bool) { + t.Helper() + mc := newMoveCluster(t) + if !archived { + mc.s.archives, mc.t.archives = nil, nil + } + sb := mustClaim(t, mc.s, testKey) + other := mustClaim(t, mc.s, testKey) + cred := Cred{Token: sb.Token} + mc.p.roundHook = func(round int) error { + if round != 2 { + return nil + } + _, forkErr := mc.s.Fork(t.Context(), sb.ID, cred, 1, time.Minute, "", "") + _, ckErr := mc.s.Checkpoint(t.Context(), sb.ID, cred, "ck", "") + _, _, promoteErr := mc.s.Promote(t.Context(), sb.ID, cred, "tpl", "") + for op, err := range map[string]error{ + "fork": forkErr, "checkpoint": ckErr, "promote": promoteErr, + "hibernate": mc.s.Hibernate(t.Context(), sb.ID, cred), + "release": mc.s.Release(t.Context(), sb.ID, cred), + } { + if !errors.Is(err, ErrEvacuating) { + t.Errorf("%s during a move: %v, want ErrEvacuating", op, err) + } + } + mc.s.mu.Lock() + replicable := mc.s.replicable(sb) + mc.s.mu.Unlock() + if replicable { + t.Error("a replica round may capture the moving claim") + } + expire(mc.s, sb) + mc.s.reapOnce(t.Context()) + if _, held := mc.s.byID(sb.ID); !held { + t.Error("the reap took the moving claim") + } + tuneView(mc.s, func(v *configView) { v.moveConcurrency = 1 }) + if _, err := mc.s.Move(t.Context(), other.ID, "b", time.Minute); !errors.Is(err, ErrQuota) { + t.Errorf("a second move past a lowered move_concurrency: %v, want ErrQuota", err) + } + wantEvac := errArchiveBusy + if !archived { + wantEvac = errNoFailover + } + mc.s.mu.Lock() + mc.s.evac = &evacuation{pending: map[string]*types.Sandbox{sb.ID: sb}, stop: func() {}} + mc.s.mu.Unlock() + evacErr := mc.s.evacuateClaim(t.Context(), sb) + mc.s.mu.Lock() + mc.s.evac = nil + mc.s.mu.Unlock() + if !errors.Is(evacErr, wantEvac) { + t.Errorf("evacuation of the moving claim: %v, want %v", evacErr, wantEvac) + } + if _, held := mc.s.byID(sb.ID); !held || sb.ArchiveCk != "" { + t.Error("the evacuation took the moving claim") + } + return nil + } + if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Minute); err != nil { + t.Fatalf("the move the lowered budget did not stop: %v", err) + } + if _, held := mc.t.byID(sb.ID); !held { + t.Error("the target does not hold the moved claim") + } +} + type moveCluster struct { *failoverCluster moveReg *movestest.Mem @@ -1711,14 +1824,18 @@ func failRound(n int) func(int) error { } } -func abortRecord(mc *moveCluster) { - ctx := context.Background() - for _, rec := range mustStale(mc) { - _, _ = mc.moveReg.Advance(ctx, rec.ID, rec.Epoch, rec.State, moves.Aborted) +func abortRecord(t *testing.T, mc *moveCluster) { + t.Helper() + for _, rec := range mustStale(t, mc) { + _, _ = mc.moveReg.Advance(t.Context(), rec.ID, rec.Epoch, rec.State, moves.Aborted) } } -func mustStale(mc *moveCluster) []moves.Record { - recs, _ := mc.moveReg.Stale(context.Background(), time.Now().Add(time.Hour)) +func mustStale(t *testing.T, mc *moveCluster) []moves.Record { + t.Helper() + recs, err := mc.moveReg.Stale(t.Context(), time.Now().Add(time.Hour)) + if err != nil { + t.Errorf("stale move records: %v", err) + } return recs } diff --git a/sandboxd/pool/operator.go b/sandboxd/pool/operator.go index 231a6091..931f6a58 100644 --- a/sandboxd/pool/operator.go +++ b/sandboxd/pool/operator.go @@ -39,7 +39,9 @@ func (m *Manager) Wake(ctx context.Context, id string, cred Cred) error { if err := m.restartFailed(ctx, sb); err != nil { return err } - return m.wake(ctx, sb) + sb.Touch() + _, err := m.wakeResolved(ctx, sb) + return err } // SetInstanceMetadata replaces a running sandbox's instance-metadata document; a paused or archived one is refused, not woken. @@ -93,9 +95,3 @@ func (m *Manager) authed(id, token string) (*types.Sandbox, bool) { } return sb, true } - -func (m *Manager) wake(ctx context.Context, sb *types.Sandbox) error { - sb.Touch() - _, err := m.wakeResolved(ctx, sb) - return err -} diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 9319f26c..4175ed2b 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -63,8 +63,9 @@ const ( refillFailStreak = 8 refillBackoffBase = 250 * time.Millisecond refillBackoffMax = buildRetryDelay - // Fallbacks when a Manager is built from a Config that skipped config.Load's defaulting - defaultRefill = 4 + // Fallback when a Manager is built from a Config that skipped config.Load's defaulting + defaultRefill = 4 + defaultMoveConcurrency = 2 // A heal pulls a whole guest memory image, so a few saturate disk and NIC. @@ -230,7 +231,7 @@ type Gauges struct { 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 marks refill parked by a refused VM or a low disk or memory reserve. AtCapacity bool AtCapacityReason string MemoryLow bool diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index 33e02f9f..dc258622 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -892,9 +892,6 @@ func TestHeldConnForwardsCloseWrite(t *testing.T) { inner := &closeWriteConn{} held := &heldConn{Conn: inner, release: func() {}} - if _, ok := any(held).(interface{ CloseWrite() error }); !ok { - t.Fatal("heldConn does not expose CloseWrite; a relay half-close would be dropped") - } if err := held.CloseWrite(); err != nil { t.Fatalf("CloseWrite: %v", err) } @@ -1316,6 +1313,7 @@ type fakeEngine struct { liveVolumeSupportErr error liveVolumeHook func(string) error inspectHook func() + inspectFails int startedReaders []string attachedCopies []string @@ -1436,7 +1434,7 @@ func (f *fakeEngine) Start(_ context.Context, name string) error { if _, ok := f.vms[name]; !ok { return fmt.Errorf("vm %s not found", name) } - f.stopped[name] = false + f.stopped[name], f.paused[name] = false, false return nil } @@ -1636,10 +1634,17 @@ func (f *fakeEngine) Inspect(ctx context.Context, name string) (types.VMRecord, f.inspects++ hook := f.inspectHook f.inspectHook = nil + failing := f.inspectFails > 0 + if failing { + f.inspectFails-- + } f.mu.Unlock() if hook != nil { hook() } + if failing { + return types.VMRecord{}, false, errors.New("cocoon unreachable") + } vms, _ := f.list(ctx) i := slices.IndexFunc(vms, func(vm types.VMRecord) bool { return vm.Config.Name == name }) if i < 0 { @@ -1933,7 +1938,7 @@ func (f *fakeEngine) MigrateReceive(_ context.Context, fromDir, name string, _ t if _, err := os.Stat(filepath.Join(fromDir, "state.json")); err != nil { return types.VMRecord{}, err } - f.vms[name], f.stopped[name] = f.sockRoot+"/"+name, true + f.vms[name], f.paused[name] = f.sockRoot+"/"+name, true return f.record(name), nil } @@ -2280,6 +2285,7 @@ func newTestManagerAt(t testing.TB, eng *fakeEngine, dataDir string, pools ...co } t.Cleanup(func() { assertRunningCount(t, m) }) t.Cleanup(func() { settleGoldenBuilds(m) }) + t.Cleanup(func() { settleCaptures(m) }) t.Cleanup(func() { m.mu.Lock() defer m.mu.Unlock() @@ -2294,6 +2300,26 @@ func newTestManagerAt(t testing.TB, eng *fakeEngine, dataDir string, pools ...co return m } +func settleCaptures(m *Manager) { + start := time.Now() + for deadline := start.Add(3 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { + m.mu.Lock() + claims := slices.Collect(maps.Values(m.claimed)) + intent := slices.ContainsFunc(claims, func(sb *types.Sandbox) bool { return sb.PendingSnap != "" || sb.SaveSnap != "" }) + m.mu.Unlock() + busy := slices.ContainsFunc(claims, func(sb *types.Sandbox) bool { + if !sb.Transition.TryLock() { + return true + } + sb.Transition.Unlock() + return false + }) + if !busy && (!intent || time.Since(start) > 300*time.Millisecond) { + return + } + } +} + func settleGoldenBuilds(m *Manager) { for deadline := time.Now().Add(3 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { m.mu.Lock() diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index 0ea84f96..68a2e368 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -65,7 +65,10 @@ func (m *Manager) Reconcile(ctx context.Context) error { case ok && sb.HibernateSnap != "": // Hibernated: the VM is stopped by design and wakes on demand. sb.PendingSnap = "" - case ok && sb.PendingSnap != "" && (snapsErr != nil || slices.Contains(snaps, sb.PendingSnap)): + case ok && sb.PendingSnap != "" && snapsErr != nil: + referenced[sb.PendingSnap] = true + interrupted = append(interrupted, sb) + case ok && sb.PendingSnap != "" && slices.Contains(snaps, sb.PendingSnap): // The journaled intent names the wake image, so adopt it when the snapshot is there sb.HibernateSnap, sb.PendingSnap, sb.ReaderSafe = sb.PendingSnap, "", false adopted = append(adopted, sb) @@ -128,7 +131,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { } func (m *Manager) sweepOrphanGoldens(ctx context.Context, pooled map[string]bool) { - logger := log.WithFunc("pool.Reconcile") + logger := log.WithFunc("pool.sweepOrphanGoldens") dir := m.goldensDir() goldens, _ := os.ReadDir(dir) for _, e := range goldens { @@ -143,7 +146,7 @@ func (m *Manager) sweepOrphanGoldens(ctx context.Context, pooled map[string]bool } func (m *Manager) sweepOrphanSnaps(ctx context.Context, snaps []string, snapsErr error, referenced map[string]bool) { - logger := log.WithFunc("pool.Reconcile") + logger := log.WithFunc("pool.sweepOrphanSnaps") // An unreferenced hibernate snapshot is an orphan; fork and golden snapshots never span a restart if snapsErr != nil { logger.Warnf(ctx, "snapshot sweep skipped: %v", snapsErr) @@ -225,7 +228,6 @@ func (m *Manager) removeStaleVM(ctx context.Context, name string, rec types.VMRe // A canceled ctx must not skip the busy check guarding the forced remove switch outcome, err := m.eng.ReconcileStaleCreate(context.WithoutCancel(ctx), name); { case err != nil: - // Verb missing (cocoon < v0.5.8) or failed: keep the old sweep. logger.Warnf(ctx, "reconcile stale create %s: %v; removing", name, err) case outcome == engine.StaleCreateCollected, outcome == engine.StaleCreateNotFound: return true diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index 16667149..4e5a8848 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -92,7 +92,7 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { m.pendingCks[ckID] = struct{}{} m.mu.Unlock() defer m.untrack(m.pendingCks, ckID) - if _, _, err := m.publishCheckpoint(ctx, sb, ckID, "replica", sb.Tenant, true); err != nil { + if _, _, err := m.publishCheckpoint(ctx, sb, ckID, "replica", sb.Tenant, "", true); err != nil { return err } var prev string @@ -266,12 +266,6 @@ func (m *Manager) replicaRowed(sb *types.Sandbox) bool { return m.archives != nil && sb.ReplicaCk != "" && sb.ArchiveEpoch != 0 } -func (m *Manager) replicaHeld(sb *types.Sandbox) bool { - m.mu.Lock() - defer m.mu.Unlock() - return sb.HasReplica() -} - func (m *Manager) replicaOf(sb *types.Sandbox) string { m.mu.Lock() defer m.mu.Unlock() diff --git a/sandboxd/pool/replica_test.go b/sandboxd/pool/replica_test.go index 0a688516..7b46fb4a 100644 --- a/sandboxd/pool/replica_test.go +++ b/sandboxd/pool/replica_test.go @@ -809,7 +809,7 @@ func TestReplicaRestartAdoptsTheRowsCheckpointOfAnInterruptedRound(t *testing.T) if err := a.FlushClaims(); err != nil { t.Fatalf("flush: %v", err) } - interrupted, _, publishErr := a.publishCheckpoint(t.Context(), sb, store.CheckpointID(randHex(8)), "replica", sb.Tenant, true) + interrupted, _, publishErr := a.publishCheckpoint(t.Context(), sb, store.CheckpointID(randHex(8)), "replica", sb.Tenant, "", true) if publishErr != nil { t.Fatalf("publish: %v", publishErr) } diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index 7a0fa0e9..f7139ace 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -70,7 +70,7 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t return types.PoolKey{}, "", fmt.Errorf("stage template: %w", err) } defer func() { _ = os.RemoveAll(staging) }() - _, vmName, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) + _, vmName, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir), "") if err != nil { return types.PoolKey{}, "", fmt.Errorf("promote %s: %w", sb.ID, err) } diff --git a/sandboxd/server/move_test.go b/sandboxd/server/move_test.go index 50fadc4b..b97cf987 100644 --- a/sandboxd/server/move_test.go +++ b/sandboxd/server/move_test.go @@ -17,7 +17,7 @@ import ( func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) { var gotAddr string var gotBudget time.Duration - outcome := error(nil) + var outcome error mgr := &fakeManager{move: func(_, addr string, budget time.Duration) (pool.MoveResult, error) { gotAddr, gotBudget = addr, budget return pool.MoveResult{Rounds: 3, Downtime: 180 * time.Millisecond}, outcome @@ -34,24 +34,24 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) { t.Errorf("move to %s within %s answered %+v", gotAddr, gotBudget, body) } for _, tt := range []struct { - token, body string - err error - want int + body string + err error + want int }{ - {"root", `{"to":"10.0.0.9:7777"}`, nil, http.StatusBadRequest}, - {"root", `{"to":"10.0.0.2:7777","max_downtime_ms":-1}`, nil, http.StatusBadRequest}, - {"root", `{"to":"10.0.0.2:7777","max_downtime_ms":400}`, nil, http.StatusBadRequest}, - {"root", `{"to":"10.0.0.2:7777","max_downtime_ms":401}`, nil, http.StatusOK}, - {"root", `{"to":"10.0.0.2:7777"}`, pool.ErrUnknownSandbox, http.StatusNotFound}, - {"root", `{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: it has volumes", pool.ErrNotMovable), http.StatusConflict}, - {"root", `{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: target: 3 moves in flight", pool.ErrQuota), http.StatusTooManyRequests}, - {"root", `{"to":"10.0.0.2:7777"}`, pool.ErrMoveTarget, http.StatusBadGateway}, - {"root", `{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: %w", pool.ErrMoveStoreDown, pool.ErrMoveTarget), http.StatusServiceUnavailable}, - {"root", `{"to":"10.0.0.2:7777"}`, pool.ErrMoveDowntime, http.StatusServiceUnavailable}, - {"root", `{"to":"10.0.0.2:7777"}`, pool.ErrLeaseLost, http.StatusServiceUnavailable}, + {`{"to":"10.0.0.9:7777"}`, nil, http.StatusBadRequest}, + {`{"to":"10.0.0.2:7777","max_downtime_ms":-1}`, nil, http.StatusBadRequest}, + {`{"to":"10.0.0.2:7777","max_downtime_ms":400}`, nil, http.StatusBadRequest}, + {`{"to":"10.0.0.2:7777","max_downtime_ms":401}`, nil, http.StatusOK}, + {`{"to":"10.0.0.2:7777"}`, pool.ErrUnknownSandbox, http.StatusNotFound}, + {`{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: it has volumes", pool.ErrNotMovable), http.StatusConflict}, + {`{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: target: 3 moves in flight", pool.ErrQuota), http.StatusTooManyRequests}, + {`{"to":"10.0.0.2:7777"}`, pool.ErrMoveTarget, http.StatusBadGateway}, + {`{"to":"10.0.0.2:7777"}`, fmt.Errorf("%w: %w", pool.ErrMoveStoreDown, pool.ErrMoveTarget), http.StatusServiceUnavailable}, + {`{"to":"10.0.0.2:7777"}`, pool.ErrMoveDowntime, http.StatusServiceUnavailable}, + {`{"to":"10.0.0.2:7777"}`, pool.ErrLeaseLost, http.StatusServiceUnavailable}, } { outcome = tt.err - resp := doReq(t, http.MethodPost, srv.URL+"/v1/sandboxes/sb_1/move", tt.token, tt.body) + resp := doReq(t, http.MethodPost, srv.URL+"/v1/sandboxes/sb_1/move", "root", tt.body) _ = resp.Body.Close() if resp.StatusCode != tt.want { t.Errorf("move %s with %v: %d, want %d", tt.body, tt.err, resp.StatusCode, tt.want) diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 3e7fbac3..2ba424aa 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -1,4 +1,4 @@ -// Package server exposes the v0 control plane: claim/release/info over HTTP/JSON, plus the data plane — an HTTP Upgrade relayed byte-for-byte between the client and the guest's silkd vsock port. +// Package server exposes the /v1 control plane over HTTP/JSON, plus the data plane — an HTTP Upgrade relayed byte-for-byte between the client and the guest's silkd vsock port. package server import ( diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index e690a9a6..d883a3e6 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -480,17 +480,20 @@ func TestSandboxByIDCarriesTheTokenOnlyForTheRootCredential(t *testing.T) { }} for _, tt := range []struct { name, apiToken, bearer, want string + status int + specs []tenants.TenantSpec }{ - {"open node", "", "", ""}, - {"root", "root", "root", "victim"}, + {name: "open node", status: http.StatusOK}, + {name: "root", apiToken: "root", bearer: "root", want: "victim", status: http.StatusOK}, + {name: "tenant", apiToken: "root", bearer: "acme-tok", status: http.StatusForbidden, specs: []tenants.TenantSpec{{Name: "acme", Token: "acme-tok"}}}, } { t.Run(tt.name, func(t *testing.T) { - ts := newTestServer(t, tt.apiToken, mgr, nil) + ts := newTenantTestServer(t, tt.apiToken, tt.specs, mgr, nil) resp := doReq(t, http.MethodGet, ts.URL+"/v1/sandboxes/sb_1", tt.bearer, "") defer resp.Body.Close() got := decodeJSON[pool.SandboxSummary](t, resp.Body) - if resp.StatusCode != http.StatusOK || got.Token != tt.want { - t.Errorf("status %d token %q, want 200 and %q", resp.StatusCode, got.Token, tt.want) + if resp.StatusCode != tt.status || got.Token != tt.want { + t.Errorf("status %d token %q, want %d and %q", resp.StatusCode, got.Token, tt.status, tt.want) } }) } @@ -793,6 +796,7 @@ func TestReleaseAndHibernateMapEachOutcomeToItsStatus(t *testing.T) { }{ {"release", func(f *fakeManager, h sandboxVerbFunc) { f.release = h }}, {"hibernate", func(f *fakeManager, h sandboxVerbFunc) { f.hibernate = h }}, + {"wake", func(f *fakeManager, h sandboxVerbFunc) { f.wake = h }}, } tests := []struct { name string @@ -896,11 +900,7 @@ func TestAgentErrorPaths(t *testing.T) { t.Run(tt.name, func(t *testing.T) { mgr := &fakeManager{socket: func(string, string) (string, error) { return "/v/sock", tt.sockErr }} dialer := &fakeDialer{dial: func(context.Context, string) (net.Conn, error) { - if tt.dialErr != nil { - return nil, tt.dialErr - } - c, _ := net.Pipe() - return c, nil + return nil, tt.dialErr }} ts := newTestServer(t, "", mgr, dialer) req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, ts.URL+"/v1/sandboxes/sb_1/agent", nil) @@ -2828,12 +2828,7 @@ func postJSON(t *testing.T, url, token, body string) *http.Response { func scrapeMetrics(t *testing.T, base string) string { t.Helper() - resp, err := http.Get(base + "/metrics") - if err != nil { - t.Fatalf("metrics: %v", err) - } - defer resp.Body.Close() - raw, _ := io.ReadAll(resp.Body) + _, raw := bodyOf(t, http.MethodGet, base+"/metrics", "", "") return string(raw) } diff --git a/sandboxd/store/peer/broadcast.go b/sandboxd/store/peer/broadcast.go index 8ca2915f..11ba0c6d 100644 --- a/sandboxd/store/peer/broadcast.go +++ b/sandboxd/store/peer/broadcast.go @@ -27,7 +27,7 @@ func (b *Broadcaster) Delete(ctx context.Context, id string) { var wg sync.WaitGroup for _, addr := range addrs { wg.Go(func() { - // 404 means the peer never healed a copy — not a failure worth a retry. + // a 404 means the peer never healed a copy if err := peerCall(ctx, http.MethodDelete, checkpointURL(addr, id, "?no_forward=1"), b.Token, nil, peerCallTimeout, http.StatusNoContent, http.StatusNotFound); err != nil { logger.Warnf(ctx, "broadcast delete %s to %s: %v", id, addr, err) } diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index b208cdb7..febb9aa1 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -1,4 +1,4 @@ -// Package peer pulls a record from the node that holds it and finds, heals and deletes records across peers. +// Package peer moves records and live-move rounds between nodes: it pulls, finds, heals, deletes and adopts them across peers. package peer import ( @@ -21,7 +21,7 @@ import ( ) const ( - maxRecordBytes = 1 << 40 // 1 TiB + maxRecordBytes = 1 << 40 exportPrefix = store.ExportDir + "/" @@ -115,7 +115,7 @@ func Untar(r io.Reader, dst string) error { return fmt.Errorf("untar: %w", err) } if hdr.Name == recordTrailer { - complete = true // a protocol marker, never written to disk + complete = true continue } target, err := safeJoin(dst, hdr.Name) diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index c86cddb3..999b189a 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -459,10 +459,12 @@ func (r VMRecord) TapDevice() string { return r.NetworkConfigs[0].TAP } -// MoveOut is the source's journal of a live move: the record's first epoch and cocoon's series. +// MoveOut is the source's journal of a live move: the record's first epoch, cocoon's series and the target. type MoveOut struct { Epoch int64 `json:"epoch"` Series string `json:"series,omitempty"` + Addr string `json:"addr,omitempty"` + To string `json:"to,omitempty"` } // MigrateFile is one file a migrate round wrote, named relative to the round's Dir.