From 6abae1112255ca0dc4aacfbee53e0a8fbba6d537 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 10 Feb 2026 16:57:50 +1100 Subject: [PATCH] feat: real snapshots (#32) --- .claude/CLAUDE.md | 5 +- docs/development/feature/snapshot.md | 164 +++++++ go.mod | 2 +- go.sum | 6 +- hive/config/config.go | 11 + hive/daemon/daemon.go | 2 +- hive/handlers/ec2/instance/service_impl.go | 93 ++-- hive/handlers/ec2/snapshot/service_impl.go | 62 ++- .../ec2/snapshot/service_impl_test.go | 2 +- hive/handlers/ec2/volume/service_impl.go | 192 ++++++++- hive/handlers/ec2/volume/service_impl_test.go | 408 ++++++++++++++++++ hive/services/viperblockd/viperblockd.go | 89 +++- tests/e2e/TEST_COVERAGE.md | 18 +- tests/e2e/run-e2e.sh | 78 ++-- tests/e2e/run-multinode-e2e.sh | 70 ++- 15 files changed, 1045 insertions(+), 157 deletions(-) create mode 100644 docs/development/feature/snapshot.md diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 492c2b9e5..85b35d571 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -28,8 +28,9 @@ Each component can be used independently or as part of the integrated Hive stack - **All unit tests MUST pass** before committing changes - **No exceptions** - failing tests block commits - Tests must complete without errors or panics -- Use `make test` command to run the full test suite -- If tests fail, fix issues before proceeding with git operations +- Use `make test` command to run all tests +- use `make security` command to run all security tests +- If tests fail, fix issues before proceeding ## Architecture Notes diff --git a/docs/development/feature/snapshot.md b/docs/development/feature/snapshot.md new file mode 100644 index 000000000..4f6080688 --- /dev/null +++ b/docs/development/feature/snapshot.md @@ -0,0 +1,164 @@ +# Viperblock Snapshot Feature + +## Overview + +Viperblock snapshots enable instant, zero-copy point-in-time captures of block storage volumes. A snapshot freezes the block-to-object mapping at a moment in time without copying any data. Both the snapshot and the live volume share the same immutable chunk files on S3. Volumes created from snapshots use a copy-on-write (COW) layered read path, reading unmodified blocks from the source volume's chunks and writing new data to their own chunks. + +This is the foundation for AMI-backed instance launches -- `aws ec2 run-instances --image-id ami-ubuntu` will boot a VM that reads Ubuntu blocks on demand from the AMI's frozen snapshot, with zero image copying. + +## Key Design Decisions + +### 1. Snapshots are frozen block maps, not data copies + +Viperblock's chunk files are immutable -- once `chunk.00000001.bin` is written, it never changes. New writes always produce new chunks with higher ObjectIDs. This means a snapshot only needs to capture the current `BlocksToObject` mapping (which block lives at which offset in which chunk file). The actual block data stays in place. + +**Trade-off**: This makes snapshot creation O(mapping size) instead of O(data size), but it means the source volume's chunk files cannot be garbage collected while any snapshot references them. + +### 2. Copy-on-write via layered block maps + +A clone volume created from a snapshot has two block maps: + +- **Own map** (`BlocksToObject`) -- blocks written since clone creation +- **Base map** (`BaseBlockMap`) -- frozen read-only map from the parent snapshot + +The read path checks own map first, then falls back to the base map, then returns zeros: + +``` +Read block N: + 1. Own BlocksToObject / BlockStore -> found? return data from own chunks + 2. BaseBlockMap -> found? return data from source volume's chunks via ReadFrom() + 3. Neither -> zero block (never written) +``` + +Writes always go to the clone's own chunks. Once block N is written in the clone, it shadows the snapshot's version permanently. + +**Trade-off**: Reads of unmodified blocks require cross-volume I/O (reading from the source volume's S3 prefix). This is the same latency as a normal backend read but targets a different path. The BlockStore caches fetched base blocks so repeated reads are fast. + +### 3. Backend interface extended with ReadFrom/WriteTo + +The existing `Backend.Read()` and `Backend.Write()` methods construct S3 keys using the backend's configured volume name. Snapshots need to read/write across volume boundaries: + +- `CreateSnapshot` writes checkpoint and config to `{snapshotID}/` prefix +- Clone volumes read chunk data from `{sourceVolumeName}/` prefix + +Rather than temporarily swapping the backend's volume name (error-prone, not thread-safe), we added two new methods to the `Backend` interface: + +```go +ReadFrom(volumeName string, fileType FileType, objectId uint64, offset uint32, length uint32) ([]byte, error) +WriteTo(volumeName string, fileType FileType, objectId uint64, headers *[]byte, data *[]byte) error +``` + +These are identical to `Read`/`Write` but use the provided volume name for path construction instead of the backend's configured name. + +**Trade-off**: This adds two methods to an interface that all backends must implement. The alternative was adding a `volumeName` parameter to the existing methods (breaks all callers) or using a separate backend instance per snapshot operation (heavier). The `ReadFrom`/`WriteTo` approach is minimal and backward-compatible. + +### 4. Snapshot state persisted in VBState + +Two fields were added to `VBState` (the serialized volume config): + +```go +SnapshotID string // ID of parent snapshot (empty for non-clones) +SourceVolumeName string // Volume whose chunks the base map references +``` + +On `LoadState()`, if `SnapshotID` is set, `OpenFromSnapshot()` is called automatically to reload the base block map. This means a clone volume survives restarts -- the snapshot relationship is restored from the persisted config. + +### 5. Both read paths modified (BlockStore and legacy) + +Viperblock has two read paths: the optimized `readBlockStore()` (O(1) sharded index) and the legacy `read()` (map rebuilding). Both were modified identically: + +- When a block is not found in the volume's own data structures, check `BaseBlockMap` +- Blocks found in the base map are collected into a separate `baseConsecutiveBlocks` list +- These are fetched via `fetchBaseBlocksFromBackend()` which uses `ReadFrom(sourceVolumeName, ...)` +- Fetched base blocks are cached in the BlockStore/LRU for subsequent reads + +The two-list approach (own blocks vs base blocks) avoids mixing `Read()` and `ReadFrom()` calls in the same fetch loop. + +### 6. Snapshot checkpoint reuses existing serialization + +The snapshot's frozen block map is serialized in the same binary format as `SaveBlockState()` -- a header followed by 26-byte `BlockLookup` entries with CRC32 checksums. `LoadSnapshotBlockMap()` deserializes using the same `readBlockWalChunk()` function. + +This means snapshot checkpoints are validated with the same integrity checks as regular checkpoints, and no new serialization code was needed. + +## S3 Layout + +``` +vol-abc123/ # Source volume + config.json + chunks/chunk.00000001.bin # Shared -- snapshot references these + chunks/chunk.00000002.bin + checkpoints/blocks.00000001.bin + +snap-xyz789/ # Snapshot + config.json # SnapshotState metadata + checkpoints/blocks.00000000.bin # Frozen block map at snapshot time + +vol-newclone/ # Clone volume + config.json # VBState with SnapshotID set + chunks/chunk.00000001.bin # Only new writes after clone creation + checkpoints/blocks.00000001.bin # Only clone's own block mappings +``` + +## Files Changed (viperblock repo) + +| File | Change | +|------|--------| +| `types/types.go` | Added `ReadFrom`, `WriteTo` to `Backend` interface | +| `viperblock/viperblock.go` | Added `BaseBlockMap`, `SourceVolumeName`, `SnapshotID` to VB struct. Added snapshot fields to `VBState`. Modified `readBlockStore()` and legacy `read()` for base map fallback. Added `fetchBaseBlocksFromBackend()`. Updated `SaveState`/`LoadState`. | +| `viperblock/snapshot.go` | New file: `SnapshotState` type, `CreateSnapshot()`, `LoadSnapshotBlockMap()`, `OpenFromSnapshot()`, `LookupBaseBlockToObject()` | +| `viperblock/backends/file/file.go` | Implemented `ReadFrom()`, `WriteTo()` | +| `viperblock/backends/s3/s3.go` | Implemented `ReadFrom()`, `WriteTo()` | +| `viperblock/backends/memory/memory.go` | Updated stub to satisfy new interface | +| `viperblock/snapshot_test.go` | 6 test cases across file and S3 backends | + +## Test Coverage + +| Test | What it validates | +|------|-------------------| +| `TestCreateSnapshot` | Flush + checkpoint + config written to backend; round-trip via LoadSnapshotBlockMap | +| `TestSnapshotReadFallback` | Clone reads unmodified blocks from source volume's chunks | +| `TestSnapshotCopyOnWrite` | Overwritten blocks return new data; unmodified blocks return snapshot data | +| `TestSnapshotZeroBlocks` | Blocks never written in snapshot or clone return zeros | +| `TestSnapshotMultiChunk` | Snapshot spanning multiple 4MB chunk files | +| `TestLoadFromSnapshot` | Clone survives save/reset/reload cycle with base map restored | + +All tests run against 4 backend configurations: file, file_nocache, s3, s3_nocache. + +## Hive API Integration + +### Request Flow + +``` +AWS SDK (create-snapshot --volume-id vol-xxx) + -> Gateway (port 9999): parse AWS query params, validate VolumeId + -> NATS ec2.CreateSnapshot: routed to daemon via queue group + -> Daemon handleEC2CreateSnapshot -> SnapshotServiceImpl.CreateSnapshot() + 1. Fetch volume config from Predastore (validate volume exists) + 2. NATS ebs.snapshot -> viperblockd: flush + freeze block map + 3. Write snapshot metadata to Predastore ({snapId}/metadata.json) + <- NATS response: ec2.Snapshot JSON + <- Gateway: wrap in XML + <- AWS SDK: Snapshot response +``` + +### NATS Topics + +| Topic | Direction | Purpose | +|-------|-----------|---------| +| `ec2.CreateSnapshot` | Gateway -> Daemon | AWS API entry point | +| `ebs.snapshot` | Daemon -> Viperblockd | Flush live VB instance and create frozen block map checkpoint | +| `ec2.DescribeSnapshots` | Gateway -> Daemon | List/filter snapshots | +| `ec2.DeleteSnapshot` | Gateway -> Daemon | Delete snapshot metadata and checkpoint | +| `ec2.CopySnapshot` | Gateway -> Daemon | Copy snapshot metadata (metadata-only) | + +### Files Changed (hive repo) + +| File | Change | +|------|--------| +| `hive/handlers/ec2/snapshot/service.go` | `SnapshotService` interface | +| `hive/handlers/ec2/snapshot/service_impl.go` | Direct implementation with S3 storage + NATS `ebs.snapshot` call | +| `hive/handlers/ec2/snapshot/service_nats.go` | NATS proxy (gateway -> daemon) | +| `hive/gateway/ec2/snapshot/CreateSnapshot.go` | Gateway validation + NATS dispatch | +| `hive/daemon/daemon_handlers.go` | `handleEC2CreateSnapshot` wiring | +| `hive/services/viperblockd/viperblockd.go` | `ebs.snapshot` subscription handler | +| `hive/config/config.go` | `EBSSnapshotRequest` / `EBSSnapshotResponse` types | diff --git a/go.mod b/go.mod index f616a7876..424ed8ce6 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/google/uuid v1.6.0 github.com/kdomanski/iso9660 v0.4.0 github.com/mulgadc/predastore v1.0.0 - github.com/mulgadc/viperblock v1.0.5 + github.com/mulgadc/viperblock v1.0.10 github.com/nats-io/nats-server/v2 v2.11.6 github.com/nats-io/nats.go v1.43.0 github.com/pelletier/go-toml/v2 v2.2.4 diff --git a/go.sum b/go.sum index ce854a44f..bd0d8f152 100644 --- a/go.sum +++ b/go.sum @@ -274,10 +274,8 @@ github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lN github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= github.com/mulgadc/predastore v1.0.0 h1:IZ5V/EL+k+zz0zGoTYp+1BgODb43JxHsb/kvORuydKI= github.com/mulgadc/predastore v1.0.0/go.mod h1:x46A1z091hlZkMt3kkcuLSy2PAzSgROr6ft04qcYOU4= -github.com/mulgadc/viperblock v1.0.4 h1:fH4jqwnnvWXcEpRk73oiJPyVuTCuufWaV2mO0R91h7o= -github.com/mulgadc/viperblock v1.0.4/go.mod h1:CuoSKPcR50M11DmadqXYAeOo5HNI6Ix5B/OvSzuLWxA= -github.com/mulgadc/viperblock v1.0.5 h1:TWLWWWHT9r25CEMHN7UxLDc6QGsi8T6l7hlTHmJb4nk= -github.com/mulgadc/viperblock v1.0.5/go.mod h1:CuoSKPcR50M11DmadqXYAeOo5HNI6Ix5B/OvSzuLWxA= +github.com/mulgadc/viperblock v1.0.10 h1:nQJGD44cQ6gW2oWt7IZGjovLNdhLpxjX+WO1EDT+lgY= +github.com/mulgadc/viperblock v1.0.10/go.mod h1:1ZKEGGbjmiMi3rfmQ5jMZJB3gIws0pMnAreDbjoCodE= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI= diff --git a/hive/config/config.go b/hive/config/config.go index ae81c2ee9..c18bc756f 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -142,6 +142,17 @@ type EBSDeleteResponse struct { Error string } +type EBSSnapshotRequest struct { + Volume string + SnapshotID string +} + +type EBSSnapshotResponse struct { + SnapshotID string + Success bool + Error string +} + // EC2, TODO: Move to vm.go or more applicable place type EC2StartInstancesRequest struct { InstanceID string diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index bd2611167..acba2fc0e 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -498,7 +498,7 @@ func (d *Daemon) Start() error { d.keyService = handlers_ec2_key.NewKeyServiceImpl(d.config) d.imageService = handlers_ec2_image.NewImageServiceImpl(d.config) d.volumeService = handlers_ec2_volume.NewVolumeServiceImpl(d.config, d.natsConn) - d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config) + d.snapshotService = handlers_ec2_snapshot.NewSnapshotServiceImpl(d.config, d.natsConn) d.tagsService = handlers_ec2_tags.NewTagsServiceImpl(d.config) d.restoreInstances() diff --git a/hive/handlers/ec2/instance/service_impl.go b/hive/handlers/ec2/instance/service_impl.go index ff661d95a..eeab8ad17 100644 --- a/hive/handlers/ec2/instance/service_impl.go +++ b/hive/handlers/ec2/instance/service_impl.go @@ -305,99 +305,54 @@ func (s *InstanceServiceImpl) prepareRootVolume(input *ec2.RunInstancesInput, im return nil } -// cloneAMIToVolume clones an AMI to a new volume +// cloneAMIToVolume creates a new volume from an AMI using snapshot-based +// zero-copy cloning. The destination volume points at the AMI's frozen block +// map and reads on-demand from the AMI's chunks (copy-on-write). func (s *InstanceServiceImpl) cloneAMIToVolume(input *ec2.RunInstancesInput, size int, volumeConfig viperblock.VolumeConfig, destVb *viperblock.VB) error { - // Open WALs for destination volume - err := destVb.OpenWAL(&destVb.WAL, fmt.Sprintf("%s/%s", destVb.WAL.BaseDir, types.GetFilePath(types.FileTypeWALChunk, destVb.WAL.WallNum.Load(), destVb.GetVolume()))) - if err != nil { - slog.Error("Failed to load WAL", "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - err = destVb.OpenWAL(&destVb.BlockToObjectWAL, fmt.Sprintf("%s/%s", destVb.WAL.BaseDir, types.GetFilePath(types.FileTypeWALBlock, destVb.BlockToObjectWAL.WallNum.Load(), destVb.GetVolume()))) - if err != nil { - slog.Error("Failed to load block WAL", "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - // Setup source AMI Viperblock + // Load AMI state to get the snapshot ID amiVb, err := s.newViperblock(*input.ImageId, size, volumeConfig) if err != nil { slog.Error("Failed to connect to Viperblock store for AMI", "err", err) return errors.New(awserrors.ErrorServerInternal) } - // Initialize AMI backend - slog.Debug("Initializing AMI Viperblock store backend") err = amiVb.Backend.Init() if err != nil { slog.Error("Could not connect to AMI Viperblock store", "err", err) return errors.New(awserrors.ErrorServerInternal) } - slog.Debug("Loading state for AMI Viperblock store") - err = amiVb.LoadState() + amiState, err := amiVb.LoadStateRequest("") if err != nil { - slog.Error("Could not load state for AMI Viperblock store", "err", err) + slog.Error("Could not load state for AMI", "err", err) return errors.New(awserrors.ErrorServerInternal) } - err = amiVb.LoadBlockState() - if err != nil { - slog.Error("Failed to load block state", "err", err) + snapshotID := amiState.VolumeConfig.AMIMetadata.SnapshotID + if snapshotID == "" { + slog.Error("AMI has no snapshot ID, cannot perform zero-copy clone", "imageId", *input.ImageId) return errors.New(awserrors.ErrorServerInternal) } - slog.Debug("Starting to clone AMI to new volume") + slog.Info("Cloning AMI via snapshot", "imageId", *input.ImageId, "snapshotID", snapshotID) - // Clone blocks from AMI to new volume - var block uint64 = 0 - nullBlock := make([]byte, destVb.BlockSize) - - for { - if block*uint64(destVb.BlockSize) >= amiVb.VolumeSize { - slog.Debug("Reached end of AMI") - break - } - - // Read 1MB - data, err := amiVb.ReadAt(block*uint64(destVb.BlockSize), uint64(destVb.BlockSize)*1024) - if err != nil && err != viperblock.ErrZeroBlock { - slog.Error("Failed to read block from AMI source", "err", err) - return errors.New(awserrors.ErrorServerInternal) - } - - numBlocks := len(data) / int(destVb.BlockSize) - - // Write individual blocks to the new volume - for i := range numBlocks { - // Check if the input is a Zero block - if bytes.Equal(data[i*int(destVb.BlockSize):(i+1)*int(destVb.BlockSize)], nullBlock) { - block++ - continue - } - - if err := destVb.WriteAt(block*uint64(destVb.BlockSize), data[i*int(destVb.BlockSize):(i+1)*int(destVb.BlockSize)]); err != nil { - slog.Error("Failed to write block", "block", block, "err", err) - } - block++ - - // Flush every 4MB - if block%uint64(destVb.BlockSize) == 0 { - // slog.Debug("Flush", "block", block) - if err := destVb.Flush(); err != nil { - slog.Error("Failed to flush", "err", err) - } - if err := destVb.WriteWALToChunk(true); err != nil { - slog.Error("Failed to write WAL to chunk", "err", err) - } - } - } + // Set up destination volume from the snapshot (zero-copy) + err = destVb.OpenFromSnapshot(snapshotID) + if err != nil { + slog.Error("Failed to open from snapshot", "snapshotID", snapshotID, "err", err) + return errors.New(awserrors.ErrorServerInternal) } - err = destVb.Close() + // Persist the snapshot relationship to the backend + err = destVb.SaveState() if err != nil { - slog.Error("Failed to close Viperblock store", "err", err) + slog.Error("Failed to save state", "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + err = destVb.SaveBlockState() + if err != nil { + slog.Error("Failed to save block state", "err", err) return errors.New(awserrors.ErrorServerInternal) } diff --git a/hive/handlers/ec2/snapshot/service_impl.go b/hive/handlers/ec2/snapshot/service_impl.go index 8209f9346..07bbc05dc 100644 --- a/hive/handlers/ec2/snapshot/service_impl.go +++ b/hive/handlers/ec2/snapshot/service_impl.go @@ -18,6 +18,7 @@ import ( "github.com/mulgadc/hive/hive/objectstore" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" + "github.com/nats-io/nats.go" ) // Ensure SnapshotServiceImpl implements SnapshotService @@ -25,9 +26,10 @@ var _ SnapshotService = (*SnapshotServiceImpl)(nil) // SnapshotServiceImpl implements SnapshotService with S3-backed storage type SnapshotServiceImpl struct { - config *config.Config - store objectstore.ObjectStore - mutex sync.RWMutex + config *config.Config + store objectstore.ObjectStore + natsConn *nats.Conn + mutex sync.RWMutex } // SnapshotConfig represents snapshot metadata stored in S3 @@ -46,7 +48,7 @@ type SnapshotConfig struct { } // NewSnapshotServiceImpl creates a new snapshot service implementation -func NewSnapshotServiceImpl(cfg *config.Config) *SnapshotServiceImpl { +func NewSnapshotServiceImpl(cfg *config.Config, natsConn *nats.Conn) *SnapshotServiceImpl { store := objectstore.NewS3ObjectStoreFromConfig( cfg.Predastore.Host, cfg.Predastore.Region, @@ -55,22 +57,26 @@ func NewSnapshotServiceImpl(cfg *config.Config) *SnapshotServiceImpl { ) return &SnapshotServiceImpl{ - config: cfg, - store: store, + config: cfg, + store: store, + natsConn: natsConn, } } // NewSnapshotServiceImplWithStore creates a snapshot service with a custom ObjectStore (for testing) -func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore) *SnapshotServiceImpl { +func NewSnapshotServiceImplWithStore(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn) *SnapshotServiceImpl { return &SnapshotServiceImpl{ - config: cfg, - store: store, + config: cfg, + store: store, + natsConn: natsConn, } } -// getSnapshotKey returns the S3 key for storing snapshot config +// getSnapshotKey returns the S3 key for storing snapshot metadata. +// Uses metadata.json to avoid conflicting with viperblock's config.json +// which stores the SnapshotState (block map references, source volume, etc). func getSnapshotKey(snapshotID string) string { - return fmt.Sprintf("%s/config.json", snapshotID) + return fmt.Sprintf("%s/metadata.json", snapshotID) } // getSnapshotConfig retrieves snapshot config from S3 @@ -179,10 +185,40 @@ func (s *SnapshotServiceImpl) CreateSnapshot(input *ec2.CreateSnapshotInput) (*e return nil, errors.New(awserrors.ErrorServerInternal) } + // Trigger viperblock to flush data and create a frozen block map checkpoint. + // This sends a NATS message to the EBS daemon that owns the volume, which + // calls vb.CreateSnapshot() on the live viperblock instance. + if s.natsConn != nil { + snapReq := config.EBSSnapshotRequest{Volume: volumeID, SnapshotID: snapshotID} + snapData, err := json.Marshal(snapReq) + if err != nil { + slog.Error("CreateSnapshot: failed to marshal ebs.snapshot request", "volumeId", volumeID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + msg, err := s.natsConn.Request(fmt.Sprintf("ebs.snapshot.%s", volumeID), snapData, 30*time.Second) + if err != nil { + slog.Error("CreateSnapshot: ebs.snapshot NATS request failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + var snapResp config.EBSSnapshotResponse + if err := json.Unmarshal(msg.Data, &snapResp); err != nil { + slog.Error("CreateSnapshot: failed to unmarshal ebs.snapshot response", "volumeId", volumeID, "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + if !snapResp.Success || snapResp.Error != "" { + slog.Error("CreateSnapshot: viperblock snapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", snapResp.Error) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("CreateSnapshot: viperblock snapshot created", "volumeId", volumeID, "snapshotId", snapshotID) + } else { + slog.Warn("CreateSnapshot: natsConn is nil, skipping viperblock snapshot (metadata-only)", "volumeId", volumeID) + } + now := time.Now() - // Mark as completed immediately -- Hive v1 stores snapshot metadata - // pointing to the source volume rather than copying actual block data. snapshotCfg := &SnapshotConfig{ SnapshotID: snapshotID, VolumeID: volumeID, diff --git a/hive/handlers/ec2/snapshot/service_impl_test.go b/hive/handlers/ec2/snapshot/service_impl_test.go index 202facb26..81b52fff6 100644 --- a/hive/handlers/ec2/snapshot/service_impl_test.go +++ b/hive/handlers/ec2/snapshot/service_impl_test.go @@ -26,7 +26,7 @@ func setupTestSnapshotService(t *testing.T) (*SnapshotServiceImpl, *objectstore. }, } - svc := NewSnapshotServiceImplWithStore(cfg, store) + svc := NewSnapshotServiceImplWithStore(cfg, store, nil) return svc, store } diff --git a/hive/handlers/ec2/volume/service_impl.go b/hive/handlers/ec2/volume/service_impl.go index 5f2c64e3e..164cb0b79 100644 --- a/hive/handlers/ec2/volume/service_impl.go +++ b/hive/handlers/ec2/volume/service_impl.go @@ -75,12 +75,6 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol return nil, errors.New(awserrors.ErrorInvalidParameterValue) } - // Validate size (1-16384 GiB) - if input.Size == nil || *input.Size < 1 || *input.Size > 16384 { - return nil, errors.New(awserrors.ErrorInvalidParameterValue) - } - size := *input.Size - // Validate volume type: only gp3 supported (or empty defaults to gp3) if input.VolumeType != nil && *input.VolumeType != "" && *input.VolumeType != "gp3" { return nil, errors.New(awserrors.ErrorInvalidParameterValue) @@ -95,6 +89,40 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol return nil, errors.New(awserrors.ErrorInvalidAvailabilityZone) } + // If creating from snapshot, read snapshot metadata to get defaults + var snapshotID string + var sourceVolumeName string + var snapshotSizeGiB int64 + + if input.SnapshotId != nil && *input.SnapshotId != "" { + snapshotID = *input.SnapshotId + snapMeta, err := s.getSnapshotMetadata(snapshotID) + if err != nil { + slog.Error("CreateVolume: snapshot not found", "snapshotId", snapshotID, "err", err) + return nil, errors.New(awserrors.ErrorInvalidSnapshotNotFound) + } + sourceVolumeName = snapMeta.VolumeID + snapshotSizeGiB = snapMeta.VolumeSize + } + + // Validate size (1-16384 GiB). When creating from snapshot, size can be + // omitted (defaults to snapshot size) or must be >= snapshot size. + var size int64 + if input.Size != nil { + if *input.Size < 1 || *input.Size > 16384 { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + if snapshotSizeGiB > 0 && *input.Size < snapshotSizeGiB { + slog.Error("CreateVolume: requested size smaller than snapshot", "size", *input.Size, "snapshotSize", snapshotSizeGiB) + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + size = *input.Size + } else if snapshotSizeGiB > 0 { + size = snapshotSizeGiB + } else { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + // Generate volume ID randomNumber, err := rand.Int(rand.Reader, big.NewInt(100_000_000)) if err != nil { @@ -107,7 +135,8 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol iops := defaultGP3IOPS - slog.Info("CreateVolume", "volumeId", volumeID, "size", size, "type", volumeType, "az", *input.AvailabilityZone) + slog.Info("CreateVolume", "volumeId", volumeID, "size", size, "type", volumeType, + "az", *input.AvailabilityZone, "snapshotId", snapshotID) // Volume size in bytes for viperblock sizeGiB := utils.SafeInt64ToUint64(size) @@ -124,6 +153,7 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol VolumeType: volumeType, IOPS: iops, IsEncrypted: false, + SnapshotID: snapshotID, }, } @@ -146,6 +176,13 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol VolumeConfig: volumeConfig, } + // 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) if err != nil { slog.Error("CreateVolume failed to create viperblock instance", "err", err) @@ -168,7 +205,7 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol slog.Info("CreateVolume completed", "volumeId", volumeID, "size", size, "type", volumeType) - return &ec2.Volume{ + vol := &ec2.Volume{ VolumeId: aws.String(volumeID), Size: aws.Int64(size), VolumeType: aws.String(volumeType), @@ -177,7 +214,13 @@ func (s *VolumeServiceImpl) CreateVolume(input *ec2.CreateVolumeInput) (*ec2.Vol CreateTime: aws.Time(now), Iops: aws.Int64(int64(iops)), Encrypted: aws.Bool(false), - }, nil + } + + if snapshotID != "" { + vol.SnapshotId = aws.String(snapshotID) + } + + return vol, nil } // DescribeVolumes lists EBS volumes by reading config.json files from S3 @@ -671,6 +714,13 @@ func (s *VolumeServiceImpl) DeleteVolume(input *ec2.DeleteVolumeInput) (*ec2.Del return nil, errors.New(awserrors.ErrorVolumeInUse) } + // Check if any snapshots reference this volume. Snapshot-backed clones + // read chunk files from the source volume's S3 prefix via ReadFrom(). + // Deleting the source volume would silently break all clones. + if err := s.checkVolumeHasNoSnapshots(volumeID); err != nil { + return nil, err + } + // Notify viperblockd to stop nbdkit/WAL syncer (best-effort) if s.natsConn != nil { deleteReq := config.EBSDeleteRequest{Volume: volumeID} @@ -753,3 +803,127 @@ func (s *VolumeServiceImpl) deleteS3Prefix(prefix string) error { return nil } + +// snapshotMetadata holds the subset of snapshot metadata needed by CreateVolume. +// Matches the JSON written by the snapshot service's SnapshotConfig. +type snapshotMetadata struct { + VolumeID string `json:"volume_id"` + VolumeSize int64 `json:"volume_size"` +} + +// getSnapshotMetadata reads snapshot metadata.json from S3 for CreateVolume. +func (s *VolumeServiceImpl) getSnapshotMetadata(snapshotID string) (*snapshotMetadata, error) { + key := snapshotID + "/metadata.json" + + getResult, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(key), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + return nil, errors.New(awserrors.ErrorInvalidSnapshotNotFound) + } + return nil, fmt.Errorf("failed to get snapshot metadata: %w", err) + } + defer getResult.Body.Close() + + body, err := io.ReadAll(getResult.Body) + if err != nil { + return nil, fmt.Errorf("failed to read snapshot metadata: %w", err) + } + + var meta snapshotMetadata + if err := json.Unmarshal(body, &meta); err != nil { + return nil, fmt.Errorf("failed to decode snapshot metadata: %w", err) + } + + return &meta, nil +} + +// snapshotVolumeRef reads VolumeID from hive metadata.json (volume_id) +// or viperblock config.json (SourceVolumeName). +type snapshotVolumeRef struct { + VolumeID string `json:"volume_id"` + SourceVolumeName string `json:"SourceVolumeName"` +} + +func (r snapshotVolumeRef) referencesVolume(volumeID string) bool { + return r.VolumeID == volumeID || r.SourceVolumeName == volumeID +} + +// checkVolumeHasNoSnapshots scans all snap-* prefixes in S3 and returns an +// error if any snapshot references the given volume. Checks both hive's +// metadata.json and viperblock's config.json to handle either format. +func (s *VolumeServiceImpl) checkVolumeHasNoSnapshots(volumeID string) error { + listResult, err := s.store.ListObjects(&s3.ListObjectsInput{ + Bucket: aws.String(s.bucketName), + Prefix: aws.String("snap-"), + Delimiter: aws.String("/"), + }) + if err != nil { + slog.Error("checkVolumeHasNoSnapshots: failed to list snapshots", "volumeId", volumeID, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + for _, prefix := range listResult.CommonPrefixes { + if prefix.Prefix == nil { + continue + } + + snapshotID := strings.TrimSuffix(*prefix.Prefix, "/") + + if err := s.snapshotReferencesVolume(snapshotID, volumeID); err != nil { + return err + } + } + + return nil +} + +// snapshotReferencesVolume checks if a snapshot references the given volume. +// Returns an error if the snapshot references the volume or if a non-recoverable +// S3/IO error occurs. Returns nil if the snapshot does not reference the volume. +func (s *VolumeServiceImpl) snapshotReferencesVolume(snapshotID, volumeID string) error { + // Try metadata.json first (hive format), then config.json (viperblock format) + for _, filename := range []string{"metadata.json", "config.json"} { + key := snapshotID + "/" + filename + + getResult, err := s.store.GetObject(&s3.GetObjectInput{ + Bucket: aws.String(s.bucketName), + Key: aws.String(key), + }) + if err != nil { + if objectstore.IsNoSuchKeyError(err) { + continue // Expected: this format file doesn't exist for this snapshot + } + slog.Error("checkVolumeHasNoSnapshots: failed to read snapshot metadata", + "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + body, err := io.ReadAll(getResult.Body) + if closeErr := getResult.Body.Close(); closeErr != nil { + slog.Error("checkVolumeHasNoSnapshots: failed to close response body", + "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", closeErr) + } + if err != nil { + slog.Error("checkVolumeHasNoSnapshots: failed to read snapshot body", + "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) + return errors.New(awserrors.ErrorServerInternal) + } + + var ref snapshotVolumeRef + if err := json.Unmarshal(body, &ref); err != nil { + slog.Warn("checkVolumeHasNoSnapshots: failed to parse snapshot metadata, skipping file", + "volumeId", volumeID, "snapshotId", snapshotID, "key", key, "err", err) + continue // JSON parse errors for individual files are not fatal -- try next file + } + + if ref.referencesVolume(volumeID) { + slog.Error("DeleteVolume blocked: volume has snapshots", "volumeId", volumeID, "snapshotId", snapshotID) + return fmt.Errorf("%s: volume %s has existing snapshot %s. Delete snapshots first", awserrors.ErrorVolumeInUse, volumeID, snapshotID) + } + } + + return nil +} diff --git a/hive/handlers/ec2/volume/service_impl_test.go b/hive/handlers/ec2/volume/service_impl_test.go index cc21796a2..14a6833a8 100644 --- a/hive/handlers/ec2/volume/service_impl_test.go +++ b/hive/handlers/ec2/volume/service_impl_test.go @@ -1,13 +1,17 @@ package handlers_ec2_volume import ( + "encoding/json" + "strings" "testing" "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/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/objectstore" + "github.com/mulgadc/viperblock/viperblock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -250,3 +254,407 @@ func TestDescribeVolumeStatus_WithVolumeIDs(t *testing.T) { require.Error(t, err) assert.Equal(t, awserrors.ErrorInvalidVolumeNotFound, err.Error()) } + +// newTestVolumeServiceWithStore creates a volume service with a specific memory store +func newTestVolumeServiceWithStore(az string, store *objectstore.MemoryObjectStore) *VolumeServiceImpl { + cfg := &config.Config{ + AZ: az, + Predastore: config.PredastoreConfig{ + Bucket: "test-bucket", + Region: "ap-southeast-2", + Host: "localhost:9000", + AccessKey: "testkey", + SecretKey: "testsecret", + }, + WalDir: "/tmp/test-wal", + } + return NewVolumeServiceImplWithStore(cfg, store, nil) +} + +func TestDeleteVolume_BlockedBySnapshot(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + volumeID := "vol-test123" + + // Create volume config in store + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Create a snapshot referencing this volume + snapCfg := snapshotVolumeRef{VolumeID: volumeID} + snapData, err := json.Marshal(snapCfg) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("snap-abc123/config.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // DeleteVolume should be blocked + _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) + assert.Contains(t, err.Error(), "snap-abc123") +} + +func TestDeleteVolume_AllowedWithoutSnapshots(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + volumeID := "vol-test456" + + // Create volume config in store + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Create a snapshot referencing a DIFFERENT volume + snapCfg := snapshotVolumeRef{VolumeID: "vol-other"} + snapData, err := json.Marshal(snapCfg) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("snap-xyz789/config.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // DeleteVolume should succeed (no snapshots reference this volume) + _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.NoError(t, err) +} + +func TestCreateVolume_FromSnapshot_PassesValidation(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + snapshotID := "snap-test123" + + // Create snapshot metadata in store (matches hive snapshot service format) + snapMeta := snapshotMetadata{ + VolumeID: "vol-source", + VolumeSize: 50, + } + snapData, err := json.Marshal(snapMeta) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(snapshotID + "/metadata.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // CreateVolume from snapshot without explicit size passes validation + // (fails later at viperblock backend init because no S3 server in tests) + _, err = svc.CreateVolume(&ec2.CreateVolumeInput{ + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String(snapshotID), + }) + if err != nil { + // Should not be a snapshot or validation error - those are the paths we're testing + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidSnapshotNotFound) + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidParameterValue) + } +} + +func TestCreateVolume_FromSnapshot_WithExplicitSize(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + snapshotID := "snap-test456" + + snapMeta := snapshotMetadata{ + VolumeID: "vol-source", + VolumeSize: 50, + } + snapData, err := json.Marshal(snapMeta) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(snapshotID + "/metadata.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // CreateVolume from snapshot with explicit larger size passes validation + _, err = svc.CreateVolume(&ec2.CreateVolumeInput{ + Size: aws.Int64(100), + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String(snapshotID), + }) + if err != nil { + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidSnapshotNotFound) + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidParameterValue) + } +} + +func TestCreateVolume_FromSnapshot_NotFound(t *testing.T) { + svc := newTestVolumeService("ap-southeast-2a") + + _, err := svc.CreateVolume(&ec2.CreateVolumeInput{ + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String("snap-nonexistent"), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorInvalidSnapshotNotFound) +} + +func TestCreateVolume_FromSnapshot_SizeSmallerThanSnapshot(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + snapshotID := "snap-sizecheck" + + snapMeta := snapshotMetadata{ + VolumeID: "vol-source", + VolumeSize: 50, + } + snapData, err := json.Marshal(snapMeta) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(snapshotID + "/metadata.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // Size 10 < snapshot size 50 -- must be rejected + _, err = svc.CreateVolume(&ec2.CreateVolumeInput{ + Size: aws.Int64(10), + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String(snapshotID), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue) +} + +func TestCreateVolume_FromSnapshot_SizeEqualToSnapshot(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + snapshotID := "snap-sizeequal" + + snapMeta := snapshotMetadata{ + VolumeID: "vol-source", + VolumeSize: 50, + } + snapData, err := json.Marshal(snapMeta) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(snapshotID + "/metadata.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // Size == snapshot size should pass validation (may fail at backend init) + _, err = svc.CreateVolume(&ec2.CreateVolumeInput{ + Size: aws.Int64(50), + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String(snapshotID), + }) + if err != nil { + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidParameterValue) + assert.NotContains(t, err.Error(), awserrors.ErrorInvalidSnapshotNotFound) + } +} + +func TestDeleteVolume_BlockedBySourceVolumeName(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + volumeID := "vol-srcname" + + // Create volume config + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Snapshot references volume via SourceVolumeName (viperblock config.json format) + snapCfg := snapshotVolumeRef{SourceVolumeName: volumeID} + snapData, err := json.Marshal(snapCfg) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("snap-vb001/config.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // DeleteVolume should be blocked via SourceVolumeName match + _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) + assert.Contains(t, err.Error(), "snap-vb001") +} + +func TestDeleteVolume_BlockedByMetadataJson(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + volumeID := "vol-metablock" + + // Create volume config + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Snapshot references volume via metadata.json (hive format, volume_id field) + snapCfg := snapshotVolumeRef{VolumeID: volumeID} + snapData, err := json.Marshal(snapCfg) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("snap-meta001/metadata.json"), + Body: strings.NewReader(string(snapData)), + }) + require.NoError(t, err) + + // DeleteVolume should be blocked via metadata.json match + _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorVolumeInUse) + assert.Contains(t, err.Error(), "snap-meta001") +} + +func TestCreateVolume_FromSnapshot_CorruptMetadata(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + snapshotID := "snap-corrupt" + + // Put invalid JSON as snapshot metadata + _, err := store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(snapshotID + "/metadata.json"), + Body: strings.NewReader("not valid json{{{"), + }) + require.NoError(t, err) + + _, err = svc.CreateVolume(&ec2.CreateVolumeInput{ + AvailabilityZone: aws.String("ap-southeast-2a"), + SnapshotId: aws.String(snapshotID), + }) + require.Error(t, err) +} + +func TestDeleteVolume_CorruptSnapshotMetadata_SkipsFile(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + svc := newTestVolumeServiceWithStore("ap-southeast-2a", store) + + volumeID := "vol-corruptsnap" + + // Create volume config + volumeState := viperblock.VBState{ + VolumeConfig: viperblock.VolumeConfig{ + VolumeMetadata: viperblock.VolumeMetadata{ + VolumeID: volumeID, + SizeGiB: 10, + State: "available", + }, + }, + } + data, err := json.Marshal(volumeState) + require.NoError(t, err) + + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String(volumeID + "/config.json"), + Body: strings.NewReader(string(data)), + }) + require.NoError(t, err) + + // Put corrupt metadata.json (invalid JSON) -- should be skipped + _, err = store.PutObject(&s3.PutObjectInput{ + Bucket: aws.String("test-bucket"), + Key: aws.String("snap-corrupt/metadata.json"), + Body: strings.NewReader("not json{{{"), + }) + require.NoError(t, err) + + // No valid config.json either, so no snapshot blocks this volume. + // DeleteVolume should succeed since corrupt metadata is skipped. + _, err = svc.DeleteVolume(&ec2.DeleteVolumeInput{ + VolumeId: aws.String(volumeID), + }) + require.NoError(t, err) +} diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 6e80f07ed..e143d7a23 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -24,12 +24,13 @@ import ( var serviceName = "viperblock" type MountedVolume struct { - Name string - Port int // TCP port (when using TCP transport) - Socket string // Unix socket path (when using socket transport) - NBDURI string // Full NBD URI (nbd:unix:/path.sock or nbd://host:port) - PID int - VB *viperblock.VB // Reference to viperblock instance for state sync/flush + Name string + Port int // TCP port (when using TCP transport) + Socket string // Unix socket path (when using socket transport) + NBDURI string // Full NBD URI (nbd:unix:/path.sock or nbd://host:port) + PID int + VB *viperblock.VB // Reference to viperblock instance for state sync/flush + SnapshotSub *nats.Subscription // Per-volume snapshot subscription (ebs.snapshot.{volumeID}) } type Config struct { @@ -66,6 +67,46 @@ func New(config any) (svc *Service, err error) { return svc, nil } +// makeSnapshotHandler returns a NATS handler for volume-specific snapshot requests (ebs.snapshot.{volumeID}). +func makeSnapshotHandler(vb *viperblock.VB, volumeName string) nats.MsgHandler { + return func(msg *nats.Msg) { + var snapRequest config.EBSSnapshotRequest + if err := json.Unmarshal(msg.Data, &snapRequest); err != nil { + slog.Error("Failed to unmarshal ebs.snapshot message", "volume", volumeName, "err", err) + errResp, _ := json.Marshal(config.EBSSnapshotResponse{Error: fmt.Sprintf("bad request: %v", err)}) + if err := msg.Respond(errResp); err != nil { + slog.Error("Failed to respond to ebs.snapshot request", "err", err) + } + return + } + + slog.Info("ebs.snapshot: processing snapshot request", "volume", volumeName, "snapshotId", snapRequest.SnapshotID) + + snapResponse := config.EBSSnapshotResponse{SnapshotID: snapRequest.SnapshotID} + + if _, err := vb.CreateSnapshot(snapRequest.SnapshotID); err != nil { + snapResponse.Error = fmt.Sprintf("snapshot failed: %v", err) + slog.Error("ebs.snapshot: CreateSnapshot failed", "volume", volumeName, "snapshotId", snapRequest.SnapshotID, "err", err) + } else { + snapResponse.Success = true + slog.Info("ebs.snapshot: snapshot created", "volume", volumeName, "snapshotId", snapRequest.SnapshotID) + } + + response, err := json.Marshal(snapResponse) + if err != nil { + slog.Error("Failed to marshal ebs.snapshot response", "err", err) + if err := msg.Respond([]byte(`{"Error":"internal marshal failure"}`)); err != nil { + slog.Error("Failed to respond to ebs.snapshot request", "err", err) + } + return + } + + if err := msg.Respond(response); err != nil { + slog.Error("Failed to respond to ebs.snapshot request", "err", err) + } + } +} + func (svc *Service) Start() (int, error) { if err := utils.WritePidFile(serviceName, os.Getpid()); err != nil { @@ -143,6 +184,12 @@ func launchService(cfg *Config) (err error) { cfg.mu.Unlock() if matchIdx >= 0 { + // Unsubscribe from volume-specific snapshot topic + if matched.SnapshotSub != nil { + if err := matched.SnapshotSub.Unsubscribe(); err != nil { + slog.Error("Failed to unsubscribe snapshot topic", "volume", ebsRequest.Volume, "err", err) + } + } // Stop WAL syncer and kill nbdkit process if matched.VB != nil { matched.VB.StopWALSyncer() @@ -215,6 +262,13 @@ func launchService(cfg *Config) (err error) { Mounted: false, } + // Unsubscribe from volume-specific snapshot topic + if matched.SnapshotSub != nil { + if err := matched.SnapshotSub.Unsubscribe(); err != nil { + slog.Error("Failed to unsubscribe snapshot topic", "volume", ebsRequest.Name, "err", err) + } + } + // Clean up the VB instance's background goroutine. // This VB is state-only (LoadState/sync) — actual I/O is in the nbdkit plugin process. if matched.VB != nil { @@ -312,6 +366,10 @@ func launchService(cfg *Config) (err error) { slog.Error("Failed to subscribe to ebs.sync", "err", err) } + // Note: ebs.snapshot is handled per-volume via ebs.snapshot.{volumeID} topics, + // subscribed at mount time and unsubscribed at unmount time. This ensures + // snapshot requests are routed to the node that owns the volume. + if _, err := nc.QueueSubscribe("ebs.mount", "hive-workers", func(msg *nats.Msg) { slog.Info("Received message:", "data", string(msg.Data)) @@ -577,14 +635,21 @@ func launchService(cfg *Config) (err error) { ebsResponse.Mounted = true ebsResponse.URI = nbdURI + // Subscribe to volume-specific snapshot topic so requests route to this node + snapSub, err := nc.Subscribe(fmt.Sprintf("ebs.snapshot.%s", ebsRequest.Name), makeSnapshotHandler(vb, ebsRequest.Name)) + if err != nil { + slog.Error("Failed to subscribe to volume snapshot topic", "volume", ebsRequest.Name, "err", err) + } + cfg.mu.Lock() cfg.MountedVolumes = append(cfg.MountedVolumes, MountedVolume{ - Name: ebsRequest.Name, - Port: nbdPort, - Socket: nbdSocket, - NBDURI: nbdURI, - PID: pid, - VB: vb, + Name: ebsRequest.Name, + Port: nbdPort, + Socket: nbdSocket, + NBDURI: nbdURI, + PID: pid, + VB: vb, + SnapshotSub: snapSub, }) cfg.mu.Unlock() diff --git a/tests/e2e/TEST_COVERAGE.md b/tests/e2e/TEST_COVERAGE.md index 8fdcf90a8..709252931 100644 --- a/tests/e2e/TEST_COVERAGE.md +++ b/tests/e2e/TEST_COVERAGE.md @@ -38,7 +38,8 @@ - `delete-volume` (verify gone) ### Phase 5c: Snapshot Lifecycle -- `create-snapshot` (from volume, with description) +- Uses root volume already attached to running instance (snapshots require a mounted VB instance) +- `create-snapshot` (from attached root volume, with description) - Verify create response fields (VolumeId, VolumeSize, State, Progress) - `describe-snapshots` (by ID, verify VolumeId/Size/Description) - `copy-snapshot` (with new description, verify distinct ID) @@ -46,6 +47,12 @@ - `delete-snapshot` (original, verify gone while copy survives) - `delete-snapshot` (copy, cleanup) +### Phase 5d: Verify Snapshot-Backed Instance Launch +- All `run-instances` calls use the snapshot path (`cloneAMIToVolume` -> `OpenFromSnapshot`), so the Phase 5 instance is already snapshot-backed +- Verify AMI snapshot exists in Predastore (`snap-{amiId}/config.json`) +- Read Phase 5 root volume's `config.json` from Predastore +- Verify `SnapshotID` and `SourceVolumeName` are set (proves zero-copy clone) + ### Phase 6: Tag Management - `create-tags` (3 tags on instance) - `describe-tags` (filter by resource-id) @@ -106,7 +113,8 @@ - `delete-volume` #### Test 1c: Snapshot Lifecycle -- `create-snapshot` (from volume, with description) +- Uses root volume of first instance (snapshots require a mounted VB instance) +- `create-snapshot` (from attached root volume, with description) - Verify create response fields (VolumeId, VolumeSize, State) - `describe-snapshots` (by ID, verify fields) - `copy-snapshot` (with new description) @@ -114,6 +122,12 @@ - `delete-snapshot` (original, verify copy survives) - `delete-snapshot` (copy, cleanup) +#### Test 1c-ii: Verify Snapshot-Backed Instance Launch +- All `run-instances` calls use the snapshot path (`cloneAMIToVolume` -> `OpenFromSnapshot`), so the Test 1 instances are already snapshot-backed +- Verify AMI snapshot exists in Predastore (`snap-{amiId}/config.json`) +- Read first instance's root volume `config.json` from Predastore +- Verify `SnapshotID` and `SourceVolumeName` are set (proves zero-copy clone) + #### Test 1d: Tag Management - `create-tags` (3 tags on instance) - `describe-tags` (filter by resource-id) diff --git a/tests/e2e/run-e2e.sh b/tests/e2e/run-e2e.sh index 6ccdabcf6..e6553fc6b 100755 --- a/tests/e2e/run-e2e.sh +++ b/tests/e2e/run-e2e.sh @@ -381,20 +381,16 @@ echo "Volume lifecycle test passed (create -> resize -> attach -> detach -> dele echo "Phase 5c: Snapshot Lifecycle" echo "Testing snapshot create -> describe -> copy -> delete..." -# Create a volume to snapshot (snapshots need a source volume) -echo "Creating source volume for snapshot tests..." -SNAP_VOL_OUTPUT=$($AWS_EC2 create-volume --size 10 --availability-zone ap-southeast-2a) -SNAP_VOL_ID=$(echo "$SNAP_VOL_OUTPUT" | jq -r '.VolumeId') +# Use the root volume from Phase 5 — it's already attached and mounted in +# viperblockd, which is required for create-snapshot (the ebs.snapshot handler +# needs a live VB instance to flush). +echo "Using root volume $VOLUME_ID (already attached to $INSTANCE_ID)" +ROOT_VOL_SIZE=$($AWS_EC2 describe-volumes --volume-ids "$VOLUME_ID" \ + --query 'Volumes[0].Size' --output text) -if [ -z "$SNAP_VOL_ID" ] || [ "$SNAP_VOL_ID" == "null" ]; then - echo "Failed to create source volume for snapshot tests" - exit 1 -fi -echo "Created source volume: $SNAP_VOL_ID" - -# Create a snapshot from the volume -echo "Creating snapshot from volume $SNAP_VOL_ID..." -SNAP_OUTPUT=$($AWS_EC2 create-snapshot --volume-id "$SNAP_VOL_ID" --description "e2e-test-snapshot") +# Create a snapshot from the attached root volume +echo "Creating snapshot from volume $VOLUME_ID..." +SNAP_OUTPUT=$($AWS_EC2 create-snapshot --volume-id "$VOLUME_ID" --description "e2e-test-snapshot") SNAPSHOT_ID=$(echo "$SNAP_OUTPUT" | jq -r '.SnapshotId') if [ -z "$SNAPSHOT_ID" ] || [ "$SNAPSHOT_ID" == "null" ]; then @@ -410,12 +406,12 @@ SNAP_VOL_REF=$(echo "$SNAP_OUTPUT" | jq -r '.VolumeId') SNAP_SIZE=$(echo "$SNAP_OUTPUT" | jq -r '.VolumeSize') SNAP_PROGRESS=$(echo "$SNAP_OUTPUT" | jq -r '.Progress') -if [ "$SNAP_VOL_REF" != "$SNAP_VOL_ID" ]; then - echo "Snapshot VolumeId mismatch: expected $SNAP_VOL_ID, got $SNAP_VOL_REF" +if [ "$SNAP_VOL_REF" != "$VOLUME_ID" ]; then + echo "Snapshot VolumeId mismatch: expected $VOLUME_ID, got $SNAP_VOL_REF" exit 1 fi -if [ "$SNAP_SIZE" -ne 10 ]; then - echo "Snapshot VolumeSize mismatch: expected 10, got $SNAP_SIZE" +if [ "$SNAP_SIZE" -ne "$ROOT_VOL_SIZE" ]; then + echo "Snapshot VolumeSize mismatch: expected $ROOT_VOL_SIZE, got $SNAP_SIZE" exit 1 fi echo "Snapshot create response verified (State=$SNAP_STATE, VolumeId=$SNAP_VOL_REF, Size=$SNAP_SIZE, Progress=$SNAP_PROGRESS)" @@ -448,12 +444,12 @@ DESC_VOL_ID=$(echo "$DESCRIBE_SNAP" | jq -r '.Snapshots[0].VolumeId') DESC_SIZE=$(echo "$DESCRIBE_SNAP" | jq -r '.Snapshots[0].VolumeSize') DESC_DESC=$(echo "$DESCRIBE_SNAP" | jq -r '.Snapshots[0].Description') -if [ "$DESC_VOL_ID" != "$SNAP_VOL_ID" ]; then - echo "Describe snapshot VolumeId mismatch: expected $SNAP_VOL_ID, got $DESC_VOL_ID" +if [ "$DESC_VOL_ID" != "$VOLUME_ID" ]; then + echo "Describe snapshot VolumeId mismatch: expected $VOLUME_ID, got $DESC_VOL_ID" exit 1 fi -if [ "$DESC_SIZE" -ne 10 ]; then - echo "Describe snapshot VolumeSize mismatch: expected 10, got $DESC_SIZE" +if [ "$DESC_SIZE" -ne "$ROOT_VOL_SIZE" ]; then + echo "Describe snapshot VolumeSize mismatch: expected $ROOT_VOL_SIZE, got $DESC_SIZE" exit 1 fi if [ "$DESC_DESC" != "e2e-test-snapshot" ]; then @@ -540,12 +536,44 @@ echo "Copy snapshot still intact after original deletion" echo "Deleting copy snapshot $COPY_SNAPSHOT_ID..." $AWS_EC2 delete-snapshot --snapshot-id "$COPY_SNAPSHOT_ID" -# Clean up the source volume -echo "Deleting source volume $SNAP_VOL_ID..." -$AWS_EC2 delete-volume --volume-id "$SNAP_VOL_ID" - echo "Snapshot lifecycle test passed (create -> describe -> copy -> delete)" +# Phase 5d: Verify Snapshot-Backed Instance Launch +echo "Phase 5d: Verify Snapshot-Backed Instance Launch" +echo "All run-instances calls go through cloneAMIToVolume() -> OpenFromSnapshot()," +echo "so the Phase 5 instance is already snapshot-backed. Verify its volume config." + +AWS_S3="aws --endpoint-url https://localhost:8443 s3" + +# Verify the AMI snapshot exists in Predastore +echo "Checking AMI snapshot in Predastore..." +SNAP_PREFIX="snap-$AMI_ID" +SNAP_FILES=$($AWS_S3 ls "s3://predastore/$SNAP_PREFIX/" 2>&1 || echo "") +if echo "$SNAP_FILES" | grep -q "config.json"; then + echo "AMI snapshot config found at $SNAP_PREFIX/" +else + echo "AMI snapshot config not found at $SNAP_PREFIX/" + exit 1 +fi + +# Verify the Phase 5 instance's root volume has SnapshotID and SourceVolumeName +echo "Verifying root volume $VOLUME_ID is snapshot-backed via Predastore config..." +VOL_CONFIG=$($AWS_S3 cp "s3://predastore/$VOLUME_ID/config.json" - 2>/dev/null || echo "{}") +VOL_SNAPSHOT_ID=$(echo "$VOL_CONFIG" | jq -r '.SnapshotID // empty') +VOL_SOURCE_NAME=$(echo "$VOL_CONFIG" | jq -r '.SourceVolumeName // empty') + +if [ -z "$VOL_SNAPSHOT_ID" ]; then + echo "Volume config missing SnapshotID — launch was NOT snapshot-backed" + exit 1 +fi +if [ -z "$VOL_SOURCE_NAME" ]; then + echo "Volume config missing SourceVolumeName — launch was NOT snapshot-backed" + exit 1 +fi +echo "Volume is snapshot-backed (SnapshotID=$VOL_SNAPSHOT_ID, SourceVolumeName=$VOL_SOURCE_NAME)" + +echo "Snapshot-backed instance launch verified" + # Phase 6: Tag Management echo "Phase 6: Tag Management" diff --git a/tests/e2e/run-multinode-e2e.sh b/tests/e2e/run-multinode-e2e.sh index 222121c96..87475bd01 100755 --- a/tests/e2e/run-multinode-e2e.sh +++ b/tests/e2e/run-multinode-e2e.sh @@ -388,16 +388,13 @@ echo "Test 1c: Snapshot Lifecycle" echo "----------------------------------------" echo "Testing snapshot create -> describe -> copy -> delete..." -# Create a volume to snapshot -echo " Creating source volume for snapshot tests..." -SNAP_VOL_OUTPUT=$($AWS_EC2 create-volume --size 10 --availability-zone ap-southeast-2a) -SNAP_VOL_ID=$(echo "$SNAP_VOL_OUTPUT" | jq -r '.VolumeId') - -if [ -z "$SNAP_VOL_ID" ] || [ "$SNAP_VOL_ID" == "null" ]; then - echo " ERROR: Failed to create source volume for snapshot tests" - exit 1 -fi -echo " Created source volume: $SNAP_VOL_ID" +# Use the root volume of the first instance — it's already attached and mounted +# in viperblockd, which is required for create-snapshot. +SNAP_VOL_ID=$($AWS_EC2 describe-instances --instance-ids "${INSTANCE_IDS[0]}" \ + --query 'Reservations[0].Instances[0].BlockDeviceMappings[0].Ebs.VolumeId' --output text) +echo " Using root volume $SNAP_VOL_ID (attached to ${INSTANCE_IDS[0]})" +SNAP_VOL_SIZE=$($AWS_EC2 describe-volumes --volume-ids "$SNAP_VOL_ID" \ + --query 'Volumes[0].Size' --output text) # Create a snapshot echo " Creating snapshot from volume $SNAP_VOL_ID..." @@ -420,8 +417,8 @@ if [ "$SNAP_VOL_REF" != "$SNAP_VOL_ID" ]; then echo " ERROR: Snapshot VolumeId mismatch: expected $SNAP_VOL_ID, got $SNAP_VOL_REF" exit 1 fi -if [ "$SNAP_SIZE" -ne 10 ]; then - echo " ERROR: Snapshot VolumeSize mismatch: expected 10, got $SNAP_SIZE" +if [ "$SNAP_SIZE" -ne "$SNAP_VOL_SIZE" ]; then + echo " ERROR: Snapshot VolumeSize mismatch: expected $SNAP_VOL_SIZE, got $SNAP_SIZE" exit 1 fi echo " Create response verified (State=$SNAP_STATE, VolumeId=$SNAP_VOL_REF, Size=$SNAP_SIZE)" @@ -458,8 +455,8 @@ if [ "$DESC_VOL_ID" != "$SNAP_VOL_ID" ]; then echo " ERROR: Describe VolumeId mismatch: expected $SNAP_VOL_ID, got $DESC_VOL_ID" exit 1 fi -if [ "$DESC_SIZE" -ne 10 ]; then - echo " ERROR: Describe VolumeSize mismatch: expected 10, got $DESC_SIZE" +if [ "$DESC_SIZE" -ne "$SNAP_VOL_SIZE" ]; then + echo " ERROR: Describe VolumeSize mismatch: expected $SNAP_VOL_SIZE, got $DESC_SIZE" exit 1 fi if [ "$DESC_DESC" != "multinode-e2e-snapshot" ]; then @@ -541,15 +538,52 @@ if [ "$COPY_STATE" != "completed" ]; then fi echo " Copy snapshot intact after original deletion" -# Delete copy and source volume +# Delete copy echo " Deleting copy snapshot $COPY_SNAPSHOT_ID..." $AWS_EC2 delete-snapshot --snapshot-id "$COPY_SNAPSHOT_ID" -echo " Deleting source volume $SNAP_VOL_ID..." -$AWS_EC2 delete-volume --volume-id "$SNAP_VOL_ID" - echo " Snapshot lifecycle test passed (create -> describe -> copy -> delete)" +# Test 1c-ii: Verify Snapshot-Backed Instance Launch +echo "" +echo "Test 1c-ii: Verify Snapshot-Backed Instance Launch" +echo "----------------------------------------" +echo "All run-instances calls go through cloneAMIToVolume() -> OpenFromSnapshot()," +echo "so the Test 1 instances are already snapshot-backed. Verify their volume configs." + +AWS_S3="aws --endpoint-url https://${NODE1_IP}:${PREDASTORE_PORT} s3" + +# Verify the AMI snapshot exists in Predastore +echo " Checking AMI snapshot in Predastore..." +SNAP_PREFIX="snap-$AMI_ID" +SNAP_FILES=$($AWS_S3 ls "s3://predastore/$SNAP_PREFIX/" 2>&1 || echo "") +if echo "$SNAP_FILES" | grep -q "config.json"; then + echo " AMI snapshot config found at $SNAP_PREFIX/" +else + echo " ERROR: AMI snapshot config not found at $SNAP_PREFIX/" + exit 1 +fi + +# Verify the first instance's root volume has SnapshotID and SourceVolumeName +SNAP_ROOT_VOL=$($AWS_EC2 describe-instances --instance-ids "${INSTANCE_IDS[0]}" \ + --query 'Reservations[0].Instances[0].BlockDeviceMappings[0].Ebs.VolumeId' --output text) +echo " Verifying root volume $SNAP_ROOT_VOL is snapshot-backed via Predastore config..." +VOL_CONFIG=$($AWS_S3 cp "s3://predastore/$SNAP_ROOT_VOL/config.json" - 2>/dev/null || echo "{}") +VOL_SNAPSHOT_ID=$(echo "$VOL_CONFIG" | jq -r '.SnapshotID // empty') +VOL_SOURCE_NAME=$(echo "$VOL_CONFIG" | jq -r '.SourceVolumeName // empty') + +if [ -z "$VOL_SNAPSHOT_ID" ]; then + echo " ERROR: Volume config missing SnapshotID — launch was NOT snapshot-backed" + exit 1 +fi +if [ -z "$VOL_SOURCE_NAME" ]; then + echo " ERROR: Volume config missing SourceVolumeName — launch was NOT snapshot-backed" + exit 1 +fi +echo " Volume is snapshot-backed (SnapshotID=$VOL_SNAPSHOT_ID, SourceVolumeName=$VOL_SOURCE_NAME)" + +echo " Snapshot-backed instance launch verified" + # Test 1d: Tag Management echo "" echo "Test 1d: Tag Management"