diff --git a/docs/deploy.md b/docs/deploy.md index f8020f0e5..bcab7cd46 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -119,7 +119,8 @@ gets the intersection of the two. | `advertise_addr` | = `listen` | internal HTTP host:port for peer traffic and preview forwarding; also used by clients when `client_advertise` is unset. Must be routable when `listen` is a wildcard; a node with `mesh` set refuses to load while it names an unspecified host | | `client_advertise` | unset | client-facing HTTP(S) origin for this node, e.g. `https://node-a.sandbox.example.com`; published in owner, redirect, and peer responses. No path, query, fragment, or userinfo | | `bridges` / `networks` | unset | egress-lane attachment: a list of host bridge devices, or a list of CNI conflist names. Mutually exclusive; with neither set the node serves only the no-network lane. A Linux bridge holds at most 1024 ports (kernel `BR_MAX_PORTS`), so an N-entry list raises the node's egress ceiling to N×1024 — VMs spread over the list by a stable hash of the VM name, so size it with headroom (the spread is statistical, not exact). `bridges` keeps the raw TAP-on-bridge attachment (taps in the root netns, no per-VM network namespace or CNI plugin execution); `networks` runs the CNI chain per VM. [Guarded egress](egress.md) on the egress lane (an egress-lane pool policy or any tenant policy) needs `bridges` and rejects a CNI network at load; none-lane pool policies ride the proxy on either | -| `volumes` | unset | node-local catalog of operator-managed dataset images: `[ {"name":"imagenet","path":"/srv/datasets/imagenet.img","directio":"off","tenants":["acme"]}, {"name":"scratch-db","path":"/srv/datasets/scratch.img","writable":true} ]`. Names match `^[a-z][a-z0-9_-]{0,19}$` and cannot start with `cocoon-`; paths are absolute; `directio` is `on`, `off`, or `auto` and defaults to `off` for both read-only and writable entries. `tenants` is an optional access list: empty means every authenticated scope, and a listed name the tenant set does not hold grants nothing and logs a warning at boot (the [tenant API](sandboxd-api.md#tenants-v1tenants) may add it later); root always has access. `writable` (default `false`) lets a claim request `mode: "rw"` on that entry — see [Dataset volumes](#dataset-volumes). The catalog is intentionally not part of the cluster digest | +| `volumes` | unset | node-local catalog of operator-managed dataset images: `[ {"name":"imagenet","path":"/srv/datasets/imagenet.img","directio":"off","tenants":["acme"]}, {"name":"scratch-db","path":"/srv/datasets/scratch.img","writable":true} ]`. Names match `^[a-z][a-z0-9_-]{0,19}$` and cannot start with `cocoon-`; paths are absolute; `directio` is `on`, `off`, or `auto` and defaults to `off` for both read-only and writable entries. `tenants` is an optional [access list](sandboxd-api.md#volume-access-lists); a listed name the tenant set does not hold grants nothing and logs a warning at boot (the [tenant API](sandboxd-api.md#tenants-v1tenants) may add it later). `writable` (default `false`) lets a claim request `mode: "rw"` on that entry — see [Dataset volumes](#dataset-volumes). The catalog is intentionally not part of the cluster digest | +| `volume_roots` | unset | absolute directories that bound [runtime registration](sandboxd-api.md#put-v1volumesname): the registered path, once cleaned, must equal a root or sit under one, or the PUT returns 400. The check is lexical: `..` is resolved but symlinks are not followed, so a symlink under a root can point outside it. Static `volumes` entries are not checked. A saved registration outside the roots prevents startup. Unregister it before narrowing the roots, or widen the roots to include it. DELETE always rewrites the catalog, including for an absent name. Keep `data_dir` outside every root. Unset or an empty list accepts any absolute path | | `secrets` | unset | credentials the egress proxy injects by name: `[{"name": "gh", "header": "Authorization", "value_env": "GH_TOKEN"}]`. A pool or egress-class rule references the name (on a claim where both apply, only the pool rule's secret is injected); the value comes from the claim's `guest: false` env, then the node's environment, never this file. Without `value_env` the secret's name is the env to read, and only claims supply it. See [egress](egress.md#claim-env-and-secrets) | | `egress_internal_allow` | unset | CIDR prefixes re-admitted through the egress proxy's SSRF guard, node-wide (every pool and tenant); `prefix:port,port` scopes an entry to those ports, a bare prefix admits every port. Prefer the port form. Prefixes, not a permit-private switch — the guest bridges are themselves ULA/RFC1918. See [egress](egress.md) | | `egress_upstream` | unset | per-claim upstream proxies: `{"claim_env": "EGRESS_UPSTREAM", "allow": ["res.example.com", "10.1.0.0/16"]}`. A claim's `guest: false` entry under `claim_env` (`http://` or `socks5://`, optional `user:pass@`, or `direct`) routes its egress through that upstream, which must be on `allow` (host names, IPs, CIDR prefixes). See [upstream proxies](egress.md#upstream-proxies) | @@ -323,10 +324,9 @@ Warm candidates retain their normal ranking, but a volume claim's candidate set, promoted-template intersection, and redirect follow the fleet-wide rule in [cluster](cluster.md#volumes-and-placement). -An empty catalog `tenants` list allows every authenticated scope; a nonempty -list limits the image to those tenants, while root always bypasses it. A -listed tenant the set no longer holds grants nothing and logs a warning at -boot. +Catalog `tenants` lists follow the +[access-list rule](sandboxd-api.md#volume-access-lists). A listed tenant the +set no longer holds grants nothing and logs a warning at boot. `GET /v1/volumes` reports the caller-visible fleet union and holder count, plus the answering node's current local availability and whether the entry is `writable`, without exposing host paths or node addresses. Applied names, diff --git a/docs/live-volumes.md b/docs/live-volumes.md index 790e0d21e..43fa04cba 100644 --- a/docs/live-volumes.md +++ b/docs/live-volumes.md @@ -9,6 +9,13 @@ file. Keep each name bound to the same immutable image while held. Use Cocoon v0.7.2 or newer and set `SANDBOX_ADDR` and `SANDBOX_API_TOKEN` for the client. A sandbox token alone cannot manage volumes. +A configured node root token can +[register backing paths](sandboxd-api.md#put-v1volumesname) without editing +static config. Registration persists only node-local catalog metadata and does +not manage disk files. Runtime `tenants` lists follow the +[access-list rule](sandboxd-api.md#volume-access-lists). Live changes remain +limited to mounted read-only entries. + Pause application reads before replacement. Close files, mmap mappings, cwd, and bind mount references on the old disk. Replacement unmounts the old disk before mounting the new one, so the path has a short unavailable interval. diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 31a255cd7..e16ef79ac 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -8,8 +8,8 @@ All bodies are JSON. Three token kinds: preview mint) and the tenant-scoped listings/deletes below; everything a tenant creates is stamped with its name. Operator surfaces (the per-id sandbox reads, `PUT /v1/sandboxes/{id}/instance-metadata`, `GET /v1/info`, - `PUT /v1/pools`, `POST/DELETE /v1/drain`, `GET /metrics`, - `GET /v1/checkpoints/{id}/blob`) + `PUT /v1/pools`, `PUT/DELETE /v1/volumes/{name}`, `POST/DELETE /v1/drain`, + `GET /metrics`, `GET /v1/checkpoints/{id}/blob`) answer a tenant token `403` — authenticated but not authorized; an unknown token stays `401`. On a node with a shared [`meta_store`](deploy.md#shared-tenant-database), a tenant token the node @@ -255,6 +255,70 @@ list; the field is emitted (as `true`) only for a writable entry this node declares and omitted otherwise (peer-only rows never carry it), so a read-only entry's response is byte-identical to v1. +## PUT /v1/volumes/{name} + +Requires the configured root API token. A tenant token or a node without an +`api_token` returns 403. Sandbox, unknown, and missing tokens return 401 when +`api_token` is configured. The URL supplies the name. Unknown body fields, +duplicate fields, and trailing JSON return 400. + +```json +{"path": "/srv/datasets/data.img", "directio": "off", + "writable": false, "tenants": ["acme"]} +``` + +- `name` matches `[a-z][a-z0-9_-]{0,19}` and cannot start with `cocoon-`. +- `path` uses static config's absolute-path validation. When the node sets + [`volume_roots`](deploy.md#configuration), the cleaned path must equal or sit + under one of them. Registration does not open the backing or check its type + or availability. A missing backing keeps its registration but cannot be + attached. +- `directio` is `on`, `off`, or `auto`, and defaults to `off`. +- `writable` defaults to `false` and controls writable claim admission. +- `tenants` is the entry's [access list](#volume-access-lists). + +Success is 204. An identical normalized registration is idempotent. A different +binding, changed fields, or a static name or cleaned path collision returns 409. +Invalid fields return 400. GET keeps its existing projection without host paths +or tenant lists. Runtime entries support the existing live volume API. + +Persistence failure returns 503 with `Retry-After`. Retry the same request; +after a restart, either state can load. + +The node stores metadata in `data_dir/runtime-volumes.json` independently of +`sync_claims` and loads it before claim recovery. New records use mode 0600. +Malformed or conflicting metadata, including a saved path outside `volume_roots`, +prevents startup. The catalog is node-local with one sandboxd writer. + +Sandboxd does not create, check, or delete backings. Keep them outside its +cleanup directories: `/readers`, `/goldens`, and +`/fork-*`. + +### Volume access lists + +Static and runtime entries share one `tenants` rule. An omitted or empty list +permits every authenticated scope. A nonempty list permits only claims owned +by the listed tenants, and root-owned claims can use any entry. Live additions +check the claim owner's list even when root sends the request; held names +finish repair or removal without a new grant. Runtime registration rejects +tenant names that do not match the tenant API syntax. Static config preserves +its supplied tenant lists. An invalid static name matches no tenant, so a +nonempty list containing only invalid names permits no tenant. A valid name +for a tenant that does not exist yet grants nothing until the tenant is created. + +## DELETE /v1/volumes/{name} + +Auth and name validation match PUT. Deletes only runtime registration metadata. +Success, including an absent unreferenced name, is 204. Static names and names +referenced by a claim, an in-flight operation, or cleanup return 409. +DELETE always writes the absence, including when the name is already absent. +Persistence failure returns 503 with `Retry-After`. Retry the same DELETE, +even if GET no longer lists the name; a restart can restore the old registration. + +Before downgrading to a daemon without runtime registration, drain runtime +references or supply equivalent static volume config and finish pending live +volume changes. + ## GET /v1/sandboxes/{id}/volumes Returns the persisted desired set (`volumes_want`), confirmed applied set @@ -277,7 +341,8 @@ when the volume is in both sets, `pending` when desired but not yet confirmed, or `detaching` when applied but no longer desired. An interrupted device call can leave the actual guest state different from the confirmed set. A failed PUT or PATCH adds a top-level `error` naming the refusal, and the affected -state may carry its own `error`. `converging` is true only while a request worker is running. +state may carry its own `error`. `converging` is true while a request worker +is running. `pending` with `converging: false` requires an explicit retry; there is no background convergence. GET retains the last safe per-volume error for the current daemon lifetime. After daemon restart, unfinished states report an diff --git a/docs/security.md b/docs/security.md index 6c519f498..0cd239c7d 100644 --- a/docs/security.md +++ b/docs/security.md @@ -94,15 +94,25 @@ rather than confirming existence, and operator surfaces answer tenants 403. The volume catalog is an operator-owned data boundary. A volume name and its access list must mean the same thing fleet-wide, although membership is -node-local. An empty entry `tenants` list permits every authenticated scope; a -nonempty list permits only those named tenants, while the root token always -has access. Config load rejects an access-list name that is not a configured -tenant. Claim lookup returns byte-identical errors for an unknown and a -forbidden name, and catalog discovery filters before replying, so a tenant -cannot enumerate restricted entries by probing. Gossip carries names only. +node-local. Entry `tenants` lists follow the [access-list +rule](sandboxd-api.md#volume-access-lists). Claim lookup returns byte-identical +errors for an unknown and a forbidden name, and catalog discovery filters before +replying, so a tenant cannot enumerate restricted entries by probing. Gossip +carries names only. Neither gossip, discovery, persisted claims, usage events, nor the sandbox index exposes host image paths or access lists. +Runtime [registration and unregistration](sandboxd-api.md#put-v1volumesname) +require a configured root `api_token`; +these endpoints return 403 when the token is unset. New catalog records use +mode 0600. Reads and replacements use the daemon's filesystem permissions. +Registration uses static configuration's path validation plus, when set, the +[`volume_roots`](deploy.md#configuration) bound. It does not verify backing type +or pin backing identity. Without +`volume_roots` the root token can register any absolute host path, writable if +requested, for every scope its [access list](sandboxd-api.md#volume-access-lists) +permits. Protect the root token like host root, or set `volume_roots`. + Read-only is integrity protection for the shared image, not confidentiality. Mounting a dataset into an egress-lane sandbox gives that sandbox an export path to every destination its tenant egress policy permits; review the volume access @@ -116,8 +126,8 @@ immutable and publish a new name/path for new content. Changing a running claim's volumes needs the root token or the owning tenant's API token. A sandbox token cannot, so a guest holding its own sandbox token -cannot attach another dataset, and every addition passes the same access list -as a claim. +cannot attach another dataset, and every addition passes the same +[access list](sandboxd-api.md#volume-access-lists) as a claim. A writable entry (`writable: true`) adds a channel the read-only model does not have: whichever tenant is permitted to claim it `rw` changes what every diff --git a/e2e/volume_catalog_test.go b/e2e/volume_catalog_test.go new file mode 100644 index 000000000..3858f3e5c --- /dev/null +++ b/e2e/volume_catalog_test.go @@ -0,0 +1,74 @@ +package e2e + +import ( + "encoding/json/v2" + "net/http" + "slices" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/tenants" + sandbox "github.com/cocoonstack/sandbox/sdk/go" +) + +func TestRuntimeVolumeRegistrationThroughHTTP(t *testing.T) { + st := startTenantStack(t, "node-token", []tenants.TenantSpec{ + {Name: "acme", Token: "acme-token"}, {Name: "beta", Token: "beta-token"}, + }, []config.VolumeSpec{{Name: "public", Path: writeVolumeImage(t, "public.img", "shared")}}) + for name, body := range map[string]map[string]any{ + "home": {"path": writeVolumeImage(t, "home.img", "shared"), "writable": true}, + "private": {"path": writeVolumeImage(t, "private.img", "private"), "tenants": []string{"acme"}}, + } { + raw, err := json.Marshal(body) + if err != nil { + t.Fatal(err) + } + if status, resp := rawJSON(t, st, http.MethodPut, "/v1/volumes/"+name, string(raw)); status != http.StatusNoContent { + t.Fatalf("register %s: status=%d body=%s", name, status, resp) + } + } + home := sandbox.Volume{Name: "home", Mount: "/mnt/home"} + for _, tc := range []struct { + token string + want []string + private bool + }{{"acme-token", []string{"home", "private", "public"}, true}, {"beta-token", []string{"home", "public"}, false}} { + client, err := sandbox.Connect(st.addr, sandbox.WithAPIToken(tc.token)) + if err != nil { + t.Fatal(err) + } + volumes, err := client.Volumes(t.Context()) + names := make([]string, 0, len(volumes)) + for _, volume := range volumes { + names = append(names, volume.Name) + } + if err != nil || !slices.Equal(names, tc.want) { + t.Fatalf("catalog for %s: names=%v want=%v err=%v", tc.token, names, tc.want, err) + } + claim, err := client.New(t.Context(), "rt:24.04", sandbox.WithVolumes(home)) + if err != nil || !slices.Equal(claim.Volumes, []sandbox.Volume{home}) { + t.Fatalf("claim with public runtime volume: %+v, %v", claim, err) + } + defer claim.Close() + if _, err = claim.SetVolumes(t.Context(), []sandbox.Volume{}); err != nil { + t.Fatal(err) + } + view, err := claim.PatchVolumes(t.Context(), sandbox.VolumePatch{Add: []sandbox.Volume{home, {Name: "private"}}}) + if (err == nil) != tc.private || tc.private && !slices.Contains(view.Volumes, home) { + t.Fatalf("live attach for %s: view=%+v err=%v", tc.token, view, err) + } + } + caller := *st + caller.token = "beta-token" + var responses []string + for _, name := range []string{"private", "unknown"} { + status, body := rawJSON(t, &caller, http.MethodPost, "/v1/claim", `{"template":"rt:24.04","volumes":[{"name":"`+name+`"}]}`) + if status != http.StatusBadRequest { + t.Fatalf("claim privacy for %s: status=%d body=%s", name, status, body) + } + responses = append(responses, string(body)) + } + if responses[0] != responses[1] { + t.Fatalf("forbidden and unknown responses differ: %s versus %s", responses[0], responses[1]) + } +} diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index fba3efba3..6310beacd 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -275,8 +275,10 @@ type Config struct { DiskReserve *DiskReserve `json:"disk_reserve,omitempty"` Volumes []VolumeSpec `json:"volumes,omitempty"` - Mesh *MeshConfig `json:"mesh,omitempty"` - Pools []PoolSpec `json:"pools"` + // VolumeRoots bounds runtime registration paths only; static volumes stay unchecked. + VolumeRoots []string `json:"volume_roots,omitempty"` + Mesh *MeshConfig `json:"mesh,omitempty"` + Pools []PoolSpec `json:"pools"` } // HasEgress reports whether the node can attach egress-lane VMs. @@ -398,28 +400,13 @@ func (c *Config) validateReplication() error { } func (c *Config) validateVolumes() error { - names := make(map[string]struct{}, len(c.Volumes)) - paths := make(map[string]string, len(c.Volumes)) - for _, volume := range c.Volumes { - if !types.ValidVolumeName(volume.Name) { - return fmt.Errorf("volume name %q must match %s and not start with cocoon-", volume.Name, types.VolumeNameRe) - } - if _, ok := names[volume.Name]; ok { - return fmt.Errorf("duplicate volume name %q", volume.Name) - } - names[volume.Name] = struct{}{} - if !filepath.IsAbs(volume.Path) { - return fmt.Errorf("volume %q path must be absolute", volume.Name) - } - if other, ok := paths[filepath.Clean(volume.Path)]; ok { - return fmt.Errorf("volume %q shares its path with volume %q: admission is keyed by name", volume.Name, other) - } - paths[filepath.Clean(volume.Path)] = volume.Name - if !types.ValidDirectIO(volume.DirectIO) { - return fmt.Errorf("volume %q directio must be on, off, or auto, got %q", volume.Name, volume.DirectIO) + for i, root := range c.VolumeRoots { + if !filepath.IsAbs(root) { + return fmt.Errorf("volume_roots %q must be an absolute path", root) } + c.VolumeRoots[i] = filepath.Clean(root) } - return nil + return ValidateVolumeCatalog(c.Volumes) } func (c *Config) validatePreview() error { @@ -637,6 +624,32 @@ func (c *Config) validateMetaStore() error { return nil } +// ValidateVolumeCatalog checks the shared static and runtime catalog rules without I/O. +func ValidateVolumeCatalog(volumes []VolumeSpec) error { + names := make(map[string]struct{}, len(volumes)) + paths := make(map[string]string, len(volumes)) + for _, volume := range volumes { + if !types.ValidVolumeName(volume.Name) { + return fmt.Errorf("volume name %q must match %s and not start with cocoon-", volume.Name, types.VolumeNameRe) + } + if _, ok := names[volume.Name]; ok { + return fmt.Errorf("duplicate volume name %q", volume.Name) + } + names[volume.Name] = struct{}{} + if !filepath.IsAbs(volume.Path) { + return fmt.Errorf("volume %q path must be absolute", volume.Name) + } + if other, ok := paths[filepath.Clean(volume.Path)]; ok { + return fmt.Errorf("volume %q shares its path with volume %q: admission is keyed by name", volume.Name, other) + } + paths[filepath.Clean(volume.Path)] = volume.Name + if !types.ValidDirectIO(volume.DirectIO) { + return fmt.Errorf("volume %q directio must be on, off, or auto, got %q", volume.Name, volume.DirectIO) + } + } + return nil +} + // Load reads a JSON config file, applies defaults, and validates. func Load(path string) (*Config, error) { raw, err := os.ReadFile(path) //nolint:gosec // path is the operator-supplied -config flag diff --git a/sandboxd/config/config_test.go b/sandboxd/config/config_test.go index 06c4fe3c5..1b5411bef 100644 --- a/sandboxd/config/config_test.go +++ b/sandboxd/config/config_test.go @@ -167,6 +167,8 @@ func TestLoadRejectsInvalid(t *testing.T) { {"duplicate volume name", `{"pools":[],"volumes":[{"name":"data","path":"/srv/datasets/a.img"},{"name":"data","path":"/srv/datasets/b.img"}]}`, "duplicate volume"}, {"relative volume path", `{"pools":[],"volumes":[{"name":"data","path":"datasets/a.img"}]}`, "path must be absolute"}, {"bad volume directio", `{"pools":[],"volumes":[{"name":"data","path":"/srv/datasets/a.img","directio":"yes"}]}`, "directio must be"}, + {"relative volume root", `{"pools":[],"volume_roots":["srv/datasets"]}`, "volume_roots"}, + {"empty volume root", `{"pools":[],"volume_roots":[""]}`, "volume_roots"}, {"reader without warm", `{"pools":[{"template":"rt:24.04","warm":0,"reader":true}]}`, "reader needs warm"}, } { t.Run(tt.name, func(t *testing.T) { @@ -292,7 +294,7 @@ func TestLoadReadsReader(t *testing.T) { } func TestLoadAcceptsVolumes(t *testing.T) { - path := writeConfig(t, `{"pools":[],"volumes":[ + path := writeConfig(t, `{"pools":[],"volume_roots":["/srv/models/"],"volumes":[ {"name":"imagenet","path":"/srv/datasets/imagenet.img"}, {"name":"weights-llama","path":"/srv/datasets/llama.img","directio":"on"}]}`) cfg, err := Load(path) @@ -303,6 +305,9 @@ func TestLoadAcceptsVolumes(t *testing.T) { cfg.Volumes[1].DirectIO != types.DirectIOOn { t.Errorf("volumes = %+v", cfg.Volumes) } + if !slices.Equal(cfg.VolumeRoots, []string{"/srv/models"}) { + t.Errorf("volume_roots = %q, want the cleaned root with static volumes outside it accepted", cfg.VolumeRoots) + } } func TestLoadAcceptsVolumeTenantAccessList(t *testing.T) { diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index f9427792c..1301cded2 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -371,7 +371,7 @@ func (m *Manager) overQuota(extra int, tenant string) error { // admitVolumes takes one claim's holds; a refusal on either check leaves nothing held. func (m *Manager) admitVolumes(tenant string, volumes []resolvedVolume) ([]types.Volume, error) { applied := appliedVolumes(volumes) - if err := m.admitClaim(tenant, applied); err != nil { + if err := m.admitClaim(tenant, volumes, applied); err != nil { return nil, err } if err := m.confirmVolumesClean(volumes); err != nil { @@ -382,13 +382,17 @@ func (m *Manager) admitVolumes(tenant string, volumes []resolvedVolume) ([]types } // admitClaim refuses a busy volume or an over-quota tenant before any VM is built. -func (m *Manager) admitClaim(tenant string, volumes []types.Volume) error { +func (m *Manager) admitClaim(tenant string, volumes []resolvedVolume, applied []types.Volume) error { m.mu.Lock() defer m.mu.Unlock() if err := m.quotaErr(1, tenant); err != nil { return err } - return m.reserveVolumes(volumes) + if err := m.checkResolvedLocked(volumes); err != nil { + return err + } + m.adoptVolumes(applied) + return nil } // quotaErr answers ErrQuota when the node takes no more claims; callers hold m.mu. @@ -603,15 +607,18 @@ func (m *Manager) reapOnce(ctx context.Context) { m.mu.Lock() keep := expired[:0] var purgeCks, purged []string + var purgeHolds []types.Volume for _, v := range expired { if v.action == reapPurge { if m.claimed[v.id] != v.sb || v.sb.ArchiveCk != v.ck { continue } m.pendingCks[v.ck] = struct{}{} + v.holds = m.cancelVolumeRun(v.sb) m.dropClaimLocked(v.sb) purgeCks = append(purgeCks, v.ck) purged = append(purged, v.id) + purgeHolds = append(purgeHolds, v.holds...) } keep = append(keep, v) } @@ -646,6 +653,7 @@ func (m *Manager) reapOnce(ctx context.Context) { logger.Error(ctx, saveErr, "persist reap; rolled back") return } + m.unreserveVolumes(purgeHolds) // engine teardown can take minutes, so fan out bounded and never stall the ticker loop m.runBounded(ctx, len(expired), func(ctx context.Context, i int) { v := expired[i] diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index b02b20dd1..05d30a20c 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -64,6 +64,7 @@ func (m *Manager) adoptArchived(ctx context.Context, id, token string) (bool, er m.mu.Unlock() return m.adoptHeld(ctx, cur, epoch), nil } + m.adoptVolumes(heldVolumes(sb)) m.addClaimLocked(sb) js := m.store.set(sb) m.mu.Unlock() @@ -411,6 +412,9 @@ func (m *Manager) forgetClaim(ctx context.Context, sb *types.Sandbox) bool { m.mu.Unlock() return false } + if sb.ArchiveCk != "" { + m.releaseVolumes(heldVolumes(sb)) + } m.dropClaimLocked(sb) js := m.store.del(sb.ID) m.mu.Unlock() diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index 505dc55c9..507571ebb 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -111,6 +111,67 @@ func TestFailoverOwnerWakeWithTheRegistryDownFailsClosed(t *testing.T) { } } +func TestFailoverLegacyVolumeTransferPreservesOtherReader(t *testing.T) { + c := newFailoverCluster(t) + b := c.node(t, "b", t.TempDir(), keepAnHour) + d := c.node(t, "d", t.TempDir(), keepAnHour) + backing := writeVolumeImage(t, "data", "volume bytes") + for _, m := range []*Manager{b, d} { + if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: backing, Writable: true}); err != nil { + t.Fatal(err) + } + } + raw, readErr := os.ReadFile("testdata/legacy-volume-replica.json") + if readErr != nil { + t.Fatal(readErr) + } + var fixture struct { + Token string + Record archives.Record + } + if err := json.Unmarshal(raw, &fixture); err != nil { + t.Fatal(err) + } + if _, err := c.reg.Put(t.Context(), fixture.Record); err != nil { + t.Fatal(err) + } + c.expire(fixture.Record.Owner) + for range 2 { + if adopted, err := b.AdoptArchived(t.Context(), fixture.Record.ID, fixture.Token); err != nil || !adopted { + t.Fatalf("adopt recorded legacy row: adopted=%v err=%v", adopted, err) + } + } + writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} + if _, err := b.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("adopted reader failed to exclude writer: %v", err) + } + other, err := b.ClaimProvision(t.Context(), plainKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) + if err != nil { + t.Fatal(err) + } + c.expire("b") + if adopted, err := d.AdoptArchived(t.Context(), fixture.Record.ID, fixture.Token); err != nil || !adopted { + t.Fatalf("transfer recorded row: adopted=%v err=%v", adopted, err) + } + for range 2 { + if err := b.Release(t.Context(), fixture.Record.ID, Cred{Token: fixture.Token}); !errors.Is(err, ErrUnknownSandbox) { + t.Fatalf("former owner release: %v", err) + } + if _, err := b.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("former owner released independent reader: %v", err) + } + } + if err := b.Release(t.Context(), other.ID, Cred{Token: other.Token}); err != nil { + t.Fatal(err) + } + if _, err := b.ClaimProvision(t.Context(), plainKey, writer); err != nil { + t.Fatalf("former owner leaked reader hold: %v", err) + } + if _, err := d.ClaimProvision(t.Context(), plainKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("new owner failed to exclude writer: %v", err) + } +} + func TestFailoverAdoptionMovesTheClaimAndTheOldOwnerLetsGo(t *testing.T) { c := newFailoverCluster(t) a, b := c.node(t, "a", t.TempDir(), keepAnHour), c.node(t, "b", t.TempDir(), keepAnHour) diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 9e0124cbd..f6a4a2957 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -102,6 +102,10 @@ var ( ErrNoEgressFork = errors.New("egress-lane sandboxes cannot fork, checkpoint, or promote: a resumed guest egresses before its fresh tap can be locked") ErrVolumeCapture = errors.New("sandboxes with volumes cannot hibernate, fork, checkpoint, or promote") ErrVolumeBusy = errors.New("volume is held by another claim") + + ErrVolumeRegistrationConflict = errors.New("volume registration conflicts with the catalog") + ErrVolumeCatalogUnavailable = errors.New("volume catalog requires a successful persistence retry") + // Replaying a journal takes a writable mount, so readers stay out. ErrVolumeNeedsRecovery = errors.New("volume needs recovery by a writable claim") ErrArchived = errors.New("sandbox is archived; an exec or file call wakes it first") @@ -321,16 +325,14 @@ type PeerDeleteFunc func(ctx context.Context, id string) // Manager owns the node's pools, claims, and their persistence. type Manager struct { - eng Engine - dataDir string - egress bool - maxFork int - store *claimStore - volumes map[string]catalogVolume - // volumeAdmission counts the live holders of each volume name; guarded by m.mu. - volumeAdmission map[string]volumeHolders - volumeRuns map[*types.Sandbox]*volumeRun - volumeErrors map[*types.Sandbox]map[string]string + eng Engine + dataDir string + egress bool + maxFork int + store *claimStore + catalog volumeCatalog + volumeRuns map[*types.Sandbox]*volumeRun + volumeErrors map[*types.Sandbox]map[string]string poolStore *poolStore // poolShared is the cell's pool set with a meta store; poolVersion, under m.mu, is its newest version applied. @@ -448,8 +450,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg cocoondSocket: cfg.CocoondSocket, maxFork: cmp.Or(cfg.MaxForkCount, defaultMaxFork), store: newClaimStore(cfg.DataDir, cfg.SyncClaims), - volumes: make(map[string]catalogVolume, len(cfg.Volumes)), - volumeAdmission: map[string]volumeHolders{}, + catalog: volumeCatalog{entries: make(map[string]*catalogVolume, len(cfg.Volumes)), roots: cfg.VolumeRoots, holders: map[string]volumeHolders{}}, volumeRuns: map[*types.Sandbox]*volumeRun{}, volumeErrors: map[*types.Sandbox]map[string]string{}, poolStore: newPoolStore(cfg.DataDir), @@ -478,12 +479,15 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg healSem: make(chan struct{}, maxConcurrentHeals), } for _, volume := range cfg.Volumes { - m.volumes[volume.Name] = catalogVolume{ + m.catalog.entries[volume.Name] = &catalogVolume{ disk: engine.VolumeSpec{Name: volume.Name, Path: volume.Path, DirectIO: volume.DirectIO}, tenants: slices.Clone(volume.Tenants), writable: volume.Writable, } } + if err := m.loadRuntimeVolumes(); err != nil { + return nil, err + } if err := os.MkdirAll(m.goldensDir(), 0o750); err != nil { return nil, fmt.Errorf("create goldens dir: %w", err) } diff --git a/sandboxd/pool/reconcile.go b/sandboxd/pool/reconcile.go index a116b5800..a776546d6 100644 --- a/sandboxd/pool/reconcile.go +++ b/sandboxd/pool/reconcile.go @@ -47,6 +47,7 @@ func (m *Manager) Reconcile(ctx context.Context) error { case sb.ArchiveCk != "": // archived: no local VM by design, so adopt the stub and let the first exec wake it m.addClaimLocked(sb) + m.adoptVolumes(heldVolumes(sb)) continue case ok && rec.State == vmStateRunning: sb.VsockSocket = rec.VsockSocket diff --git a/sandboxd/pool/testdata/legacy-volume-replica.json b/sandboxd/pool/testdata/legacy-volume-replica.json new file mode 100644 index 000000000..b246bf44b --- /dev/null +++ b/sandboxd/pool/testdata/legacy-volume-replica.json @@ -0,0 +1,15 @@ +{ + "Source": "5cd5c4caa0b85e90489e62e25b8c8e7277040a78", + "Token": "6892970471eef08bd0c28f34aa10cb09", + "Record": { + "ID": "sb_9200d2ad82fc1f9c", + "Tenant": "", + "TokenSHA256": "eaf458d283c8c6fcf8f647f20336395939a074412bef97a19f9ca604caa977eb", + "Owner": "a", + "Epoch": 1, + "Consumed": false, + "Checkpoint": "ck_96a0c036f8fafaca", + "Deadline": "0001-01-01T00:00:00Z", + "Claim": {"id":"sb_9200d2ad82fc1f9c","vm_name":"","key":{"template":"rt:24.04","net":"none","size":"small"},"claimed_at":"2026-10-29T00:50:38.542878594Z","lease_seconds":300,"policy_layer":"pooled","volumes":[{"name":"data","mount":"/volumes/data"}],"archive_ck":"ck_96a0c036f8fafaca"} + } +} diff --git a/sandboxd/pool/volume.go b/sandboxd/pool/volume.go index 5a1e8eedd..7467696c0 100644 --- a/sandboxd/pool/volume.go +++ b/sandboxd/pool/volume.go @@ -31,6 +31,7 @@ type volumeHolders struct { } type resolvedVolume struct { + entry *catalogVolume disk engine.VolumeSpec applied types.Volume } @@ -42,19 +43,21 @@ type volumeTeardown struct { } type catalogVolume struct { + runtime bool disk engine.VolumeSpec tenants []string writable bool } -func (v catalogVolume) allowed(tenant string) bool { +func (v *catalogVolume) allowed(tenant string) bool { return tenant == "" || len(v.tenants) == 0 || slices.Contains(v.tenants, tenant) } // Volumes reports the caller-visible fleet catalog; an empty tenant means root. func (m *Manager) Volumes(tenant string, holders map[string]int) []types.VolumeInfo { - infos := make([]types.VolumeInfo, 0, len(m.volumes)) - for name, volume := range m.volumes { + catalog := m.catalogSnapshot() + infos := make([]types.VolumeInfo, 0, len(catalog)) + for name, volume := range catalog { if !volume.allowed(tenant) { continue } @@ -74,7 +77,7 @@ func (m *Manager) Volumes(tenant string, holders map[string]int) []types.VolumeI // root can see peer-only names; a tenant needs local catalog metadata to apply the access list if tenant == "" { for name, nodes := range holders { - if _, ok := m.volumes[name]; !ok { + if _, ok := catalog[name]; !ok { infos = append(infos, types.VolumeInfo{ Name: name, DefaultMount: types.DefaultVolumeMount(name), @@ -89,8 +92,9 @@ func (m *Manager) Volumes(tenant string, holders map[string]int) []types.VolumeI // VolumeNames returns the sorted names this node can currently serve. func (m *Manager) VolumeNames() []string { - names := make([]string, 0, len(m.volumes)) - for name, volume := range m.volumes { + catalog := m.catalogSnapshot() + names := make([]string, 0, len(catalog)) + for name, volume := range catalog { if _, err := os.Stat(volume.disk.Path); err == nil { names = append(names, name) } @@ -108,8 +112,9 @@ func (m *Manager) VolumePlacement(key types.PoolKey, tenant string, names []stri return true, nil } local := true + catalog := m.catalogSnapshot() for _, name := range names { - volume, ok := m.volumes[name] + volume, ok := catalog[name] if !ok { if tenant == "" { local = false @@ -136,13 +141,14 @@ func (m *Manager) resolveVolumes(ctx context.Context, tenant string, requested [ if err != nil { return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) } + catalog := m.catalogSnapshot() resolved := make([]resolvedVolume, 0, len(applied)) for _, volume := range applied { - entry, ok := m.volumes[volume.Name] - if !ok || !entry.allowed(tenant) { + entry := catalog[volume.Name] + if entry == nil || !entry.allowed(tenant) { return nil, ErrVolumeUnavailable } - resolvedEntry, resolveErr := m.resolveCatalogVolume(ctx, volume) + resolvedEntry, resolveErr := m.resolveCatalogEntry(ctx, volume, entry) if resolveErr != nil { return nil, resolveErr } @@ -151,9 +157,8 @@ func (m *Manager) resolveVolumes(ctx context.Context, tenant string, requested [ return resolved, nil } -func (m *Manager) resolveCatalogVolume(ctx context.Context, volume types.Volume) (resolvedVolume, error) { - entry, ok := m.volumes[volume.Name] - if !ok { +func (m *Manager) resolveCatalogEntry(ctx context.Context, volume types.Volume, entry *catalogVolume) (resolvedVolume, error) { + if entry == nil { return resolvedVolume{}, ErrVolumeUnavailable } if volume.RW() && !entry.writable { @@ -161,13 +166,13 @@ func (m *Manager) resolveCatalogVolume(ctx context.Context, volume types.Volume) } if _, statErr := os.Stat(entry.disk.Path); statErr != nil { // the operator's path stays server-side; the refusal matches the unknown one - log.WithFunc("pool.resolveCatalogVolume").Warnf(ctx, "stat volume %q image %s: %v", volume.Name, entry.disk.Path, statErr) + log.WithFunc("pool.resolveCatalogEntry").Warnf(ctx, "stat volume %q image %s: %v", volume.Name, entry.disk.Path, statErr) return resolvedVolume{}, ErrVolumeUnavailable } disk := entry.disk disk.RW = volume.RW() volume.AttachOnly = false // past validation the empty mount carries it - return resolvedVolume{disk: disk, applied: volume}, nil + return resolvedVolume{entry: entry, disk: disk, applied: volume}, nil } // confirmVolumesClean refuses a read-only claim of a dirty image; run after holds are taken. @@ -232,6 +237,7 @@ func (m *Manager) quiesceVolumes(ctx context.Context, sb *types.Sandbox) volumeT if mounts == 0 { return td } + catalog := m.catalogSnapshot() logger := log.WithFunc("pool.quiesceVolumes") // cancellation-immune like removal: a caller hanging up must not skip the flush. ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), quiesceBudget(mounts)) @@ -246,7 +252,7 @@ func (m *Manager) quiesceVolumes(ctx context.Context, sb *types.Sandbox) volumeT stuck = true continue } - if entry, ok := m.volumes[volume.Name]; ok { + if entry, ok := catalog[volume.Name]; ok { td.clears = append(td.clears, entry.disk.Path) } } @@ -275,45 +281,33 @@ func (m *Manager) finishVolumeTeardown(ctx context.Context, td volumeTeardown) { m.unreserveVolumes(td.holds) } -// reserveVolumes admits one claim's volumes atomically; callers hold m.mu. -func (m *Manager) reserveVolumes(volumes []types.Volume) error { - for _, volume := range volumes { - holders := m.volumeAdmission[volume.Name] - if holders.writers > 0 || (volume.RW() && holders.readers > 0) { - return fmt.Errorf("%w: volume %q", ErrVolumeBusy, volume.Name) - } - } - m.adoptVolumes(volumes) - return nil -} - // adoptVolumes counts volumes an adopted claim already holds; callers hold m.mu. func (m *Manager) adoptVolumes(volumes []types.Volume) { for _, volume := range volumes { - holders := m.volumeAdmission[volume.Name] + holders := m.catalog.holders[volume.Name] if volume.RW() { holders.writers++ } else { holders.readers++ } - m.volumeAdmission[volume.Name] = holders + m.catalog.holders[volume.Name] = holders } } // releaseVolumes drops one claim's admission holds; callers hold m.mu. func (m *Manager) releaseVolumes(volumes []types.Volume) { for _, volume := range volumes { - holders := m.volumeAdmission[volume.Name] + holders := m.catalog.holders[volume.Name] if volume.RW() { holders.writers-- } else { holders.readers-- } if holders.writers < 1 && holders.readers < 1 { - delete(m.volumeAdmission, volume.Name) + delete(m.catalog.holders, volume.Name) continue } - m.volumeAdmission[volume.Name] = holders + m.catalog.holders[volume.Name] = holders } } @@ -326,6 +320,26 @@ func (m *Manager) unreserveVolumes(volumes []types.Volume) { m.releaseVolumes(volumes) } +func (m *Manager) catalogSnapshot() map[string]*catalogVolume { + m.mu.Lock() + defer m.mu.Unlock() + return m.catalog.entries +} + +func (m *Manager) checkResolvedLocked(volumes []resolvedVolume) error { + for _, resolved := range volumes { + volume := resolved.applied + if m.catalog.entries[volume.Name] != resolved.entry { + return ErrVolumeUnavailable + } + holders := m.catalog.holders[volume.Name] + if holders.writers > 0 || (volume.RW() && holders.readers > 0) { + return fmt.Errorf("%w: volume %q", ErrVolumeBusy, volume.Name) + } + } + return nil +} + // quiesceBudget bounds one quiesce: a slice per writable mount plus one for the sync fallback. func quiesceBudget(mounts int) time.Duration { return min(time.Duration(mounts+1)*engine.VolumeCallTimeout, volumeQuiesceMax) diff --git a/sandboxd/pool/volume_acl_test.go b/sandboxd/pool/volume_acl_test.go new file mode 100644 index 000000000..f4136667f --- /dev/null +++ b/sandboxd/pool/volume_acl_test.go @@ -0,0 +1,55 @@ +package pool + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestRuntimeCatalogEmptyACLIsPublic(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, nil) + spec := types.VolumeRegistration{Path: writeVolumeImage(t, "home.img", "private"), Writable: true, Tenants: []string{}} + if err := m.RegisterVolume(t.Context(), "home", spec); err != nil { + t.Fatal(err) + } + for _, current := range []*Manager{m, newVolumeManagerAt(t, eng, m.dataDir, nil)} { + if got := current.Volumes("acme", nil); len(got) != 1 || got[0].Name != "home" { + t.Fatalf("public runtime volume hidden from a tenant: %+v", got) + } + if local, err := current.VolumePlacement(testKey, "acme", []string{"home"}); err != nil || !local { + t.Fatalf("tenant placement: %v %v", local, err) + } + sb, err := current.ClaimProvision(t.Context(), testKey, ClaimOptions{Tenant: "acme", Volumes: []types.Volume{{Name: "home", Mount: "/home"}}}) + if err != nil { + t.Fatalf("tenant attachment: %v", err) + } + if err = current.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatal(err) + } + } +} + +func TestInvalidStaticTenantListNeverBecomesPublic(t *testing.T) { + path := writeVolumeImage(t, "data", "private contents") + file := filepath.Join(t.TempDir(), "config.json") + raw := fmt.Sprintf(`{"pools":[],"volumes":[{"name":"data","path":%q,"tenants":["has space"]}]}`, path) + if err := os.WriteFile(file, []byte(raw), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := config.Load(file) + if err != nil { + t.Fatalf("legacy config stopped booting: %v", err) + } + m := newVolumeManager(t, newFakeEngine(), cfg.Volumes) + if got := m.Volumes("acme", nil); len(got) != 0 { + t.Fatalf("invalid restricted ACL became public: %+v", got) + } + if err = m.RegisterVolume(t.Context(), "other", types.VolumeRegistration{Path: path + ".other"}); err != nil { + t.Fatalf("invalid legacy ACL blocked unrelated registration: %v", err) + } +} diff --git a/sandboxd/pool/volume_bench_test.go b/sandboxd/pool/volume_bench_test.go new file mode 100644 index 000000000..561afd8dc --- /dev/null +++ b/sandboxd/pool/volume_bench_test.go @@ -0,0 +1,56 @@ +package pool + +import ( + "os" + "path/filepath" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func BenchmarkVolumeAdmission(b *testing.B) { + m := volumeBenchManager(b) + requested := []types.Volume{{Name: "bench-data", Mount: "/mnt/data"}} + b.ReportAllocs() + for b.Loop() { + resolved, err := m.resolveVolumes(b.Context(), "", requested) + if err != nil { + b.Fatal(err) + } + if len(resolved) != 1 || resolved[0].applied != requested[0] { + b.Fatalf("resolved volumes: %+v", resolved) + } + applied := appliedVolumes(resolved) + m.mu.Lock() + err = m.checkResolvedLocked(resolved) + if err == nil { + m.adoptVolumes(applied) + m.releaseVolumes(applied) + } + m.mu.Unlock() + if err != nil { + b.Fatal(err) + } + } + if len(m.catalog.holders) != 0 { + b.Fatalf("unreleased admission holds: %v", m.catalog.holders) + } +} + +func volumeBenchManager(b *testing.B) *Manager { + b.Helper() + m := newTestManager(b, newFakeEngine(), config.PoolSpec{PoolKey: testKey}) + m.claimed = checkpointBenchClaims(false) + if err := m.store.save(m.claimed); err != nil { + b.Fatal(err) + } + path := filepath.Join(b.TempDir(), "data.raw") + if err := os.WriteFile(path, []byte("benchmark backing"), 0o600); err != nil { + b.Fatal(err) + } + if err := m.RegisterVolume(b.Context(), "bench-data", types.VolumeRegistration{Path: path}); err != nil { + b.Fatal(err) + } + return m +} diff --git a/sandboxd/pool/volume_catalog.go b/sandboxd/pool/volume_catalog.go new file mode 100644 index 000000000..feeffb00f --- /dev/null +++ b/sandboxd/pool/volume_catalog.go @@ -0,0 +1,182 @@ +package pool + +import ( + "cmp" + "context" + "encoding/json/v2" + "errors" + "fmt" + "maps" + "os" + "path/filepath" + "slices" + "strings" + "sync" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/engine" + "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/utils" +) + +const runtimeVolumesFile = "runtime-volumes.json" + +type volumeCatalog struct { + entries map[string]*catalogVolume + roots []string + holders map[string]volumeHolders + writeMu sync.Mutex +} + +func (c *volumeCatalog) withinRoots(path string) bool { + return len(c.roots) == 0 || slices.ContainsFunc(c.roots, func(root string) bool { + return path == root || strings.HasPrefix(path, strings.TrimSuffix(root, string(filepath.Separator))+string(filepath.Separator)) + }) +} + +// RegisterVolume durably binds a root-managed name to a backing path. +func (m *Manager) RegisterVolume(ctx context.Context, name string, spec types.VolumeRegistration) error { + volume, err := normalizeRegistration(name, spec) + if err != nil { + return err + } + if !m.catalog.withinRoots(volume.disk.Path) { + return fmt.Errorf("%w: volume %q path is outside volume_roots", ErrBadVolume, name) + } + m.catalog.writeMu.Lock() + defer m.catalog.writeMu.Unlock() + m.mu.Lock() + entries := m.catalog.entries + if entry := entries[name]; entry != nil { + m.mu.Unlock() + if entry.runtime && entry.disk == volume.disk && entry.writable == volume.writable && slices.Equal(entry.tenants, volume.tenants) { + return nil + } + return fmt.Errorf("%w: volume %q already has different semantics or is static", ErrVolumeRegistrationConflict, name) + } + if m.volumeReferenced(name) { + m.mu.Unlock() + return fmt.Errorf("%w: volume %q is referenced", ErrVolumeRegistrationConflict, name) + } + m.mu.Unlock() + if err := catalogPathConflict(entries, volume.disk.Path); err != nil { + return fmt.Errorf("%w: %w", ErrVolumeRegistrationConflict, err) + } + entries = maps.Clone(entries) + entries[name] = volume + if err := m.persistRuntimeCatalog(entries); err != nil { + return err + } + m.mu.Lock() + m.catalog.entries = entries + m.mu.Unlock() + m.recordAudit(ctx, "", auditFrame{Op: "volumes", Added: []string{name}}) + return nil +} + +// UnregisterVolume retires an unreferenced runtime binding without touching its backing. +func (m *Manager) UnregisterVolume(ctx context.Context, name string) error { + if !types.ValidVolumeName(name) { + return fmt.Errorf("%w: invalid volume name", ErrBadVolume) + } + m.catalog.writeMu.Lock() + defer m.catalog.writeMu.Unlock() + m.mu.Lock() + entry := m.catalog.entries[name] + if (entry != nil && !entry.runtime) || m.volumeReferenced(name) { + m.mu.Unlock() + return fmt.Errorf("%w: volume %q is static or referenced", ErrVolumeRegistrationConflict, name) + } + entries := maps.Clone(m.catalog.entries) + delete(entries, name) + m.catalog.entries = entries + m.mu.Unlock() + if err := m.persistRuntimeCatalog(entries); err != nil { + return err + } + m.recordAudit(ctx, "", auditFrame{Op: "volumes", Removed: []string{name}}) + return nil +} + +func (m *Manager) loadRuntimeVolumes() error { + raw, err := os.ReadFile(filepath.Join(m.dataDir, runtimeVolumesFile)) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("read runtime volume catalog: %w", err) + } + var record map[string]types.VolumeRegistration + if err = json.Unmarshal(raw, &record); err != nil { + return fmt.Errorf("parse runtime volume catalog: %w", err) + } + entries := maps.Clone(m.catalog.entries) + for name, spec := range record { + volume, normalizeErr := normalizeRegistration(name, spec) + if normalizeErr != nil { + return fmt.Errorf("runtime volume catalog: %w", normalizeErr) + } + if !m.catalog.withinRoots(volume.disk.Path) { + return fmt.Errorf("runtime volume catalog: volume %q path is outside volume_roots", name) + } + if m.catalog.entries[name] != nil { + return fmt.Errorf("runtime volume catalog duplicates static name %q", name) + } + if err := catalogPathConflict(entries, volume.disk.Path); err != nil { + return fmt.Errorf("runtime volume catalog: %w", err) + } + entries[name] = volume + } + m.catalog.entries = entries + return nil +} + +func (m *Manager) volumeReferenced(name string) bool { + return m.catalog.holders[name] != (volumeHolders{}) +} + +func (m *Manager) persistRuntimeCatalog(entries map[string]*catalogVolume) error { + record := map[string]types.VolumeRegistration{} + for name, entry := range entries { + if entry.runtime { + record[name] = types.VolumeRegistration{Path: entry.disk.Path, DirectIO: entry.disk.DirectIO, Writable: entry.writable, Tenants: entry.tenants} + } + } + raw, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("%w: marshal runtime volume catalog: %w", ErrVolumeCatalogUnavailable, err) + } + if err := utils.WriteFileSync(filepath.Join(m.dataDir, runtimeVolumesFile), raw, 0o600); err != nil { + return fmt.Errorf("%w: %w", ErrVolumeCatalogUnavailable, err) + } + return nil +} + +func catalogPathConflict(entries map[string]*catalogVolume, path string) error { + for name, entry := range entries { + if filepath.Clean(entry.disk.Path) == path { + return fmt.Errorf("backing path already registered as %q", name) + } + } + return nil +} + +func normalizeRegistration(name string, spec types.VolumeRegistration) (*catalogVolume, error) { + spec.DirectIO = cmp.Or(spec.DirectIO, types.DirectIOOff) + for _, tenant := range spec.Tenants { + if !types.NameRe.MatchString(tenant) { + return nil, fmt.Errorf("%w: volume %q tenant %q must match %s", ErrBadVolume, name, tenant, types.NameRe) + } + } + if err := config.ValidateVolumeCatalog([]config.VolumeSpec{{ + Name: name, Path: spec.Path, DirectIO: spec.DirectIO, Writable: spec.Writable, Tenants: spec.Tenants, + }}); err != nil { + return nil, fmt.Errorf("%w: %w", ErrBadVolume, err) + } + return &catalogVolume{ + runtime: true, + disk: engine.VolumeSpec{Name: name, Path: filepath.Clean(spec.Path), DirectIO: spec.DirectIO}, + tenants: slices.Compact(slices.Sorted(slices.Values(spec.Tenants))), + writable: spec.Writable, + }, nil +} diff --git a/sandboxd/pool/volume_catalog_ordering_test.go b/sandboxd/pool/volume_catalog_ordering_test.go new file mode 100644 index 000000000..1eda02a31 --- /dev/null +++ b/sandboxd/pool/volume_catalog_ordering_test.go @@ -0,0 +1,128 @@ +package pool + +import ( + "errors" + "os" + "slices" + "strings" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestRuntimeCatalogFailedDeleteRetriesAbsence(t *testing.T) { + for _, replace := range []bool{false, true} { + t.Run(map[bool]string{false: "delete retry", true: "changed-file put"}[replace], func(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, nil) + spec := types.VolumeRegistration{Path: writeVolumeImage(t, "old", "bytes")} + if err := m.RegisterVolume(t.Context(), "data", spec); err != nil { + t.Fatal(err) + } + restore := blockCatalogRename(t, m) + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatal(err) + } + if got := m.VolumeNames(); len(got) != 0 { + t.Fatalf("failed delete remained published: %v", got) + } + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatalf("missing delete skipped persistence: %v", err) + } + restore() + if got := newVolumeManagerAt(t, eng, m.dataDir, nil).VolumeNames(); !slices.Equal(got, []string{"data"}) { + t.Fatalf("restart lost old file: %v", got) + } + if !replace { + if err := m.UnregisterVolume(t.Context(), "data"); err != nil { + t.Fatal(err) + } + if got := newVolumeManagerAt(t, eng, m.dataDir, nil).VolumeNames(); len(got) != 0 { + t.Fatalf("retry did not persist absence: %v", got) + } + return + } + spec.Path = writeVolumeImage(t, "new", "replacement") + if err := m.RegisterVolume(t.Context(), "data", spec); err != nil { + t.Fatal(err) + } + restarted := newVolumeManagerAt(t, eng, m.dataDir, nil) + resolved, err := restarted.resolveVolumes(t.Context(), "", []types.Volume{{Name: "data"}}) + if err != nil || resolved[0].disk.Path != spec.Path { + t.Fatalf("replacement after failed delete: %v, %v", resolved, err) + } + }) + } +} + +func TestRuntimeCatalogFailedPutIsNotReplayed(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + spec := types.VolumeRegistration{Path: writeVolumeImage(t, "data", "bytes")} + restore := blockCatalogRename(t, m) + if err := m.RegisterVolume(t.Context(), "data", spec); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatal(err) + } + restore() + if err := m.RegisterVolume(t.Context(), "replacement", spec); err != nil { + t.Fatal(err) + } + if got := newVolumeManagerAt(t, newFakeEngine(), m.dataDir, nil).VolumeNames(); !slices.Equal(got, []string{"replacement"}) { + t.Fatalf("failed put was replayed: %v", got) + } +} + +func TestRuntimeCatalogRenameSucceededBeforeDirectoryOpenFailed(t *testing.T) { + for _, remove := range []bool{false, true} { + t.Run(map[bool]string{false: "register", true: "unregister"}[remove], func(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, nil) + spec := types.VolumeRegistration{Path: writeVolumeImage(t, "data", "bytes")} + if remove { + if err := m.RegisterVolume(t.Context(), "data", spec); err != nil { + t.Fatal(err) + } + } + if err := os.Chmod(m.dataDir, 0o300); err != nil { + t.Fatal(err) + } + restore := func() { + if err := os.Chmod(m.dataDir, 0o700); err != nil { + t.Fatal(err) + } + } + t.Cleanup(restore) + if dir, err := os.Open(m.dataDir); err == nil { + dir.Close() + t.Skip("requires directory read permission enforcement; drop CAP_DAC_OVERRIDE and CAP_DAC_READ_SEARCH when running as root") + } + var err error + if remove { + err = m.UnregisterVolume(t.Context(), "data") + } else { + err = m.RegisterVolume(t.Context(), "data", spec) + } + if !errors.Is(err, ErrVolumeCatalogUnavailable) || !strings.Contains(err.Error(), "open dir") { + t.Fatalf("expected failure after rename: %v", err) + } + if got := m.VolumeNames(); len(got) != 0 { + t.Fatalf("failed operation published data: %v", got) + } + restore() + restarted := newVolumeManagerAt(t, eng, m.dataDir, nil) + if got := slices.Contains(restarted.VolumeNames(), "data"); got == remove { + t.Fatalf("restart did not load renamed file: %v", restarted.VolumeNames()) + } + if remove { + err = m.UnregisterVolume(t.Context(), "data") + } else { + err = m.RegisterVolume(t.Context(), "data", spec) + } + if err != nil { + t.Fatalf("retry: %v", err) + } + if got := slices.Contains(m.VolumeNames(), "data"); got == remove { + t.Fatalf("retry publication: %v", m.VolumeNames()) + } + }) + } +} diff --git a/sandboxd/pool/volume_catalog_test.go b/sandboxd/pool/volume_catalog_test.go new file mode 100644 index 000000000..7c0c42078 --- /dev/null +++ b/sandboxd/pool/volume_catalog_test.go @@ -0,0 +1,616 @@ +package pool + +import ( + "context" + "encoding/json/jsontext" + "encoding/json/v2" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "testing/synctest" + + "github.com/cocoonstack/sandbox/sandboxd/config" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestRuntimeCatalogPersistsNormalizedSemantics(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, nil) + path := writeVolumeImage(t, "runtime", "backing bytes stay intact") + spec := types.VolumeRegistration{Path: filepath.Dir(path) + "/./runtime", Writable: true, Tenants: []string{"beta", "acme", "acme"}} + before := m.catalogSnapshot() + if err := m.RegisterVolume(t.Context(), "runtime", spec); err != nil { + t.Fatal(err) + } + if before["runtime"] != nil || spec.Tenants[0] != "beta" { + t.Fatal("registration mutated a captured catalog or the caller's spec") + } + normalized := types.VolumeRegistration{Path: path, DirectIO: types.DirectIOOff, Writable: true, Tenants: []string{"acme", "beta"}} + if err := m.RegisterVolume(t.Context(), "runtime", normalized); err != nil { + t.Fatalf("normalized retry: %v", err) + } + raw, err := os.ReadFile(filepath.Join(m.dataDir, runtimeVolumesFile)) + if err != nil { + t.Fatal(err) + } + var saved map[string]types.VolumeRegistration + if err = json.Unmarshal(raw, &saved); err != nil || len(saved) != 1 || saved["runtime"].Path != path || !slices.Equal(saved["runtime"].Tenants, normalized.Tenants) { + t.Fatalf("saved registrations: %s, %v", raw, err) + } + if st, statErr := os.Stat(filepath.Join(m.dataDir, runtimeVolumesFile)); statErr != nil || st.Mode().Perm() != 0o600 { + t.Fatalf("metadata mode: %v %v", st, statErr) + } + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "runtime"}}}) + if err != nil { + t.Fatal(err) + } + m2 := newVolumeManagerAt(t, eng, m.dataDir, nil) + if got := m2.Volumes("acme", nil); len(got) != 1 || got[0].Name != "runtime" || !got[0].Available || !got[0].Writable { + t.Fatalf("catalog absent before reconcile: %+v", got) + } + if got := m2.Volumes("forbidden", nil); len(got) != 0 { + t.Fatalf("ACL changed at restart: %+v", got) + } + if err = m2.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + if _, err = m2.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "runtime", Mode: types.VolumeModeRW}}}); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("recovered holder missing: %v", err) + } + changed := normalized + changed.DirectIO = "on" + if err = m2.RegisterVolume(t.Context(), "runtime", changed); !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("changed semantics: %v", err) + } + if err = m2.UnregisterVolume(t.Context(), "runtime"); !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("unregister of a held volume: %v", err) + } + if err = m2.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err != nil { + t.Fatal(err) + } + for range 2 { + if err = m2.UnregisterVolume(t.Context(), "runtime"); err != nil { + t.Fatal(err) + } + } + if got, err := os.ReadFile(path); err != nil || string(got) != "backing bytes stay intact" { + t.Fatalf("backing changed: %q %v", got, err) + } + if got := newVolumeManagerAt(t, eng, m.dataDir, nil).VolumeNames(); len(got) != 0 { + t.Fatalf("unregister not durable: %v", got) + } +} + +func TestRuntimeCatalogRecoversLegacyArchivedVolumeReference(t *testing.T) { + for _, tc := range []struct { + name string + finish func(*testing.T, *Manager, *types.Sandbox) error + }{ + {"release", func(t *testing.T, m *Manager, sb *types.Sandbox) error { + return m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}) + }}, + {"ownership transfer", func(t *testing.T, m *Manager, sb *types.Sandbox) error { + m.dropAdopted(t.Context(), sb) + return nil + }}, + {"reap", func(t *testing.T, m *Manager, sb *types.Sandbox) error { + expire(m, sb) + m.reapOnce(t.Context()) + return nil + }}, + {"reap with canceled cleanup", func(t *testing.T, m *Manager, sb *types.Sandbox) error { + for range cap(m.maintSem) { + m.maintSem <- struct{}{} + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + expire(m, sb) + m.reapOnce(ctx) + synctest.Wait() + for range cap(m.maintSem) { + <-m.maintSem + } + return nil + }}, + } { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + path := writeVolumeImage(t, "data", "bytes") + if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: path, Writable: true}); err != nil { + t.Fatal(err) + } + sb := &types.Sandbox{ + ID: "sb_legacy", Key: testKey, Token: "legacy-token", ArchiveCk: "ck_00000000000000aa", + Volumes: []types.Volume{{Name: "data", Mount: "/volumes/data"}}, + } + if err := m.store.save(map[string]*types.Sandbox{sb.ID: sb}); err != nil { + t.Fatal(err) + } + m = newVolumeManagerAt(t, newFakeEngine(), m.dataDir, nil) + if err := m.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + recovered, ok := m.byID(sb.ID) + if !ok { + t.Fatal("archived claim did not survive restart") + } + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("unregister recovered reference: %v", err) + } + writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} + if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("writer beside recovered reference: %v", err) + } + reader, claimErr := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) + if claimErr != nil { + t.Fatal(claimErr) + } + for range 2 { + if err := tc.finish(t, m, recovered); err != nil && !errors.Is(err, ErrUnknownSandbox) { + t.Fatal(err) + } + synctest.Wait() + if _, ok := m.byID(sb.ID); ok { + t.Fatal("archived claim survived retirement") + } + if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("writer beside remaining reader: %v", err) + } + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("unregister beside remaining reader: %v", err) + } + } + if err := m.Release(t.Context(), reader.ID, Cred{Token: reader.Token}); err != nil { + t.Fatal(err) + } + writable, err := m.ClaimProvision(t.Context(), testKey, writer) + if err != nil { + t.Fatalf("writer after both readers retire: %v", err) + } + if err := m.Release(t.Context(), writable.ID, Cred{Token: writable.Token}); err != nil { + t.Fatal(err) + } + if err := m.UnregisterVolume(t.Context(), "data"); err != nil { + t.Fatalf("unregister after claim removal: %v", err) + } + }) + }) + } +} + +func TestRuntimeCatalogRetainsArchivedVolumeReferenceAfterFailedReap(t *testing.T) { + for _, tc := range []struct { + name string + breakStore func(*testing.T, *Manager) func() + }{ + {"delete marker", func(t *testing.T, m *Manager) func() { + blocker := filepath.Join(m.dataDir, archiveDeleteDir) + if err := os.WriteFile(blocker, nil, 0o600); err != nil { + t.Fatal(err) + } + return func() { + if err := os.Remove(blocker); err != nil { + t.Fatal(err) + } + } + }}, + {"claim store", func(t *testing.T, m *Manager) func() { + breakStore(t, m) + return func() { healStore(t, m) } + }}, + } { + t.Run(tc.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + path := writeVolumeImage(t, "data", "bytes") + if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: path, Writable: true}); err != nil { + t.Fatal(err) + } + sb := &types.Sandbox{ + ID: "sb_legacy", Key: testKey, Token: "legacy-token", ArchiveCk: "ck_00000000000000aa", + Volumes: []types.Volume{{Name: "data", Mount: "/volumes/data"}}, + } + if err := m.store.save(map[string]*types.Sandbox{sb.ID: sb}); err != nil { + t.Fatal(err) + } + m = newVolumeManagerAt(t, newFakeEngine(), m.dataDir, nil) + if err := m.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + recovered, ok := m.byID(sb.ID) + if !ok { + t.Fatal("archived claim did not survive restart") + } + restore := sync.OnceFunc(tc.breakStore(t, m)) + defer restore() + expire(m, recovered) + m.reapOnce(t.Context()) + if _, ok := m.byID(sb.ID); !ok { + t.Fatal("failed reap removed the archived claim") + } + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("unregister after failed reap: %v", err) + } + writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} + if _, err := m.ClaimProvision(t.Context(), testKey, writer); !errors.Is(err, ErrVolumeBusy) { + t.Fatalf("writer after failed reap: %v", err) + } + restore() + m.reapOnce(t.Context()) + synctest.Wait() + if err := m.UnregisterVolume(t.Context(), "data"); err != nil { + t.Fatalf("unregister after reap retry: %v", err) + } + }) + }) + } +} + +func TestRuntimeCatalogChangesAreAudited(t *testing.T) { + cfg := &config.Config{DataDir: t.TempDir(), AuditLog: true} + m, initErr := NewManager(t.Context(), cfg, newFakeEngine(), testSecrets(t), testBackends(t, cfg)) + if initErr != nil { + t.Fatal(initErr) + } + spec := types.VolumeRegistration{Path: "/images/data"} + restore := blockCatalogRename(t, m) + if err := m.RegisterVolume(t.Context(), "data", spec); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatalf("failed register: %v", err) + } + restore() + for range 2 { + if err := m.RegisterVolume(t.Context(), "data", spec); err != nil { + t.Fatal(err) + } + } + restore = blockCatalogRename(t, m) + if err := m.UnregisterVolume(t.Context(), "data"); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatalf("failed unregister: %v", err) + } + restore() + for range 2 { + if err := m.UnregisterVolume(t.Context(), "data"); err != nil { + t.Fatal(err) + } + } + raw, err := os.ReadFile(filepath.Join(m.dataDir, "audit.jsonl")) + if err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(string(raw)), "\n") + want := []auditFrame{{Op: "volumes", Added: []string{"data"}}, {Op: "volumes", Removed: []string{"data"}}, {Op: "volumes", Removed: []string{"data"}}} + if len(lines) != len(want) { + t.Fatalf("audit records: %s", raw) + } + for i, line := range lines { + var got auditFrame + if err := json.Unmarshal([]byte(line), &got); err != nil || got.Op != want[i].Op || !slices.Equal(got.Added, want[i].Added) || !slices.Equal(got.Removed, want[i].Removed) { + t.Fatalf("audit record %d: %s, %v", i, line, err) + } + } +} + +func TestRuntimeCatalogRegistersUncheckedBacking(t *testing.T) { + link := filepath.Join(t.TempDir(), "backing-link") + if err := os.Symlink(writeVolumeImage(t, "backing", "bytes"), link); err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + name, path string + wantErr error + }{{"symlink", link, nil}, {"missing", filepath.Join(t.TempDir(), "missing"), ErrVolumeUnavailable}} { + t.Run(tc.name, func(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + registration := types.VolumeRegistration{Path: tc.path, Writable: true} + if err := m.RegisterVolume(t.Context(), "runtime", registration); err != nil { + t.Fatalf("register: %v", err) + } + m2 := newVolumeManagerAt(t, newFakeEngine(), m.dataDir, nil) + if err := m2.RegisterVolume(t.Context(), "runtime", registration); err != nil { + t.Fatalf("idempotent retry after restart: %v", err) + } + if infos := m2.Volumes("", nil); len(infos) != 1 || infos[0].Available != (tc.wantErr == nil) { + t.Fatalf("availability: %+v", infos) + } + if _, err := m2.resolveVolumes(t.Context(), "", []types.Volume{{Name: "runtime"}}); !errors.Is(err, tc.wantErr) { + t.Fatalf("resolve = %v, want %v", err, tc.wantErr) + } + if err := m2.UnregisterVolume(t.Context(), "runtime"); err != nil { + t.Fatal(err) + } + }) + } +} + +func TestRuntimeCatalogRejectsCollisionsAndInvalidRegistrations(t *testing.T) { + static := liveCatalog(t) + eng := newFakeEngine() + other := writeVolumeImage(t, "other-runtime", "bytes") + m := newVolumeManager(t, eng, append(slices.Clone(static), config.VolumeSpec{Name: "unexplained", Path: other})) + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "unexplained"}}}); err != nil { + t.Fatal(err) + } + m = newVolumeManagerAt(t, eng, m.dataDir, static) + if err := m.Reconcile(t.Context()); err != nil { + t.Fatal(err) + } + alias := filepath.Dir(static[0].Path) + "/./" + filepath.Base(static[0].Path) + for _, tc := range []struct { + name string + spec types.VolumeRegistration + want error + }{ + {"data", types.VolumeRegistration{Path: other}, ErrVolumeRegistrationConflict}, + {"runtime", types.VolumeRegistration{Path: static[0].Path}, ErrVolumeRegistrationConflict}, + {"runtime", types.VolumeRegistration{Path: alias}, ErrVolumeRegistrationConflict}, + {"unexplained", types.VolumeRegistration{Path: other}, ErrVolumeRegistrationConflict}, + {"cocoon-reserved", types.VolumeRegistration{Path: other}, ErrBadVolume}, + {"bad/name", types.VolumeRegistration{Path: other}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: "relative"}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: other, DirectIO: "invalid"}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: other, Tenants: []string{"has space"}}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: other, Tenants: []string{""}}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: other, Tenants: []string{"_leading"}}, ErrBadVolume}, + {"runtime", types.VolumeRegistration{Path: other, Tenants: []string{strings.Repeat("a", 64)}}, ErrBadVolume}, + } { + err := m.RegisterVolume(t.Context(), tc.name, tc.spec) + if !errors.Is(err, tc.want) { + t.Errorf("register %s %+v = %v, want %v", tc.name, tc.spec, err, tc.want) + } + } + if _, err := os.Stat(filepath.Join(m.dataDir, runtimeVolumesFile)); !errors.Is(err, os.ErrNotExist) || m.catalog.entries["runtime"] != nil { + t.Fatalf("rejected registration persisted: stat=%v entry=%+v", err, m.catalog.entries["runtime"]) + } + for _, tc := range []struct { + name string + want error + }{{"data", ErrVolumeRegistrationConflict}, {"unexplained", ErrVolumeRegistrationConflict}, {"bad/name", ErrBadVolume}} { + if err := m.UnregisterVolume(t.Context(), tc.name); !errors.Is(err, tc.want) { + t.Errorf("unregister %s = %v, want %v", tc.name, err, tc.want) + } + } + if err := m.RegisterVolume(t.Context(), "runtime", types.VolumeRegistration{Path: "/images/data", Tenants: []string{"future-tenant", "team/project:v1"}}); err != nil { + t.Fatalf("future tenant: %v", err) + } +} + +func TestRuntimeCatalogMarshalFailureDoesNotPublish(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + err := m.RegisterVolume(t.Context(), "runtime", types.VolumeRegistration{Path: "/srv/\xff"}) + _, wrapped := errors.AsType[*jsontext.SyntacticError](err) + if !errors.Is(err, ErrVolumeCatalogUnavailable) || !wrapped { + t.Fatalf("marshal error was not preserved: %v", err) + } + if _, err := os.Stat(filepath.Join(m.dataDir, runtimeVolumesFile)); !errors.Is(err, os.ErrNotExist) || m.catalog.entries["runtime"] != nil { + t.Fatalf("failed registration persisted: stat=%v entry=%+v", err, m.catalog.entries["runtime"]) + } +} + +func TestRuntimeCatalogVolumeRootsBoundRegistration(t *testing.T) { + newRootedManager := func(t *testing.T, eng *fakeEngine, dataDir string, roots ...string) (*Manager, error) { + t.Helper() + cfg := &config.Config{DataDir: dataDir, VolumeRoots: roots} + return NewManager(t.Context(), cfg, eng, testSecrets(t), testBackends(t, cfg)) + } + for _, tc := range []struct { + name, path string + roots []string + want error + }{ + {"inside a root", "/srv/data/a.img", []string{"/mnt/other", "/srv/data"}, nil}, + {"equal to a root", "/srv/data/", []string{"/srv/data"}, nil}, + {"sibling sharing a prefix", "/srv/data2/a.img", []string{"/srv/data"}, ErrBadVolume}, + {"dot-dot out of a root", "/srv/data/../etc/a.img", []string{"/srv/data"}, ErrBadVolume}, + {"outside every root", "/etc/a.img", []string{"/srv/data"}, ErrBadVolume}, + {"filesystem root", "/etc/a.img", []string{"/"}, nil}, + {"no roots", "/etc/a.img", nil, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + m, err := newRootedManager(t, newFakeEngine(), t.TempDir(), tc.roots...) + if err != nil { + t.Fatal(err) + } + if err := m.RegisterVolume(t.Context(), "runtime", types.VolumeRegistration{Path: tc.path}); !errors.Is(err, tc.want) { + t.Fatalf("register %s under %v = %v, want %v", tc.path, tc.roots, err, tc.want) + } + }) + } + eng := newFakeEngine() + m := newVolumeManager(t, eng, nil) + images := map[string]string{"outside": writeVolumeImage(t, "outside", "bytes"), "inside": writeVolumeImage(t, "inside", "bytes")} + for name, path := range images { + if err := m.RegisterVolume(t.Context(), name, types.VolumeRegistration{Path: path}); err != nil { + t.Fatal(err) + } + } + outside, inside := filepath.Dir(images["outside"]), filepath.Dir(images["inside"]) + wide, err := newRootedManager(t, eng, m.dataDir, outside, inside) + if err != nil || len(wide.VolumeNames()) != 2 { + t.Fatalf("restart with both registrations inside volume_roots: %v", err) + } + before, err := os.ReadFile(filepath.Join(m.dataDir, runtimeVolumesFile)) + if err != nil { + t.Fatal(err) + } + if _, err = newRootedManager(t, eng, m.dataDir, inside); err == nil { + t.Fatal("restart accepted a persisted registration outside volume_roots") + } + after, err := os.ReadFile(filepath.Join(m.dataDir, runtimeVolumesFile)) + if err != nil || string(after) != string(before) { + t.Fatalf("refused startup changed saved registrations: %s, %v", after, err) + } + if err := wide.UnregisterVolume(t.Context(), "outside"); err != nil { + t.Fatal(err) + } + if narrowed, err := newRootedManager(t, eng, m.dataDir, inside); err != nil || !slices.Equal(narrowed.VolumeNames(), []string{"inside"}) { + t.Fatalf("restart after unregistering the out-of-root entry: %v", err) + } +} + +func TestRuntimeCatalogFailedWriteLeavesUnrelatedAdmissionsAvailable(t *testing.T) { + unregister := func(m *Manager, _ types.VolumeRegistration) error { return m.UnregisterVolume(t.Context(), "affected") } + register := func(m *Manager, spec types.VolumeRegistration) error { + return m.RegisterVolume(t.Context(), "affected", spec) + } + for _, tc := range []struct { + name string + before []string + write func(*Manager, types.VolumeRegistration) error + resolved []types.Volume + wantResolved, want error + }{ + {"register", []string{"healthy"}, register, []types.Volume{{Name: "healthy"}}, nil, ErrVolumeUnavailable}, + {"unregister", []string{"healthy", "affected"}, unregister, []types.Volume{{Name: "healthy"}, {Name: "affected"}}, ErrVolumeUnavailable, ErrVolumeUnavailable}, + } { + t.Run(tc.name, func(t *testing.T) { + eng := newFakeEngine() + m, sb := newTenantVolumeManager(t, eng, liveCatalog(t)) + specs := map[string]types.VolumeRegistration{ + "healthy": {Path: writeVolumeImage(t, "healthy", "bytes")}, + "affected": {Path: writeVolumeImage(t, "affected", "bytes"), Tenants: []string{"other"}}, + } + for _, name := range tc.before { + if err := m.RegisterVolume(t.Context(), name, specs[name]); err != nil { + t.Fatal(err) + } + } + resolved, err := m.resolveVolumes(t.Context(), "", tc.resolved) + if err != nil { + t.Fatal(err) + } + blockCatalogRename(t, m) + if err = tc.write(m, specs["affected"]); !errors.Is(err, ErrVolumeCatalogUnavailable) { + t.Fatalf("obstructed catalog write: %v", err) + } + holds, err := m.admitVolumes("", resolved) + m.unreserveVolumes(holds) + if !errors.Is(err, tc.wantResolved) { + t.Fatalf("admission resolved before the write = %v, want %v", err, tc.wantResolved) + } + for _, volumes := range [][]types.Volume{{{Name: "affected"}}, {{Name: "healthy"}, {Name: "affected"}}} { + if _, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: volumes}); !errors.Is(err, tc.want) { + t.Errorf("root claim %v = %v, want %v", volumes, err, tc.want) + } + } + _, deniedErr := m.SetVolumes(t.Context(), sb.ID, "acme", []types.Volume{{Name: "data"}, {Name: "affected"}}) + _, missingErr := m.SetVolumes(t.Context(), sb.ID, "acme", []types.Volume{{Name: "data"}, {Name: "missing"}}) + if !errors.Is(deniedErr, ErrVolumeUnavailable) || !errors.Is(missingErr, ErrVolumeUnavailable) || deniedErr.Error() != missingErr.Error() { + t.Errorf("private name distinguishable: denied=%v missing=%v", deniedErr, missingErr) + } + if len(eng.volumeSpecs) != 0 { + t.Fatal("refused request reached device attachment") + } + for _, name := range []string{"data", "healthy", "affected", "missing"} { + if got := volumeHoldersOf(m, name); got != (volumeHolders{}) { + t.Fatalf("refused request reserved %s: %+v", name, got) + } + } + for _, name := range []string{"data", "healthy"} { + if err := claimAndRelease(t, m, types.Volume{Name: name}); err != nil { + t.Errorf("unrelated %s claim: %v", name, err) + } + } + if _, err := m.SetVolumes(t.Context(), sb.ID, "acme", []types.Volume{{Name: "data"}, {Name: "healthy"}}); err != nil { + t.Errorf("unrelated live addition: %v", err) + } + }) + } +} + +func TestResolvedCatalogBindingRequiresSameEntry(t *testing.T) { + m := newVolumeManager(t, newFakeEngine(), nil) + spec := types.VolumeRegistration{Path: writeVolumeImage(t, "runtime", "bytes")} + if err := m.RegisterVolume(t.Context(), "runtime", spec); err != nil { + t.Fatal(err) + } + resolved, err := m.resolveVolumes(t.Context(), "", []types.Volume{{Name: "runtime"}}) + if err != nil { + t.Fatal(err) + } + if err = m.RegisterVolume(t.Context(), "other", types.VolumeRegistration{Path: writeVolumeImage(t, "other", "bytes")}); err != nil { + t.Fatal(err) + } + holds, err := m.admitVolumes("", resolved) + if err != nil { + t.Fatalf("unrelated registration invalidated resolution: %v", err) + } + m.unreserveVolumes(holds) + if err = m.UnregisterVolume(t.Context(), "runtime"); err != nil { + t.Fatal(err) + } + if err = m.RegisterVolume(t.Context(), "runtime", spec); err != nil { + t.Fatal(err) + } + if _, err = m.admitVolumes("", resolved); !errors.Is(err, ErrVolumeUnavailable) { + t.Fatalf("same-file reuse admitted stale resolution: %v", err) + } + if got := volumeHoldersOf(m, "runtime"); got != (volumeHolders{}) { + t.Fatalf("stale binding leaked reservation: %+v", got) + } +} + +func TestRuntimeCatalogRejectsInvalidMetadata(t *testing.T) { + static := liveCatalog(t) + for _, tc := range []struct{ name, raw string }{ + {"malformed", "{"}, + {"name", `{"bad/name":{"path":"/data"}}`}, + {"relative", `{"runtime":{"path":"relative"}}`}, + {"directio", `{"runtime":{"path":"/data","directio":"bad"}}`}, + {"duplicate-name", `{"runtime":{"path":"/one"},"runtime":{"path":"/two"}}`}, + {"duplicate-path", `{"runtime":{"path":"/one"},"second":{"path":"/one"}}`}, + {"static-name", `{"data":{"path":"/other"}}`}, + {"static-path", fmt.Sprintf(`{"runtime":{"path":%q}}`, static[0].Path)}, + } { + t.Run(tc.name, func(t *testing.T) { + eng := newFakeEngine() + m := newVolumeManager(t, eng, static) + if err := os.WriteFile(filepath.Join(m.dataDir, runtimeVolumesFile), []byte(tc.raw), 0o600); err != nil { + t.Fatal(err) + } + cfg := &config.Config{DataDir: m.dataDir, Volumes: static} + if _, err := NewManager(t.Context(), cfg, eng, testSecrets(t), testBackends(t, cfg)); err == nil { + t.Fatalf("accepted %s metadata", tc.name) + } + }) + } +} + +func blockCatalogRename(t *testing.T, m *Manager) func() { + t.Helper() + record := filepath.Join(m.dataDir, runtimeVolumesFile) + raw, err := os.ReadFile(record) + if err != nil && !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + existed := err == nil + if err := errors.Join(os.RemoveAll(record), os.Mkdir(record, 0o700)); err != nil { + t.Fatal(err) + } + restore := sync.OnceFunc(func() { + if err := os.Remove(record); err != nil { + t.Fatal(err) + } + if existed { + if err := os.WriteFile(record, raw, 0o600); err != nil { + t.Fatal(err) + } + } + }) + t.Cleanup(restore) + return restore +} + +func claimAndRelease(t *testing.T, m *Manager, volumes ...types.Volume) error { + t.Helper() + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: volumes}) + if err != nil { + return err + } + return m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}) +} diff --git a/sandboxd/pool/volume_cleanup_test.go b/sandboxd/pool/volume_cleanup_test.go new file mode 100644 index 000000000..8728f50cb --- /dev/null +++ b/sandboxd/pool/volume_cleanup_test.go @@ -0,0 +1,59 @@ +package pool + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestDispatchedCleanupRetainsCatalogReference(t *testing.T) { + base := newFakeEngine() + m := newVolumeManager(t, base, nil) + if err := m.RegisterVolume(t.Context(), "data", types.VolumeRegistration{Path: writeVolumeImage(t, "runtime", "bytes")}); err != nil { + t.Fatal(err) + } + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Volumes: []types.Volume{{Name: "data"}}}) + if err != nil { + t.Fatal(err) + } + base.removeErrFor = sb.VMName + if err = m.Release(t.Context(), sb.ID, Cred{Token: sb.Token}); err == nil { + t.Fatal("expected survivor") + } + wg := dispatchBlocked(t.Context(), t, m) + err = m.UnregisterVolume(t.Context(), "data") + unblock(m, wg) + if !errors.Is(err, ErrVolumeRegistrationConflict) { + t.Fatalf("dispatched cleanup lost guard: %v", err) + } + base.removeErrFor = "" + m.retryRemovals(t.Context()).Wait() + if err := m.UnregisterVolume(t.Context(), "data"); err != nil { + t.Fatal(err) + } +} + +// dispatchBlocked moves the removal queue into workers parked on a full maintenance semaphore. +func dispatchBlocked(ctx context.Context, t *testing.T, m *Manager) *sync.WaitGroup { + t.Helper() + for range cap(m.maintSem) { + m.maintSem <- struct{}{} + } + wg := m.retryRemovals(ctx) + m.mu.Lock() + defer m.mu.Unlock() + if len(m.pendingRemovals) != 0 { + t.Fatal("cleanup was not dispatched") + } + return wg +} + +func unblock(m *Manager, wg *sync.WaitGroup) { + for range cap(m.maintSem) { + <-m.maintSem + } + wg.Wait() +} diff --git a/sandboxd/pool/volume_convergence_test.go b/sandboxd/pool/volume_convergence_test.go index 7c002ecc5..3913c596b 100644 --- a/sandboxd/pool/volume_convergence_test.go +++ b/sandboxd/pool/volume_convergence_test.go @@ -22,7 +22,7 @@ func TestSettledVolumesDoNotCallDevicesOrResolveCatalog(t *testing.T) { if _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, []types.Volume{{Name: "data"}}); err != nil { t.Fatal(err) } - delete(m.volumes, "data") + delete(m.catalog.entries, "data") before := len(eng.volumeSpecs) if _, err := m.SetVolumes(t.Context(), sb.ID, sb.Tenant, []types.Volume{{Name: "data"}, {Name: "other"}}); err != nil { t.Fatal(err) @@ -57,9 +57,8 @@ func TestRemovedTenantRetainsAndRepairsHeldVolumes(t *testing.T) { t.Fatal("expected detaching state") } eng.detachVolumeErr = nil - entry := m.volumes["data"] + entry := m.catalog.entries["data"] entry.tenants = []string{"other-tenant"} - m.volumes["data"] = entry if _, err := m.SetVolumes(t.Context(), sb.ID, "", []types.Volume{{Name: "data"}}); err != nil { t.Fatalf("held repair checked new authorization: %v", err) } @@ -330,12 +329,9 @@ func TestVolumeHoldsRejectCaptureBeforeTrim(t *testing.T) { t.Fatal(err) } m.mu.Lock() - err = m.reserveVolumes(held) + m.adoptVolumes(held) run.holds = held m.mu.Unlock() - if err != nil { - t.Fatal(err) - } defer m.finishVolumeRun(sb, run) } if _, err := m.Checkpoint(t.Context(), sb.ID, Cred{Token: sb.Token}, "blocked", ""); !errors.Is(err, ErrVolumeCapture) { diff --git a/sandboxd/pool/volume_mutation.go b/sandboxd/pool/volume_mutation.go index 33dc50e2b..6b7caf2bf 100644 --- a/sandboxd/pool/volume_mutation.go +++ b/sandboxd/pool/volume_mutation.go @@ -139,14 +139,14 @@ func (m *Manager) changeVolumes(ctx context.Context, id, tenant string, change v m.clearVolumeError(sb, run, v.Name) } m.mu.Lock() - pending := volumeDifference(sb.VolumesWant, sb.Volumes) + pending, catalog := volumeDifference(sb.VolumesWant, sb.Volumes), m.catalog.entries m.mu.Unlock() for _, v := range pending { failedName = v.Name if err = m.checkVolumeRun(ctx, sb, run); err != nil { return view, err } - repair, resolveErr := m.resolveCatalogVolume(ctx, v) + repair, resolveErr := m.resolveCatalogEntry(ctx, v, catalog[v.Name]) if resolveErr != nil { return view, resolveErr } @@ -267,10 +267,11 @@ func (m *Manager) stageVolumeTarget(ctx context.Context, sb *types.Sandbox, run m.mu.Unlock() return claimSnapshot{}, false, ErrVolumeCapture } - if err := m.reserveVolumes(additions); err != nil { + if err := m.checkResolvedLocked(resolved); err != nil { m.mu.Unlock() return claimSnapshot{}, false, err } + m.adoptVolumes(additions) run.holds = append(run.holds, additions...) m.mu.Unlock() cleanErr := m.confirmVolumesClean(resolved) diff --git a/sandboxd/pool/volume_mutation_test.go b/sandboxd/pool/volume_mutation_test.go index 4f94b749e..e1925af35 100644 --- a/sandboxd/pool/volume_mutation_test.go +++ b/sandboxd/pool/volume_mutation_test.go @@ -193,6 +193,101 @@ func TestLiveVolumePersistenceFailureRetainsRecoverableState(t *testing.T) { } } +func TestLiveDetachResultWriteFailureSurvivesTwoRestarts(t *testing.T) { + eng := newFakeEngine() + catalog := liveCatalog(t) + m := newVolumeManager(t, eng, catalog) + a := mustClaim(t, m, testKey) + if _, err := m.SetVolumes(t.Context(), a.ID, a.Tenant, []types.Volume{{Name: "data"}}); err != nil { + t.Fatal(err) + } + loadClaims := func() map[string]*types.Sandbox { + t.Helper() + claims, err := newClaimStore(m.dataDir, false).load() + if err != nil { + t.Fatal(err) + } + return claims + } + journalPath := m.store.path + m.store.recommitting.Store(true) + t.Cleanup(func() { + eng.liveVolumeHook = nil + m.store.path = journalPath + m.store.recommitting.Store(false) + }) + eng.liveVolumeHook = func(verb string) error { + if verb == "detach" { + m.store.path = journalPath + "/unavailable" + } + return nil + } + view, err := m.SetVolumes(t.Context(), a.ID, a.Tenant, nil) + eng.liveVolumeHook = nil + m.store.path = journalPath + m.store.recommitting.Store(false) + if err == nil || len(view.Volumes) != 0 || len(view.VolumesWant) != 0 { + t.Fatalf("failed detach result write: view=%+v err=%v", view, err) + } + if ops := eng.volumeOpsLog(); !slices.Contains(ops, "detach:data") { + t.Fatalf("result write failed before device detach: %v", ops) + } + if got := volumeHoldersOf(m, "data"); got != (volumeHolders{}) { + t.Fatalf("successful detach retained live holders: %+v", got) + } + stale := loadClaims()[a.ID] + wantReader := []types.Volume{{Name: "data", Mount: "/volumes/data"}} + if stale == nil || !slices.Equal(stale.Volumes, wantReader) || stale.VolumesWant == nil || len(stale.VolumesWant) != 0 { + t.Fatalf("journal lost stale applied reader or empty target: %+v", stale) + } + + m2 := newVolumeManagerAt(t, eng, m.dataDir, catalog) + if reconcileErr := m2.Reconcile(t.Context()); reconcileErr != nil { + t.Fatal(reconcileErr) + } + writer := ClaimOptions{Volumes: []types.Volume{{Name: "data", Mode: types.VolumeModeRW}}} + if _, claimErr := m2.ClaimProvision(t.Context(), testKey, writer); !errors.Is(claimErr, ErrVolumeBusy) { + t.Fatalf("stale reader admitted writer after restart: %v", claimErr) + } + if got := volumeHoldersOf(m2, "data"); got != (volumeHolders{readers: 1}) { + t.Fatalf("restart did not restore one conservative reader: %+v", got) + } + view, err = m2.SetVolumes(t.Context(), a.ID, a.Tenant, nil) + if err != nil || len(view.Volumes) != 0 || len(view.VolumesWant) != 0 || volumeHoldersOf(m2, "data") != (volumeHolders{}) { + t.Fatalf("retry empty target: view=%+v err=%v", view, err) + } + beforeWriter := len(eng.volumeOpsLog()) + b, err := m2.ClaimProvision(t.Context(), testKey, writer) + if err != nil { + t.Fatalf("writer after recovered detach: %v", err) + } + claims := loadClaims() + wantWriter := []types.Volume{{Name: "data", Mount: "/volumes/data", Mode: types.VolumeModeRW}} + if savedA, savedB := claims[a.ID], claims[b.ID]; savedA == nil || savedB == nil || len(savedA.Volumes) != 0 || len(savedA.VolumesWant) != 0 || !slices.Equal(savedB.Volumes, wantWriter) { + t.Fatalf("successful writer kept obsolete reader journal: A=%+v B=%+v", savedA, savedB) + } + + m3 := newVolumeManagerAt(t, eng, m.dataDir, catalog) + if reconcileErr := m3.Reconcile(t.Context()); reconcileErr != nil { + t.Fatal(reconcileErr) + } + if _, claimErr := m3.ClaimProvision(t.Context(), testKey, writer); !errors.Is(claimErr, ErrVolumeBusy) { + t.Fatalf("second restart admitted competing writer: %v", claimErr) + } + if got := volumeHoldersOf(m3, "data"); got != (volumeHolders{writers: 1}) { + t.Fatalf("second restart did not restore only one writer: %+v", got) + } + attaches := 0 + for _, op := range eng.volumeOpsLog()[beforeWriter:] { + if op == "attach:data" { + attaches++ + } + } + if attaches != 1 { + t.Fatalf("writer and second restart attached data %d times, want 1", attaches) + } +} + func TestLiveBatchPartialFailureRetriesWithoutDoubleHolds(t *testing.T) { eng := newFakeEngine() m := newVolumeManager(t, eng, liveCatalog(t)) @@ -853,9 +948,7 @@ func beginAdmittedVolumeRun(t *testing.T, m *Manager, sb *types.Sandbox) *volume m.mu.Lock() defer m.mu.Unlock() transient := []types.Volume{{Name: "data", Mount: "/volumes/data"}} - if err = m.reserveVolumes(transient); err != nil { - t.Fatal(err) - } + m.adoptVolumes(transient) run.holds = append(run.holds, transient...) return run } diff --git a/sandboxd/pool/volume_rw_test.go b/sandboxd/pool/volume_rw_test.go index a81ff1bb1..7fc072947 100644 --- a/sandboxd/pool/volume_rw_test.go +++ b/sandboxd/pool/volume_rw_test.go @@ -152,10 +152,7 @@ func TestConfirmVolumesCleanCatchesMarkerAfterAdmission(t *testing.T) { if markErr := markVolumeDirty(path); markErr != nil { t.Fatalf("mark dirty: %v", markErr) } - m.mu.Lock() - reserveErr := m.reserveVolumes(appliedVolumes(resolved)) - m.mu.Unlock() - if reserveErr != nil { + if reserveErr := m.admitClaim("", resolved, appliedVolumes(resolved)); reserveErr != nil { t.Fatalf("reserve once the writer released: %v", reserveErr) } @@ -219,17 +216,21 @@ func TestVolumeAdmissionMatrix(t *testing.T) { } } -func TestReserveVolumesExcludesReaderUnderWriter(t *testing.T) { +func TestAdmitClaimExcludesReaderUnderWriter(t *testing.T) { path := writeVolumeImage(t, "scratch.img", "data") m := newVolumeManager(t, newFakeEngine(), []config.VolumeSpec{{Name: "scratch", Path: path, Writable: true}}) + resolved, err := m.resolveVolumes(t.Context(), "", []types.Volume{{Name: "scratch", Mount: "/volumes/scratch"}}) + if err != nil { + t.Fatal(err) + } m.mu.Lock() - defer m.mu.Unlock() m.adoptVolumes([]types.Volume{{Name: "scratch", Mount: "/volumes/scratch", Mode: types.VolumeModeRW}}) + m.mu.Unlock() - if err := m.reserveVolumes([]types.Volume{{Name: "scratch", Mount: "/volumes/scratch"}}); !errors.Is(err, ErrVolumeBusy) { + if err := m.admitClaim("", resolved, appliedVolumes(resolved)); !errors.Is(err, ErrVolumeBusy) { t.Errorf("read-only reservation under a writer: %v, want ErrVolumeBusy", err) } - if holders := m.volumeAdmission["scratch"]; holders != (volumeHolders{writers: 1}) { + if holders := volumeHoldersOf(m, "scratch"); holders != (volumeHolders{writers: 1}) { t.Errorf("registry after the refusal=%+v, want the writer alone", holders) } } @@ -580,5 +581,5 @@ func newVolumeManagerAt(t *testing.T, eng *fakeEngine, dataDir string, volumes [ func volumeHoldersOf(m *Manager, name string) volumeHolders { m.mu.Lock() defer m.mu.Unlock() - return m.volumeAdmission[name] + return m.catalog.holders[name] } diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 78b44fd65..286abcddd 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -57,12 +57,14 @@ var poolErrHTTP = []struct { {pool.ErrTenantStoreDown, http.StatusServiceUnavailable, "tenant store down; retry"}, {pool.ErrPoolStoreDown, http.StatusServiceUnavailable, "pool set store down; retry"}, {pool.ErrArchiveStoreDown, http.StatusServiceUnavailable, "archive registry down; retry"}, + {pool.ErrVolumeCatalogUnavailable, http.StatusServiceUnavailable, "volume catalog unavailable; retry"}, {pool.ErrLeaseLost, http.StatusServiceUnavailable, "lease lost; retry"}, {pool.ErrNoEgress, http.StatusConflict, ""}, {pool.ErrNoEgressHibernate, http.StatusConflict, ""}, {pool.ErrNoEgressFork, http.StatusConflict, ""}, {pool.ErrVolumeCapture, http.StatusConflict, ""}, {pool.ErrVolumeBusy, http.StatusConflict, ""}, + {pool.ErrVolumeRegistrationConflict, http.StatusConflict, pool.ErrVolumeRegistrationConflict.Error()}, {pool.ErrVolumeNeedsRecovery, http.StatusConflict, ""}, {pool.ErrArchived, http.StatusConflict, ""}, {pool.ErrPaused, http.StatusConflict, ""}, @@ -97,6 +99,8 @@ type Manager interface { TenantClaims() map[string]int VolumePlacement(key types.PoolKey, tenant string, names []string) (bool, error) Volumes(tenant string, holders map[string]int) []types.VolumeInfo + RegisterVolume(ctx context.Context, name string, spec types.VolumeRegistration) error + UnregisterVolume(ctx context.Context, name string) error GetVolumes(id, tenant string) (types.SandboxVolumesResponse, error) SetVolumes(ctx context.Context, id, tenant string, volumes []types.Volume) (types.SandboxVolumesResponse, error) PatchVolumes(ctx context.Context, id, tenant string, patch types.VolumePatch) (types.SandboxVolumesResponse, error) @@ -269,6 +273,8 @@ func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.HandleFunc("POST /v1/claim", s.requireToken(s.handleClaim)) mux.HandleFunc("GET /v1/volumes", s.requireToken(s.handleVolumes)) + mux.HandleFunc("PUT /v1/volumes/{name}", s.requireRoot(s.handleVolumeRegistration)) + mux.HandleFunc("DELETE /v1/volumes/{name}", s.requireRoot(s.handleVolumeRegistration)) mux.HandleFunc("POST /v1/sandboxes/{id}/release", s.handleSandboxVerb("release", s.mgr.Release)) mux.HandleFunc("POST /v1/sandboxes/{id}/hibernate", s.handleSandboxVerb("hibernate", s.mgr.Hibernate)) mux.HandleFunc("POST /v1/sandboxes/{id}/wake", s.handleSandboxVerb("wake", s.mgr.Wake)) diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index 6debba66d..f24769d1b 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -2482,6 +2482,8 @@ type checkpointClaimFunc func(string) (*types.Sandbox, error) type idActionFunc func(string) error type fakeManager struct { + registerVolume func(context.Context, string, types.VolumeRegistration) error + unregisterVolume func(context.Context, string) error getVolumes func(string, string) (types.SandboxVolumesResponse, error) setVolumes func(string, string, []types.Volume) (types.SandboxVolumesResponse, error) patchVolumes func(string, string, types.VolumePatch) (types.SandboxVolumesResponse, error) @@ -2557,6 +2559,14 @@ type fakeManager struct { reader *fakeReader } +func (f *fakeManager) RegisterVolume(ctx context.Context, name string, spec types.VolumeRegistration) error { + return f.registerVolume(ctx, name, spec) +} + +func (f *fakeManager) UnregisterVolume(ctx context.Context, name string) error { + return f.unregisterVolume(ctx, name) +} + func (f *fakeManager) GetVolumes(id, tenant string) (types.SandboxVolumesResponse, error) { return f.getVolumes(id, tenant) } diff --git a/sandboxd/server/volume_registration.go b/sandboxd/server/volume_registration.go new file mode 100644 index 000000000..d4290bccf --- /dev/null +++ b/sandboxd/server/volume_registration.go @@ -0,0 +1,28 @@ +package server + +import ( + "net/http" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func (s *Server) handleVolumeRegistration(w http.ResponseWriter, r *http.Request) { + if s.apiToken == "" { + writeErr(w, http.StatusForbidden, "volume registration requires a configured api token") + return + } + name := r.PathValue("name") + var err error + op := "unregister volume" + if r.Method == http.MethodPut { + req, ok := decodeBodyStrict[types.VolumeRegistration](w, r) + if !ok { + return + } + op = "register volume" + err = s.mgr.RegisterVolume(r.Context(), name, req) + } else { + err = s.mgr.UnregisterVolume(r.Context(), name) + } + writeResult(w, r, op, name, op+" failed", err, noContent(w)) +} diff --git a/sandboxd/server/volume_registration_test.go b/sandboxd/server/volume_registration_test.go new file mode 100644 index 000000000..ed85d9b25 --- /dev/null +++ b/sandboxd/server/volume_registration_test.go @@ -0,0 +1,144 @@ +package server + +import ( + "bytes" + "context" + "encoding/json/v2" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strconv" + "strings" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/pool" + "github.com/cocoonstack/sandbox/sandboxd/tenants" + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestVolumeRegistrationRoutesRequireRootBeforeReadingBody(t *testing.T) { + for _, method := range []string{http.MethodPut, http.MethodDelete} { + for _, tt := range []struct { + name, apiToken, token string + code int + wantCalled bool + }{ + {"root", "root", "root", http.StatusNoContent, true}, + {"tenant", "root", "acme-token", http.StatusForbidden, false}, + {"sandbox", "root", "sandbox-token", http.StatusUnauthorized, false}, + {"unknown", "root", "unknown", http.StatusUnauthorized, false}, + {"missing", "root", "", http.StatusUnauthorized, false}, + {"no api token", "", "", http.StatusForbidden, false}, + } { + t.Run(method+"/"+tt.name, func(t *testing.T) { + called := false + mgr := &fakeManager{ + registerVolume: func(context.Context, string, types.VolumeRegistration) error { called = true; return nil }, + unregisterVolume: func(context.Context, string) error { called = true; return nil }, + tenants: []tenants.TenantSpec{{Name: "acme", Token: "acme-token"}}, + } + body := bytes.NewBufferString(`{"path":"/private/data.img","writable":true}`) + unread := body.Len() + req := httptest.NewRequest(method, "/v1/volumes/data", body) + req.Header.Set("Authorization", "Bearer "+tt.token) + w := httptest.NewRecorder() + New(tt.apiToken, "node:7777", mgr, &fakeDialer{}, nil, nil, nil, nil).Handler().ServeHTTP(w, req) + if w.Code != tt.code || called != tt.wantCalled || !tt.wantCalled && body.Len() != unread || w.Header().Get("Retry-After") != "" { + t.Fatalf("status=%d called=%v unread=%d/%d body=%s headers=%v, want %d", w.Code, called, body.Len(), unread, w.Body, w.Header(), tt.code) + } + }) + } + } +} + +func TestVolumeRegistrationRoutesForwardBindingAndContext(t *testing.T) { + type requestKey struct{} + const body = `{"path":"/private/images/../data.img","directio":"auto","writable":true,"tenants":["beta","acme"]}` + want := types.VolumeRegistration{Path: "/private/images/../data.img", DirectIO: "auto", Writable: true, Tenants: []string{"beta", "acme"}} + var registered types.VolumeRegistration + var registeredName, removedName string + checkContext := func(ctx context.Context) { + if ctx.Value(requestKey{}) != "request-value" { + t.Error("manager did not receive the request context") + } + } + mgr := &fakeManager{ + registerVolume: func(ctx context.Context, name string, spec types.VolumeRegistration) error { + checkContext(ctx) + registeredName, registered = name, spec + return nil + }, + unregisterVolume: func(ctx context.Context, name string) error { + checkContext(ctx) + removedName = name + return nil + }, + } + handler := New("root", "", mgr, nil, nil, nil, nil, nil).Handler() + for _, method := range []string{http.MethodPut, http.MethodDelete} { + req := httptest.NewRequest(method, "/v1/volumes/data-v1", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer root") + req = req.WithContext(context.WithValue(req.Context(), requestKey{}, "request-value")) + w := httptest.NewRecorder() + handler.ServeHTTP(w, req) + if w.Code != http.StatusNoContent || w.Body.Len() != 0 { + t.Fatalf("%s status=%d body=%s, want empty 204", method, w.Code, w.Body) + } + } + if registeredName != "data-v1" || removedName != "data-v1" || !reflect.DeepEqual(registered, want) { + t.Fatalf("registered %q %+v, removed %q, want data-v1 %+v", registeredName, registered, removedName, want) + } +} + +func TestVolumeRegistrationRejectsBodyName(t *testing.T) { + called := false + mgr := &fakeManager{registerVolume: func(context.Context, string, types.VolumeRegistration) error { + called = true + return nil + }} + ts := newTestServer(t, "root", mgr, nil) + resp := doReq(t, http.MethodPut, ts.URL+"/v1/volumes/data", "root", `{"name":"other","path":"/private/data.img"}`) + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusBadRequest || called { + t.Fatalf("status=%d called=%v body=%s, want 400 without registration", resp.StatusCode, called, body) + } +} + +func TestVolumeRegistrationRoutesMapSafeErrors(t *testing.T) { + for _, method := range []string{http.MethodPut, http.MethodDelete} { + for _, tt := range []struct { + name string + err error + code int + }{ + {"invalid registration", fmt.Errorf("%w: invalid directio", pool.ErrBadVolume), http.StatusBadRequest}, + {"binding conflict", fmt.Errorf("%w: /private/data.img", pool.ErrVolumeRegistrationConflict), http.StatusConflict}, + {"catalog unavailable", fmt.Errorf("%w: sync /private/runtime-volumes.json", pool.ErrVolumeCatalogUnavailable), http.StatusServiceUnavailable}, + {"unexpected failure", errors.New("open /private/data.img: permission denied"), http.StatusInternalServerError}, + } { + t.Run(method+"/"+tt.name, func(t *testing.T) { + mgr := &fakeManager{ + registerVolume: func(context.Context, string, types.VolumeRegistration) error { return tt.err }, + unregisterVolume: func(context.Context, string) error { return tt.err }, + } + ts := newTestServer(t, "root", mgr, nil) + resp := doReq(t, method, ts.URL+"/v1/volumes/data", "root", `{"path":"/private/data.img"}`) + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + var got map[string]string + if err := json.Unmarshal(body, &got); err != nil || len(got) != 1 || got["error"] == "" { + t.Fatalf("error body = %s, decode error = %v", body, err) + } + retry := tt.code == http.StatusServiceUnavailable + seconds, _ := strconv.Atoi(resp.Header.Get("Retry-After")) + if resp.StatusCode != tt.code || strings.Contains(string(body), "/private") || (seconds >= 1) != retry || strings.Contains(got["error"], "retry") != retry { + t.Fatalf("status=%d headers=%v body=%s, want safe %d with retry=%t", resp.StatusCode, resp.Header, body, tt.code, retry) + } + }) + } + } +} diff --git a/sandboxd/types/api.go b/sandboxd/types/api.go index ac0cd6b4d..0d81b5407 100644 --- a/sandboxd/types/api.go +++ b/sandboxd/types/api.go @@ -22,6 +22,14 @@ type VolumeInfo struct { Writable bool `json:"writable,omitzero"` } +// VolumeRegistration binds a backing path to a runtime catalog name. +type VolumeRegistration struct { + Path string `json:"path"` + DirectIO string `json:"directio,omitempty"` + Writable bool `json:"writable,omitzero"` + Tenants []string `json:"tenants,omitempty"` +} + // VolumePatch is the wire body of PATCH /v1/sandboxes/{id}/volumes. type VolumePatch struct { Add []Volume `json:"add,omitzero"`