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

refactor: establish cluster discovery contract

Browse Source
This commit is contained in:
joshsiv-mulga
2026-09-30 15:35:33 +10:00
parent c239fd8973
commit b49e55a586
12 changed files with 71 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/cluster/v1/README.md
+10
View File
@@ -0,0 +1,10 @@
# Cluster v1 contracts
This package owns the versioned NATS contracts for cluster-membership control
plane traffic. It is not an AWS tenant-facing API: it lets platform consumers
identify the daemon nodes currently responding to a cluster fan-out.
`NodesDiscoverSubject` retains the deployed `spinifex.nodes.discover` route.
`NodeDiscoverResponse.Node` is the node identifier used for responder-set
coverage. Do not change either incompatibly in this package; add a new version
when a wire change is required.
contracts/cluster/v1/discovery.go
+16
View File
@@ -0,0 +1,16 @@
// Package clusterv1 defines version 1 of the cross-process cluster-membership
// contract. These messages establish which daemon nodes are responding to the
// control plane; their subject and JSON fields are wire compatibility.
package clusterv1
const (
// NodesDiscoverSubject fans out a membership discovery request to every
// active daemon. Its reply payload is NodeDiscoverResponse.
NodesDiscoverSubject = "spinifex.nodes.discover"
)
// NodeDiscoverResponse identifies one daemon that responded to a membership
// discovery request.
type NodeDiscoverResponse struct {
Node string `json:"node"`
}
contracts/cluster/v1/discovery_test.go
+18
View File
@@ -0,0 +1,18 @@
package clusterv1
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
func TestNodeDiscoverResponseJSONContract(t *testing.T) {
data, err := json.Marshal(NodeDiscoverResponse{Node: "node-a"})
require.NoError(t, err)
require.JSONEq(t, `{"node":"node-a"}`, string(data))
}
func TestNodesDiscoverSubject(t *testing.T) {
require.Equal(t, "spinifex.nodes.discover", NodesDiscoverSubject)
}
docs/PACKAGE_BOUNDARY_MIGRATION.md
+2 -1
View File
@@ -183,8 +183,9 @@ while the branch is in flight; they do not describe the umbrella repository's
| `248f0e005`, `8c4af3030` | `spinifex/formation` | `spinifex/runtime/formation` | Formation owns its protocol payloads and remains runtime membership/bootstrap work. |
| `b8b97279c`, `c0c64cd53` | `spinifex/instancecache` | `spinifex/runtime/compute/cache` | Cache policy is decoupled from callers before becoming node-local compute runtime. |
| `aa0c36ff4` | `spinifex/types/ec2.go` | `contracts/ec2/v1` | The EC2 instance-command subject and JSON payload are an explicit versioned cross-process contract; `commands_test.go` is its compatibility evidence. |
| `c239fd897` | `spinifex/types/eni.go` | `spinifex/runtime/compute/vm/eni_requests.go` | PCIe hot-plug slot allocation is mutex-bearing, per-VM runtime state rather than a shared type or wire contract; `record_test.go` preserves its restart semantics. |
The final row is intentionally different from the directory moves: it creates
The EC2 contract row is intentionally different from the directory moves: it creates
a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS
subscription and permission shape, while callers use
`InstanceCommandSubject` rather than reconstructing the subject. The
spinifex/daemon/daemon.go
+2 -1
View File
@@ -31,6 +31,7 @@ import (
"github.com/go-chi/chi/v5"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/mulgadc/bluebottle/pkg/tlsconfig"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/admin"
"github.com/mulgadc/spinifex/spinifex/bootstrap/preflight"
@@ -1017,7 +1018,7 @@ func (d *Daemon) subscribeAll() error {
{"ec2.EnableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.EnableSerialConsoleAccess), "spinifex-workers"},
{"ec2.DisableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.DisableSerialConsoleAccess), "spinifex-workers"},
{fmt.Sprintf("spinifex.admin.%s.health", d.node), d.handleHealthCheck, ""},
{"spinifex.nodes.discover", d.handleNodeDiscover, ""},
{clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""},
{"spinifex.node.status", d.handleNodeStatus, ""},
{"spinifex.node.vms", d.handleNodeVMs, ""},
{"spinifex.storage.config", d.handleStorageConfig, ""},
spinifex/daemon/daemon_handlers.go
+2 -1
View File
@@ -14,6 +14,7 @@ import (
"sync"
"time"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
@@ -361,7 +362,7 @@ func (d *Daemon) handleHealthCheck(msg *nats.Msg) string {
// handleNodeDiscover responds to node discovery requests with this node's ID
// Used by the gateway to dynamically discover active spinifex nodes in the cluster.
func (d *Daemon) handleNodeDiscover(msg *nats.Msg) string {
response := types.NodeDiscoverResponse{
response := clusterv1.NodeDiscoverResponse{
Node: d.node,
}
spinifex/daemon/daemon_handlers_test.go
+4 -3
View File
@@ -17,6 +17,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
awss3 "github.com/aws/aws-sdk-go/service/s3"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/ebsfake"
@@ -344,15 +345,15 @@ func TestHandleNodeDiscover(t *testing.T) {
daemon := createTestDaemon(t, natsURL)
sub, err := daemon.natsConn.Subscribe("spinifex.nodes.discover", asMsgHandler(daemon.handleNodeDiscover))
sub, err := daemon.natsConn.Subscribe(clusterv1.NodesDiscoverSubject, asMsgHandler(daemon.handleNodeDiscover))
require.NoError(t, err)
defer sub.Unsubscribe()
reply, err := daemon.natsConn.Request("spinifex.nodes.discover", nil, 5*time.Second)
reply, err := daemon.natsConn.Request(clusterv1.NodesDiscoverSubject, nil, 5*time.Second)
require.NoError(t, err)
require.NotNil(t, reply)
var resp types.NodeDiscoverResponse
var resp clusterv1.NodeDiscoverResponse
err = json.Unmarshal(reply.Data, &resp)
require.NoError(t, err)
spinifex/daemon/daemon_test.go
+2 -1
View File
@@ -28,6 +28,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
@@ -4623,7 +4624,7 @@ func TestNodeIDNamespace_Agrees(t *testing.T) {
assert.Equal(t, daemon.node, reply.Header.Get(utils.NodeIDHeader),
"the reply header must carry the same node ID as everything else")
var resp types.NodeDiscoverResponse
var resp clusterv1.NodeDiscoverResponse
require.NoError(t, json.Unmarshal(reply.Data, &resp))
assert.Equal(t, daemon.node, resp.Node)
}
spinifex/gateway/discover_nodes_test.go
+4 -4
View File
@@ -7,9 +7,9 @@ import (
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/gateway"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/gateway"
"github.com/nats-io/nats.go"
)
@@ -21,11 +21,11 @@ func respondAsNodes(t *testing.T, nc *nats.Conn, n int) *atomic.Int64 {
var rounds atomic.Int64
for i := range n {
node := string(rune('a' + i))
sub, err := nc.Subscribe("spinifex.nodes.discover", func(msg *nats.Msg) {
sub, err := nc.Subscribe(clusterv1.NodesDiscoverSubject, func(msg *nats.Msg) {
if node == "a" {
rounds.Add(1)
}
payload, err := json.Marshal(types.NodeDiscoverResponse{Node: node})
payload, err := json.Marshal(clusterv1.NodeDiscoverResponse{Node: node})
if err != nil {
return
}
spinifex/gateway/gateway.go
+4 -4
View File
@@ -24,8 +24,10 @@ import (
bbotel "github.com/mulgadc/bluebottle/pkg/otelsetup"
"github.com/mulgadc/bluebottle/pkg/ratelimit"
"github.com/mulgadc/bluebottle/pkg/sigv4"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/spinifex/accountteardown"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock"
gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr"
@@ -36,8 +38,6 @@ import (
handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota"
handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
@@ -1039,7 +1039,7 @@ func (gw *GatewayConfig) DiscoverActiveNodes(ctx context.Context) int {
return count
}
frames, _, err := utils.Gather(ctx, gw.NATSConn, "spinifex.nodes.discover", []byte("{}"),
frames, _, err := utils.Gather(ctx, gw.NATSConn, clusterv1.NodesDiscoverSubject, []byte("{}"),
utils.GatherOpts{Timeout: 500 * time.Millisecond})
if err != nil {
slog.ErrorContext(ctx, "DiscoverActiveNodes: fan-out failed, using ExpectedNodes fallback", "err", err, "fallback", gw.ExpectedNodes)
@@ -1048,7 +1048,7 @@ func (gw *GatewayConfig) DiscoverActiveNodes(ctx context.Context) int {
nodesSeen := make(map[string]bool)
for _, frame := range frames {
var response types.NodeDiscoverResponse
var response clusterv1.NodeDiscoverResponse
if err := json.Unmarshal(frame.Data, &response); err != nil {
slog.DebugContext(ctx, "DiscoverActiveNodes: Failed to unmarshal response", "err", err)
continue
spinifex/gateway/gateway_test.go
+7 -7
View File
@@ -18,10 +18,10 @@ import (
awssession "github.com/aws/aws-sdk-go/aws/session"
awsec2 "github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/bluebottle/pkg/ratelimit"
clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -509,8 +509,8 @@ func TestDiscoverActiveNodes_WithResponders(t *testing.T) {
for _, nodeName := range []string{"node-1", "node-2"} {
name := nodeName
_, err := nc.Subscribe("spinifex.nodes.discover", func(msg *nats.Msg) {
resp := types.NodeDiscoverResponse{Node: name}
_, err := nc.Subscribe(clusterv1.NodesDiscoverSubject, func(msg *nats.Msg) {
resp := clusterv1.NodeDiscoverResponse{Node: name}
data, _ := json.Marshal(resp)
msg.Respond(data)
})
@@ -530,7 +530,7 @@ func TestDiscoverActiveNodes_WithResponders(t *testing.T) {
func TestDiscoverActiveNodes_InvalidJSON(t *testing.T) {
nc := startTestNATS(t)
_, err := nc.Subscribe("spinifex.nodes.discover", func(msg *nats.Msg) {
_, err := nc.Subscribe(clusterv1.NodesDiscoverSubject, func(msg *nats.Msg) {
msg.Respond([]byte("not json"))
})
require.NoError(t, err)
@@ -549,8 +549,8 @@ func TestDiscoverActiveNodes_DuplicateNodes(t *testing.T) {
nc := startTestNATS(t)
for range 2 {
_, err := nc.Subscribe("spinifex.nodes.discover", func(msg *nats.Msg) {
resp := types.NodeDiscoverResponse{Node: "same-node"}
_, err := nc.Subscribe(clusterv1.NodesDiscoverSubject, func(msg *nats.Msg) {
resp := clusterv1.NodeDiscoverResponse{Node: "same-node"}
data, _ := json.Marshal(resp)
msg.Respond(data)
})
spinifex/types/node.go
-5
View File
@@ -1,10 +1,5 @@
package types
// NodeDiscoverResponse is the response for node discovery requests.
type NodeDiscoverResponse struct {
Node string `json:"node"`
}
// GPUSliceInfo describes a single MIG slice within a physical GPU.
type GPUSliceInfo struct {
GIID int `json:"gi_id"`