gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-15 09:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: establish legacy Viperblockd sync contract

Browse Source
This commit is contained in:
joshsiv-mulga
2026-10-01 12:12:01 +10:00
parent 94a6e7af25
commit 0b82e14049
10 changed files with 96 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/viperblockd/legacy/v1/README.md
+18 -5
View File
@@ -1,10 +1,10 @@
# Viperblockd legacy contracts, version 1
This package owns the deployed, transitional request/reply protocol between
Spinifex's legacy control-plane EBS routes and `viperblockd`. It makes the
existing compatibility boundary explicit while the generic `providers/ebs`
contract is adopted; it is not the EBS provider abstraction and new provider
integrations must not depend on it.
This package owns the deployed, transitional configuration-update and
state-reload request/reply protocols between Spinifex's legacy control plane
and `viperblockd`. It makes those existing compatibility boundaries explicit
while the generic `providers/ebs` contract is adopted; it is not the EBS
provider abstraction and new provider integrations must not depend on it.
## Configuration update routes
@@ -22,6 +22,19 @@ Both routes carry `EBSConfigUpdateRequest` and respond with
request/reply semantics are deployed compatibility behaviour. Incompatible
changes require a new contract version or an explicit migration plan.
## State reload route
`SyncSubject` is `ebs.sync`. After a successful EC2 `ModifyVolume`, the daemon
uses request/reply to ask a `spinifex-workers` queue-group member to reload the
volume's state from its backend. The selected Viperblockd worker responds with
`EBSSyncResponse`; it returns an error when it does not own a live mount of the
volume. The daemon records that error as a warning after the API response has
already been returned. This is deployed legacy behaviour, not a guarantee that
every modification has reached its owning mount.
The route carries `EBSSyncRequest` and `EBSSyncResponse`. Their JSON shape and
the queue-group route are compatibility behaviour.
This package does not authorize requests, validate a volume ID, store volume
state, or define the generic storage-provider API. Those concerns remain with
the consuming daemon, the EC2 domain, and `providers/ebs` respectively.
contracts/viperblockd/legacy/v1/sync.go
+22
View File
@@ -0,0 +1,22 @@
package viperblocklegacyv1
const (
// SyncSubject is the legacy queue-group route that asks a viperblockd
// worker to reload a mounted volume's state from its backend.
SyncSubject = "ebs.sync"
)
// EBSSyncRequest identifies the volume whose mounted Viperblock state should
// be reloaded by the selected legacy worker.
type EBSSyncRequest struct {
Volume string `json:"Volume"`
}
// EBSSyncResponse reports the selected worker's reload result. A worker that
// does not own the mounted volume returns an error rather than forwarding the
// request.
type EBSSyncResponse struct {
Volume string `json:"Volume"`
Synced bool `json:"Synced"`
Error string `json:"Error"`
}
contracts/viperblockd/legacy/v1/sync_test.go
+28
View File
@@ -0,0 +1,28 @@
package viperblocklegacyv1
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
func TestSyncSubject(t *testing.T) {
require.Equal(t, "ebs.sync", SyncSubject)
}
func TestSyncRequestJSONContract(t *testing.T) {
data, err := json.Marshal(EBSSyncRequest{Volume: "vol-123"})
require.NoError(t, err)
require.JSONEq(t, `{"Volume":"vol-123"}`, string(data))
}
func TestSyncResponseJSONContract(t *testing.T) {
data, err := json.Marshal(EBSSyncResponse{
Volume: "vol-123",
Synced: true,
Error: "",
})
require.NoError(t, err)
require.JSONEq(t, `{"Volume":"vol-123","Synced":true,"Error":""}`, string(data))
}
docs/PACKAGE_BOUNDARY_MIGRATION.md
+1
View File
@@ -191,6 +191,7 @@ while the branch is in flight; they do not describe the umbrella repository's
| `08bd335f0` | `spinifex/types/node.go` (`NodeStatusResponse`, capacity and GPU inventory) | `contracts/cluster/v1/status.go` | Node status is a versioned fan-out observation contract for operator inspection and EC2/EKS scheduling consumers; `status_test.go` pins its subject and JSON shape. |
| `f4cb550c9` | `spinifex/types/ebs.go` (GP3 policy constants) | `spinifex/domains/ec2/ebs/policy/gp3.go` | The supported GP3 product envelope is EC2 EBS domain policy, not generic runtime or Viperblockd wire state; `gp3_test.go` pins the deployed values. |
| `adb55de33` | `spinifex/types/ebs.go` (`NBDTransport`) | `spinifex/services/viperblockd/nbd_transport.go` | NBD endpoint transport selection is Viperblockd-local delivery configuration, not generic EBS state or a provider contract; `nbd_transport_test.go` pins the socket and TCP forms. |
| `94a6e7af2` | `spinifex/types/ebs.go` (`EBSConfigUpdateRequest`, `EBSConfigUpdateResponse`) | `contracts/viperblockd/legacy/v1/config_update.go` | Legacy configuration updates have deployed queue and volume-addressed NATS routes; the versioned compatibility contract is expressly transitional and separate from `providers/ebs`. |
The EC2 contract row is intentionally different from the directory moves: it creates
a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS
docs/service-interfaces.yaml
+7 -5
View File
@@ -169,9 +169,10 @@ services:
- "vpc.dhcp.acquire"
- "vpc.dhcp.release"
# → viperblockd
# Legacy configuration-update payloads and routes are owned by
# contracts/viperblockd/legacy/v1; they are transitional and distinct
# from the generic providers/ebs protocol.
# The migrated legacy ebs.config and ebs.sync payloads and routes are
# owned by contracts/viperblockd/legacy/v1. They are transitional and
# distinct from the generic providers/ebs protocol; other legacy ebs.*
# routes remain to be migrated separately.
- "ebs.mount"
- "ebs.unmount"
- "ebs.delete"
@@ -227,8 +228,9 @@ services:
# mulga/viperblock submodule. Submodule pointer bump → selector forces
# "run all" (contract changes can be invisible at the spinifex source
# level).
# The legacy ebs.config routes are versioned by
# contracts/viperblockd/legacy/v1, not by providers/ebs.
# The migrated legacy ebs.config and ebs.sync routes are versioned by
# contracts/viperblockd/legacy/v1, not by providers/ebs. Other legacy
# ebs.* routes remain to be migrated separately.
subscribes:
- "ebs.mount"
- "ebs.unmount"
spinifex/daemon/daemon_handlers_test.go
+2 -1
View File
@@ -19,6 +19,7 @@ import (
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/ebsfake"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
@@ -2675,7 +2676,7 @@ func TestHandleEC2ModifyVolume_Success(t *testing.T) {
})
// Subscribe a dummy ebs.sync handler so the NATS Request doesn't time out
syncSub, err := daemon.natsConn.Subscribe("ebs.sync", func(msg *nats.Msg) {
syncSub, err := daemon.natsConn.Subscribe(viperblocklegacyv1.SyncSubject, func(msg *nats.Msg) {
_ = msg.Respond([]byte(`{}`))
})
require.NoError(t, err)
spinifex/daemon/daemon_handlers_volume.go
+3 -3
View File
@@ -9,10 +9,10 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
viperblocklegacyv1 "github.com/mulgadc/spinifex/contracts/viperblockd/legacy/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
@@ -241,11 +241,11 @@ func (d *Daemon) handleEC2ModifyVolume(msg *nats.Msg) string {
// Notify viperblockd to reload state after volume modification (e.g. resize)
if modifyVolumeInput.VolumeId != nil {
syncData, err := json.Marshal(types.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId})
syncData, err := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: *modifyVolumeInput.VolumeId})
if err != nil {
slog.ErrorContext(ctx, "failed to marshal ebs.sync request", "volumeId", *modifyVolumeInput.VolumeId, "err", err)
} else {
_, syncErr := d.natsConn.Request("ebs.sync", syncData, 5*time.Second)
_, syncErr := d.natsConn.Request(viperblocklegacyv1.SyncSubject, syncData, 5*time.Second)
if syncErr != nil {
slog.WarnContext(ctx, "ebs.sync notification failed (volume may not be mounted)",
"volumeId", *modifyVolumeInput.VolumeId, "err", syncErr)
spinifex/services/viperblockd/viperblockd.go
+4 -4
View File
@@ -895,20 +895,20 @@ func launchService(cfg *Config) (err error) {
return fmt.Errorf("failed to subscribe to %s: %w", abandonTopic, err)
}
if _, err := nc.QueueSubscribe("ebs.sync", "spinifex-workers", func(msg *nats.Msg) {
if _, err := nc.QueueSubscribe(viperblocklegacyv1.SyncSubject, "spinifex-workers", func(msg *nats.Msg) {
ctx, span := utils.StartConsumerSpan(msg)
defer span.End()
slog.InfoContext(ctx, "Received ebs.sync message")
var syncRequest types.EBSSyncRequest
var syncRequest viperblocklegacyv1.EBSSyncRequest
if err := json.Unmarshal(msg.Data, &syncRequest); err != nil {
slog.ErrorContext(ctx, "Failed to unmarshal ebs.sync message", "err", err)
utils.MarkSpanError(span, err)
respondJSON(msg, types.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)})
respondJSON(msg, viperblocklegacyv1.EBSSyncResponse{Error: fmt.Sprintf("bad request: %v", err)})
return
}
syncResponse := types.EBSSyncResponse{Volume: syncRequest.Volume}
syncResponse := viperblocklegacyv1.EBSSyncResponse{Volume: syncRequest.Volume}
defer func() {
if syncResponse.Error != "" {
utils.MarkSpanError(span, errors.New(syncResponse.Error))
spinifex/services/viperblockd/viperblockd_handlers_test.go
+11 -11
View File
@@ -324,11 +324,11 @@ func TestIntegration_EBSSyncVolumeNotMounted(t *testing.T) {
require.NoError(t, err)
defer nc.Close()
reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-not-here"})
msg, err := nc.Request("ebs.sync", reqData, 3*time.Second)
reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-not-here"})
msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second)
require.NoError(t, err)
var resp types.EBSSyncResponse
var resp viperblocklegacyv1.EBSSyncResponse
require.NoError(t, json.Unmarshal(msg.Data, &resp))
assert.False(t, resp.Synced)
assert.Contains(t, resp.Error, "not mounted")
@@ -354,11 +354,11 @@ func TestIntegration_EBSSyncVolumeNoVBInstance(t *testing.T) {
require.NoError(t, err)
defer nc.Close()
reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-no-vb"})
msg, err := nc.Request("ebs.sync", reqData, 3*time.Second)
reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-no-vb"})
msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second)
require.NoError(t, err)
var resp types.EBSSyncResponse
var resp viperblocklegacyv1.EBSSyncResponse
require.NoError(t, json.Unmarshal(msg.Data, &resp))
assert.False(t, resp.Synced)
assert.Contains(t, resp.Error, "not mounted")
@@ -381,10 +381,10 @@ func TestIntegration_EBSSyncInvalidJSON(t *testing.T) {
require.NoError(t, err)
defer nc.Close()
msg, err := nc.Request("ebs.sync", []byte("garbage"), 3*time.Second)
msg, err := nc.Request(viperblocklegacyv1.SyncSubject, []byte("garbage"), 3*time.Second)
require.NoError(t, err)
var resp types.EBSSyncResponse
var resp viperblocklegacyv1.EBSSyncResponse
require.NoError(t, json.Unmarshal(msg.Data, &resp))
assert.Contains(t, resp.Error, "bad request:")
}
@@ -491,11 +491,11 @@ func TestIntegration_EBSSyncWithVBInstance(t *testing.T) {
require.NoError(t, err)
defer nc.Close()
reqData, _ := json.Marshal(types.EBSSyncRequest{Volume: "vol-sync-vb"})
msg, err := nc.Request("ebs.sync", reqData, 3*time.Second)
reqData, _ := json.Marshal(viperblocklegacyv1.EBSSyncRequest{Volume: "vol-sync-vb"})
msg, err := nc.Request(viperblocklegacyv1.SyncSubject, reqData, 3*time.Second)
require.NoError(t, err)
var resp types.EBSSyncResponse
var resp viperblocklegacyv1.EBSSyncResponse
require.NoError(t, json.Unmarshal(msg.Data, &resp))
assert.True(t, resp.Synced)
assert.Empty(t, resp.Error)
spinifex/types/ebs.go
-10
View File
@@ -44,16 +44,6 @@ type EBSUnMountResponse struct {
Reaped bool `json:"Reaped,omitempty"`
}
type EBSSyncRequest struct {
Volume string `json:"Volume"`
}
type EBSSyncResponse struct {
Volume string `json:"Volume"`
Synced bool `json:"Synced"`
Error string `json:"Error"`
}
type EBSDeleteRequest struct {
Volume string `json:"Volume"`
}