From cae76ba1b633c84ea0d2469645b126c332fb2960 Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Wed, 22 Jul 2026 17:02:20 +1000 Subject: [PATCH] test: complete the e2e live-vs-mock rebalance (epic mulga-8oyg3) (#566) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test(volume): assert chunk GC stays wired on in CreateVolume GCEnabled was zero-valued at every viperblock construction site for a long time and no test noticed. It is not part of viperblock.VBState, so it never round-trips through a saved config.json and cannot be observed by reading volume state back. Extract the viperblock.VB assembly out of CreateVolume into buildVBConfig so the GC wiring is assertable directly against the returned config, without a live viperblock or S3 backend. The config is returned by pointer because viperblock.VB carries atomic.Uint64 fields that vet's copylocks check rightly refuses to see copied. * test(e2e): snapshot backend byte accounting before volume purge DeleteVolume purges a volume's entire backend prefix, including any chunks a leak left behind, before anything measures it. That is the same masking that hid the storage-growth leak in production: a passing E2E cannot fail on backend growth when the evidence is deleted first. Snapshot the volume's backend chunk footprint in EnsureVolume's cleanup closure, immediately before DeleteVolume, so every volume the shared teardown path purges is measured rather than silently discarded. The snapshot is always taken and logged; the hard ratio check additionally runs for callers that registered a guest-written-bytes baseline, since most callers never attach or write to the volume they create and so have no baseline to compare against. The tolerance is the RS(2,1) erasure floor widened for per-chunk header and tail-chunk overhead. It is sized to distinguish unbounded growth from a footprint settling at the floor, not to catch bounded residual garbage. The check is non-fatal inside the cleanup callback: t.Fatalf's runtime.Goexit would abort the DeleteVolume call queued behind it and turn a reported leak into a real one. * test(e2e): add memory-constrained concurrent-write durability regression The existing durability tests write one 4MiB sentinel, sequentially, on a comfortably sized guest, which is the exact inverse of the known-open data-loss bug: acked writes read back zero-filled after a cold reread when a guest is memory-starved during concurrent writeback. No happy-path test can reach it, because the fault lives in the intersection of memory pressure, write concurrency and cache-cold reread. Add a suite that encodes those three conditions together: a guest discovered at or below the working set, N parallel large writers launched as background jobs in a single command so they genuinely overlap, swap disabled, then sync and drop_caches before rereading. Assertions are per stream rather than aggregate so a failure names the losing writer instead of leaving whoever reads it to re-derive which of N writers lost data from one combined digest. GuestExec gains a timeout-parameterised variant. Its fixed two-minute default is sized for sub-second payloads, and a multi-gigabyte write to a network-backed volume would exceed it, so the suite would have failed by timeout while looking like it had caught the corruption. The budget scales with the working set, and a deadline is reported distinctly from a data mismatch so the two red states are never confused. The suite is dispatchable but deliberately absent from the default SUITES: the bug it reproduces is still open, so it is expected to fail and must not sit in a merge-gating path. It is registered in the workflow allow-list so it stays discoverable rather than being buried behind a skip. Also record on the live IAM instance-profile test why its assertions were not moved down to the integration tier, so the trace is not repeated. * test(e2e): drop backend byte accounting, assert a node-wide delta instead The accounting helper listed the internal predastore bucket to attribute bytes per volume. That listing is refused unconditionally: predastore authorises bucket access by same-account ownership, and the bucket belongs to the service account, so a test caller gets 403 no matter what it holds. Granting the test a service credential would invert the trust boundary to buy a measurement, which is the wrong trade. Replace it with what the S3 surface actually permits: sample node-wide allocated bytes either side of the write and bound the delta against the bytes the guest wrote. Coarser, but it measures something real rather than depending on a call that cannot succeed. The tolerance moves into the suite that uses it. It encodes RS(2,1)'s 1.5x expansion floor plus headroom; a cluster running the recommended RS(3,2) has a 1.667x floor and only ~12% of margin left, so the constant needs revisiting if the encoding changes. * revert: drop memdurability e2e suite The suite asserted the guest read back what it wrote after a memory-constrained concurrent write, but the assertion could not fail for the reason it claimed: it compared the guest's own read against the guest's own write with the page cache still warm, so it passed whether or not the data reached the backend. A test that is green on a build with a known-open durability bug is worse than no test — it advertises coverage that does not exist. Removes the suite and its Makefile/workflow wiring, plus DiscoverInstanceTypeAtLeastMemory, which had no other caller. GuestExecTimeout stays: GuestExec now delegates to it, and the old two-minute hard cap it replaced was a latent trap for any long guest command. * test(e2e): add partial-block concurrent-write durability suite Replaces the removed memdurability suite, which passed on a build that still carried the bug it claimed to test. The engine reaches its read-modify-write path only for a partial block; a fully covered block takes a full-overwrite branch that reads nothing and can lose nothing. The old workload wrote through ext4 with 1 MiB buffers, which the filesystem coalesced into aligned full-block requests: block-layer tracing recorded zero unaligned and zero sub-4096 requests, so the path under test never ran, and writers 4 MiB apart could not collide on a block even in principle. This suite writes the shape the defect needs. Two writers issue concurrent O_DIRECT writes into disjoint halves of the same 4096-byte blocks on a raw, unformatted device, stamping every half with the round number, then drop the page cache and reread with O_DIRECT asserting every half carries the final round. One half reading back at an earlier round is the signature of the lost update. The old suite also could not fail even had it reached the path: its write phase ended in 'wait; sync; sha256sum' and reported sha256sum's exit status, while bare wait returns 0 unconditionally, so writer errors were invisible and a backend ENOSPC presented as a pass. Here each writer's exit status and exact byte count are asserted before any durability claim is made. Working set is 8 MiB rather than 2048 MiB, and the volume is deleted on teardown; the predecessor left roughly 3 GiB of backend per run and twice drove a node under the nearfull watermark. Dispatchable but not in the default suite list until a green run and a demonstrated fail-first against a pre-fix engine exist. * test(e2e): let the fixture own the working-set volume delete The suite deleted the volume in its own cleanup and the fixture deleted it again on close, so every run ended with an InvalidVolume.NotFound from the second attempt. Detach stays — DeleteVolume against a still attached volume is what leaks volumes on teardown — and the delete is left to the fixture so the volume is released exactly once. * test(e2e): gate merges on the partial-block durability suite Verified on one node with the same test binary and workload, varying only the engine: 1758 of 8192 halves reverted a generation against the pre-fix build, 0 of 8192 against the fixed one. That is the fail-first evidence the suite needed before it could be trusted to gate anything. Adding it to the default suite list rather than leaving it dispatchable is the point. Its predecessor sat outside every path that runs, which is how it survived long enough to be discovered green on a build that still lost data. * test(integration): cover the ECR data plane in the front gate The live ecr/ suite drove docker, crane and skopeo against a real registry, yet none of it boots a guest: it needs a registry, an object store and an HTTP client. The in-process gateway already serves the OCI Distribution v2 surface on an ephemeral port with a memory-backed object store, so these run here in 0.18s instead of behind full provisioning. Covers the three claims the live tests made beyond authorization, which is already down here: a push/pull round trip asserting blobs and manifests read back byte-identical, refusal to auto-create a repository on push, and retagging an existing manifest without disturbing the original tag. Content is compared rather than status codes alone. A registry that stores a blob under the wrong digest, truncates it, or serves a different one answers 200 to every request in the sequence. The capacity/load leg deliberately stays live: it exists to put real parallel pressure on a real predastore, and a memory-backed object store cannot make that claim. * fix(e2e): make volume teardown release the volume before deleting it Fixture cleanups drain LIFO, so a suite that creates its instance before its volume tears the volume down first, while it is still attached. DeleteVolume answered VolumeInUse, nothing retried, and the volume was stranded on the node. Reordering the chain cannot fix this — registration order has no way to express the instance/volume dependency. Instead teardown is now self-sufficient for any creation order: it force-detaches, waits for the volume to reach available, and only then deletes. Instance teardown waits for terminated for the same reason, since TerminateInstances returns as soon as the request is accepted. Two things kept the leak invisible. A failing cleanup was logged and forgotten, so a run stayed green with resources left behind; failures are now reported as leaks, failing the bound test or surfacing through Close() for TestMain to fail the run. And a panic in one cleanup aborted the whole chain, stranding every resource behind it; each callback is now isolated. RegisterVolumeTeardown and the fixture had grown separate teardown paths, only one of which detached. Both now share it. --- .github/workflows/e2e.yml | 21 +- spinifex/handlers/ec2/volume/service_impl.go | 55 +-- .../handlers/ec2/volume/service_impl_test.go | 34 ++ tests/e2e/Makefile | 5 +- tests/e2e/gpu/main_test.go | 8 +- tests/e2e/harness/ec2helpers.go | 141 ++++++-- tests/e2e/harness/fixtures.go | 65 +++- tests/e2e/harness/fixtures_test.go | 162 ++++++++- tests/e2e/harness/guest_data.go | 31 +- tests/e2e/iam/instance_iam_profile_test.go | 22 ++ tests/e2e/iam/main_test.go | 8 +- tests/e2e/multinode/main_test.go | 7 +- tests/e2e/natuplink/natuplink_test.go | 6 +- tests/e2e/partialblock/main_test.go | 155 ++++++++ .../partial_block_durability_test.go | 334 ++++++++++++++++++ tests/e2e/single/main_test.go | 8 +- tests/e2e/storagegrowth/image_pull_test.go | 35 +- tests/e2e/storagegrowth/main_test.go | 55 ++- .../e2e/storagegrowth/storage_growth_test.go | 130 ++++++- tests/integration/ecr_registry_test.go | 173 +++++++++ 20 files changed, 1347 insertions(+), 108 deletions(-) create mode 100644 tests/e2e/partialblock/main_test.go create mode 100644 tests/e2e/partialblock/partial_block_durability_test.go create mode 100644 tests/integration/ecr_registry_test.go diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index db382fbe5..a3f2c0159 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -29,7 +29,7 @@ permissions: env: GIT_BRANCH: ${{ github.ref_name }} - SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-iam' }} + SUITES: ${{ inputs.suites || 'e2e-single e2e-multinode e2e-cert e2e-lb e2e-iam e2e-partialblock' }} TEST_RUN: ${{ inputs.test_run || '' }} # OTLP export to the central Elastic sink (wattle). Repo variable overrides; # empty string disables telemetry entirely. @@ -78,11 +78,12 @@ jobs: # here, before this job provisions a tofu env or pays the Spinifex install. # # The dispatch check gates on exactly the suites this job's Run step - # allow-lists (single|iam|cert|eks|ecs|storagegrowth). Gating on a suite the - # job won't run — lb, which only executes in the multinode job — would - # provision a tofu VM and pay the Spinifex install only to run nothing and - # tear down. Push triggers keep running via the null-check on inputs.suites. - if: needs.integration.result == 'success' && (github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-iam') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-eks') || contains(github.event.inputs.suites, 'e2e-ecs') || contains(github.event.inputs.suites, 'e2e-storagegrowth')) + # allow-lists (single|iam|cert|eks|ecs|storagegrowth|partialblock). Gating + # on a suite the job won't run — lb, which only executes in the multinode + # job — would provision a tofu VM and pay the Spinifex install only to run + # nothing and tear down. Push triggers keep running via the null-check on + # inputs.suites. + if: needs.integration.result == 'success' && (github.event_name != 'workflow_dispatch' || contains(github.event.inputs.suites, 'e2e-single') || contains(github.event.inputs.suites, 'e2e-iam') || contains(github.event.inputs.suites, 'e2e-cert') || contains(github.event.inputs.suites, 'e2e-eks') || contains(github.event.inputs.suites, 'e2e-ecs') || contains(github.event.inputs.suites, 'e2e-storagegrowth') || contains(github.event.inputs.suites, 'e2e-partialblock')) runs-on: [self-hosted, ci-single] timeout-minutes: 50 env: @@ -318,7 +319,13 @@ jobs: # Allow-list, not a deny-list: a suite runs here only if named. Deny-lists # silently fan a newly added suite out to every job, which is how lb ended up # running twice per e2e run. - SUITES_RUN=$(echo "${SUITES}" | sed 's/e2e-//g' | tr ' ' '\n' | grep -xE 'single|iam|cert|eks|ecs|storagegrowth' | tr '\n' ' ') + # + # partialblock gates merges because it earned it, not because it is + # new: on one node, same binary and workload, it fails against the + # pre-fix engine (1758 of 8192 halves reverted a generation) and + # passes against the fixed one (0 of 8192). A suite that has never + # been shown to go red for the reason it claims does not belong here. + SUITES_RUN=$(echo "${SUITES}" | sed 's/e2e-//g' | tr ' ' '\n' | grep -xE 'single|iam|cert|eks|ecs|storagegrowth|partialblock' | tr '\n' ' ') if [ -z "$(echo "${SUITES_RUN}" | tr -d ' ')" ]; then echo "no in-VM suites requested (SUITES=${SUITES}); skipping" echo "rc=0" >> "$GITHUB_OUTPUT" diff --git a/spinifex/handlers/ec2/volume/service_impl.go b/spinifex/handlers/ec2/volume/service_impl.go index 9f9792daa..62d38088c 100644 --- a/spinifex/handlers/ec2/volume/service_impl.go +++ b/spinifex/handlers/ec2/volume/service_impl.go @@ -17,6 +17,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/predastore/pkg/masterkey" "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" @@ -202,29 +203,9 @@ func (s *VolumeServiceImpl) CreateVolume(ctx context.Context, input *ec2.CreateV return nil, errors.New(awserrors.ErrorServerInternal) } - // GCEnabled: default false unless explicitly set to true, matching the - // nbdkit plugin and viperblockd resolution of the same config field. - gcEnabled := s.config.Viperblock.GCEnabled != nil && *s.config.Viperblock.GCEnabled + vbconfig := s.buildVBConfig(volumeID, volumeSizeBytes, volumeConfig, mkey, snapshotID, sourceVolumeName) - vbconfig := viperblock.VB{ - VolumeName: volumeID, - VolumeSize: volumeSizeBytes, - BaseDir: s.config.WalDir, - Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, - VolumeConfig: volumeConfig, - MasterKey: mkey, - EncryptionEnabled: mkey != nil, - GCEnabled: gcEnabled, - } - - // If created from a snapshot, set the snapshot fields so viperblock's - // LoadState will call OpenFromSnapshot to load the base block map. - if snapshotID != "" { - vbconfig.SnapshotID = snapshotID - vbconfig.SourceVolumeName = sourceVolumeName - } - - vb, err := viperblock.New(&vbconfig, "s3", cfg) + vb, err := viperblock.New(vbconfig, "s3", cfg) if err != nil { slog.ErrorContext(ctx, "CreateVolume failed to create viperblock instance", "err", err) return nil, errors.New(awserrors.ErrorServerInternal) @@ -269,6 +250,36 @@ func (s *VolumeServiceImpl) CreateVolume(ctx context.Context, input *ec2.CreateV return vol, nil } +// buildVBConfig assembles the viperblock.VB config CreateVolume hands to +// viperblock.New. Split out from CreateVolume so the GC-enablement wiring can +// be asserted directly against the returned config, without needing a live +// viperblock/S3 backend to observe it: GCEnabled is not part of +// viperblock.VBState, so it never round-trips through a saved config.json and +// can't be checked by reading volume state back. +func (s *VolumeServiceImpl) buildVBConfig(volumeID string, volumeSizeBytes uint64, volumeConfig viperblock.VolumeConfig, mkey *masterkey.Key, snapshotID, sourceVolumeName string) *viperblock.VB { + vbconfig := &viperblock.VB{ + VolumeName: volumeID, + VolumeSize: volumeSizeBytes, + BaseDir: s.config.WalDir, + Cache: viperblock.Cache{Config: viperblock.CacheConfig{Size: 0}}, + VolumeConfig: volumeConfig, + MasterKey: mkey, + EncryptionEnabled: mkey != nil, + // GCEnabled: default false unless explicitly set to true, matching the + // nbdkit plugin and viperblockd resolution of the same config field. + GCEnabled: s.config.Viperblock.GCEnabled != nil && *s.config.Viperblock.GCEnabled, + } + + // If created from a snapshot, set the snapshot fields so viperblock's + // LoadState will call OpenFromSnapshot to load the base block map. + if snapshotID != "" { + vbconfig.SnapshotID = snapshotID + vbconfig.SourceVolumeName = sourceVolumeName + } + + return vbconfig +} + // describeVolumesValidFilters defines the set of filter names accepted by DescribeVolumes. var describeVolumesValidFilters = map[string]bool{ "volume-id": true, diff --git a/spinifex/handlers/ec2/volume/service_impl_test.go b/spinifex/handlers/ec2/volume/service_impl_test.go index d803f5e0b..e66c91636 100644 --- a/spinifex/handlers/ec2/volume/service_impl_test.go +++ b/spinifex/handlers/ec2/volume/service_impl_test.go @@ -233,6 +233,40 @@ func TestCreateVolume_PassesValidation(t *testing.T) { } } +// TestCreateVolume_BuildVBConfig_GCEnabled asserts that CreateVolume's +// viperblock config carries GCEnabled through from spinifex.toml's +// [viperblock] gc_enabled key (config.Config.Viperblock.GCEnabled). This is +// the config-level seam: GCEnabled isn't part of viperblock.VBState, so it +// never persists to config.json and can't be observed by reading a created +// volume back — the only place it's checkable is the config CreateVolume +// hands to viperblock.New, before construction ever reaches the backend. +// A regression that unwires GC in buildVBConfig, or a future refactor of +// CreateVolume that stops routing through buildVBConfig, fails this test. +func TestCreateVolume_BuildVBConfig_GCEnabled(t *testing.T) { + tests := []struct { + name string + gcEnabled *bool + want bool + }{ + {name: "NilDefaultsToDisabled", gcEnabled: nil, want: false}, + {name: "ExplicitFalse", gcEnabled: aws.Bool(false), want: false}, + {name: "ExplicitTrue", gcEnabled: aws.Bool(true), want: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := newTestVolumeService("ap-southeast-2a") + svc.config.Viperblock.GCEnabled = tt.gcEnabled + + vbconfig := svc.buildVBConfig("vol-gc-test", 10*1024*1024*1024, + viperblock.VolumeConfig{}, nil, "", "") + + assert.Equal(t, tt.want, vbconfig.GCEnabled, + "buildVBConfig GCEnabled must follow config.Viperblock.GCEnabled") + }) + } +} + func TestDeleteVolume_Validation(t *testing.T) { tests := []struct { name string diff --git a/tests/e2e/Makefile b/tests/e2e/Makefile index e02fd58dd..23c99cb77 100644 --- a/tests/e2e/Makefile +++ b/tests/e2e/Makefile @@ -18,9 +18,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG)))) # named suite into BIN_DIR/.test so CI can ship binaries (not source + # go toolchain) to the remote VM. BIN_DIR ?= $(CURDIR)/_bin -SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth +SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth partialblock -.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-all clean e2e-build +.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-partialblock e2e-all clean e2e-build RUN_FLAG = $(if $(RUN),-run '$(RUN)',) @@ -56,6 +56,7 @@ e2e-tofu: ; $(MAKE) e2e PKG=./tofu/... e2e-gpu: ; $(MAKE) e2e PKG=./gpu/... e2e-natuplink: ; $(MAKE) e2e PKG=./natuplink/... e2e-storagegrowth: ; $(MAKE) e2e PKG=./storagegrowth/... +e2e-partialblock: ; $(MAKE) e2e PKG=./partialblock/... e2e-all: ; $(MAKE) e2e e2e-build: diff --git a/tests/e2e/gpu/main_test.go b/tests/e2e/gpu/main_test.go index 02361efb2..3f6245853 100644 --- a/tests/e2e/gpu/main_test.go +++ b/tests/e2e/gpu/main_test.go @@ -3,6 +3,7 @@ package gpu import ( + "fmt" "os" "sync" "testing" @@ -23,7 +24,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/harness/ec2helpers.go b/tests/e2e/harness/ec2helpers.go index 67ec1827e..0d7727064 100644 --- a/tests/e2e/harness/ec2helpers.go +++ b/tests/e2e/harness/ec2helpers.go @@ -4,6 +4,7 @@ package harness import ( "bytes" + "fmt" "os/exec" "strconv" "strings" @@ -13,6 +14,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/ec2/ec2iface" ) // SSHTarget is the minimal addressing tuple used by LsblkRootGiB. Unlike @@ -261,33 +263,132 @@ func DetachVolumeWait(t *testing.T, c *AWSClient, volID string) { WaitForVolumeState(t, c, volID, "available", WithPoll(500*time.Millisecond)) } -// RegisterVolumeTeardown best-effort force-detaches then deletes volID at test -// end. Non-fatal — it polls for "available" within a bounded window and never -// calls t.Fatal so a cleanup hiccup doesn't mask the test's real result. +// RegisterVolumeTeardown force-detaches then deletes volID at test end. It does +// not call t.Fatal, so a cleanup hiccup can't mask the test's real result, but a +// teardown that actually leaves the volume behind fails the test — a silent leak +// is worse than a noisy one. func RegisterVolumeTeardown(t *testing.T, c *AWSClient, volID string) { t.Helper() t.Cleanup(func() { - _, _ = c.EC2.DetachVolume(&ec2.DetachVolumeInput{ - VolumeId: aws.String(volID), - Force: aws.Bool(true), - }) - deadline := time.Now().Add(90 * time.Second) - for time.Now().Before(deadline) { - out, err := c.EC2.DescribeVolumes(&ec2.DescribeVolumesInput{ - VolumeIds: []*string{aws.String(volID)}, - }) - if err != nil || len(out.Volumes) == 0 { - return - } - if aws.StringValue(out.Volumes[0].State) == "available" { - break - } - time.Sleep(2 * time.Second) + if err := teardownVolume(c.EC2, volID); err != nil { + t.Errorf("LEAKED volume %s: %v", volID, err) } - _, _ = c.EC2.DeleteVolume(&ec2.DeleteVolumeInput{VolumeId: aws.String(volID)}) }) } +// Teardown windows. Detach and terminate are both asynchronous, and a volume is +// only released once the guest holding it is down, so these have to outlast a +// slow QEMU shutdown rather than a single API round trip. +const ( + volumeTeardownTimeout = 90 * time.Second + instanceTeardownTimeout = 3 * time.Minute + teardownPollInterval = 2 * time.Second +) + +// teardownVolume force-detaches volID if it is still attached, waits for it to +// reach "available", then deletes it. It returns an error only when the volume +// is still present afterwards — that is, only when it has genuinely leaked. +// +// Deletion has to be self-sufficient rather than rely on teardown ordering. +// Fixture cleanups drain LIFO, so a suite that creates its instance before its +// volume tears the volume down first, while it is still attached: DeleteVolume +// answers VolumeInUse and the volume is stranded. Registration order cannot +// express that dependency, so the teardown resolves it itself and works for any +// creation order. +func teardownVolume(ec2c ec2iface.EC2API, volID string) error { + forced := false + deadline := time.Now().Add(volumeTeardownTimeout) + for { + out, err := ec2c.DescribeVolumes(&ec2.DescribeVolumesInput{ + VolumeIds: []*string{aws.String(volID)}, + }) + if err != nil { + if volumeGone(err) { + return nil + } + return fmt.Errorf("describe volume %s: %w", volID, err) + } + if len(out.Volumes) == 0 { + return nil + } + + switch state := aws.StringValue(out.Volumes[0].State); state { + case ec2.VolumeStateDeleting, ec2.VolumeStateDeleted: + return nil + case ec2.VolumeStateAvailable: + if _, err := ec2c.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volID), + }); err != nil && !volumeGone(err) { + return fmt.Errorf("delete volume %s: %w", volID, err) + } + return nil + default: + // Force, because by teardown the attaching guest is usually already + // terminating and there is nothing left to acknowledge a graceful + // detach. Issued once, then we wait for the state to settle. + if !forced { + _, _ = ec2c.DetachVolume(&ec2.DetachVolumeInput{ + VolumeId: aws.String(volID), + Force: aws.Bool(true), + }) + forced = true + continue + } + if time.Now().After(deadline) { + return fmt.Errorf("volume %s still %s after %s, not deleted", volID, state, volumeTeardownTimeout) + } + } + time.Sleep(teardownPollInterval) + } +} + +// teardownInstance terminates instID and waits until it reports "terminated". +// The wait is the point: TerminateInstances returns as soon as the request is +// accepted, so a volume teardown running straight afterwards would still find +// its volume attached to a guest that has not gone away yet. +func teardownInstance(ec2c ec2iface.EC2API, instID string) error { + if _, err := ec2c.TerminateInstances(&ec2.TerminateInstancesInput{ + InstanceIds: []*string{aws.String(instID)}, + }); err != nil && !instanceGone(err) { + return fmt.Errorf("terminate instance %s: %w", instID, err) + } + + deadline := time.Now().Add(instanceTeardownTimeout) + for { + out, err := ec2c.DescribeInstances(&ec2.DescribeInstancesInput{ + InstanceIds: []*string{aws.String(instID)}, + }) + if err != nil { + if instanceGone(err) { + return nil + } + return fmt.Errorf("describe instance %s: %w", instID, err) + } + if len(out.Reservations) == 0 || len(out.Reservations[0].Instances) == 0 { + return nil + } + state := aws.StringValue(out.Reservations[0].Instances[0].State.Name) + if state == ec2.InstanceStateNameTerminated { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf("instance %s still %s after %s", instID, state, instanceTeardownTimeout) + } + time.Sleep(teardownPollInterval) + } +} + +// volumeGone and instanceGone report whether err is the API's way of saying the +// resource no longer exists, which is success for a teardown. +func volumeGone(err error) bool { return errCodeIs(err, "InvalidVolume.NotFound") } + +func instanceGone(err error) bool { return errCodeIs(err, "InvalidInstanceID.NotFound") } + +func errCodeIs(err error, code string) bool { + var ae awserr.Error + return asErr(err, &ae) && ae.Code() == code +} + // asErr is a thin errors.As that lets the helper stay testify-free. func asErr(err error, target *awserr.Error) bool { if a, ok := err.(awserr.Error); ok { diff --git a/tests/e2e/harness/fixtures.go b/tests/e2e/harness/fixtures.go index 256ce5bce..066a810ae 100644 --- a/tests/e2e/harness/fixtures.go +++ b/tests/e2e/harness/fixtures.go @@ -15,6 +15,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -68,8 +69,12 @@ type Fixture struct { // processCleanups holds teardown callbacks for process-mode fixtures. // Close() runs them LIFO. Unused (and untouched) when parent != nil. + // + // leaks accumulates teardown failures for Close() to return; test-mode + // fixtures report them on the bound *testing.T instead. processMu sync.Mutex processCleanups []func() + leaks []string } // Compile-time interface checks. @@ -132,10 +137,12 @@ func newFixture(t *testing.T, ec2c ec2iface.EC2API, elbc elbv2iface.ELBV2API) *F } } -// Close drains the process-mode cleanup chain LIFO. No-op for test-mode -// fixtures (parent != nil) since those route teardown through t.Cleanup. +// Close drains the process-mode cleanup chain LIFO and returns a non-nil error +// naming every resource it could not tear down, so a caller in TestMain can +// fail the run. No-op returning nil for test-mode fixtures (parent != nil) +// since those route teardown through t.Cleanup. // Safe to call multiple times; subsequent calls drain an empty slice. -func (f *Fixture) Close() { +func (f *Fixture) Close() error { f.processMu.Lock() cleanups := f.processCleanups f.processCleanups = nil @@ -143,18 +150,54 @@ func (f *Fixture) Close() { for i := len(cleanups) - 1; i >= 0; i-- { cleanups[i]() } + + f.processMu.Lock() + leaks := f.leaks + f.leaks = nil + f.processMu.Unlock() + if len(leaks) == 0 { + return nil + } + return fmt.Errorf("%d resource(s) left behind:\n %s", len(leaks), strings.Join(leaks, "\n ")) } // registerCleanup routes a teardown callback to the appropriate sink based // on construction mode. Test mode: t.Cleanup on the bound parent. Process // mode: append to the LIFO drained by Close(). +// +// Each callback is isolated from the rest: a panic in one teardown would +// otherwise abort the whole chain and strand every resource behind it. func (f *Fixture) registerCleanup(fn func()) { + isolated := func() { + defer func() { + if r := recover(); r != nil { + f.reportLeak("cleanup panicked: %v", r) + } + }() + fn() + } if f.parent != nil { - f.parent.Cleanup(fn) + f.parent.Cleanup(isolated) return } f.processMu.Lock() - f.processCleanups = append(f.processCleanups, fn) + f.processCleanups = append(f.processCleanups, isolated) + f.processMu.Unlock() +} + +// reportLeak records a teardown failure. A cleanup that fails has left a real +// resource behind on the node, so this must not be a log line nobody reads: +// test-mode fixtures fail the bound test, process-mode fixtures accumulate the +// failures for Close() to return. +func (f *Fixture) reportLeak(format string, args ...any) { + msg := fmt.Sprintf(format, args...) + if f.parent != nil { + f.parent.Errorf("harness.Fixture: LEAKED resource, teardown failed: %s", msg) + return + } + fmt.Fprintf(os.Stderr, "harness.Fixture: LEAKED resource, teardown failed: %s\n", msg) + f.processMu.Lock() + f.leaks = append(f.leaks, msg) f.processMu.Unlock() } @@ -222,7 +265,7 @@ func (f *Fixture) ensureOnce(t *testing.T, key string, create func() (string, fu if !dup && cleanup != nil { f.registerCleanup(func() { if cerr := cleanup(); cerr != nil { - f.logf("fixture cleanup %s: %v", key, cerr) + f.reportLeak("%s (%s): %v", key, id, cerr) } }) } @@ -631,10 +674,7 @@ func EnsureInstance(t *testing.T, fx *Fixture, spec InstanceSpec) string { } return instID, func() error { - _, derr := fx.EC2.TerminateInstances(&ec2.TerminateInstancesInput{ - InstanceIds: []*string{aws.String(instID)}, - }) - return derr + return teardownInstance(fx.EC2, instID) }, nil }) if err != nil { @@ -683,10 +723,7 @@ func EnsureVolume(t *testing.T, fx *Fixture, az string, sizeGiB int64) string { } return volID, func() error { - _, derr := fx.EC2.DeleteVolume(&ec2.DeleteVolumeInput{ - VolumeId: aws.String(volID), - }) - return derr + return teardownVolume(fx.EC2, volID) }, nil }) if err != nil { diff --git a/tests/e2e/harness/fixtures_test.go b/tests/e2e/harness/fixtures_test.go index 9e970c8b0..c0cbf6977 100644 --- a/tests/e2e/harness/fixtures_test.go +++ b/tests/e2e/harness/fixtures_test.go @@ -5,12 +5,14 @@ package harness import ( "errors" "path/filepath" + "strings" "sync" "sync/atomic" "testing" "time" "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/ec2/ec2iface" "github.com/aws/aws-sdk-go/service/elbv2/elbv2iface" @@ -28,6 +30,56 @@ type fakeEC2 struct { // holdCreate, when non-nil, blocks CreateKeyPair until closed. Drives // the concurrent-callers test deterministically. holdCreate chan struct{} + + // Volume teardown state. volState is what DescribeVolumes reports; + // DetachVolume flips it to "available", modelling a node that releases the + // volume once the attachment is forced off. deleteVolumeErr, when set, is + // what DeleteVolume returns — used to drive the leak-reporting path. + volMu sync.Mutex + volState string + detachVolume atomic.Int64 + deleteVolume atomic.Int64 + deleteVolumeErr error +} + +func (f *fakeEC2) CreateVolume(*ec2.CreateVolumeInput) (*ec2.Volume, error) { + return &ec2.Volume{VolumeId: aws.String("vol-fake")}, nil +} + +func (f *fakeEC2) DescribeVolumes(*ec2.DescribeVolumesInput) (*ec2.DescribeVolumesOutput, error) { + f.volMu.Lock() + defer f.volMu.Unlock() + return &ec2.DescribeVolumesOutput{ + Volumes: []*ec2.Volume{{VolumeId: aws.String("vol-fake"), State: aws.String(f.volState)}}, + }, nil +} + +func (f *fakeEC2) DetachVolume(*ec2.DetachVolumeInput) (*ec2.VolumeAttachment, error) { + f.detachVolume.Add(1) + f.volMu.Lock() + defer f.volMu.Unlock() + f.volState = ec2.VolumeStateAvailable + return &ec2.VolumeAttachment{}, nil +} + +func (f *fakeEC2) DeleteVolume(*ec2.DeleteVolumeInput) (*ec2.DeleteVolumeOutput, error) { + f.deleteVolume.Add(1) + if f.deleteVolumeErr != nil { + return nil, f.deleteVolumeErr + } + // The refusal the real API gives, and the whole reason teardown has to + // detach first: an attached volume cannot be deleted. + f.volMu.Lock() + defer f.volMu.Unlock() + if f.volState != ec2.VolumeStateAvailable { + return nil, awserr.New("VolumeInUse", + "The specified Amazon EBS volume is attached to an instance.", nil) + } + return &ec2.DeleteVolumeOutput{}, nil +} + +func (f *fakeEC2) CreateTags(*ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error) { + return &ec2.CreateTagsOutput{}, nil } func (f *fakeEC2) CreateKeyPair(in *ec2.CreateKeyPairInput) (*ec2.CreateKeyPairOutput, error) { @@ -151,7 +203,9 @@ func TestProcessFixture_CloseRunsCleanupsLIFO(t *testing.T) { fx.RegisterCleanup(func() { order = append(order, 2) }) fx.RegisterCleanup(func() { order = append(order, 3) }) - fx.Close() + if err := fx.Close(); err != nil { + t.Fatalf("Close err = %v, want nil", err) + } want := []int{3, 2, 1} if len(order) != len(want) { t.Fatalf("cleanup count = %d, want %d", len(order), len(want)) @@ -163,12 +217,116 @@ func TestProcessFixture_CloseRunsCleanupsLIFO(t *testing.T) { } // Second Close is a no-op — no duplicate firings, no panic. - fx.Close() + if err := fx.Close(); err != nil { + t.Fatalf("second Close err = %v, want nil", err) + } if len(order) != len(want) { t.Fatalf("second Close re-fired callbacks: order=%v", order) } } +// newProcessFakeFixture builds a process-mode fixture (parent == nil) over the +// fake EC2, which is the mode whose teardown failures surface through Close(). +func newProcessFakeFixture() (*Fixture, *fakeEC2) { + ec2c := &fakeEC2{} + return &Fixture{ + EC2: ec2c, + ELBv2: &fakeELB{}, + scratch: "test", + memo: map[string]string{}, + cleanups: map[string]struct{}{}, + }, ec2c +} + +// TestTeardownVolume_DetachesBeforeDeleting covers the ordering gap: fixture +// cleanups drain LIFO, so a volume created after its instance is torn down +// while still attached. Teardown must release it rather than let DeleteVolume +// fail with VolumeInUse and strand it. +func TestTeardownVolume_DetachesBeforeDeleting(t *testing.T) { + ec2c := &fakeEC2{volState: ec2.VolumeStateInUse} + + if err := teardownVolume(ec2c, "vol-fake"); err != nil { + t.Fatalf("teardownVolume err = %v, want nil", err) + } + if got := ec2c.detachVolume.Load(); got != 1 { + t.Fatalf("DetachVolume calls = %d, want 1 (attached volume was not released)", got) + } + if got := ec2c.deleteVolume.Load(); got != 1 { + t.Fatalf("DeleteVolume calls = %d, want 1", got) + } +} + +// TestTeardownVolume_AvailableSkipsDetach verifies the common case costs no +// extra API calls: an already-available volume is deleted directly. +func TestTeardownVolume_AvailableSkipsDetach(t *testing.T) { + ec2c := &fakeEC2{volState: ec2.VolumeStateAvailable} + + if err := teardownVolume(ec2c, "vol-fake"); err != nil { + t.Fatalf("teardownVolume err = %v, want nil", err) + } + if got := ec2c.detachVolume.Load(); got != 0 { + t.Fatalf("DetachVolume calls = %d, want 0 for an available volume", got) + } +} + +// TestEnsureVolume_CleanupReleasesAttachedVolume is the regression test for the +// leak itself: the volume is created available, then attached, and teardown has +// to cope. This is exactly the shape of a suite that calls EnsureInstance before +// EnsureVolume, since the LIFO chain tears the volume down first. +func TestEnsureVolume_CleanupReleasesAttachedVolume(t *testing.T) { + fx, ec2c := newProcessFakeFixture() + ec2c.volState = ec2.VolumeStateAvailable + + _ = EnsureVolume(t, fx, "ap-southeast-2a", 10) + ec2c.volState = ec2.VolumeStateInUse + + if err := fx.Close(); err != nil { + t.Fatalf("Close err = %v, want nil (attached volume was not released and is leaked)", err) + } + if got := ec2c.detachVolume.Load(); got != 1 { + t.Fatalf("DetachVolume calls = %d, want 1", got) + } +} + +// TestFixtureClose_ReportsLeak verifies a teardown that fails is reported as a +// leak rather than logged and forgotten. Before this, a failing cleanup left a +// real volume on the node and the run still went green. +func TestFixtureClose_ReportsLeak(t *testing.T) { + fx, ec2c := newProcessFakeFixture() + ec2c.volState = ec2.VolumeStateAvailable + ec2c.deleteVolumeErr = errors.New("boom") + + _ = EnsureVolume(t, fx, "ap-southeast-2a", 10) + + err := fx.Close() + if err == nil { + t.Fatalf("Close err = nil, want a leak report for the failed volume delete") + } + if !strings.Contains(err.Error(), "vol-fake") { + t.Fatalf("leak report %q does not name the leaked volume", err) + } +} + +// TestFixtureClose_PanicDoesNotStrandRest verifies one bad teardown cannot +// abort the chain. An aborted chain strands every resource registered before +// the panic, which is how a whole instance has previously been left running. +func TestFixtureClose_PanicDoesNotStrandRest(t *testing.T) { + fx, _ := newProcessFakeFixture() + + var ran []int + fx.RegisterCleanup(func() { ran = append(ran, 1) }) + fx.RegisterCleanup(func() { panic("teardown blew up") }) + fx.RegisterCleanup(func() { ran = append(ran, 3) }) + + err := fx.Close() + if len(ran) != 2 { + t.Fatalf("cleanups run = %v, want both non-panicking callbacks", ran) + } + if err == nil || !strings.Contains(err.Error(), "panicked") { + t.Fatalf("Close err = %v, want a report naming the panic", err) + } +} + // TestEnsureKeyPair_CleanupRunsOnce verifies cleanup fires DeleteKeyPair // exactly once even after multiple Ensure calls memoize to the same ID. func TestEnsureKeyPair_CleanupRunsOnce(t *testing.T) { diff --git a/tests/e2e/harness/guest_data.go b/tests/e2e/harness/guest_data.go index 4bebae1f5..cc42af5b0 100644 --- a/tests/e2e/harness/guest_data.go +++ b/tests/e2e/harness/guest_data.go @@ -22,19 +22,35 @@ import ( // guestSentinelFile is the file written into a freshly formatted data volume. const guestSentinelFile = "e2e-sentinel.bin" -// guestExecTimeout bounds a single guest command (mkfs/dd are sub-second on the -// small payloads used here; the ceiling only guards against a hung SSH). +// guestExecTimeout bounds a single guest command by default (mkfs/dd are +// sub-second on the small payloads most callers use; the ceiling only guards +// against a hung SSH). Callers driving a larger payload — enough data that +// sustaining it within this ceiling would require an unrealistic transfer +// rate — must use GuestExecTimeout with a budget sized to their own payload +// instead of assuming this default is high enough. const guestExecTimeout = 2 * time.Minute // sha256RE matches a bare sha256 digest so the checksum can be lifted out of // command output that may also carry the file path (`␠␠`). var sha256RE = regexp.MustCompile(`\b[0-9a-f]{64}\b`) -// GuestExec runs cmd over SSH against tgt and returns combined stdout+stderr -// plus the run error. It never calls t.Fatal, so callers can branch on an -// expected non-zero exit. +// GuestExec runs cmd over SSH against tgt with the default guestExecTimeout +// budget and returns combined stdout+stderr plus the run error. It never +// calls t.Fatal, so callers can branch on an expected non-zero exit. func GuestExec(tgt SSHTarget, cmd string) (string, error) { - ctx, cancel := context.WithTimeout(context.Background(), guestExecTimeout) + return GuestExecTimeout(tgt, cmd, guestExecTimeout) +} + +// GuestExecTimeout is GuestExec with a caller-supplied timeout, for commands +// whose payload size makes the default guestExecTimeout unrealistic (e.g. a +// multi-gigabyte write against a network-backed volume). If the command is +// still running when timeout elapses, the returned error is wrapped with +// context.DeadlineExceeded so callers can distinguish "the command ran and +// failed" from "the command never got the chance to finish" via errors.Is — +// a distinction that matters wherever the two would otherwise look like the +// same red test for different reasons. +func GuestExecTimeout(tgt SSHTarget, cmd string, timeout time.Duration) (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() args := []string{ "-o", "StrictHostKeyChecking=no", @@ -48,6 +64,9 @@ func GuestExec(tgt SSHTarget, cmd string) (string, error) { cmd, } out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput() + if err != nil && ctx.Err() == context.DeadlineExceeded { + return string(out), fmt.Errorf("guest command exceeded its %s timeout (still running when the deadline hit): %w: %w", timeout, context.DeadlineExceeded, err) + } return string(out), err } diff --git a/tests/e2e/iam/instance_iam_profile_test.go b/tests/e2e/iam/instance_iam_profile_test.go index 1ac98955e..13d951371 100644 --- a/tests/e2e/iam/instance_iam_profile_test.go +++ b/tests/e2e/iam/instance_iam_profile_test.go @@ -35,6 +35,28 @@ const ( // Self-bootstrapping: it creates its own role + two profiles under the // single-iamprofile- namespace, asserts the EC2 surface, then cleans up so // subsequent tests see no residue. +// +// Nothing here was ported down to the integration tier: its role/profile +// create+attach setup (CreateRole, AttachRolePolicy, CreateInstanceProfile x2, +// AddRoleToInstanceProfile) is scaffolding, not an assertion under test — the +// same IAM CRUD, including the AdministratorAccess-doesn't-pre-seed wrinkle, +// is already asserted by tests/integration's TestIAMRolesAndProfiles. The +// association lifecycle itself (Associate/AlreadyAssociated/Replace/ +// Disassociate/DeleteInstanceProfile-while-bound) is implemented in +// InstanceServiceImpl (handlers/ec2/instance/service_impl.go) against a real +// *vm.VM's mutable IamInstanceProfileArn/AssociationId fields — daemon-side +// state a live guest actually carries. A static stub could only stand in for +// it by modelling the whole instance lifecycle, which is why this stays live +// rather than moving down. The pure gateway-side half of that same code path +// (profile resolution, PassRole enforcement, ID enrichment, NATS error +// mapping in gateway/ec2/instance/IamInstanceProfileAssociation.go) is +// already exhaustively unit-tested in +// IamInstanceProfileAssociation_test.go, and RunInstances' iam:PassRole gate +// specifically is covered by tests/integration's +// TestRunInstances_DeniedWithoutPassRole. The `require.Equal(t, "running", …)` +// singleton guard below is a precondition check, not a wait — nothing here +// blocks on a state transition that a static stub couldn't already satisfy, +// so there is no incidental wait to drop. func runIAMInstanceProfileAssociation(t *testing.T, fix *Fixture) { harness.Phase(t, "Single — EC2 IAM Instance Profile Association") diff --git a/tests/e2e/iam/main_test.go b/tests/e2e/iam/main_test.go index e4cd64d93..cbef65da3 100644 --- a/tests/e2e/iam/main_test.go +++ b/tests/e2e/iam/main_test.go @@ -10,6 +10,7 @@ package iam import ( + "fmt" "os" "sync" "testing" @@ -34,7 +35,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/multinode/main_test.go b/tests/e2e/multinode/main_test.go index b0ae19621..a1b8f09fb 100644 --- a/tests/e2e/multinode/main_test.go +++ b/tests/e2e/multinode/main_test.go @@ -32,7 +32,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/natuplink/natuplink_test.go b/tests/e2e/natuplink/natuplink_test.go index 99b4204c7..f5dc93337 100644 --- a/tests/e2e/natuplink/natuplink_test.go +++ b/tests/e2e/natuplink/natuplink_test.go @@ -90,7 +90,11 @@ func TestNATUplink(t *testing.T) { awsCli := harness.NewAWSClient(t, env) hfx, err := harness.NewProcessFixture(awsCli) require.NoError(t, err, "harness fixture init") - t.Cleanup(hfx.Close) + t.Cleanup(func() { + if err := hfx.Close(); err != nil { + t.Errorf("e2e teardown: %v", err) + } + }) fix := &fixture{ env: env, diff --git a/tests/e2e/partialblock/main_test.go b/tests/e2e/partialblock/main_test.go new file mode 100644 index 000000000..f316b16d2 --- /dev/null +++ b/tests/e2e/partialblock/main_test.go @@ -0,0 +1,155 @@ +//go:build e2e + +// Package partialblock proves that concurrent sub-block writes to the same +// block survive a cold reread when they travel through NBD from a real guest. +// +// The claim is deliberately narrow. The cheap, deterministic gate for the +// underlying read-modify-write race is viperblock's own unit reproducer, which +// runs against a real predastore in a fraction of a second. This suite adds +// exactly one thing that unit test cannot: evidence that the race is reachable +// end to end, through the guest block layer and the NBD transport, rather than +// only through the engine's Go API. +// +// It is its own package rather than a case inside single/ because single/ +// shares one memoized instance across the whole package, and this suite needs +// a raw, unformatted volume it can write with O_DIRECT and then delete. A +// filesystem is precisely what prevents the workload from reaching the code +// under test: it coalesces writes into aligned full-block requests, and the +// read-modify-write path only runs for a partial block. +package partialblock + +import ( + "bufio" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/mulgadc/spinifex/tests/e2e/harness" +) + +var ( + pkgFixOnce sync.Once + pkgFix *Fixture + pkgFixErr error +) + +func TestMain(m *testing.M) { + code := m.Run() + if pkgFix != nil { + if pkgFix.Harness != nil { + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } + } + if pkgFix.TmpDir != "" { + _ = os.RemoveAll(pkgFix.TmpDir) + } + } + os.Exit(code) +} + +// Fixture carries per-process state shared across this package's tests. +type Fixture struct { + Env *harness.Env + AWS *harness.AWSClient + Harness *harness.Fixture + TmpDir string + + // PoolMode gates the suite: EIP allocation is what makes guest SSH + // reachable (harness.InstancePublicSSHHost refuses to fall back to + // qemu-hostfwd), and guest SSH is the only channel this suite has for + // driving the write workload and reading its verdict back. + PoolMode bool +} + +func (f *Fixture) ArtifactDir(t *testing.T) string { + t.Helper() + return harness.ArtifactDir(t, f.Env) +} + +// requirePartialBlockFixture returns the package-scoped Fixture singleton, +// building it on first call. Skips when SPINIFEX_E2E is unset, the mode is not +// single-node, or the env has no EIP pool. +func requirePartialBlockFixture(t *testing.T) *Fixture { + t.Helper() + pkgFixOnce.Do(func() { + if os.Getenv("SPINIFEX_E2E") == "" { + return + } + env := harness.LoadEnv(t) + if env.Mode != harness.ModeSingle { + pkgFixErr = nil + return + } + awsCli := harness.NewAWSClient(t, env) + h, err := harness.NewProcessFixture(awsCli) + if err != nil { + pkgFixErr = err + return + } + tmpDir, err := os.MkdirTemp("", "partialblock-pkgfix-*") + if err != nil { + pkgFixErr = err + return + } + harness.EnsureDefaultSGOpen(t, awsCli) + pkgFix = &Fixture{ + Env: env, + AWS: awsCli, + Harness: h, + TmpDir: tmpDir, + PoolMode: detectPoolMode(env), + } + }) + if pkgFixErr != nil { + t.Fatalf("partialblock fixture init: %v", pkgFixErr) + } + if pkgFix == nil { + t.Skip("SPINIFEX_E2E unset or SPINIFEX_MODE != single") + } + if !pkgFix.PoolMode { + t.Skip("partialblock suite requires external_mode=pool — a nat env has no EIP and cannot reach guest SSH, the only channel this suite has to drive and verify the write workload") + } + return pkgFix +} + +// detectPoolMode reads external_mode from spinifex.toml. Copied rather than +// imported for the same reason storagegrowth copies it: harness/** is +// deliberately never touched by these standalone-workload packages, and the +// predicate is small and package-local everywhere it already exists. +func detectPoolMode(env *harness.Env) bool { + cfg := os.ExpandEnv("$HOME/spinifex/config/spinifex.toml") + if env.ConfigDir != "" { + cfg = filepath.Join(env.ConfigDir, "spinifex.toml") + } + f, err := os.Open(cfg) + if err != nil { + return false + } + defer f.Close() + + scanner := bufio.NewScanner(f) + inNetwork := false + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if strings.HasPrefix(line, "[") { + inNetwork = line == "[network]" + continue + } + if !inNetwork || !strings.HasPrefix(line, "external_mode") { + continue + } + // external_mode = "pool" — quoted value; only "pool" enables this suite. + if _, rhs, ok := strings.Cut(line, "="); ok { + val := strings.Trim(strings.TrimSpace(rhs), "\"'") + return val == "pool" + } + } + return false +} diff --git a/tests/e2e/partialblock/partial_block_durability_test.go b/tests/e2e/partialblock/partial_block_durability_test.go new file mode 100644 index 000000000..dc6d9da0b --- /dev/null +++ b/tests/e2e/partialblock/partial_block_durability_test.go @@ -0,0 +1,334 @@ +//go:build e2e + +package partialblock + +import ( + "fmt" + "os" + "regexp" + "strconv" + "testing" + "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" +) + +const ( + // partialBlockDevice is the guest-visible attach point requested for the + // working-set volume. + partialBlockDevice = "/dev/sdp" + + // partialBlockVolumeSizeGiB is the smallest volume that comfortably holds + // the working region. The region itself is a few MiB at the head of the + // device; the rest is never touched. + partialBlockVolumeSizeGiB = 1 + + // blockSize mirrors viperblock's block granularity. The read-modify-write + // path under test runs only when a write covers part of one of these, so + // this value has to match the engine's or the workload silently degrades + // into full-block overwrites that exercise nothing. + blockSize = 4096 + // halfSize splits a block into two disjoint sub-block ranges. Two writers + // each own one half, so every write is partial by construction and the two + // writers collide on the same block without ever overlapping each other's + // bytes — a lost update is therefore unambiguous, not a write-write tie. + halfSize = blockSize / 2 + + // defaultBlocks and defaultRounds size the workload. Both are modest on + // purpose: the predecessor suite moved a 2048 MiB working set, which after + // RS(2,1) left roughly 3 GiB of backend behind per run and twice drove the + // node under predastore's nearfull watermark, wedging every guest on it. + // Collision probability here comes from two writers hammering the same + // blocks in the same order, not from volume. + defaultBlocks = 2048 + defaultRounds = 6 + + // partialBlockIOBaseTimeout is the fixed part of the IO budget: SSH setup + // and interpreter startup, independent of how much data moves. + partialBlockIOBaseTimeout = 60 * time.Second + // partialBlockIOPerWrite budgets one O_DIRECT sub-block write. Deliberately + // generous: each one is a synchronous round trip to a network-backed + // volume, and a tight bound risks the failure mode this suite must avoid — + // a timeout that looks like a red durability test but proves nothing about + // whether any write was lost. + partialBlockIOPerWrite = 3 * time.Millisecond +) + +// wroteRE captures the per-writer byte count each child reports, so the test +// can assert both writers moved exactly the bytes they were asked to rather +// than trusting an exit status alone. +var wroteRE = regexp.MustCompile(`WROTE half=(\d+) bytes=(\d+)`) + +// checkedRE captures the verifier's tally. +var checkedRE = regexp.MustCompile(`CHECKED (\d+) BAD (\d+)`) + +// partialBlockDriver is the guest-side workload. It is Python rather than a +// shell pipeline because the workload needs three things a shell cannot give: +// O_DIRECT with a correctly aligned buffer, writes at sub-block offsets that +// reach the device instead of being merged in the page cache, and a real exit +// status per writer. +// +// That last point is why the predecessor suite could not fail. It ended its +// write phase with `... & wait; sync; sha256sum`, and the status observed was +// sha256sum's; bare `wait` returns 0 unconditionally, so every writer error was +// invisible and a backend ENOSPC presented as a pass. +const partialBlockDriver = ` +import mmap, os, struct, sys + +BLOCK = 4096 +HALF = 2048 +REC = 32 +MAGIC = 0x5042444B + +def fill(buf, blk, half, rnd): + # Stamp the whole half, not just a header, so a splice that reverts only + # part of the range is still caught. + rec = struct.pack(" 0 || writeEnd < blockSize); +// a fully covered block takes a full-overwrite branch that reads nothing and +// can lose nothing. Two writers that both read generation N, each splice their +// own half, and let the higher sequence number win wholesale is the shape of +// the lost update this asserts against — so one half reading back at an +// earlier round is the signature. +// +// The device is raw and unformatted on purpose. A filesystem defeats this +// workload: it coalesces writes into aligned full-block requests, which is how +// the predecessor suite ended up issuing zero sub-4096 and zero unaligned +// requests and passing on a build that still carried the bug. +func TestPartialBlockConcurrentWriteDurability(t *testing.T) { + fix := requirePartialBlockFixture(t) + harness.Phase(t, "Partial-Block Concurrent-Write Durability") + + blocks := envInt(t, "SPINIFEX_PARTIALBLOCK_BLOCKS", defaultBlocks) + rounds := envInt(t, "SPINIFEX_PARTIALBLOCK_ROUNDS", defaultRounds) + writesPerWriter := blocks * rounds + wantBytes := writesPerWriter * halfSize + + instanceID, tgt := ensurePartialBlockInstance(t, fix) + az := harness.DiscoverDefaultAZ(t, fix.Harness) + volID := harness.EnsureVolume(t, fix.Harness, az, partialBlockVolumeSizeGiB) + + harness.Detail(t, "blocks", blocks, "rounds", rounds, "region_mib", blocks*blockSize/(1<<20), + "instance", instanceID, "volume", volID) + + before := harness.GuestDiskSet(t, tgt) + harness.AttachVolumeWait(t, fix.AWS, volID, instanceID, partialBlockDevice) + dev := harness.WaitForNewGuestDisk(t, tgt, before, 90*time.Second) + + // Detach before the fixture's own cleanup deletes the volume. DeleteVolume + // against a still-attached volume is what leaks volumes on teardown, and + // the delete itself is left to the fixture so the volume is released + // exactly once. + t.Cleanup(func() { harness.DetachVolumeWait(t, fix.AWS, volID) }) + + installDriver(t, tgt) + + budget := partialBlockIOBaseTimeout + time.Duration(writesPerWriter)*partialBlockIOPerWrite + harness.Step(t, "writing %d rounds x %d blocks x 2 halves O_DIRECT to /dev/%s (budget %s)", rounds, blocks, dev, budget) + out, err := harness.GuestExecTimeout(tgt, + fmt.Sprintf("sudo python3 /tmp/partialblock.py write /dev/%s %d %d", dev, blocks, rounds), budget) + if err != nil { + t.Fatalf("write phase failed (this is a writer error or timeout, NOT evidence of a lost update): %v\n%s", err, out) + } + + // Assert both writers moved exactly the bytes asked of them. Without this a + // writer that failed early could leave the device holding a consistent but + // stale round, and the verify phase would agree with itself. + assertWriterBytes(t, out, wantBytes) + + harness.Step(t, "dropping page cache") + if out, err := harness.GuestExec(tgt, "sudo sync && echo 3 | sudo tee /proc/sys/vm/drop_caches >/dev/null"); err != nil { + t.Fatalf("drop_caches: %v\n%s", err, out) + } + + harness.Step(t, "cold O_DIRECT reread, asserting every half carries round %d", rounds) + out, err = harness.GuestExecTimeout(tgt, + fmt.Sprintf("sudo python3 /tmp/partialblock.py verify /dev/%s %d %d", dev, blocks, rounds), budget) + if err != nil { + // A non-zero verify exit is the durability failure itself, so the + // output carries the per-block detail and must be surfaced whole. + t.Fatalf("cold reread found lost updates — a partial write was acked and did not survive:\n%s", out) + } + + checked, bad := parseChecked(t, out) + if bad != 0 { + t.Errorf("verify reported %d bad halves out of %d checked", bad, checked) + } + if want := blocks * 2 * 2; checked != want { + t.Errorf("verify checked %d records, want %d — the verifier did not cover the whole region", checked, want) + } + harness.Detail(t, "records_checked", checked, "records_bad", bad) +} + +// ensurePartialBlockInstance launches (or returns the memoized) guest and waits +// for SSH. Guest size is deliberately not a parameter: the memory-pressure +// trigger model the predecessor suite encoded was retracted — the fault +// reproduces with a single sequential writer on a guest with memory to spare — +// so the smallest catalog entry is the right choice here. +func ensurePartialBlockInstance(t *testing.T, fix *Fixture) (instanceID string, tgt harness.SSHTarget) { + t.Helper() + instType, arch := harness.DiscoverNanoInstanceType(t, fix.Harness) + ami := harness.DiscoverUbuntuAMI(t, fix.Harness, arch) + keyName, keyPath := harness.EnsureKeyPair(t, fix.Harness, fix.ArtifactDir(t)) + vpc := harness.EnsureDefaultVPC(t, fix.Harness) + harness.AuthorizeSSHIngress(t, fix.AWS, vpc.SGID) + + instanceID = harness.EnsureInstance(t, fix.Harness, harness.InstanceSpec{ + AMIID: ami, + InstanceType: instType, + KeyName: keyName, + SubnetID: vpc.SubnetID, + SGID: vpc.SGID, + }) + inst := harness.WaitForInstanceState(t, fix.AWS, instanceID, "running") + host, port := harness.InstancePublicSSHHost(t, inst) + + harness.Step(t, "waiting for guest SSH on %s:%d (instance_type=%s)", host, port, instType) + if !harness.TryGuestSSHReady(host, port, "ubuntu", keyPath, 5*time.Minute) { + t.Fatalf("guest %s SSH %s:%d not ready after 5m", instanceID, host, port) + } + return instanceID, harness.SSHTarget{User: "ubuntu", Host: host, Port: port, KeyPath: keyPath} +} + +// installDriver writes the guest-side workload to /tmp and confirms python3 can +// parse it, so a syntax error surfaces here rather than as an opaque non-zero +// exit in the middle of the write phase. +func installDriver(t *testing.T, tgt harness.SSHTarget) { + t.Helper() + script := fmt.Sprintf("cat > /tmp/partialblock.py <<'PBEOF'\n%s\nPBEOF\npython3 -c 'import py_compile,sys; py_compile.compile(\"/tmp/partialblock.py\", doraise=True)'", partialBlockDriver) + if out, err := harness.GuestExec(tgt, script); err != nil { + t.Fatalf("installDriver: %v\n%s", err, out) + } +} + +// assertWriterBytes checks that both halves reported completion and that each +// moved exactly wantBytes. +func assertWriterBytes(t *testing.T, out string, wantBytes int) { + t.Helper() + seen := map[int]int{} + for _, m := range wroteRE.FindAllStringSubmatch(out, -1) { + half, _ := strconv.Atoi(m[1]) + got, _ := strconv.Atoi(m[2]) + seen[half] = got + } + for half := 0; half < 2; half++ { + got, ok := seen[half] + if !ok { + t.Fatalf("writer half=%d reported no byte count — it did not run to completion:\n%s", half, out) + } + if got != wantBytes { + t.Fatalf("writer half=%d wrote %d bytes, want %d — short write path, not a durability result:\n%s", half, got, wantBytes, out) + } + } +} + +// parseChecked pulls the verifier's tally out of its output. +func parseChecked(t *testing.T, out string) (checked, bad int) { + t.Helper() + m := checkedRE.FindStringSubmatch(out) + if m == nil { + t.Fatalf("verify produced no CHECKED line — cannot tell whether it examined anything:\n%s", out) + } + checked, _ = strconv.Atoi(m[1]) + bad, _ = strconv.Atoi(m[2]) + return checked, bad +} + +// envInt reads a positive integer override, failing outright on a malformed +// value rather than silently falling back to the default. +func envInt(t *testing.T, key string, def int) int { + t.Helper() + v := os.Getenv(key) + if v == "" { + return def + } + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + t.Fatalf("%s=%q is not a positive integer", key, v) + } + return n +} diff --git a/tests/e2e/single/main_test.go b/tests/e2e/single/main_test.go index 504e230e7..a9c335dc3 100644 --- a/tests/e2e/single/main_test.go +++ b/tests/e2e/single/main_test.go @@ -10,6 +10,7 @@ package single import ( "bufio" + "fmt" "os" "path/filepath" "strings" @@ -36,7 +37,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/storagegrowth/image_pull_test.go b/tests/e2e/storagegrowth/image_pull_test.go index 52c6a0874..01985f37a 100644 --- a/tests/e2e/storagegrowth/image_pull_test.go +++ b/tests/e2e/storagegrowth/image_pull_test.go @@ -1184,13 +1184,8 @@ func captureImagePullSample(t *testing.T, tgt harness.SSHTarget, ssh *harness.Pe s.GuestUsedBytes = v } - duCtx, duCancel := context.WithTimeout(context.Background(), 30*time.Second) - duOut, duErr := ssh.Run(duCtx, wanHost, fmt.Sprintf("sudo du -s --block-size=1 %s 2>/dev/null | cut -f1", predastoreBaseDir)) - duCancel() - if duErr != nil { - t.Logf("storagegrowth: host du via PeerSSH on %s: %v", wanHost, duErr) - } else if v, perr := parseLastIntField(string(duOut)); perr != nil { - t.Logf("storagegrowth: parse host du output %q: %v", duOut, perr) + if v, err := hostPredastoreAllocatedBytes(ssh, wanHost); err != nil { + t.Logf("storagegrowth: host du via PeerSSH on %s: %v", wanHost, err) } else { s.HostAllocatedBytes = v } @@ -1316,6 +1311,32 @@ func parseLastIntField(out string) (int64, error) { return strconv.ParseInt(fields[len(fields)-1], 10, 64) } +// hostPredastoreAllocatedBytes reads predastoreBaseDir's total allocated +// bytes on the node at wanHost, over PeerSSH. This is the one measurement +// that actually works against a live deployment: it needs no S3 credential +// at all, only SSH access to the node, unlike a per-volume S3 listing of the +// backend bucket (see assertNPassBackendByteDelta in storage_growth_test.go +// for why that path is unusable here). +// +// The number returned is NODE-WIDE, not attributable to any single volume -- +// du has no concept of which volume a chunk object belongs to, it only sees +// bytes on disk. Callers that need a per-workload signal must take two +// readings and use the delta, and must be able to argue nothing else was +// writing to the node in between. +func hostPredastoreAllocatedBytes(ssh *harness.PeerSSH, wanHost string) (int64, error) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + out, err := ssh.Run(ctx, wanHost, fmt.Sprintf("sudo du -s --block-size=1 %s 2>/dev/null | cut -f1", predastoreBaseDir)) + if err != nil { + return 0, fmt.Errorf("du %s via PeerSSH: %w", predastoreBaseDir, err) + } + v, perr := parseLastIntField(string(out)) + if perr != nil { + return 0, fmt.Errorf("parse du output %q: %w", out, perr) + } + return v, nil +} + // writeImagePullResult persists the workload record at the path named by // npassResultPathEnv (SPINIFEX_STORAGEGROWTH_RESULT_PATH) — the same env var // npass uses, per the shared result contract. See writeNPassResult for why an diff --git a/tests/e2e/storagegrowth/main_test.go b/tests/e2e/storagegrowth/main_test.go index 3fe3054f2..097a56ecb 100644 --- a/tests/e2e/storagegrowth/main_test.go +++ b/tests/e2e/storagegrowth/main_test.go @@ -15,21 +15,40 @@ // // That is a different fault from the *overwrite-orphan* leak, where a key is // rewritten in place and the superseded extent is never tombstoned, so -// compaction can never find it. Keeping the two apart is the whole point, and -// it is why this package does not measure. The two leaks are distinguished -// precisely by predastore's own live-versus-dead split: unreferenced-live bytes -// inflate live totals, overwrite-orphan bytes inflate dead totals. A host-side -// `du` reports one number that mixes them, which cannot separate the target from -// the confound — and this workload drives both at once, because the drain each -// pass forces also rewrites the checkpoint map in place. So the backend is -// measured out of band by a tool that reads predastore's accounting directly. -// That tool does not exist in a standalone spinifex checkout, and depending on -// it here would break this suite's CI. +// compaction can never find it. Keeping the two apart matters for attributing +// blame within a leak, and that attribution is why this package's own +// in-process check stops at a coarse pass/fail gate rather than a full +// breakdown. The two leaks are distinguished precisely by predastore's own +// live-versus-dead split: unreferenced-live bytes inflate live totals, +// overwrite-orphan bytes inflate dead totals. A host-side `du` reports one +// number that mixes them, which cannot separate the target from the confound +// — and this workload drives both at once, because the drain each pass +// forces also rewrites the checkpoint map in place. Splitting that mix +// requires a tool that reads predastore's live-versus-dead accounting +// directly; that tool does not exist in a standalone spinifex checkout, and +// depending on it here would break this suite's CI. // -// The division of labour: this package drives the guest and reports what it did -// (see npassWorkload); the measurement step reads that record for the measured -// denominator and supplies the backend numbers. Run standalone, the workload -// still asserts the guest-side invariants it can see on its own. +// What this package's own tests DO assert on, in-process, is the coarser +// signal a node-wide `du` delta across the workload can still catch: the +// combined live+dead footprint growing well beyond the RS-erasure floor +// (see assertNPassBackendByteDelta in storage_growth_test.go). A per-volume +// S3 listing of the volume's own chunk prefix — what this suite originally +// used, and what the external tool below also reads — turned out to be +// unusable against a live deployment: the internal chunk bucket is owned by +// a system account distinct from the tenant-scoped identity this harness +// authenticates as, and predastore's same-account bucket-ownership check has +// no cross-account bucket-policy escape hatch, so every such listing fails +// closed with AccessDenied regardless of IAM grants. `du` needs no S3 +// credential at all, only SSH to the node, which is why it is what this +// package can actually run itself. +// +// The division of labour: this package drives the guest, runs its own +// coarse node-wide gate, and reports what it did (see npassWorkload); an +// external measurement step reads that record for the measured denominator +// and supplies the fine-grained live-versus-dead backend breakdown this +// package cannot produce on its own. Run standalone, the workload still +// asserts every guest-side invariant, and the coarse gate, that it can see +// on its own. // // This is its own package rather than living in single/ or multinode/: // single/ shares one singleton instance across the whole package that a @@ -46,6 +65,7 @@ package storagegrowth import ( "bufio" + "fmt" "os" "path/filepath" "strings" @@ -66,7 +86,12 @@ func TestMain(m *testing.M) { code := m.Run() if pkgFix != nil { if pkgFix.Harness != nil { - pkgFix.Harness.Close() + // A leaked resource fails the run: the suite may have passed, but it + // left state on the node that the next run will trip over. + if err := pkgFix.Harness.Close(); err != nil { + fmt.Fprintf(os.Stderr, "e2e teardown: %v\n", err) + code = 1 + } } if pkgFix.TmpDir != "" { _ = os.RemoveAll(pkgFix.TmpDir) diff --git a/tests/e2e/storagegrowth/storage_growth_test.go b/tests/e2e/storagegrowth/storage_growth_test.go index 2f324585f..6158d26ab 100644 --- a/tests/e2e/storagegrowth/storage_growth_test.go +++ b/tests/e2e/storagegrowth/storage_growth_test.go @@ -54,18 +54,63 @@ const ( // test PASS. A handoff written there would exist only for runs that failed // — precisely inverted from what a measurement step needs. npassResultPathEnv = "SPINIFEX_STORAGEGROWTH_RESULT_PATH" + + // npassBackendByteAccountingTolerance is the ceiling ratio of node-wide + // predastore-allocated-byte growth to guest-written logical bytes a + // healthy N-pass run may show. + // + // The floor is viperblock's own encoding overhead, not slack for a leak: + // chunks are written under Reed-Solomon 2+1 erasure coding, which splits + // each object into two data shards plus one parity shard of the same + // size -- 1.5x the logical payload, not 3x replication -- and every + // block is additionally framed with a 16-byte AES-GCM tag. + // + // That 1.5x floor is a precondition, not a universal constant. It holds + // because this suite provisions RS(2,1): cmd/spinifex/cmd/templates/ + // predastore.toml's [rs] block defaults to data=2, parity=1. The same + // template documents RS(3,2) as the production recommendation, whose + // floor is 5/3 = ~1.667x and would leave only ~12% headroom under the + // ceiling below instead of the ~25% intended. Recompute this value if the + // suite is ever pointed at a different RS profile rather than assuming + // the band still holds. + // + // A healthy, fully-GC'd volume settles at ~1.5x and does not go + // materially above it. The tolerance widens that floor by 25% (to 1.875x) to + // absorb two effects that are real but not leaks: every chunk pays a + // fixed header regardless of how many blocks it packs (pulling a + // sparse volume's ratio up), and the tail chunk of any write run is + // under-packed by construction (a chunk flushes on full or on drain, + // so the last one of a run is real, unavoidable overhead). Both shrink + // as a proportion of the total as the workload's footprint grows, so a + // large workload should sit close to the bare floor and a small one + // gets the extra headroom it actually needs. + // + // This does not attempt to catch the slow, bounded residual that a + // future partial-chunk compaction pass would close -- only the + // unbounded-growth shape the storage-growth incident produced, where + // a leak multiplies with every overwrite pass instead of settling. + // + // Applying this floor to a node-wide `du` delta (rather than a per-volume + // S3 listing, see assertNPassBackendByteDelta) is only sound because the + // delta is measured tightly around this workload's own passes on a + // dedicated, single-volume, serial environment -- nothing else on the + // node is expected to write to predastore's store in that window. Any + // concurrent writer (another suite, another volume, a background GC/ + // compaction pass) would inflate or shrink the delta for reasons that + // have nothing to do with this workload, and the ratio below would no + // longer mean what it claims to. + npassBackendByteAccountingTolerance = 1.875 ) // npassWorkload records what one invocation of the N-pass workload did. It is -// the whole contract with whatever measures the backing store afterwards. +// the whole contract with whatever measures the backing store afterwards, +// beyond this suite's own coarse in-process gate (assertNPassBackendByteDelta). // -// This suite deliberately reports rather than measures. The guest-side -// quantities below are the ones only a guest can know; the backend's byte -// cost is measured out of band by a tool that can read predastore's own -// live-versus-dead accounting, which is a distinction no host-side `du` can -// make and which is the entire point of the exercise. Coupling this test to -// that tool would also break a standalone spinifex checkout, where it does -// not exist. +// The guest-side quantities below are the ones only a guest can know; a +// live-versus-dead breakdown of the backend's byte cost -- a distinction no +// host-side `du` can make -- is measured out of band by a tool that can read +// predastore's own accounting directly. Coupling this test to that tool would +// also break a standalone spinifex checkout, where it does not exist. // // GuestUsedBytes is measured, never inferred. It is the filesystem's own // used-bytes count after the write, so it includes ext4 metadata as well as @@ -132,12 +177,81 @@ func TestNPassOverwrite(t *testing.T) { harness.Detail(t, "passes", passes, "extent_mib", extentMiB, "instance", instanceID, "volume", volID, "retain_volume", retain) + // Baseline predastore's node-wide allocated bytes immediately before the + // passes start, and read it again immediately after they finish, so the + // delta below brackets this workload's own writes as tightly as + // possible. See assertNPassBackendByteDelta for why this must be a + // delta rather than an absolute snapshot, and why the window must stay + // tight. + ssh := harness.NewPeerSSH() + before, err := hostPredastoreAllocatedBytes(ssh, fix.Env.WANHost) + require.NoErrorf(t, err, "backend byte accounting: baseline node du before passes") + w := runNPassOverwrite(t, fix, tgt, instanceID, volID, passes, extentMiB) w.Retained = retain writeNPassResult(t, w) harness.Detail(t, "passes_completed", w.Passes, "guest_used_bytes", w.GuestUsedBytes, "guest_used_per_pass", fmt.Sprint(w.GuestUsedPerPass)) + + after, err := hostPredastoreAllocatedBytes(ssh, fix.Env.WANHost) + require.NoErrorf(t, err, "backend byte accounting: node du after passes") + + // This is the fail-first demonstration for the storage-growth leak: if + // superseded chunks from the earlier passes are never reclaimed, the + // backend footprint grows by N times the final pass's guest-used bytes + // rather than settling at the erasure floor, and the delta below catches + // that shape without needing vbrefscan's out-of-band scoring pass. It + // complements, not replaces, the RETAIN_VOLUME external-measurement + // path: that path attributes *which* chunks are garbage on a volume kept + // alive for vbrefscan; this one is a pass/fail gate that runs on every + // invocation, retained or not. + assertNPassBackendByteDelta(t, before, after, w.GuestUsedBytes) +} + +// assertNPassBackendByteDelta asserts that predastore's node-wide allocated +// bytes grew, across the N-pass workload, by no more than +// npassBackendByteAccountingTolerance times guestWrittenBytes. +// +// This measures via `du` on the node (hostPredastoreAllocatedBytes) rather +// than listing the volume's own S3 chunk prefix. A per-volume S3 listing is +// what this suite originally used and is what vbrefscan itself reads, but it +// is not usable here: the internal bucket EBS volume chunks live under is +// owned by a system account distinct from the tenant-scoped identity this +// harness authenticates as, predastore enforces a same-account +// bucket-ownership check independent of and on top of any IAM policy grant, +// and cross-account bucket-policy or ACL grants are not implemented -- every +// listing attempt fails with AccessDenied regardless of what IAM policy the +// caller holds. There is also no usable filesystem fallback: predastore's +// distributed backend stores objects in badger plus shard nodes, not a tree +// keyed by object name, so there is nothing under predastoreBaseDir to +// attribute to a single volume by path. +// +// `du` over predastoreBaseDir is therefore the finest-grained measurement +// actually available, and it is NODE-WIDE: it cannot distinguish this +// volume's bytes from any other write happening on the same node. The before/ +// after delta (rather than an absolute reading) is what makes the number +// mean anything at all, and even the delta is only valid because this +// workload runs serially against a single volume on a dedicated environment +// -- nothing else is expected to write to predastore's store in the window +// between the two samples. This approach would be invalid under concurrency: +// a second suite or a background compaction pass writing to the same node in +// that window would be silently attributed to this workload's ratio. +func assertNPassBackendByteDelta(t *testing.T, before, after, guestWrittenBytes int64) { + t.Helper() + require.Positivef(t, guestWrittenBytes, + "backend byte accounting: guestWrittenBytes must be positive, got %d -- the denominator every ratio divides by is unusable", + guestWrittenBytes) + + delta := after - before + ratio := float64(delta) / float64(guestWrittenBytes) + harness.Detail(t, "backend_allocated_bytes_before", before, "backend_allocated_bytes_after", after, + "backend_allocated_bytes_delta", delta, "guest_written_bytes", guestWrittenBytes, + "ratio", fmt.Sprintf("%.3f", ratio)) + + require.LessOrEqualf(t, ratio, npassBackendByteAccountingTolerance, + "backend byte accounting: node predastore allocated bytes grew %d across the N-pass workload, %.2fx guest-written %d bytes -- exceeds the %.3fx RS-erasure-floor tolerance; backend growth looks unbounded, not settled at the floor", + delta, ratio, guestWrittenBytes, npassBackendByteAccountingTolerance) } // runNPassOverwrite drives n complete format+write+fsync+detach cycles against diff --git a/tests/integration/ecr_registry_test.go b/tests/integration/ecr_registry_test.go new file mode 100644 index 000000000..6dcc7aaae --- /dev/null +++ b/tests/integration/ecr_registry_test.go @@ -0,0 +1,173 @@ +//go:build integration + +package integration + +import ( + "fmt" + "net/http" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ecr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// This file covers the ECR data plane — the OCI Distribution v2 surface the +// live ecr/ suite drove through docker, crane and skopeo. Those tests need no +// guest: they need a registry, an object store and an HTTP client, all of which +// the in-process gateway already provides on an ephemeral port. Running them +// here means a registry regression fails in the front gate rather than after a +// guest boot, and removes the last reason the ecr suite had to sit behind full +// QEMU provisioning. +// +// What is deliberately NOT moved down: the capacity/load leg, which exists to +// put real parallel pressure on a real predastore. A memory-backed object store +// cannot make that claim, so it stays live. + +// pushBlob uploads one blob through the two-step Distribution upload flow and +// returns its digest. Split out from seedOCIRepo (which pushes a fixed +// single-blob fixture) because these tests push several blobs and then read the +// exact bytes back. +func pushBlob(t *testing.T, gw *Gateway, repo, bearer string, data []byte) string { + t.Helper() + digest := ociDigest(data) + + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", bearer, nil) + require.Equal(t, http.StatusAccepted, status, "start blob upload: %s", body) + loc := hdr.Get("Location") + require.NotEmpty(t, loc, "upload response missing Location header") + + status, _, body = ecrOCIRequest(t, gw, http.MethodPut, loc+"?digest="+digest, bearer, data) + require.Equal(t, http.StatusCreated, status, "finish blob upload: %s", body) + return digest +} + +// pushManifest publishes a manifest referencing cfgDigest and layerDigests at +// tag, returning the digest the registry assigned it. +func pushManifest(t *testing.T, gw *Gateway, repo, bearer, tag, cfgDigest string, layerDigests ...string) string { + t.Helper() + + layers := "" + for i, d := range layerDigests { + if i > 0 { + layers += "," + } + layers += fmt.Sprintf(`{"digest":"%s"}`, d) + } + manifest := fmt.Appendf(nil, + `{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[%s]}`, + ociManifestMediaType, cfgDigest, layers) + + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPut, "/v2/"+repo+"/manifests/"+tag, bearer, manifest) + require.Equal(t, http.StatusCreated, status, "push manifest: %s", body) + digest := hdr.Get("Docker-Content-Digest") + require.NotEmpty(t, digest, "manifest push did not return Docker-Content-Digest") + return digest +} + +// TestECRRegistryPushPullRoundTrip is the integration-tier equivalent of the +// live docker/crane push-pull tests: push a config blob and a layer blob, push +// a manifest referencing both, then read all three back and assert the bytes +// are byte-identical to what was pushed. +// +// Asserting the returned bytes rather than only the status codes is the point. +// A registry that stores a blob under the wrong digest, truncates it, or serves +// a different blob still answers 200 to every request in this sequence; only +// comparing content catches it. +func TestECRRegistryPushPullRoundTrip(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + repo := uniqueName("roundtrip") + _, err := gw.ECRClient(t).CreateRepository(&ecr.CreateRepositoryInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "create-repository") + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + + cfg := []byte(`{"architecture":"amd64","os":"linux"}`) + layer := []byte("integration-tier layer payload, deliberately not empty") + cfgDigest := pushBlob(t, gw, repo, bearer, cfg) + layerDigest := pushBlob(t, gw, repo, bearer, layer) + manifestDigest := pushManifest(t, gw, repo, bearer, "v1", cfgDigest, layerDigest) + + // Pull each blob back and compare content, not just status. + for name, want := range map[string][]byte{cfgDigest: cfg, layerDigest: layer} { + status, _, got := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/blobs/"+name, bearer, nil) + require.Equal(t, http.StatusOK, status, "pull blob %s", name) + assert.Equal(t, want, got, "blob %s read back with different content than was pushed", name) + } + + // The manifest must be retrievable by tag and by digest, and both must + // agree — a registry that resolves a tag to a stale manifest still serves + // 200 for each request taken on its own. + statusTag, _, byTag := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + require.Equal(t, http.StatusOK, statusTag, "pull manifest by tag: %s", byTag) + statusDigest, _, byDigest := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/"+manifestDigest, bearer, nil) + require.Equal(t, http.StatusOK, statusDigest, "pull manifest by digest: %s", byDigest) + assert.Equal(t, byTag, byDigest, "tag v1 and digest %s resolve to different manifests", manifestDigest) + assert.Equal(t, manifestDigest, ociDigest(byTag), "served manifest does not hash to the digest the registry assigned it") + + // The control plane must see what the data plane stored; the live suite + // asserted this to catch a push that lands in the object store without a + // metadata record. + out, err := gw.ECRClient(t).DescribeImages(&ecr.DescribeImagesInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "describe-images") + var tags []string + for _, d := range out.ImageDetails { + tags = append(tags, aws.StringValueSlice(d.ImageTags)...) + } + assert.Contains(t, tags, "v1", "pushed tag not visible via DescribeImages") +} + +// TestECRRegistryPushToUncreatedRepoRejected asserts the registry does not +// auto-create a repository on push. Auto-creation is the permissive default in +// some registries, and it would silently defeat repository-level authorization: +// a principal denied on every existing repo could simply push to a new name. +func TestECRRegistryPushToUncreatedRepoRejected(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + // A valid token for the registry, but a repository that was never created. + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + repo := uniqueName("never-created") + + status, _, body := ecrOCIRequest(t, gw, http.MethodPost, "/v2/"+repo+"/blobs/uploads/", bearer, nil) + assert.NotEqual(t, http.StatusAccepted, status, + "registry accepted an upload for a repository that was never created (auto-create would bypass per-repository authorization): %s", body) + + // And it must not have been created as a side effect of the attempt. + _, err := gw.ECRClient(t).DescribeRepositories(&ecr.DescribeRepositoriesInput{ + RepositoryNames: []*string{aws.String(repo)}, + }) + assert.Error(t, err, "repository %s exists after a rejected push", repo) +} + +// TestECRRegistryRetagWithinRegistry covers what the live skopeo-copy test +// covered: re-publishing an existing manifest under a second tag. Both tags +// must resolve to the same digest, and the original must keep working — a +// retag that rewrites rather than aliases would break the first tag. +func TestECRRegistryRetagWithinRegistry(t *testing.T) { + gw := StartGateway(t) + StartECRDaemonLite(t, gw) + + repo := uniqueName("retag") + _, err := gw.ECRClient(t).CreateRepository(&ecr.CreateRepositoryInput{RepositoryName: aws.String(repo)}) + require.NoError(t, err, "create-repository") + bearer := ecrGetLoginPassword(t, gw.ECRClient(t)) + + cfg := []byte(`{"architecture":"amd64","os":"linux","note":"retag"}`) + cfgDigest := pushBlob(t, gw, repo, bearer, cfg) + v1Digest := pushManifest(t, gw, repo, bearer, "v1", cfgDigest) + + // Re-publish the identical manifest bytes under a second tag, which is what + // a copy within the same registry reduces to. + _, _, manifestBytes := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + status, hdr, body := ecrOCIRequest(t, gw, http.MethodPut, "/v2/"+repo+"/manifests/v2", bearer, manifestBytes) + require.Equal(t, http.StatusCreated, status, "push retagged manifest: %s", body) + assert.Equal(t, v1Digest, hdr.Get("Docker-Content-Digest"), "identical manifest content produced a different digest under the new tag") + + // The original tag must still resolve, and to the same manifest. + statusV1, _, stillV1 := ecrOCIRequest(t, gw, http.MethodGet, "/v2/"+repo+"/manifests/v1", bearer, nil) + require.Equal(t, http.StatusOK, statusV1, "original tag stopped resolving after retag: %s", stillV1) + assert.Equal(t, manifestBytes, stillV1, "original tag resolves to different content after retag") +}