mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
refactor: split the Viperblock EBS adapter out of services/viperblockd
Move the EBS provider adapter (volume lease, fence, dirty marker, nbdkit recovery, provider handlers, encryption key loading) to providers/ebs/viperblock. services/viperblockd keeps only process lifecycle and the NATS subscription wiring that starts and stops it. The dirty-bucket read path moves to providers/ebs/viperblock/dirty as a read-only ListUnsealedVolumes capability; the adapter's write, clear and purge logic stays unexported in the adapter package. Delete services/viperblockd/vbwire now that its vocabulary lives in contracts/viperblockd/legacy/v1 and its I/O helpers live in dirty. Update operator/cli, the runtime service registry, qemunbdd's config comment and the integration test harness to the new import paths, and drop the now-stale architecture/layering allowlist and frozen-package entries for the deleted package. Behaviour, persisted keys and NATS subjects are unchanged.
This commit is contained in:
@@ -174,7 +174,5 @@ var allowlist = []allowance{
|
||||
{From: "operator/cli", To: "gateway/bedrock", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
{From: "operator/cli", To: "handlers/eks", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
{From: "operator/cli", To: "handlers/iam", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
{From: "operator/cli", To: "services/viperblockd", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
{From: "operator/cli", To: "services/viperblockd/vbwire", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
{From: "operator/cli", To: "vpcd", Debt: "operator/cli (dependency)", Reason: "command tree imports implementations; needs public capabilities"},
|
||||
}
|
||||
|
||||
@@ -35,7 +35,6 @@ var frozenLegacyPackages = []string{
|
||||
"handlers/sts",
|
||||
"lbagent",
|
||||
"services/viperblockd",
|
||||
"services/viperblockd/vbwire",
|
||||
"utils",
|
||||
"vpcd",
|
||||
}
|
||||
|
||||
@@ -50,7 +50,7 @@ RFC2-18 (`mulga/docs/specs/spinifex/rfc2-18_operator-lifecycle.md:34-53,88-104,1
|
||||
|
||||
| Mechanism | Evidence | Scope | Shape |
|
||||
|---|---|---|---|
|
||||
| EBS provider capabilities | `spinifex/providers/ebs/provider.go:55-70`, `spinifex/services/viperblockd/provider_handlers.go:236-262`, `spinifex/daemon/daemon.go:2397` | Spinifex ↔ block provider | Boolean feature set fetched at runtime; the only real capability exchange found |
|
||||
| EBS provider capabilities | `spinifex/providers/ebs/provider.go:55-70`, `spinifex/providers/ebs/viperblock/provider_handlers.go:236-262`, `spinifex/daemon/daemon.go:2397` | Spinifex ↔ block provider | Boolean feature set fetched at runtime; the only real capability exchange found |
|
||||
| EBS provider wire version | `spinifex/providers/ebs/provider.go:13-25`, `spinifex/providers/ebs/errors.go:90-95`, `spinifex/providers/ebs/nats.go:212-316` | Every provider request/response | **Exact match** (`!= SchemaVersion` → `ErrUnsupportedVersion`); no N-1 window |
|
||||
| RDS data-volume contract image tag | `scripts/images/rds-postgres/manifest.conf:49-51`, `scripts/images/rds-mariadb/manifest.conf:59`, `spinifex/handlers/rds/launch.go:39-45,532`, `spinifex/operator/imagecatalog/images.go:666,686` | Control plane → guest image | AMI filter `rds-data-volume-contract=format-auth-v1` excludes older images; a static image-tag capability, not a runtime proof |
|
||||
| RDS bootstrap envelope | `spinifex/handlers/rds/bootstrap.go:23-36,190-192,212` | Host ↔ RDS guest | `envelopeVersion` exact match |
|
||||
|
||||
@@ -268,6 +268,13 @@ services:
|
||||
# ebs.* routes remain to be migrated separately.
|
||||
# The dirty-marker JetStream KV bucket VIPERBLOCK_VOLUME_DIRTY is owned
|
||||
# by the same contract; not a NATS subject, so not listed below.
|
||||
#
|
||||
# Source layout: spinifex/services/viperblockd (role: process lifecycle,
|
||||
# NATS subscription wiring) and spinifex/providers/ebs/viperblock (the
|
||||
# Viperblock EBS provider adapter the role wires up). Selector treats the
|
||||
# adapter path as belonging to this service via `additional_paths`.
|
||||
additional_paths:
|
||||
- spinifex/providers/ebs/viperblock
|
||||
subscribes:
|
||||
- "ebs.mount"
|
||||
- "ebs.unmount"
|
||||
|
||||
@@ -41,12 +41,12 @@ import (
|
||||
"github.com/mulgadc/spinifex/spinifex/operator/imagecatalog"
|
||||
operatorprogress "github.com/mulgadc/spinifex/spinifex/operator/progress"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock/dirty"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/formation"
|
||||
hostcommand "github.com/mulgadc/spinifex/spinifex/runtime/host/command"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/host/dns"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/pterm/pterm"
|
||||
@@ -3995,7 +3995,7 @@ func runVolumesUnsealedCmd(_ *cobra.Command, _ []string) {
|
||||
}
|
||||
defer nc.Close()
|
||||
|
||||
unsealed, err := vbwire.ListUnsealedVolumes(context.Background(), nc)
|
||||
unsealed, err := dirty.ListUnsealedVolumes(context.Background(), nc)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "Could not read the unsealed volumes:", err)
|
||||
os.Exit(1)
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
|
||||
"github.com/mulgadc/spinifex/spinifex/domains/dns"
|
||||
"github.com/mulgadc/spinifex/spinifex/domains/network/external"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/awsgw"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/nats"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/northstar"
|
||||
@@ -21,7 +22,6 @@ import (
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/qmpcollector"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/spinifexui"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/service"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/viperblockd"
|
||||
"github.com/mulgadc/spinifex/spinifex/vpcd"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/spf13/viper"
|
||||
@@ -336,7 +336,7 @@ var viperblockStartCmd = &cobra.Command{
|
||||
|
||||
defer initTelemetry("viperblockd", debug)()
|
||||
|
||||
service, err := service.New("viperblock", &viperblockd.Config{
|
||||
service, err := service.New("viperblock", &viperblock.Config{
|
||||
NatsHost: nodeConfig.NATS.Host,
|
||||
NatsToken: nodeConfig.NATS.ACL.Token,
|
||||
NatsCACert: nodeConfig.NATS.CACert,
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Tests for the encrypted-volume config-update path: applyConfigUpdate,
|
||||
// makeConfigUpdateHandler (ebs.config.{volumeID}), and the ebs.config
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// The contract suite run against the provider production actually uses, over
|
||||
// a real predastore and a real nbdkit export. Until this existed viperblockd
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — covers the unexported shutdownVolumes and reuses the
|
||||
// in-package doubles captureLogs/createTestVBWithState/volumeOpenCount that
|
||||
+12
-9
@@ -1,12 +1,12 @@
|
||||
// Package vbwire is the I/O half of viperblockd's wire contract: the calls
|
||||
// that bind and read the KV bucket recording which node holds a volume's only
|
||||
// current copy. The subjects, DTOs and bucket name it reads are owned by
|
||||
// contracts/viperblockd/legacy/v1; this package has no vocabulary of its own.
|
||||
// Package dirty is a read-only capability over the dirty-volume bucket: which
|
||||
// node holds a volume's only current copy. The subjects, DTOs and bucket name
|
||||
// it reads are owned by contracts/viperblockd/legacy/v1; this package has no
|
||||
// vocabulary of its own and exposes no write, delete or purge operation.
|
||||
//
|
||||
// It is a leaf on purpose. A consumer of the contract -- an operator tool
|
||||
// listing unsealed volumes -- should not have to link the storage service to
|
||||
// listing unsealed volumes -- should not have to link the storage adapter to
|
||||
// read what it published.
|
||||
package vbwire
|
||||
package dirty
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -29,11 +29,14 @@ type UnsealedVolume struct {
|
||||
Reason string
|
||||
}
|
||||
|
||||
// BindDirtyBucket opens the dirty bucket read-only for an operator tool. It does
|
||||
// bindDirtyBucket opens the dirty bucket read-only for an operator tool. It does
|
||||
// not create the bucket: a cluster where no volume has ever been opened has no
|
||||
// bucket, and reporting that as an error would read as a fault. A nil bucket
|
||||
// with a nil error is that case.
|
||||
func BindDirtyBucket(ctx context.Context, nc *nats.Conn) (jetstream.KeyValue, error) {
|
||||
//
|
||||
// Unexported: this package offers only the read-only ListUnsealedVolumes
|
||||
// capability, never a raw bucket handle a caller could write or purge through.
|
||||
func bindDirtyBucket(ctx context.Context, nc *nats.Conn) (jetstream.KeyValue, error) {
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("jetstream: %w", err)
|
||||
@@ -55,7 +58,7 @@ func BindDirtyBucket(ctx context.Context, nc *nats.Conn) (jetstream.KeyValue, er
|
||||
// run, but it opens from the last checkpoint that did reach the backend. This
|
||||
// is the list of volumes where that trade would cost something.
|
||||
func ListUnsealedVolumes(ctx context.Context, nc *nats.Conn) ([]UnsealedVolume, error) {
|
||||
kv, err := BindDirtyBucket(ctx, nc)
|
||||
kv, err := bindDirtyBucket(ctx, nc)
|
||||
if err != nil || kv == nil {
|
||||
return nil, err
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// NBDTransport controls how Viperblockd exports a mounted volume.
|
||||
type NBDTransport string
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import "testing"
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Rebuilds cfg.MountedVolumes from nbdkit processes that survived a
|
||||
// viperblockd restart, so findMountedVolume can find them again (see
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Tests for startup recovery: rebuilding cfg.MountedVolumes from nbdkit
|
||||
// processes that survived a restart, exercised against a fabricated proc
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Tests proving a mounted volume's snapshot handlers reuse the live VB via
|
||||
// findMountedVolume rather than open a second engine, pinned on the same
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Tests for the ebs.provider.v1.* handlers registered by RegisterProviderSubjects.
|
||||
// Engine-backed cases (CreateVolume idempotency, ExpandVolume in-use, CreateSnapshot)
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — mountEntryIsStale and releaseStaleMount are unexported, and
|
||||
//exporting them would widen the package API for a test rather than for a caller.
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,187 @@
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
vbtypes "github.com/mulgadc/viperblock/types"
|
||||
"github.com/mulgadc/viperblock/viperblock"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestConfigLoadStateRetryPolicy(t *testing.T) {
|
||||
attempts, delay := (&Config{}).loadStateRetryPolicy()
|
||||
assert.Equal(t, defaultLoadStateRetryAttempts, attempts)
|
||||
assert.Equal(t, defaultLoadStateRetryBaseDelay, delay)
|
||||
|
||||
cfg := &Config{loadStateRetryAttempts: 2, loadStateRetryBaseDelay: time.Nanosecond}
|
||||
attempts, delay = cfg.loadStateRetryPolicy()
|
||||
assert.Equal(t, 2, attempts)
|
||||
assert.Equal(t, time.Nanosecond, delay)
|
||||
}
|
||||
|
||||
// TestReadVolumeState_BadRequestFailsFast covers all three retry layers with
|
||||
// the production budgets: one HTTP 400 must produce one prompt request.
|
||||
func TestReadVolumeState_BadRequestFailsFast(t *testing.T) {
|
||||
var requests atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
requests.Add(1)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
cfg := &Config{
|
||||
S3Host: srv.URL,
|
||||
Bucket: "bucket",
|
||||
Region: "us-east-1",
|
||||
AccessKey: "access-key",
|
||||
SecretKey: "secret-key",
|
||||
BaseDir: t.TempDir(),
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
_, err := readVolumeState(context.Background(), cfg, "vol-rejected")
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, vbtypes.ErrBackendNonRetryable)
|
||||
assert.Equal(t, int32(1), requests.Load())
|
||||
assert.Less(t, elapsed, time.Second)
|
||||
}
|
||||
|
||||
// TestRetryLoadState pins the retry policy: transient errors retry with backoff,
|
||||
// ErrStateNotFound fails fast, and the retry budget is honoured.
|
||||
func TestRetryLoadState(t *testing.T) {
|
||||
t.Run("success_first_attempt_no_sleep", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error { calls++; return nil })
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, calls)
|
||||
assert.Equal(t, 0, sleeps)
|
||||
})
|
||||
|
||||
t.Run("transient_recovers_within_budget", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error {
|
||||
calls++
|
||||
if calls < 3 {
|
||||
return fmt.Errorf("wrap: %w", viperblock.ErrStateBackendUnavailable)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 3, calls)
|
||||
assert.Equal(t, 2, sleeps)
|
||||
})
|
||||
|
||||
t.Run("not_found_fails_fast", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error { calls++; return viperblock.ErrStateNotFound })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, viperblock.ErrStateNotFound)
|
||||
assert.Equal(t, 1, calls, "ErrStateNotFound must not be retried")
|
||||
assert.Equal(t, 0, sleeps)
|
||||
})
|
||||
|
||||
t.Run("unclassified_fails_fast", func(t *testing.T) {
|
||||
calls := 0
|
||||
sentinel := errors.New("permission denied")
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) {},
|
||||
func() error { calls++; return sentinel })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, sentinel)
|
||||
assert.Equal(t, 1, calls)
|
||||
})
|
||||
|
||||
t.Run("exhausts_budget_on_persistent_transient", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
var lastDelay time.Duration
|
||||
err := retryLoadState("vol-x", 4, 100*time.Millisecond,
|
||||
func(d time.Duration) { sleeps++; lastDelay = d },
|
||||
func() error { calls++; return viperblock.ErrStateBackendUnavailable })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, viperblock.ErrStateBackendUnavailable)
|
||||
assert.Equal(t, 4, calls)
|
||||
assert.Equal(t, 3, sleeps, "sleep happens between attempts, not after the final one")
|
||||
// Backoff multiplier is 1.5: 100ms, 150ms, 225ms.
|
||||
assert.Equal(t, 225*time.Millisecond, lastDelay)
|
||||
})
|
||||
}
|
||||
|
||||
// TestValidVolumeName covers the character-level gate every handler that
|
||||
// turns a wire volume name into a filesystem path must pass first.
|
||||
func TestValidVolumeName(t *testing.T) {
|
||||
valid := []string{"vol-0123456789abcdef0", "vol-0123456789abcdef0-efi", "a"}
|
||||
for _, name := range valid {
|
||||
assert.True(t, validVolumeName(name), "expected %q to be valid", name)
|
||||
}
|
||||
|
||||
invalid := []string{"", ".", "..", "../..", "a/b", "/etc/passwd", "vol/../.."}
|
||||
for _, name := range invalid {
|
||||
assert.False(t, validVolumeName(name), "expected %q to be rejected", name)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLocalVolumeDir is the regression test for the ebs.delete data-destruction
|
||||
// hazard: an unvalidated volume name reaching os.RemoveAll could wipe BaseDir
|
||||
// itself (empty name) or escape it entirely ("../.."). Every case here must be
|
||||
// rejected before a caller ever gets a path back to remove.
|
||||
func TestLocalVolumeDir(t *testing.T) {
|
||||
baseDir := t.TempDir()
|
||||
|
||||
t.Run("empty_volume_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir(baseDir, "")
|
||||
require.Error(t, err, "an empty volume must never resolve to BaseDir itself")
|
||||
})
|
||||
|
||||
t.Run("dot_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir(baseDir, ".")
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("parent_traversal_rejected", func(t *testing.T) {
|
||||
for _, name := range []string{"..", "../..", "../../etc"} {
|
||||
_, err := localVolumeDir(baseDir, name)
|
||||
require.Error(t, err, "volume %q must not escape BaseDir", name)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("embedded_separator_rejected", func(t *testing.T) {
|
||||
for _, name := range []string{"a/b", "a/../../b", "/abs/path"} {
|
||||
_, err := localVolumeDir(baseDir, name)
|
||||
require.Error(t, err, "volume %q must not contain a path separator", name)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("empty_base_dir_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir("", "vol-abc123")
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("valid_name_resolves_under_base_dir", func(t *testing.T) {
|
||||
dir, err := localVolumeDir(baseDir, "vol-abc123")
|
||||
require.NoError(t, err)
|
||||
absBase, err := filepath.Abs(baseDir)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, filepath.Join(absBase, "vol-abc123"), dir)
|
||||
})
|
||||
}
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Tests for NATS message handlers: ebs.delete, ebs.sync, ebs.unmount
|
||||
|
||||
+18
-12
@@ -1,8 +1,8 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
// Integration tests for viperblockd using an embedded NATS server. Run with:
|
||||
// go test ./spinifex/services/viperblockd/... -timeout 40s
|
||||
// go test -short ./spinifex/services/viperblockd/... # unit tests only
|
||||
// Integration tests for the Viperblock adapter using an embedded NATS server. Run with:
|
||||
// go test ./spinifex/providers/ebs/viperblock/... -timeout 40s
|
||||
// go test -short ./spinifex/providers/ebs/viperblock/... # unit tests only
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
@@ -48,22 +48,28 @@ func installTestVolumeLeases(t *testing.T, cfg *Config, natsURL string) {
|
||||
cfg.leases = leases
|
||||
}
|
||||
|
||||
// startTestService launches the daemon and blocks until every subscription is
|
||||
// registered on the server. launchService returns early on a startup error, so
|
||||
// the wait is bounded rather than blocking the suite on a daemon that died.
|
||||
// startTestService runs the adapter's startup sequence (the same one
|
||||
// viperblockd's role wires up over a connection it owns) and blocks until
|
||||
// every subscription is registered on the server. Run returns early on a
|
||||
// startup error, so the wait is bounded rather than blocking the suite on an
|
||||
// adapter that died.
|
||||
func startTestService(t *testing.T, cfg *Config) {
|
||||
t.Helper()
|
||||
|
||||
nc, err := nats.Connect(cfg.NatsHost)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(nc.Close)
|
||||
|
||||
cfg.ready = make(chan struct{})
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- launchService(cfg) }()
|
||||
go func() { errCh <- cfg.Run(nc) }()
|
||||
|
||||
select {
|
||||
case <-cfg.ready:
|
||||
case err := <-errCh:
|
||||
t.Fatalf("launchService returned before becoming ready: %v", err)
|
||||
t.Fatalf("Run returned before becoming ready: %v", err)
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("launchService did not become ready")
|
||||
t.Fatal("Run did not become ready")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -189,7 +195,7 @@ func TestIntegration_ServiceStartWithEmbeddedNATS(t *testing.T) {
|
||||
err = nc.Publish("test.subject", []byte("test"))
|
||||
assert.NoError(t, err)
|
||||
|
||||
// Test completes, launchService will be cleaned up when test ends
|
||||
// Test completes, the adapter goroutine will be cleaned up when test ends
|
||||
}
|
||||
|
||||
// TestIntegration_EBSMountRequest tests EBS mount request handling.
|
||||
@@ -813,7 +819,7 @@ func TestIntegration_ServiceGracefulShutdown(t *testing.T) {
|
||||
|
||||
startTestService(t, cfg)
|
||||
|
||||
// Note: We can't easily test SIGTERM handling without refactoring launchService
|
||||
// Note: SIGTERM handling lives in the role (services/viperblockd) and is not exercised here
|
||||
// to accept a context or shutdown channel. For now, we verify the service
|
||||
// is running and leave full shutdown testing for manual integration tests.
|
||||
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — abandonVolume, the lease store and MountedVolumes are all
|
||||
//unexported, and the reason this differs from an unmount has no exported form.
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — both bucket constructors are unexported, and the replica
|
||||
//count they are created with has no exported surface to read it back from.
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
//test:in-package — the start-path bucket openers and the constructors beneath
|
||||
//them are unexported, and the point is which of the two waits.
|
||||
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"testing"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+3
-3
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — the dirty marker, its bucket and acquireVolumeLease are all
|
||||
//unexported, and the takeover they implement has no exported surface.
|
||||
@@ -6,7 +6,7 @@ package viperblockd
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire"
|
||||
vbdirty "github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock/dirty"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -257,7 +257,7 @@ func TestVolumeDirty_ListReportsTheHolder(t *testing.T) {
|
||||
dirty := newTestDirty(t, natsURL, "node-a")
|
||||
require.NoError(t, dirty.mark(t.Context(), "vol-dirtylisted", 1, "seal volume: predastore unreachable"))
|
||||
|
||||
unsealed, err := vbwire.ListUnsealedVolumes(t.Context(), nc)
|
||||
unsealed, err := vbdirty.ListUnsealedVolumes(t.Context(), nc)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, unsealed, 1)
|
||||
assert.Equal(t, "vol-dirtylisted", unsealed[0].VolumeID)
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
//test:in-package — the fence, the lease store and MountedVolumes are all
|
||||
//unexported, and the decision the fence makes has no exported surface.
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,4 +1,4 @@
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"context"
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
//test:in-package: asserts on volumeVBConfig, the unexported per-volume VB
|
||||
// config builder.
|
||||
|
||||
package viperblockd
|
||||
package viperblock
|
||||
|
||||
import (
|
||||
"testing"
|
||||
@@ -17,8 +17,8 @@ import (
|
||||
// serviceName roots this daemon's PID file: baseDir/qemunbd.pid.
|
||||
var serviceName = "qemunbd"
|
||||
|
||||
// Config is the qemunbdd service's startup configuration, mirroring
|
||||
// viperblockd.Config's NATS and base-directory fields. NodeName scopes the
|
||||
// Config is the qemunbdd service's startup configuration, mirroring the
|
||||
// Viperblock adapter's NATS and base-directory fields. NodeName scopes the
|
||||
// natsserve PublishVolume/UnpublishVolume subjects to this node.
|
||||
type Config struct {
|
||||
BaseDir string
|
||||
|
||||
@@ -4,10 +4,10 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/nats"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/predastore"
|
||||
"github.com/mulgadc/spinifex/spinifex/runtime/roles/spinifexui"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/viperblockd"
|
||||
"github.com/mulgadc/spinifex/spinifex/vpcd"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
@@ -28,7 +28,7 @@ func TestNew(t *testing.T) {
|
||||
svc, err = New(s, &predastore.Config{})
|
||||
// No special setup needed
|
||||
case "viperblock":
|
||||
svc, err = New(s, &viperblockd.Config{})
|
||||
svc, err = New(s, &viperblock.Config{})
|
||||
case "spinifex":
|
||||
svc, err = New(s, &config.ClusterConfig{})
|
||||
case "awsgw":
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,25 +1,15 @@
|
||||
package viperblockd
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
vbtypes "github.com/mulgadc/viperblock/types"
|
||||
"github.com/mulgadc/viperblock/viperblock"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestNew tests the service constructor.
|
||||
func TestNew(t *testing.T) {
|
||||
cfg := &Config{
|
||||
cfg := &viperblock.Config{
|
||||
NatsHost: "nats://localhost:4222",
|
||||
S3Host: "https://s3.amazonaws.com",
|
||||
Bucket: "test-bucket",
|
||||
@@ -45,7 +35,7 @@ func TestNew(t *testing.T) {
|
||||
func TestNewWithNilConfig(t *testing.T) {
|
||||
// This will panic if not handled, but based on the code it type asserts
|
||||
// For now we test that it accepts a Config pointer
|
||||
cfg := &Config{}
|
||||
cfg := &viperblock.Config{}
|
||||
svc, err := New(cfg)
|
||||
|
||||
assert.NoError(t, err)
|
||||
@@ -57,172 +47,3 @@ func TestNewWithNilConfig(t *testing.T) {
|
||||
func TestServiceNameConstant(t *testing.T) {
|
||||
assert.Equal(t, "viperblock", serviceName)
|
||||
}
|
||||
|
||||
func TestConfigLoadStateRetryPolicy(t *testing.T) {
|
||||
attempts, delay := (&Config{}).loadStateRetryPolicy()
|
||||
assert.Equal(t, defaultLoadStateRetryAttempts, attempts)
|
||||
assert.Equal(t, defaultLoadStateRetryBaseDelay, delay)
|
||||
|
||||
cfg := &Config{loadStateRetryAttempts: 2, loadStateRetryBaseDelay: time.Nanosecond}
|
||||
attempts, delay = cfg.loadStateRetryPolicy()
|
||||
assert.Equal(t, 2, attempts)
|
||||
assert.Equal(t, time.Nanosecond, delay)
|
||||
}
|
||||
|
||||
// TestReadVolumeState_BadRequestFailsFast covers all three retry layers with
|
||||
// the production budgets: one HTTP 400 must produce one prompt request.
|
||||
func TestReadVolumeState_BadRequestFailsFast(t *testing.T) {
|
||||
var requests atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
requests.Add(1)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
|
||||
cfg := &Config{
|
||||
S3Host: srv.URL,
|
||||
Bucket: "bucket",
|
||||
Region: "us-east-1",
|
||||
AccessKey: "access-key",
|
||||
SecretKey: "secret-key",
|
||||
BaseDir: t.TempDir(),
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
_, err := readVolumeState(context.Background(), cfg, "vol-rejected")
|
||||
elapsed := time.Since(start)
|
||||
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, vbtypes.ErrBackendNonRetryable)
|
||||
assert.Equal(t, int32(1), requests.Load())
|
||||
assert.Less(t, elapsed, time.Second)
|
||||
}
|
||||
|
||||
// TestRetryLoadState pins the retry policy: transient errors retry with backoff,
|
||||
// ErrStateNotFound fails fast, and the retry budget is honoured.
|
||||
func TestRetryLoadState(t *testing.T) {
|
||||
t.Run("success_first_attempt_no_sleep", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error { calls++; return nil })
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, calls)
|
||||
assert.Equal(t, 0, sleeps)
|
||||
})
|
||||
|
||||
t.Run("transient_recovers_within_budget", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error {
|
||||
calls++
|
||||
if calls < 3 {
|
||||
return fmt.Errorf("wrap: %w", viperblock.ErrStateBackendUnavailable)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 3, calls)
|
||||
assert.Equal(t, 2, sleeps)
|
||||
})
|
||||
|
||||
t.Run("not_found_fails_fast", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) { sleeps++ },
|
||||
func() error { calls++; return viperblock.ErrStateNotFound })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, viperblock.ErrStateNotFound)
|
||||
assert.Equal(t, 1, calls, "ErrStateNotFound must not be retried")
|
||||
assert.Equal(t, 0, sleeps)
|
||||
})
|
||||
|
||||
t.Run("unclassified_fails_fast", func(t *testing.T) {
|
||||
calls := 0
|
||||
sentinel := errors.New("permission denied")
|
||||
err := retryLoadState("vol-x", 5, 10*time.Millisecond,
|
||||
func(time.Duration) {},
|
||||
func() error { calls++; return sentinel })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, sentinel)
|
||||
assert.Equal(t, 1, calls)
|
||||
})
|
||||
|
||||
t.Run("exhausts_budget_on_persistent_transient", func(t *testing.T) {
|
||||
calls := 0
|
||||
sleeps := 0
|
||||
var lastDelay time.Duration
|
||||
err := retryLoadState("vol-x", 4, 100*time.Millisecond,
|
||||
func(d time.Duration) { sleeps++; lastDelay = d },
|
||||
func() error { calls++; return viperblock.ErrStateBackendUnavailable })
|
||||
require.Error(t, err)
|
||||
assert.ErrorIs(t, err, viperblock.ErrStateBackendUnavailable)
|
||||
assert.Equal(t, 4, calls)
|
||||
assert.Equal(t, 3, sleeps, "sleep happens between attempts, not after the final one")
|
||||
// Backoff multiplier is 1.5: 100ms, 150ms, 225ms.
|
||||
assert.Equal(t, 225*time.Millisecond, lastDelay)
|
||||
})
|
||||
}
|
||||
|
||||
// TestValidVolumeName covers the character-level gate every handler that
|
||||
// turns a wire volume name into a filesystem path must pass first.
|
||||
func TestValidVolumeName(t *testing.T) {
|
||||
valid := []string{"vol-0123456789abcdef0", "vol-0123456789abcdef0-efi", "a"}
|
||||
for _, name := range valid {
|
||||
assert.True(t, validVolumeName(name), "expected %q to be valid", name)
|
||||
}
|
||||
|
||||
invalid := []string{"", ".", "..", "../..", "a/b", "/etc/passwd", "vol/../.."}
|
||||
for _, name := range invalid {
|
||||
assert.False(t, validVolumeName(name), "expected %q to be rejected", name)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLocalVolumeDir is the regression test for the ebs.delete data-destruction
|
||||
// hazard: an unvalidated volume name reaching os.RemoveAll could wipe BaseDir
|
||||
// itself (empty name) or escape it entirely ("../.."). Every case here must be
|
||||
// rejected before a caller ever gets a path back to remove.
|
||||
func TestLocalVolumeDir(t *testing.T) {
|
||||
baseDir := t.TempDir()
|
||||
|
||||
t.Run("empty_volume_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir(baseDir, "")
|
||||
require.Error(t, err, "an empty volume must never resolve to BaseDir itself")
|
||||
})
|
||||
|
||||
t.Run("dot_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir(baseDir, ".")
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("parent_traversal_rejected", func(t *testing.T) {
|
||||
for _, name := range []string{"..", "../..", "../../etc"} {
|
||||
_, err := localVolumeDir(baseDir, name)
|
||||
require.Error(t, err, "volume %q must not escape BaseDir", name)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("embedded_separator_rejected", func(t *testing.T) {
|
||||
for _, name := range []string{"a/b", "a/../../b", "/abs/path"} {
|
||||
_, err := localVolumeDir(baseDir, name)
|
||||
require.Error(t, err, "volume %q must not contain a path separator", name)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("empty_base_dir_rejected", func(t *testing.T) {
|
||||
_, err := localVolumeDir("", "vol-abc123")
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("valid_name_resolves_under_base_dir", func(t *testing.T) {
|
||||
dir, err := localVolumeDir(baseDir, "vol-abc123")
|
||||
require.NoError(t, err)
|
||||
absBase, err := filepath.Abs(baseDir)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, filepath.Join(absBase, "vol-abc123"), dir)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -10,8 +10,8 @@ import (
|
||||
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
|
||||
ec2volume "github.com/mulgadc/spinifex/spinifex/domains/ec2/volume"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/ebs/viperblock"
|
||||
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
|
||||
"github.com/mulgadc/spinifex/spinifex/services/viperblockd"
|
||||
testpredastore "github.com/mulgadc/spinifex/tests/fixtures/predastore"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
@@ -76,7 +76,7 @@ func StartVolumeDaemonLite(t *testing.T, gw *Gateway) *ec2volume.VolumeServiceIm
|
||||
// The control plane no longer builds volumes itself; it asks a provider.
|
||||
// Serve the provider contract from viperblockd against the same fixture,
|
||||
// so the blocks are still written by real viperblock to real predastore.
|
||||
require.NoError(t, viperblockd.RegisterProviderSubjects(&viperblockd.Config{
|
||||
require.NoError(t, viperblock.RegisterProviderSubjects(&viperblock.Config{
|
||||
S3Host: "https://" + fixture.Host,
|
||||
Bucket: testVolumeBucket,
|
||||
Region: fixture.Region,
|
||||
|
||||
Reference in New Issue
Block a user