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 operator storage contract

Browse Source
This commit is contained in:
joshsiv-mulga
2026-09-30 15:56:13 +10:00
parent 53a6147178
commit 5d88a33998
11 changed files with 124 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/operator/v1/README.md
+10
View File
@@ -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.
contracts/operator/v1/storage.go
+50
View File
@@ -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"`
}
contracts/operator/v1/storage_test.go
+28
View File
@@ -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)
}
docs/DESIGN.md
+5 -1
View File
@@ -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
docs/service-interfaces.yaml
+1
View File
@@ -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"
spinifex/daemon/daemon.go
+2 -1
View File
@@ -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"},
spinifex/daemon/daemon_handlers_storage.go
+9 -9
View File
@@ -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)
spinifex/daemon/daemon_handlers_storage_test.go
+5 -5
View File
@@ -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)
spinifex/gateway/spx/storage.go
+7 -7
View File
@@ -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)
}
spinifex/gateway/spx/storage_test.go
+7 -7
View File
@@ -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"},
},
}
spinifex/types/storage.go
-38
View File
@@ -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"`
}