diff --git a/cmd/spinifex/cmd/admin_gpu.go b/cmd/spinifex/cmd/admin_gpu.go index 37094f334..ada75ad2c 100644 --- a/cmd/spinifex/cmd/admin_gpu.go +++ b/cmd/spinifex/cmd/admin_gpu.go @@ -10,10 +10,10 @@ import ( "strings" "time" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/spf13/cobra" "github.com/spf13/viper" ) @@ -111,7 +111,7 @@ func init() { } // gpuNodeStatus queries NATS and returns the NodeStatusResponse for the target node. -func gpuNodeStatus(targetNode string) (*types.NodeStatusResponse, error) { +func gpuNodeStatus(targetNode string) (*clusterv1.NodeStatusResponse, error) { cfg, nc, err := loadConfigAndConnect() if err != nil { return nil, err @@ -122,12 +122,12 @@ func gpuNodeStatus(targetNode string) (*types.NodeStatusResponse, error) { targetNode = cfg.Node } - responses, err := collectResponses(nc, "spinifex.node.status", 3*time.Second) + responses, err := collectResponses(nc, clusterv1.NodeStatusSubject, 3*time.Second) if err != nil { return nil, fmt.Errorf("collect node status: %w", err) } for _, raw := range responses { - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse if err := json.Unmarshal(raw, &resp); err != nil { continue } diff --git a/cmd/spinifex/cmd/get.go b/cmd/spinifex/cmd/get.go index a6c951491..58b2fa473 100644 --- a/cmd/spinifex/cmd/get.go +++ b/cmd/spinifex/cmd/get.go @@ -15,7 +15,6 @@ import ( clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/config" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/pterm/pterm" @@ -134,7 +133,7 @@ func formatUptime(seconds int64) string { return fmt.Sprintf("%dm", mins) } -func formatRoles(resp types.NodeStatusResponse) string { +func formatRoles(resp clusterv1.NodeStatusResponse) string { var roles []string if resp.NATSRole != "" { roles = append(roles, "nats:"+resp.NATSRole) @@ -170,16 +169,16 @@ func runGetNodes(cmd *cobra.Command, args []string) { defer nc.Close() timeout, _ := cmd.Flags().GetDuration("timeout") - responses, err := collectResponses(nc, "spinifex.node.status", timeout) + responses, err := collectResponses(nc, clusterv1.NodeStatusSubject, timeout) if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) } // Parse responses into a map by node name - respondedNodes := make(map[string]types.NodeStatusResponse) + respondedNodes := make(map[string]clusterv1.NodeStatusResponse) for _, data := range responses { - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse if err := json.Unmarshal(data, &resp); err != nil { continue } diff --git a/cmd/spinifex/cmd/get_test.go b/cmd/spinifex/cmd/get_test.go index c360ed115..fe2ff6e3b 100644 --- a/cmd/spinifex/cmd/get_test.go +++ b/cmd/spinifex/cmd/get_test.go @@ -3,7 +3,7 @@ package cmd import ( "testing" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/stretchr/testify/assert" ) diff --git a/cmd/spinifex/cmd/top.go b/cmd/spinifex/cmd/top.go index 3bacb6ec5..c77bda198 100644 --- a/cmd/spinifex/cmd/top.go +++ b/cmd/spinifex/cmd/top.go @@ -9,7 +9,7 @@ import ( "strconv" "time" - "github.com/mulgadc/spinifex/spinifex/types" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/pterm/pterm" "github.com/spf13/cobra" ) @@ -43,15 +43,15 @@ func runTopNodes(cmd *cobra.Command, args []string) { defer nc.Close() timeout, _ := cmd.Flags().GetDuration("timeout") - responses, err := collectResponses(nc, "spinifex.node.status", timeout) + responses, err := collectResponses(nc, clusterv1.NodeStatusSubject, timeout) if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) } - respondedNodes := make(map[string]types.NodeStatusResponse) + respondedNodes := make(map[string]clusterv1.NodeStatusResponse) for _, data := range responses { - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse if err := json.Unmarshal(data, &resp); err != nil { continue } diff --git a/contracts/cluster/v1/README.md b/contracts/cluster/v1/README.md index c885b78a9..9454646a1 100644 --- a/contracts/cluster/v1/README.md +++ b/contracts/cluster/v1/README.md @@ -20,3 +20,10 @@ by EKS host lookup. It is distinct from node status and capacity reporting: do not add scheduling or physical-GPU inventory merely because those callers also inspect nodes. Do not change its subject or JSON shape incompatibly; add a new version when a wire change is required. + +`NodeStatusSubject` retains the deployed `spinifex.node.status` fan-out route. +`NodeStatusResponse` carries daemon-reported status, schedulable capacity and +physical-GPU inventory for the SPX operator CLI and EC2/EKS placement callers. +It is an observation contract, not the owner of capacity, scheduling or GPU +allocation policy. Do not change its subject or JSON shape incompatibly; add +a new version when a wire change is required. diff --git a/spinifex/types/node.go b/contracts/cluster/v1/status.go similarity index 86% rename from spinifex/types/node.go rename to contracts/cluster/v1/status.go index 346d3bcea..4f9a9d589 100644 --- a/spinifex/types/node.go +++ b/contracts/cluster/v1/status.go @@ -1,4 +1,10 @@ -package types +package clusterv1 + +const ( + // NodeStatusSubject fans out a request for each responding daemon's status + // and schedulable-capacity snapshot. Its reply payload is NodeStatusResponse. + NodeStatusSubject = "spinifex.node.status" +) // GPUSliceInfo describes a single MIG slice within a physical GPU. type GPUSliceInfo struct { @@ -22,8 +28,9 @@ type GPUInfo struct { Slices []GPUSliceInfo `json:"slices,omitempty"` } -// NodeStatusResponse is returned by the spinifex.node.status NATS topic (fan-out). -// Schedulable capacity = TotalVCPU - ReservedVCPU - AllocVCPU (same for memory). +// NodeStatusResponse is one daemon's reply on NodeStatusSubject. Schedulable +// capacity = TotalVCPU - ReservedVCPU - AllocVCPU (and the same for memory). +// The subject fans out, so callers collect one response per responding node. type NodeStatusResponse struct { Node string `json:"node"` Status string `json:"status"` diff --git a/contracts/cluster/v1/status_test.go b/contracts/cluster/v1/status_test.go new file mode 100644 index 000000000..ba779a0ec --- /dev/null +++ b/contracts/cluster/v1/status_test.go @@ -0,0 +1,81 @@ +package clusterv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNodeStatusResponseJSONContract(t *testing.T) { + data, err := json.Marshal(NodeStatusResponse{ + Node: "node-a", + Status: "Ready", + Host: "10.0.0.1", + Region: "ap-southeast-2", + AZ: "ap-southeast-2a", + Uptime: 42, + Services: []string{"nats", "spinifex"}, + TotalVCPU: 16, + TotalMemGB: 64, + ReservedVCPU: 2, + ReservedMemGB: 4, + AllocVCPU: 4, + AllocMemGB: 8, + TotalGPUs: 1, + AllocGPUs: 1, + GPUCapable: true, + GPUPassthrough: true, + GPUModels: []string{"NVIDIA A10"}, + GPUs: []GPUInfo{{ + PCIAddress: "0000:01:00.0", + Model: "NVIDIA A10", + VRAMMiB: 23028, + MIGEnabled: true, + MIGProfile: "1g.6gb", + InstanceID: "i-abc123", + Slices: []GPUSliceInfo{{ + GIID: 1, + Profile: "1g.6gb", + VRAMMiB: 6144, + MdevPath: "/sys/bus/mdev/devices/example", + InstanceID: "i-abc123", + }}, + }}, + VMCount: 1, + InstanceTypes: []InstanceTypeCap{{ + Name: "g5.xlarge", VCPU: 4, MemoryGB: 16, Available: 2, + }}, + NATSRole: "leader", + PredastoreRole: "follower", + OVNNBRole: "leader", + OVNSBRole: "follower", + }) + require.NoError(t, err) + require.JSONEq(t, `{ + "node":"node-a", "status":"Ready", "host":"10.0.0.1", + "region":"ap-southeast-2", "az":"ap-southeast-2a", "uptime":42, + "services":["nats","spinifex"], + "total_vcpu":16, "total_mem_gb":64, + "reserved_vcpu":2, "reserved_mem_gb":4, + "alloc_vcpu":4, "alloc_mem_gb":8, + "total_gpus":1, "alloc_gpus":1, + "gpu_capable":true, "gpu_passthrough":true, "gpu_models":["NVIDIA A10"], + "gpus":[{ + "pci_address":"0000:01:00.0", "model":"NVIDIA A10", "vram_mib":23028, + "mig_enabled":true, "mig_profile":"1g.6gb", "instance_id":"i-abc123", + "slices":[{ + "gi_id":1, "profile":"1g.6gb", "vram_mib":6144, + "mdev_path":"/sys/bus/mdev/devices/example", "instance_id":"i-abc123" + }] + }], + "vm_count":1, + "instance_types":[{"name":"g5.xlarge", "vcpu":4, "memory_gb":16, "available":2}], + "nats_role":"leader", "predastore_role":"follower", + "ovn_nb_role":"leader", "ovn_sb_role":"follower" + }`, string(data)) +} + +func TestNodeStatusSubject(t *testing.T) { + require.Equal(t, "spinifex.node.status", NodeStatusSubject) +} diff --git a/docs/PACKAGE_BOUNDARY_MIGRATION.md b/docs/PACKAGE_BOUNDARY_MIGRATION.md index b806a7367..7e5239330 100644 --- a/docs/PACKAGE_BOUNDARY_MIGRATION.md +++ b/docs/PACKAGE_BOUNDARY_MIGRATION.md @@ -187,6 +187,7 @@ while the branch is in flight; they do not describe the umbrella repository's | `6d6275e78` | `spinifex/types/cluster.go` (`NodeHealthResponse`) | `contracts/cluster/v1/health.go` | Node health is a versioned cluster control-plane representation shared by the per-node NATS route and daemon HTTPS `/health`; `health_test.go` pins both wire forms. | | `53a614717` | `spinifex/types/cluster.go` (`SharedClusterData`) | `spinifex/daemon/config_hash.go` | Shared-config hash input is daemon-local implementation data; focused config-hash tests preserve deterministic, shared-only semantics. | | `78fd017e0` | `spinifex/types/telemetry.go` (`GuestTelemetryMeta`) | `contracts/telemetry/v1/guest_meta.go` | QMP collector sidecar JSON is a versioned file contract between VM runtime and telemetry role; `guest_meta_test.go` pins its filename and payload. | +| `4b35a2a33` | `spinifex/types/node.go` (`NodeVMsResponse`, `VMInfo`, `VMGPUInfo`) | `contracts/cluster/v1/node_vms.go` | VM inventory is a versioned cluster fan-out payload used by SPX CLI and EKS host lookup; `node_vms_test.go` pins its subject and JSON shape. | The EC2 contract row is intentionally different from the directory moves: it creates a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS diff --git a/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index 988e7a6e9..d769cf7ce 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -91,6 +91,7 @@ services: # An EIP association reaches whichever node holds the instance, which is # rarely the node that served the request (fan-out, no queue group). - "ec2.eni.public-ip-changed" + # contracts/cluster/v1 owns the node status/capacity payload and subject. - "spinifex.node.status" # contracts/cluster/v1 owns the VM inventory payload and subject. - "spinifex.node.vms" diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 52486d2be..5c65d4813 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -76,7 +76,6 @@ import ( "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" "github.com/mulgadc/spinifex/spinifex/services/viperblockd/vbwire" - "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" @@ -666,7 +665,7 @@ func (rm *ResourceManager) GetSupportedInstanceTypeInfos() []*ec2.InstanceTypeIn // GetResourceStats returns host resource figures, reservation, allocation, and // per-type capacity caps for the node status response. -func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64, reservedVCPU int, reservedMemGB float64, allocVCPU int, allocMemGB float64, caps []types.InstanceTypeCap) { +func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64, reservedVCPU int, reservedMemGB float64, allocVCPU int, allocMemGB float64, caps []clusterv1.InstanceTypeCap) { rm.mu.RLock() defer rm.mu.RUnlock() @@ -1020,7 +1019,7 @@ func (d *Daemon) subscribeAll() error { {"ec2.DisableSerialConsoleAccess", handleNATSRequest(d.node, d.accountService.DisableSerialConsoleAccess), "spinifex-workers"}, {clusterv1.NodeHealthSubject(d.node), d.handleHealthCheck, ""}, {clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""}, - {"spinifex.node.status", d.handleNodeStatus, ""}, + {clusterv1.NodeStatusSubject, d.handleNodeStatus, ""}, {clusterv1.NodeVMsSubject, d.handleNodeVMs, ""}, {operatorv1.StorageConfigSubject, d.handleStorageConfig, ""}, {"spinifex.image.promote", d.handleSpinifexPromoteImage, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_gpu_handlers_test.go b/spinifex/daemon/daemon_gpu_handlers_test.go index 18eb9fc83..4d1e5ac46 100644 --- a/spinifex/daemon/daemon_gpu_handlers_test.go +++ b/spinifex/daemon/daemon_gpu_handlers_test.go @@ -10,14 +10,13 @@ import ( clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) // gpuStatusRequest subscribes and requests handleNodeStatus on a fresh subject // so each test gets an isolated round trip. -func gpuStatusRequest(t *testing.T, daemon *Daemon, subject string) types.NodeStatusResponse { +func gpuStatusRequest(t *testing.T, daemon *Daemon, subject string) clusterv1.NodeStatusResponse { t.Helper() sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleNodeStatus)) require.NoError(t, err) @@ -26,7 +25,7 @@ func gpuStatusRequest(t *testing.T, daemon *Daemon, subject string) types.NodeSt reply, err := daemon.natsConn.Request(subject, nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse require.NoError(t, json.Unmarshal(reply.Data, &resp)) return resp } @@ -69,7 +68,7 @@ func TestHandleNodeStatus_WholeGPU(t *testing.T) { resp := gpuStatusRequest(t, daemon, "spinifex.node.status.wholegpu") require.Len(t, resp.GPUs, 2) - byPCI := make(map[string]types.GPUInfo, 2) + byPCI := make(map[string]clusterv1.GPUInfo, 2) for _, g := range resp.GPUs { byPCI[g.PCIAddress] = g } @@ -123,7 +122,7 @@ func TestHandleNodeStatus_MIG(t *testing.T) { assert.Empty(t, g.InstanceID, "whole-GPU InstanceID is unused in MIG mode") require.Len(t, g.Slices, 2) - byGI := make(map[int]types.GPUSliceInfo, 2) + byGI := make(map[int]clusterv1.GPUSliceInfo, 2) for _, s := range g.Slices { byGI[s.GIID] = s } diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index d901621f5..d1fa8c666 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -22,7 +22,6 @@ import ( "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" "github.com/mulgadc/spinifex/spinifex/runtime/formation" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "go.opentelemetry.io/otel" @@ -408,12 +407,12 @@ func (d *Daemon) handleNodeStatus(msg *nats.Msg) string { gpuModelNames = append(gpuModelNames, dev.Model) } - var gpuInventory []types.GPUInfo + var gpuInventory []clusterv1.GPUInfo if d.gpuManager != nil { gpuInventory = buildGPUInventory(d.gpuManager.Snapshot()) } - resp := types.NodeStatusResponse{ + resp := clusterv1.NodeStatusResponse{ Node: d.node, Status: "Ready", Host: d.daemonIP(), @@ -518,14 +517,14 @@ func fetchNATSRole(url string, client *http.Client) string { // buildGPUInventory converts a pool snapshot into per-physical-GPU GPUInfo // records suitable for the NodeStatusResponse. Entries are ordered by first // appearance of each PCI address in the snapshot. -func buildGPUInventory(snapshot []gpu.PoolEntry) []types.GPUInfo { - byPCI := make(map[string]*types.GPUInfo, len(snapshot)) +func buildGPUInventory(snapshot []gpu.PoolEntry) []clusterv1.GPUInfo { + byPCI := make(map[string]*clusterv1.GPUInfo, len(snapshot)) var order []string for _, e := range snapshot { pci := e.Device.PCIAddress if _, ok := byPCI[pci]; !ok { - byPCI[pci] = &types.GPUInfo{ + byPCI[pci] = &clusterv1.GPUInfo{ PCIAddress: pci, Model: e.Device.Model, VRAMMiB: e.Device.MemoryMiB, @@ -536,7 +535,7 @@ func buildGPUInventory(snapshot []gpu.PoolEntry) []types.GPUInfo { if e.MIGInstance != nil { g.MIGEnabled = true g.MIGProfile = e.MIGInstance.Profile.Name - g.Slices = append(g.Slices, types.GPUSliceInfo{ + g.Slices = append(g.Slices, clusterv1.GPUSliceInfo{ GIID: e.MIGInstance.GIID, Profile: e.MIGInstance.Profile.Name, VRAMMiB: e.MIGInstance.Profile.MemoryMiB, @@ -548,7 +547,7 @@ func buildGPUInventory(snapshot []gpu.PoolEntry) []types.GPUInfo { } } - gpus := make([]types.GPUInfo, 0, len(order)) + gpus := make([]clusterv1.GPUInfo, 0, len(order)) for _, pci := range order { gpus = append(gpus, *byPCI[pci]) } diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index cc38b538b..b321f9522 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -2169,7 +2169,7 @@ func TestHandleNodeStatus(t *testing.T) { reply, err := daemon.natsConn.Request("spinifex.node.status.test", nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse err = json.Unmarshal(reply.Data, &resp) require.NoError(t, err) @@ -2199,7 +2199,7 @@ func TestHandleNodeStatus_NoVMs(t *testing.T) { reply, err := daemon.natsConn.Request("spinifex.node.status.empty", nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeStatusResponse + var resp clusterv1.NodeStatusResponse err = json.Unmarshal(reply.Data, &resp) require.NoError(t, err) diff --git a/spinifex/daemon/gpu_admission_test.go b/spinifex/daemon/gpu_admission_test.go index b728e02f2..e884f7e7c 100644 --- a/spinifex/daemon/gpu_admission_test.go +++ b/spinifex/daemon/gpu_admission_test.go @@ -9,9 +9,9 @@ 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/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -203,7 +203,7 @@ func TestGetResourceStats_GPUTypesAreGatedOnGPUs(t *testing.T) { reservations: make(map[string]*capacityReservation), } - capFor := func() types.InstanceTypeCap { + capFor := func() clusterv1.InstanceTypeCap { t.Helper() _, _, _, _, _, _, caps := rm.GetResourceStats() for _, c := range caps { @@ -212,7 +212,7 @@ func TestGetResourceStats_GPUTypesAreGatedOnGPUs(t *testing.T) { } } t.Fatal("gpu.4x4c missing from the census") - return types.InstanceTypeCap{} + return clusterv1.InstanceTypeCap{} } assert.Equal(t, 1, capFor().Available, "four GPUs back one four-GPU instance") diff --git a/spinifex/daemon/resource.go b/spinifex/daemon/resource.go index 22d3d7ad9..5cbccb9ea 100644 --- a/spinifex/daemon/resource.go +++ b/spinifex/daemon/resource.go @@ -10,7 +10,7 @@ import ( "strings" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" ) // errInsufficientCapacity is returned by allocateForLaunch when MinCount @@ -237,7 +237,7 @@ func nbdkitChargeMiB(mainVols, auxVols, mainMiB, auxMiB int) int64 { // resourceStatsForType computes the InstanceTypeCap for a single type given // remaining resources. Clamps negative capacity to zero without logging. -func resourceStatsForType(remainVCPU int, remainMem float64, it *ec2.InstanceTypeInfo) types.InstanceTypeCap { +func resourceStatsForType(remainVCPU int, remainMem float64, it *ec2.InstanceTypeInfo) clusterv1.InstanceTypeCap { vCPUs := instanceTypeVCPUs(it) memGB := float64(instanceTypeMemoryMiB(it)) / 1024.0 @@ -253,7 +253,7 @@ func resourceStatsForType(remainVCPU int, remainMem float64, it *ec2.InstanceTyp name = *it.InstanceType } - return types.InstanceTypeCap{ + return clusterv1.InstanceTypeCap{ Name: name, VCPU: int(vCPUs), MemoryGB: memGB, diff --git a/spinifex/gateway/ec2/capacityreservation/capacityreservation_test.go b/spinifex/gateway/ec2/capacityreservation/capacityreservation_test.go index 956cd7609..b92d74a7f 100644 --- a/spinifex/gateway/ec2/capacityreservation/capacityreservation_test.go +++ b/spinifex/gateway/ec2/capacityreservation/capacityreservation_test.go @@ -8,9 +8,9 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/internal/testkit" - "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" diff --git a/spinifex/gateway/ec2/capacityreservation/fanout.go b/spinifex/gateway/ec2/capacityreservation/fanout.go index 0eada6d00..fc4f28aa3 100644 --- a/spinifex/gateway/ec2/capacityreservation/fanout.go +++ b/spinifex/gateway/ec2/capacityreservation/fanout.go @@ -6,7 +6,7 @@ import ( "log/slog" "time" - "github.com/mulgadc/spinifex/spinifex/types" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -32,7 +32,7 @@ func collectCensus(ctx context.Context, natsConn *nats.Conn, expectedNodes int, expectedNodes = 1 } - frames, _, err := utils.Gather(ctx, natsConn, "spinifex.node.status", []byte("{}"), + frames, _, err := utils.Gather(ctx, natsConn, clusterv1.NodeStatusSubject, []byte("{}"), utils.GatherOpts{Timeout: censusTimeout, ExpectedNodes: expectedNodes, AccountID: accountID}) if err != nil { return nil, err @@ -41,7 +41,7 @@ func collectCensus(ctx context.Context, natsConn *nats.Conn, expectedNodes int, seen := make(map[string]struct{}) var census []nodeCensus for _, frame := range frames { - var status types.NodeStatusResponse + var status clusterv1.NodeStatusResponse if err := json.Unmarshal(frame.Data, &status); err != nil { slog.DebugContext(ctx, "collectCensus: failed to unmarshal response", "err", err) continue diff --git a/spinifex/gateway/ec2/capacityreservation/fanout_test.go b/spinifex/gateway/ec2/capacityreservation/fanout_test.go index 9a72f4082..911f6abb6 100644 --- a/spinifex/gateway/ec2/capacityreservation/fanout_test.go +++ b/spinifex/gateway/ec2/capacityreservation/fanout_test.go @@ -5,8 +5,8 @@ import ( "encoding/json" "testing" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "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" diff --git a/spinifex/gateway/ec2/instance/placement.go b/spinifex/gateway/ec2/instance/placement.go index 88c342285..7fa70549a 100644 --- a/spinifex/gateway/ec2/instance/placement.go +++ b/spinifex/gateway/ec2/instance/placement.go @@ -13,9 +13,9 @@ 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/spinifex/foundation/aws/errors" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -67,7 +67,7 @@ func distributeInstances(ctx context.Context, input *ec2.RunInstancesInput, nats // Early-exits once expectedNodes reply; on a degraded cluster it waits the full timeout // rather than placing on a partial view. func queryNodeCapacity(ctx context.Context, natsConn *nats.Conn, instanceType string, expectedNodes int, accountID string) ([]nodeAllocation, error) { - frames, _, err := utils.Gather(ctx, natsConn, "spinifex.node.status", []byte("{}"), + frames, _, err := utils.Gather(ctx, natsConn, clusterv1.NodeStatusSubject, []byte("{}"), utils.GatherOpts{Timeout: 3 * time.Second, ExpectedNodes: expectedNodes, AccountID: accountID}) if err != nil { return nil, err @@ -75,7 +75,7 @@ func queryNodeCapacity(ctx context.Context, natsConn *nats.Conn, instanceType st var nodes []nodeAllocation for _, frame := range frames { - var status types.NodeStatusResponse + var status clusterv1.NodeStatusResponse if err := json.Unmarshal(frame.Data, &status); err != nil { slog.DebugContext(ctx, "queryNodeCapacity: failed to unmarshal response", "err", err) continue diff --git a/spinifex/gateway/ec2/instance/placement_spread_test.go b/spinifex/gateway/ec2/instance/placement_spread_test.go index 40dd17ec2..303c88fcd 100644 --- a/spinifex/gateway/ec2/instance/placement_spread_test.go +++ b/spinifex/gateway/ec2/instance/placement_spread_test.go @@ -8,10 +8,10 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/contracts/ec2/v1" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" diff --git a/spinifex/gateway/ec2/instance/placement_test.go b/spinifex/gateway/ec2/instance/placement_test.go index e012fbf83..8935e870f 100644 --- a/spinifex/gateway/ec2/instance/placement_test.go +++ b/spinifex/gateway/ec2/instance/placement_test.go @@ -9,8 +9,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" diff --git a/spinifex/gateway/ec2/spotinstance/quota_gate_order_test.go b/spinifex/gateway/ec2/spotinstance/quota_gate_order_test.go index 33ec8eca7..ea1167478 100644 --- a/spinifex/gateway/ec2/spotinstance/quota_gate_order_test.go +++ b/spinifex/gateway/ec2/spotinstance/quota_gate_order_test.go @@ -10,9 +10,9 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/internal/testkit" - "github.com/mulgadc/spinifex/spinifex/types" + handlers_quota "github.com/mulgadc/spinifex/spinifex/handlers/quota" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "github.com/stretchr/testify/assert" diff --git a/spinifex/gateway/spx/spx.go b/spinifex/gateway/spx/spx.go index 7863d4d92..05a649bea 100644 --- a/spinifex/gateway/spx/spx.go +++ b/spinifex/gateway/spx/spx.go @@ -10,7 +10,6 @@ import ( "time" clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -41,21 +40,21 @@ func GetVersion(version, commit string) (*VersionOutput, error) { // GetNodesOutput is the response for GetNodes. type GetNodesOutput struct { - Nodes []types.NodeStatusResponse `json:"nodes"` - ClusterMode string `json:"cluster_mode"` + Nodes []clusterv1.NodeStatusResponse `json:"nodes"` + ClusterMode string `json:"cluster_mode"` } // GetNodes queries all daemon nodes via NATS fan-out and returns their status. func GetNodes(ctx context.Context, nc *nats.Conn, expectedNodes int) (*GetNodesOutput, error) { - frames, _, err := utils.Gather(ctx, nc, "spinifex.node.status", []byte("{}"), + frames, _, err := utils.Gather(ctx, nc, clusterv1.NodeStatusSubject, []byte("{}"), utils.GatherOpts{Timeout: 3 * time.Second, ExpectedNodes: expectedNodes}) if err != nil { return nil, err } - nodes := make([]types.NodeStatusResponse, 0, len(frames)) + nodes := make([]clusterv1.NodeStatusResponse, 0, len(frames)) for _, frame := range frames { - var node types.NodeStatusResponse + var node clusterv1.NodeStatusResponse if json.Unmarshal(frame.Data, &node) == nil { nodes = append(nodes, node) } diff --git a/spinifex/gateway/spx/spx_test.go b/spinifex/gateway/spx/spx_test.go index 140970979..e67f2e35a 100644 --- a/spinifex/gateway/spx/spx_test.go +++ b/spinifex/gateway/spx/spx_test.go @@ -6,7 +6,6 @@ import ( clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/internal/testkit" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" @@ -54,8 +53,8 @@ func TestGetNodes_NoResponders(t *testing.T) { func TestGetNodes_SingleNode(t *testing.T) { _, nc := startEmbeddedNATS(t) - sub, err := nc.Subscribe("spinifex.node.status", func(msg *nats.Msg) { - resp := types.NodeStatusResponse{ + sub, err := nc.Subscribe(clusterv1.NodeStatusSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeStatusResponse{ Node: "node1", Status: "Ready", Host: "10.0.0.1", @@ -65,7 +64,7 @@ func TestGetNodes_SingleNode(t *testing.T) { TotalMemGB: 16.0, AllocVCPU: 2, AllocMemGB: 2.0, - InstanceTypes: []types.InstanceTypeCap{ + InstanceTypes: []clusterv1.InstanceTypeCap{ {Name: "t3.small", VCPU: 2, MemoryGB: 2.0, Available: 3}, }, } @@ -93,8 +92,8 @@ func TestGetNodes_MultiNode(t *testing.T) { for _, name := range []string{"node1", "node2", "node3"} { nodeName := name - sub, err := nc.Subscribe("spinifex.node.status", func(msg *nats.Msg) { - resp := types.NodeStatusResponse{Node: nodeName, Status: "Ready"} + sub, err := nc.Subscribe(clusterv1.NodeStatusSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeStatusResponse{Node: nodeName, Status: "Ready"} data, _ := json.Marshal(resp) msg.Respond(data) }) diff --git a/spinifex/handlers/eks/k3s_ha_control_plane.go b/spinifex/handlers/eks/k3s_ha_control_plane.go index b9f5660be..03ba09b97 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane.go +++ b/spinifex/handlers/eks/k3s_ha_control_plane.go @@ -16,7 +16,6 @@ import ( "github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes" "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -500,8 +499,8 @@ func (h *natsHostScheduler) SchedulableHosts(ctx context.Context, instanceType s var hosts []azHost seen := make(map[string]bool) - h.fanout(ctx, "spinifex.node.status", func(data []byte) { - var st types.NodeStatusResponse + h.fanout(ctx, clusterv1.NodeStatusSubject, func(data []byte) { + var st clusterv1.NodeStatusResponse if json.Unmarshal(data, &st) != nil || st.Node == "" || seen[st.Node] { return } @@ -554,7 +553,7 @@ func spreadHostsByAZ(hosts []azHost) []string { // nodeFitsCustomerInstance reports whether a node advertises at least one free // slot for the given customer instance type in its node.status capacity. -func nodeFitsCustomerInstance(st types.NodeStatusResponse, instanceType string) bool { +func nodeFitsCustomerInstance(st clusterv1.NodeStatusResponse, instanceType string) bool { for _, c := range st.InstanceTypes { if c.Name == instanceType && c.Available >= 1 { return true @@ -565,7 +564,7 @@ func nodeFitsCustomerInstance(st types.NodeStatusResponse, instanceType string) // nodeFitsSystemInstance reports whether a node's headroom (Total - Reserved - Alloc) // fits at least one VM of the given vCPU/memory footprint. -func nodeFitsSystemInstance(st types.NodeStatusResponse, vcpu int, memGB float64) bool { +func nodeFitsSystemInstance(st clusterv1.NodeStatusResponse, vcpu int, memGB float64) bool { remainVCPU := st.TotalVCPU - st.ReservedVCPU - st.AllocVCPU remainMem := st.TotalMemGB - st.ReservedMemGB - st.AllocMemGB return remainVCPU >= vcpu && remainMem >= memGB diff --git a/spinifex/handlers/eks/k3s_ha_control_plane_test.go b/spinifex/handlers/eks/k3s_ha_control_plane_test.go index 368c772a6..beefbb8cc 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane_test.go +++ b/spinifex/handlers/eks/k3s_ha_control_plane_test.go @@ -13,12 +13,12 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" + "github.com/mulgadc/spinifex/internal/testkit" "github.com/mulgadc/spinifex/spinifex/domains/ec2/instancetypes" + "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" - "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" diff --git a/tests/integration/harness.go b/tests/integration/harness.go index bd7f4f8e5..8abdc2241 100644 --- a/tests/integration/harness.go +++ b/tests/integration/harness.go @@ -32,6 +32,7 @@ import ( "github.com/aws/aws-sdk-go/service/ecr" "github.com/aws/aws-sdk-go/service/iam" "github.com/aws/aws-sdk-go/service/sts" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/gateway" gateway_bedrock "github.com/mulgadc/spinifex/spinifex/gateway/bedrock" gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr" @@ -40,7 +41,6 @@ import ( handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" handlers_sts "github.com/mulgadc/spinifex/spinifex/handlers/sts" "github.com/mulgadc/spinifex/spinifex/objectstore" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/stretchr/testify/require" @@ -202,7 +202,7 @@ func startGateway(t *testing.T, collector *conformanceCollector, opts ...Option) Config: cfg, } - nodeReply, err := json.Marshal(types.NodeDiscoverResponse{Node: "integration-test-node"}) + nodeReply, err := json.Marshal(clusterv1.NodeDiscoverResponse{Node: "integration-test-node"}) require.NoError(t, err) gw.StubSubject(t, nodeDiscoverSubject, nodeReply) diff --git a/tests/integration/launchtemplate_test.go b/tests/integration/launchtemplate_test.go index 5241f68ed..159a9ea48 100644 --- a/tests/integration/launchtemplate_test.go +++ b/tests/integration/launchtemplate_test.go @@ -10,7 +10,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" diff --git a/tests/integration/run_instances_clienttoken_test.go b/tests/integration/run_instances_clienttoken_test.go index 3a57b313d..651418eb2 100644 --- a/tests/integration/run_instances_clienttoken_test.go +++ b/tests/integration/run_instances_clienttoken_test.go @@ -9,7 +9,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/nats-io/nats.go" "github.com/stretchr/testify/require" ) diff --git a/tests/integration/run_instances_launchtemplate_test.go b/tests/integration/run_instances_launchtemplate_test.go index 799379e6d..7e001cfc0 100644 --- a/tests/integration/run_instances_launchtemplate_test.go +++ b/tests/integration/run_instances_launchtemplate_test.go @@ -10,7 +10,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/nats-io/nats.go" "github.com/stretchr/testify/require" ) diff --git a/tests/integration/run_instances_passrole_test.go b/tests/integration/run_instances_passrole_test.go index 71edc544d..76a65497f 100644 --- a/tests/integration/run_instances_passrole_test.go +++ b/tests/integration/run_instances_passrole_test.go @@ -10,7 +10,7 @@ import ( "github.com/aws/aws-sdk-go/service/ec2" "github.com/aws/aws-sdk-go/service/iam" "github.com/aws/aws-sdk-go/service/sts" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/stretchr/testify/require" ) diff --git a/tests/integration/run_instances_placementgroup_test.go b/tests/integration/run_instances_placementgroup_test.go index 417fddc9a..5c3ae85c0 100644 --- a/tests/integration/run_instances_placementgroup_test.go +++ b/tests/integration/run_instances_placementgroup_test.go @@ -7,8 +7,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/gateway" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" "github.com/stretchr/testify/require" ) diff --git a/tests/integration/run_instances_test.go b/tests/integration/run_instances_test.go index 915bed9e5..6c2b96064 100644 --- a/tests/integration/run_instances_test.go +++ b/tests/integration/run_instances_test.go @@ -9,7 +9,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/nats-io/nats.go" "github.com/stretchr/testify/require" ) diff --git a/tests/integration/spot_instance_test.go b/tests/integration/spot_instance_test.go index 2f071fb14..e7351a98d 100644 --- a/tests/integration/spot_instance_test.go +++ b/tests/integration/spot_instance_test.go @@ -13,8 +13,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + types "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/contracts/ec2/v1" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require"