diff --git a/contracts/cluster/v1/README.md b/contracts/cluster/v1/README.md new file mode 100644 index 000000000..5d21bd654 --- /dev/null +++ b/contracts/cluster/v1/README.md @@ -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. diff --git a/contracts/cluster/v1/discovery.go b/contracts/cluster/v1/discovery.go new file mode 100644 index 000000000..32696ed27 --- /dev/null +++ b/contracts/cluster/v1/discovery.go @@ -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"` +} diff --git a/contracts/cluster/v1/discovery_test.go b/contracts/cluster/v1/discovery_test.go new file mode 100644 index 000000000..29e5adf37 --- /dev/null +++ b/contracts/cluster/v1/discovery_test.go @@ -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) +} diff --git a/docs/PACKAGE_BOUNDARY_MIGRATION.md b/docs/PACKAGE_BOUNDARY_MIGRATION.md index e3c78d682..f1d4f3eac 100644 --- a/docs/PACKAGE_BOUNDARY_MIGRATION.md +++ b/docs/PACKAGE_BOUNDARY_MIGRATION.md @@ -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 diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index a419dabb8..c3cae5a77 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -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, ""}, diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index d25d0174b..9ce9130d8 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -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, } diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index 07d3884f1..79b7f82a9 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -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) diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index d069cfb7f..07b451a48 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -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) } diff --git a/spinifex/gateway/discover_nodes_test.go b/spinifex/gateway/discover_nodes_test.go index e8eb5237a..cb885e31e 100644 --- a/spinifex/gateway/discover_nodes_test.go +++ b/spinifex/gateway/discover_nodes_test.go @@ -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 } diff --git a/spinifex/gateway/gateway.go b/spinifex/gateway/gateway.go index 14d82a0eb..cb15467ab 100644 --- a/spinifex/gateway/gateway.go +++ b/spinifex/gateway/gateway.go @@ -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 diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index b1e746335..b573c51ac 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -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) }) diff --git a/spinifex/types/node.go b/spinifex/types/node.go index 88e680dc4..4ff58de9f 100644 --- a/spinifex/types/node.go +++ b/spinifex/types/node.go @@ -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"`