From 5d88a339982a1a5f0e8debcd81473dda37a95d7d Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Wed, 30 Sep 2026 15:56:13 +1000 Subject: [PATCH] refactor: establish operator storage contract --- contracts/operator/v1/README.md | 10 ++++ contracts/operator/v1/storage.go | 50 +++++++++++++++++++ contracts/operator/v1/storage_test.go | 28 +++++++++++ docs/DESIGN.md | 6 ++- docs/service-interfaces.yaml | 1 + spinifex/daemon/daemon.go | 3 +- spinifex/daemon/daemon_handlers_storage.go | 18 +++---- .../daemon/daemon_handlers_storage_test.go | 10 ++-- spinifex/gateway/spx/storage.go | 14 +++--- spinifex/gateway/spx/storage_test.go | 14 +++--- spinifex/types/storage.go | 38 -------------- 11 files changed, 124 insertions(+), 68 deletions(-) create mode 100644 contracts/operator/v1/README.md create mode 100644 contracts/operator/v1/storage.go create mode 100644 contracts/operator/v1/storage_test.go delete mode 100644 spinifex/types/storage.go diff --git a/contracts/operator/v1/README.md b/contracts/operator/v1/README.md new file mode 100644 index 000000000..db502b592 --- /dev/null +++ b/contracts/operator/v1/README.md @@ -0,0 +1,10 @@ +# Operator v1 contracts + +This package owns versioned cross-process interfaces for deployment +administration. It is separate from AWS tenant-facing service contracts. + +`StorageConfigSubject` supplies the Predastore topology consumed by SPX +`GetStorageStatus`. Its response deliberately contains Predastore-specific +roles and Reed-Solomon configuration. It is an operational status contract, +not a generic provider abstraction; do not change its subject or JSON shape +incompatibly. Add a new version when that is required. diff --git a/contracts/operator/v1/storage.go b/contracts/operator/v1/storage.go new file mode 100644 index 000000000..ef9c8b4b7 --- /dev/null +++ b/contracts/operator/v1/storage.go @@ -0,0 +1,50 @@ +// Package operatorv1 defines version 1 of cross-process operator-plane +// contracts. These are deployment-administration interfaces, not AWS +// tenant-facing APIs. +package operatorv1 + +const ( + // StorageConfigSubject returns the current Predastore topology as configured + // for this deployment. It is consumed by the SPX operator status path. + StorageConfigSubject = "spinifex.storage.config" +) + +// StorageConfigResponse is the operator-plane response on +// StorageConfigSubject. It intentionally describes the current Predastore +// implementation (Reed-Solomon and meta/blob roles); it is not a generic +// storage-provider contract. +type StorageConfigResponse struct { + Encoding StorageEncoding `json:"encoding"` + MetaNodes []StorageMetaNode `json:"meta_nodes"` + BlobNodes []StorageBlobNode `json:"blob_nodes"` + Buckets []StorageBucket `json:"buckets"` +} + +// StorageEncoding describes Predastore's Reed-Solomon erasure coding +// configuration. +type StorageEncoding struct { + DataShards int `json:"data_shards"` + ParityShards int `json:"parity_shards"` +} + +// StorageMetaNode describes a Predastore meta node, one member of the Raft +// quorum over global state. Credentials are deliberately excluded. +type StorageMetaNode struct { + ID int `json:"id"` + Host string `json:"host"` + Port int `json:"port"` +} + +// StorageBlobNode describes a Predastore blob node that holds erasure-coded +// object shards. +type StorageBlobNode struct { + ID int `json:"id"` + Host string `json:"host"` + Port int `json:"port"` +} + +// StorageBucket describes a configured S3 bucket. +type StorageBucket struct { + Name string `json:"name"` + Region string `json:"region"` +} diff --git a/contracts/operator/v1/storage_test.go b/contracts/operator/v1/storage_test.go new file mode 100644 index 000000000..ff9167d72 --- /dev/null +++ b/contracts/operator/v1/storage_test.go @@ -0,0 +1,28 @@ +package operatorv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestStorageConfigResponseJSONContract(t *testing.T) { + data, err := json.Marshal(StorageConfigResponse{ + Encoding: StorageEncoding{DataShards: 2, ParityShards: 1}, + MetaNodes: []StorageMetaNode{{ID: 1, Host: "10.0.0.1", Port: 7660}}, + BlobNodes: []StorageBlobNode{{ID: 2, Host: "10.0.0.1", Port: 6660}}, + Buckets: []StorageBucket{{Name: "objects", Region: "ap-southeast-2"}}, + }) + require.NoError(t, err) + require.JSONEq(t, `{ + "encoding":{"data_shards":2,"parity_shards":1}, + "meta_nodes":[{"id":1,"host":"10.0.0.1","port":7660}], + "blob_nodes":[{"id":2,"host":"10.0.0.1","port":6660}], + "buckets":[{"name":"objects","region":"ap-southeast-2"}] + }`, string(data)) +} + +func TestStorageConfigSubject(t *testing.T) { + require.Equal(t, "spinifex.storage.config", StorageConfigSubject) +} diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 31c2b06d4..cb8709597 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -268,7 +268,11 @@ Ports are unique within a host but repeat across the cluster, so a multi-node cl A single-node install is the same arrangement with one host: seven nodes — a gate, three blob and three meta — colocated in one process, needing seven distinct ports (8443, 6660–6662, 7660–7662) to stay unique within the host even though only the gate opens a listener. It has no code path of its own. -The daemon reports the topology on `spinifex.storage.config` by parsing the config file. It does not connect to Predastore to do so, and no Predastore node exposes a status endpoint. +The daemon reports the topology on `spinifex.storage.config` (the +`contracts/operator/v1` Predastore-topology contract) by parsing the config +file. It does not connect to Predastore to do so, and no Predastore node +exposes a status endpoint. This is an operator status interface, not a generic +storage-provider contract. ## Configuration diff --git a/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index 769938685..65bdcf852 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -101,6 +101,7 @@ services: - "spinifex.cluster.shutdown.storage" - "spinifex.cluster.shutdown.persist" - "spinifex.cluster.shutdown.infra" + # Operator-plane Predastore topology contract (contracts/operator/v1). - "spinifex.storage.config" # ELBv2 control plane (handlers/elbv2/service_nats.go topic map). - "elbv2.CreateLoadBalancer" diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index c7b7bd61c..adb40cc7a 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -33,6 +33,7 @@ import ( "github.com/mulgadc/bluebottle/pkg/tlsconfig" clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/contracts/ec2/v1" + operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/bootstrap/preflight" "github.com/mulgadc/spinifex/spinifex/config" @@ -1021,7 +1022,7 @@ func (d *Daemon) subscribeAll() error { {clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""}, {"spinifex.node.status", d.handleNodeStatus, ""}, {"spinifex.node.vms", d.handleNodeVMs, ""}, - {"spinifex.storage.config", d.handleStorageConfig, ""}, + {operatorv1.StorageConfigSubject, d.handleStorageConfig, ""}, {"spinifex.image.promote", d.handleSpinifexPromoteImage, "spinifex-workers"}, // Account creation → create default VPC for new account {"iam.account.created", d.handleAccountCreated, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_handlers_storage.go b/spinifex/daemon/daemon_handlers_storage.go index 49a6ee3af..f022ac5ca 100644 --- a/spinifex/daemon/daemon_handlers_storage.go +++ b/spinifex/daemon/daemon_handlers_storage.go @@ -4,7 +4,7 @@ import ( "log/slog" "os" - "github.com/mulgadc/spinifex/spinifex/types" + operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1" "github.com/nats-io/nats.go" toml "github.com/pelletier/go-toml/v2" ) @@ -70,8 +70,8 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) string { return outcomeError } - resp := types.StorageConfigResponse{ - Encoding: types.StorageEncoding{ + resp := operatorv1.StorageConfigResponse{ + Encoding: operatorv1.StorageEncoding{ DataShards: cfg.RS.Data, ParityShards: cfg.RS.Parity, }, @@ -83,9 +83,9 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) string { for _, n := range h.Nodes { switch n.Role { case predastoreRoleMeta: - resp.MetaNodes = append(resp.MetaNodes, types.StorageMetaNode{ID: n.ID, Host: h.Addr, Port: n.Port}) + resp.MetaNodes = append(resp.MetaNodes, operatorv1.StorageMetaNode{ID: n.ID, Host: h.Addr, Port: n.Port}) case predastoreRoleBlob: - resp.BlobNodes = append(resp.BlobNodes, types.StorageBlobNode{ID: n.ID, Host: h.Addr, Port: n.Port}) + resp.BlobNodes = append(resp.BlobNodes, operatorv1.StorageBlobNode{ID: n.ID, Host: h.Addr, Port: n.Port}) case predastoreRoleGate: // A gate serves the S3 API and holds no cluster state, so it // belongs to neither table this report carries. @@ -95,20 +95,20 @@ func (d *Daemon) handleStorageConfig(msg *nats.Msg) string { } } if resp.MetaNodes == nil { - resp.MetaNodes = []types.StorageMetaNode{} + resp.MetaNodes = []operatorv1.StorageMetaNode{} } if resp.BlobNodes == nil { - resp.BlobNodes = []types.StorageBlobNode{} + resp.BlobNodes = []operatorv1.StorageBlobNode{} } for _, b := range cfg.Buckets { - resp.Buckets = append(resp.Buckets, types.StorageBucket{ + resp.Buckets = append(resp.Buckets, operatorv1.StorageBucket{ Name: b.Name, Region: b.Region, }) } if resp.Buckets == nil { - resp.Buckets = []types.StorageBucket{} + resp.Buckets = []operatorv1.StorageBucket{} } respondWithJSON(d.node, msg, resp) diff --git a/spinifex/daemon/daemon_handlers_storage_test.go b/spinifex/daemon/daemon_handlers_storage_test.go index 79379a2ff..e408e199c 100644 --- a/spinifex/daemon/daemon_handlers_storage_test.go +++ b/spinifex/daemon/daemon_handlers_storage_test.go @@ -7,7 +7,7 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/spinifex/types" + operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1" "github.com/nats-io/nats.go" toml "github.com/pelletier/go-toml/v2" "github.com/stretchr/testify/assert" @@ -100,18 +100,18 @@ func storageConfigResponse(t *testing.T, cfg string, reply string) []byte { func TestHandleStorageConfig_SplitsNodesByRole(t *testing.T) { payload := storageConfigResponse(t, clusterTOML, "test.storage.config.ok") - var resp types.StorageConfigResponse + var resp operatorv1.StorageConfigResponse require.NoError(t, json.Unmarshal(payload, &resp)) assert.Equal(t, 2, resp.Encoding.DataShards) assert.Equal(t, 1, resp.Encoding.ParityShards) require.Len(t, resp.MetaNodes, 1) - assert.Equal(t, types.StorageMetaNode{ID: 3, Host: "10.0.0.1", Port: 7660}, resp.MetaNodes[0]) + assert.Equal(t, operatorv1.StorageMetaNode{ID: 3, Host: "10.0.0.1", Port: 7660}, resp.MetaNodes[0]) require.Len(t, resp.BlobNodes, 2) - assert.Equal(t, types.StorageBlobNode{ID: 2, Host: "10.0.0.1", Port: 6660}, resp.BlobNodes[0]) - assert.Equal(t, types.StorageBlobNode{ID: 4, Host: "10.0.0.2", Port: 6660}, resp.BlobNodes[1]) + assert.Equal(t, operatorv1.StorageBlobNode{ID: 2, Host: "10.0.0.1", Port: 6660}, resp.BlobNodes[0]) + assert.Equal(t, operatorv1.StorageBlobNode{ID: 4, Host: "10.0.0.2", Port: 6660}, resp.BlobNodes[1]) require.Len(t, resp.Buckets, 1) assert.Equal(t, "predastore", resp.Buckets[0].Name) diff --git a/spinifex/gateway/spx/storage.go b/spinifex/gateway/spx/storage.go index da5661e74..44a15de07 100644 --- a/spinifex/gateway/spx/storage.go +++ b/spinifex/gateway/spx/storage.go @@ -10,7 +10,7 @@ import ( "time" pds "github.com/mulgadc/predastore" - "github.com/mulgadc/spinifex/spinifex/types" + operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1" "github.com/nats-io/nats.go" ) @@ -31,10 +31,10 @@ type MetaNodeStatus struct { // StorageStatusOutput is the response for GetStorageStatus. type StorageStatusOutput struct { - Encoding StorageEncodingOutput `json:"encoding"` - MetaNodes []MetaNodeStatus `json:"meta_nodes"` - BlobNodes []types.StorageBlobNode `json:"blob_nodes"` - Buckets []types.StorageBucket `json:"buckets"` + Encoding StorageEncodingOutput `json:"encoding"` + MetaNodes []MetaNodeStatus `json:"meta_nodes"` + BlobNodes []operatorv1.StorageBlobNode `json:"blob_nodes"` + Buckets []operatorv1.StorageBucket `json:"buckets"` } // StorageEncodingOutput adds the type label to the encoding config. @@ -55,12 +55,12 @@ const metaNodeQueryTimeout = 2 * time.Second // TLS config only the node owning it could read) is what makes every dial // trusted. func GetStorageStatus(nc *nats.Conn, rootCAs *x509.CertPool) (*StorageStatusOutput, error) { - msg, err := nc.Request("spinifex.storage.config", []byte("{}"), 3*time.Second) + msg, err := nc.Request(operatorv1.StorageConfigSubject, []byte("{}"), 3*time.Second) if err != nil { return nil, fmt.Errorf("storage config request: %w", err) } - var cfg types.StorageConfigResponse + var cfg operatorv1.StorageConfigResponse if err := json.Unmarshal(msg.Data, &cfg); err != nil { return nil, fmt.Errorf("parse storage config: %w", err) } diff --git a/spinifex/gateway/spx/storage_test.go b/spinifex/gateway/spx/storage_test.go index a7170f474..f64c65573 100644 --- a/spinifex/gateway/spx/storage_test.go +++ b/spinifex/gateway/spx/storage_test.go @@ -7,7 +7,7 @@ import ( "time" pds "github.com/mulgadc/predastore" - "github.com/mulgadc/spinifex/spinifex/types" + operatorv1 "github.com/mulgadc/spinifex/contracts/operator/v1" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -17,17 +17,17 @@ func TestGetStorageStatus_Success(t *testing.T) { _, nc := startEmbeddedNATS(t) // Mock daemon returning storage config - sub, err := nc.Subscribe("spinifex.storage.config", func(msg *nats.Msg) { - resp := types.StorageConfigResponse{ - Encoding: types.StorageEncoding{DataShards: 2, ParityShards: 1}, - MetaNodes: []types.StorageMetaNode{ + sub, err := nc.Subscribe(operatorv1.StorageConfigSubject, func(msg *nats.Msg) { + resp := operatorv1.StorageConfigResponse{ + Encoding: operatorv1.StorageEncoding{DataShards: 2, ParityShards: 1}, + MetaNodes: []operatorv1.StorageMetaNode{ {ID: 1, Host: "127.0.0.1", Port: 1}, // nothing listens on port 1 }, - BlobNodes: []types.StorageBlobNode{ + BlobNodes: []operatorv1.StorageBlobNode{ {ID: 1, Host: "0.0.0.0", Port: 9991}, {ID: 2, Host: "0.0.0.0", Port: 9992}, }, - Buckets: []types.StorageBucket{ + Buckets: []operatorv1.StorageBucket{ {Name: "predastore", Region: "ap-southeast-2"}, }, } diff --git a/spinifex/types/storage.go b/spinifex/types/storage.go deleted file mode 100644 index a3bf8b0ad..000000000 --- a/spinifex/types/storage.go +++ /dev/null @@ -1,38 +0,0 @@ -package types - -// StorageConfigResponse is returned by the spinifex.storage.config NATS topic. -// Contains predastore topology with credentials stripped. -type StorageConfigResponse struct { - Encoding StorageEncoding `json:"encoding"` - MetaNodes []StorageMetaNode `json:"meta_nodes"` - BlobNodes []StorageBlobNode `json:"blob_nodes"` - Buckets []StorageBucket `json:"buckets"` -} - -// StorageEncoding describes the Reed-Solomon erasure coding configuration. -type StorageEncoding struct { - DataShards int `json:"data_shards"` - ParityShards int `json:"parity_shards"` -} - -// StorageMetaNode describes a predastore meta node, one member of the Raft -// quorum over global state (no credentials). -type StorageMetaNode struct { - ID int `json:"id"` - Host string `json:"host"` - Port int `json:"port"` -} - -// StorageBlobNode describes a predastore blob node, which holds erasure-coded -// object shards. -type StorageBlobNode struct { - ID int `json:"id"` - Host string `json:"host"` - Port int `json:"port"` -} - -// StorageBucket describes a configured S3 bucket. -type StorageBucket struct { - Name string `json:"name"` - Region string `json:"region"` -}