feat: real snapshots (#32)

This commit is contained in:
Julian Sommer
2026-02-10 16:57:50 +11:00
committed by GitHub
parent 8d3808c364
commit 6abae11122
15 changed files with 1045 additions and 157 deletions
+3 -2
View File
@@ -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
+164
View File
@@ -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 |
+1 -1
View File
@@ -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
+2 -4
View File
@@ -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=
+11
View File
@@ -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
+1 -1
View File
@@ -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()
+24 -69
View File
@@ -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)
}
+49 -13
View File
@@ -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,
@@ -26,7 +26,7 @@ func setupTestSnapshotService(t *testing.T) (*SnapshotServiceImpl, *objectstore.
},
}
svc := NewSnapshotServiceImplWithStore(cfg, store)
svc := NewSnapshotServiceImplWithStore(cfg, store, nil)
return svc, store
}
+183 -9
View File
@@ -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
}
@@ -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)
}
+77 -12
View File
@@ -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()
+16 -2
View File
@@ -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)
+53 -25
View File
@@ -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"
+52 -18
View File
@@ -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"