diff --git a/cmd/spinifex/cmd/cluster.go b/cmd/spinifex/cmd/cluster.go index c69a83d5e..f841690e4 100644 --- a/cmd/spinifex/cmd/cluster.go +++ b/cmd/spinifex/cmd/cluster.go @@ -10,10 +10,10 @@ import ( "strings" "time" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/daemon" - "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" "github.com/mulgadc/spinifex/spinifex/foundation/telemetry" - "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" "github.com/nats-io/nats.go" "github.com/spf13/cobra" ) @@ -376,7 +376,7 @@ func reportGuestsLeftRunning(severe bool, msg string) { defer nc.Close() node := cfg.Node - responses, err := collectResponses(nc, "spinifex.node.vms", guestEnumerationTimeout) + responses, err := collectResponses(nc, clusterv1.NodeVMsSubject, guestEnumerationTimeout) if err != nil { slog.Warn("could not enumerate local guests", "node", node, "error", err) return @@ -384,7 +384,7 @@ func reportGuestsLeftRunning(severe bool, msg string) { var running []string for _, data := range responses { - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse if err := json.Unmarshal(data, &resp); err != nil { continue } diff --git a/cmd/spinifex/cmd/cluster_test.go b/cmd/spinifex/cmd/cluster_test.go index 878ec08a0..fc04aaa7d 100644 --- a/cmd/spinifex/cmd/cluster_test.go +++ b/cmd/spinifex/cmd/cluster_test.go @@ -8,11 +8,11 @@ import ( "testing" "time" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" + "github.com/mulgadc/spinifex/internal/testkit" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/daemon" "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" - "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" @@ -287,10 +287,10 @@ func TestWarnIfGuestsLeftRunningNamesRunningGuests(t *testing.T) { _, nc, _ := testutil.StartTestJetStream(t) stubConnect(t, &config.ClusterConfig{Node: "node1"}, nc, nil) - sub, err := nc.Subscribe("spinifex.node.vms", func(msg *nats.Msg) { - resp := types.NodeVMsResponse{ + sub, err := nc.Subscribe(clusterv1.NodeVMsSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeVMsResponse{ Node: "node1", - VMs: []types.VMInfo{ + VMs: []clusterv1.VMInfo{ {InstanceID: "i-running", Status: vmStatusRunning}, {InstanceID: "i-stopped", Status: "stopped"}, }, @@ -318,10 +318,10 @@ func TestWarnIfGuestsLeftRunningNoGuestsIsSilent(t *testing.T) { _, nc, _ := testutil.StartTestJetStream(t) stubConnect(t, &config.ClusterConfig{Node: "node1"}, nc, nil) - sub, err := nc.Subscribe("spinifex.node.vms", func(msg *nats.Msg) { - resp := types.NodeVMsResponse{ + sub, err := nc.Subscribe(clusterv1.NodeVMsSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeVMsResponse{ Node: "node1", - VMs: []types.VMInfo{{InstanceID: "i-stopped", Status: "stopped"}}, + VMs: []clusterv1.VMInfo{{InstanceID: "i-stopped", Status: "stopped"}}, } body, _ := json.Marshal(resp) _ = msg.Respond(body) @@ -516,10 +516,10 @@ func TestReportGuestsLeftRunningSevereLogsErrorAndNamesGuests(t *testing.T) { _, nc, _ := testutil.StartTestJetStream(t) stubConnect(t, &config.ClusterConfig{Node: "node1"}, nc, nil) - sub, err := nc.Subscribe("spinifex.node.vms", func(msg *nats.Msg) { - resp := types.NodeVMsResponse{ + sub, err := nc.Subscribe(clusterv1.NodeVMsSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeVMsResponse{ Node: "node1", - VMs: []types.VMInfo{ + VMs: []clusterv1.VMInfo{ {InstanceID: "i-abandoned", Status: vmStatusRunning}, {InstanceID: "i-stopped", Status: "stopped"}, }, diff --git a/cmd/spinifex/cmd/get.go b/cmd/spinifex/cmd/get.go index 94a21e17b..a6c951491 100644 --- a/cmd/spinifex/cmd/get.go +++ b/cmd/spinifex/cmd/get.go @@ -13,6 +13,7 @@ import ( "strings" "time" + 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" @@ -241,14 +242,14 @@ func runGetVMs(cmd *cobra.Command, args []string) { defer nc.Close() timeout, _ := cmd.Flags().GetDuration("timeout") - responses, err := collectResponses(nc, "spinifex.node.vms", timeout) + responses, err := collectResponses(nc, clusterv1.NodeVMsSubject, timeout) if err != nil { fmt.Fprintf(os.Stderr, "Error: %v\n", err) os.Exit(1) } type vmRow struct { - types.VMInfo + clusterv1.VMInfo Node string Host string @@ -256,7 +257,7 @@ func runGetVMs(cmd *cobra.Command, args []string) { var allVMs []vmRow for _, data := range responses { - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse if err := json.Unmarshal(data, &resp); err != nil { continue } diff --git a/contracts/cluster/v1/README.md b/contracts/cluster/v1/README.md index c58c3d52d..c885b78a9 100644 --- a/contracts/cluster/v1/README.md +++ b/contracts/cluster/v1/README.md @@ -13,3 +13,10 @@ when a wire change is required. `NodeHealthResponse` is also the daemon HTTPS `/health` representation. Do not change their subject or JSON shape incompatibly in this package; add a new version when a wire change is required. + +`NodeVMsSubject` retains the deployed `spinifex.node.vms` fan-out route. +`NodeVMsResponse` is the VM inventory reply used by the SPX operator CLI and +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. diff --git a/contracts/cluster/v1/node_vms.go b/contracts/cluster/v1/node_vms.go new file mode 100644 index 000000000..f78f0deb3 --- /dev/null +++ b/contracts/cluster/v1/node_vms.go @@ -0,0 +1,44 @@ +package clusterv1 + +const ( + // NodeVMsSubject fans out a request for each responding daemon's VM + // inventory. Its reply payload is NodeVMsResponse. + NodeVMsSubject = "spinifex.node.vms" +) + +// VMGPUInfo describes one GPU attached to a VM in a NodeVMsResponse. +type VMGPUInfo struct { + Model string `json:"model"` + VRAMMiB int64 `json:"vram_mib"` + PCIAddress string `json:"pci_address,omitempty"` // whole-GPU passthrough only + Profile string `json:"profile,omitempty"` // MIG profile name; empty for whole-GPU + MdevPath string `json:"mdev_path,omitempty"` // MIG only +} + +// VMInfo describes one VM in a node's inventory response. +type VMInfo struct { + InstanceID string `json:"instance_id"` + Status string `json:"status"` + InstanceType string `json:"instance_type"` + VCPU int `json:"vcpu"` + MemoryGB float64 `json:"memory_gb"` + LaunchTime int64 `json:"launch_time"` + // ManagedBy is the Spinifex platform component that owns this VM + // (e.g. "elbv2"). Empty for customer VMs. The UI uses this to filter + // system-managed resources out of customer-facing listings. + ManagedBy string `json:"managed_by,omitempty"` + GPUs []VMGPUInfo `json:"gpus,omitempty"` + // Health is a display label for instance health: "ok", "impaired", + // "recovering", or "-" for non-running VMs. CrashCount is the lifetime + // crash tally within the current restart window. + Health string `json:"health,omitempty"` + CrashCount int `json:"crash_count,omitempty"` +} + +// NodeVMsResponse is one daemon's reply on NodeVMsSubject. The subject fans +// out, so callers collect one response per responding node. +type NodeVMsResponse struct { + Node string `json:"node"` + Host string `json:"host"` + VMs []VMInfo `json:"vms"` +} diff --git a/contracts/cluster/v1/node_vms_test.go b/contracts/cluster/v1/node_vms_test.go new file mode 100644 index 000000000..e1868fe21 --- /dev/null +++ b/contracts/cluster/v1/node_vms_test.go @@ -0,0 +1,60 @@ +package clusterv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestNodeVMsResponseJSONContract(t *testing.T) { + data, err := json.Marshal(NodeVMsResponse{ + Node: "node-a", + Host: "10.0.0.1", + VMs: []VMInfo{{ + InstanceID: "i-abc123", + Status: "running", + InstanceType: "t3.small", + VCPU: 2, + MemoryGB: 2.0, + LaunchTime: 1736935200, + ManagedBy: "elbv2", + GPUs: []VMGPUInfo{{ + Model: "NVIDIA A10", + VRAMMiB: 23028, + PCIAddress: "0000:01:00.0", + Profile: "MIG 1g.6gb", + MdevPath: "/sys/bus/mdev/devices/example", + }}, + Health: "ok", + CrashCount: 2, + }}, + }) + require.NoError(t, err) + require.JSONEq(t, `{ + "node":"node-a", + "host":"10.0.0.1", + "vms":[{ + "instance_id":"i-abc123", + "status":"running", + "instance_type":"t3.small", + "vcpu":2, + "memory_gb":2, + "launch_time":1736935200, + "managed_by":"elbv2", + "gpus":[{ + "model":"NVIDIA A10", + "vram_mib":23028, + "pci_address":"0000:01:00.0", + "profile":"MIG 1g.6gb", + "mdev_path":"/sys/bus/mdev/devices/example" + }], + "health":"ok", + "crash_count":2 + }] + }`, string(data)) +} + +func TestNodeVMsSubject(t *testing.T) { + require.Equal(t, "spinifex.node.vms", NodeVMsSubject) +} diff --git a/docs/PACKAGE_BOUNDARY_MIGRATION.md b/docs/PACKAGE_BOUNDARY_MIGRATION.md index 7918650ed..b806a7367 100644 --- a/docs/PACKAGE_BOUNDARY_MIGRATION.md +++ b/docs/PACKAGE_BOUNDARY_MIGRATION.md @@ -186,6 +186,7 @@ while the branch is in flight; they do not describe the umbrella repository's | `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. | | `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. | 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 65bdcf852..988e7a6e9 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -92,6 +92,7 @@ services: # rarely the node that served the request (fan-out, no queue group). - "ec2.eni.public-ip-changed" - "spinifex.node.status" + # contracts/cluster/v1 owns the VM inventory payload and subject. - "spinifex.node.vms" - "spinifex.nodes.discover" - "spinifex.cluster.shutdown.progress" diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index adb40cc7a..52486d2be 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1021,7 +1021,7 @@ func (d *Daemon) subscribeAll() error { {clusterv1.NodeHealthSubject(d.node), d.handleHealthCheck, ""}, {clusterv1.NodesDiscoverSubject, d.handleNodeDiscover, ""}, {"spinifex.node.status", d.handleNodeStatus, ""}, - {"spinifex.node.vms", d.handleNodeVMs, ""}, + {clusterv1.NodeVMsSubject, d.handleNodeVMs, ""}, {operatorv1.StorageConfigSubject, d.handleStorageConfig, ""}, {"spinifex.image.promote", d.handleSpinifexPromoteImage, "spinifex-workers"}, // Account creation → create default VPC for new account diff --git a/spinifex/daemon/daemon_gpu_handlers_test.go b/spinifex/daemon/daemon_gpu_handlers_test.go index d5b575c3a..18eb9fc83 100644 --- a/spinifex/daemon/daemon_gpu_handlers_test.go +++ b/spinifex/daemon/daemon_gpu_handlers_test.go @@ -7,9 +7,10 @@ import ( "testing" "time" + clusterv1 "github.com/mulgadc/spinifex/contracts/cluster/v1" "github.com/mulgadc/spinifex/spinifex/runtime/compute/gpu" - "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/runtime/compute/vm" + "github.com/mulgadc/spinifex/spinifex/types" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -31,7 +32,7 @@ func gpuStatusRequest(t *testing.T, daemon *Daemon, subject string) types.NodeSt } // gpuVMsRequest subscribes and requests handleNodeVMs on a fresh subject. -func gpuVMsRequest(t *testing.T, daemon *Daemon, subject string) types.NodeVMsResponse { +func gpuVMsRequest(t *testing.T, daemon *Daemon, subject string) clusterv1.NodeVMsResponse { t.Helper() sub, err := daemon.natsConn.Subscribe(subject, asMsgHandler(daemon.handleNodeVMs)) require.NoError(t, err) @@ -40,7 +41,7 @@ func gpuVMsRequest(t *testing.T, daemon *Daemon, subject string) types.NodeVMsRe reply, err := daemon.natsConn.Request(subject, nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse require.NoError(t, json.Unmarshal(reply.Data, &resp)) return resp } diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index a56af0514..d901621f5 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -578,10 +578,10 @@ func buildPoolLookup(mgr *gpu.Manager) (byMdev, byPCI map[string]gpu.PoolEntry) // resolveVMGPU maps a single GPUAttachment to a VMGPUInfo using the pool // lookup tables built by buildPoolLookup. Returns nil if the attachment cannot // be matched (e.g. daemon restart before pool is fully restored). -func resolveVMGPU(att gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) *types.VMGPUInfo { +func resolveVMGPU(att gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) *clusterv1.VMGPUInfo { if att.MdevPath != "" { if e, ok := byMdev[att.MdevPath]; ok && e.MIGInstance != nil { - return &types.VMGPUInfo{ + return &clusterv1.VMGPUInfo{ Model: e.Device.Model, VRAMMiB: e.MIGInstance.Profile.MemoryMiB, Profile: e.MIGInstance.Profile.Name, @@ -592,7 +592,7 @@ func resolveVMGPU(att gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) } if att.PCIAddress != "" { if e, ok := byPCI[att.PCIAddress]; ok { - return &types.VMGPUInfo{ + return &clusterv1.VMGPUInfo{ Model: e.Device.Model, VRAMMiB: e.Device.MemoryMiB, PCIAddress: att.PCIAddress, @@ -605,8 +605,8 @@ func resolveVMGPU(att gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) // resolveVMGPUs resolves every attachment it can, reporting a short list rather // than passing it off as complete. One unresolved attachment used to read as no // GPU at all; eight make a shortened list a plausible wrong answer. -func resolveVMGPUs(instanceID string, attachments []gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) []types.VMGPUInfo { - gpus := make([]types.VMGPUInfo, 0, len(attachments)) +func resolveVMGPUs(instanceID string, attachments []gpu.GPUAttachment, byMdev, byPCI map[string]gpu.PoolEntry) []clusterv1.VMGPUInfo { + gpus := make([]clusterv1.VMGPUInfo, 0, len(attachments)) for _, attachment := range attachments { if info := resolveVMGPU(attachment, byMdev, byPCI); info != nil { gpus = append(gpus, *info) @@ -627,9 +627,9 @@ func resolveVMGPUs(instanceID string, attachments []gpu.GPUAttachment, byMdev, b func (d *Daemon) handleNodeVMs(msg *nats.Msg) string { poolByMdev, poolByPCI := buildPoolLookup(d.gpuManager) - vms := make([]types.VMInfo, 0, d.vmMgr.Count()) + vms := make([]clusterv1.VMInfo, 0, d.vmMgr.Count()) d.vmMgr.ForEach(func(v *vm.VM) { - info := types.VMInfo{ + info := clusterv1.VMInfo{ InstanceID: v.ID, Status: string(v.Status), InstanceType: v.InstanceType, @@ -648,7 +648,7 @@ func (d *Daemon) handleNodeVMs(msg *nats.Msg) string { vms = append(vms, info) }) - resp := types.NodeVMsResponse{ + resp := clusterv1.NodeVMsResponse{ Node: d.node, Host: d.daemonIP(), VMs: vms, diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index d47127fbe..cc38b538b 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -2239,7 +2239,7 @@ func TestHandleNodeVMs(t *testing.T) { reply, err := daemon.natsConn.Request("spinifex.node.vms.test", nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse err = json.Unmarshal(reply.Data, &resp) require.NoError(t, err) @@ -2248,7 +2248,7 @@ func TestHandleNodeVMs(t *testing.T) { assert.Len(t, resp.VMs, 2) // Build a lookup by instance ID - vmsByID := make(map[string]types.VMInfo) + vmsByID := make(map[string]clusterv1.VMInfo) for _, v := range resp.VMs { vmsByID[v.InstanceID] = v } @@ -2276,7 +2276,7 @@ func TestHandleNodeVMs_Empty(t *testing.T) { reply, err := daemon.natsConn.Request("spinifex.node.vms.empty", nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse err = json.Unmarshal(reply.Data, &resp) require.NoError(t, err) @@ -2300,7 +2300,7 @@ func TestHandleNodeVMs_UnknownInstanceType(t *testing.T) { reply, err := daemon.natsConn.Request("spinifex.node.vms.unknown", nil, 5*time.Second) require.NoError(t, err) - var resp types.NodeVMsResponse + var resp clusterv1.NodeVMsResponse err = json.Unmarshal(reply.Data, &resp) require.NoError(t, err) diff --git a/spinifex/gateway/spx/spx.go b/spinifex/gateway/spx/spx.go index 02b9fe006..7863d4d92 100644 --- a/spinifex/gateway/spx/spx.go +++ b/spinifex/gateway/spx/spx.go @@ -9,6 +9,7 @@ import ( "strings" "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" @@ -73,7 +74,7 @@ func GetNodes(ctx context.Context, nc *nats.Conn, expectedNodes int) (*GetNodesO // VMInfoWithNode extends the daemon's VMInfo with node attribution. type VMInfoWithNode struct { - types.VMInfo + clusterv1.VMInfo Node string `json:"node"` } @@ -85,7 +86,7 @@ type GetVMsOutput struct { // GetVMs queries all daemon nodes via NATS fan-out and returns their VMs. func GetVMs(ctx context.Context, nc *nats.Conn, expectedNodes int) (*GetVMsOutput, error) { - frames, _, err := utils.Gather(ctx, nc, "spinifex.node.vms", []byte("{}"), + frames, _, err := utils.Gather(ctx, nc, clusterv1.NodeVMsSubject, []byte("{}"), utils.GatherOpts{Timeout: 3 * time.Second, ExpectedNodes: expectedNodes}) if err != nil { return nil, err @@ -93,7 +94,7 @@ func GetVMs(ctx context.Context, nc *nats.Conn, expectedNodes int) (*GetVMsOutpu allVMs := make([]VMInfoWithNode, 0) for _, frame := range frames { - var nodeResp types.NodeVMsResponse + var nodeResp clusterv1.NodeVMsResponse if json.Unmarshal(frame.Data, &nodeResp) != nil { continue } diff --git a/spinifex/gateway/spx/spx_test.go b/spinifex/gateway/spx/spx_test.go index 46bff7dc3..140970979 100644 --- a/spinifex/gateway/spx/spx_test.go +++ b/spinifex/gateway/spx/spx_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "testing" + 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" @@ -119,11 +120,11 @@ func TestGetVMs_NoResponders(t *testing.T) { func TestGetVMs_WithVMs(t *testing.T) { _, nc := startEmbeddedNATS(t) - sub, err := nc.Subscribe("spinifex.node.vms", func(msg *nats.Msg) { - resp := types.NodeVMsResponse{ + sub, err := nc.Subscribe(clusterv1.NodeVMsSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeVMsResponse{ Node: "node1", Host: "10.0.0.1", - VMs: []types.VMInfo{ + VMs: []clusterv1.VMInfo{ {InstanceID: "i-abc123", Status: "running", InstanceType: "t3.small"}, {InstanceID: "i-def456", Status: "running", InstanceType: "t3.medium"}, }, @@ -149,10 +150,10 @@ func TestGetVMs_MultiNode(t *testing.T) { for _, name := range []string{"node1", "node2"} { nodeName := name - sub, err := nc.Subscribe("spinifex.node.vms", func(msg *nats.Msg) { - resp := types.NodeVMsResponse{ + sub, err := nc.Subscribe(clusterv1.NodeVMsSubject, func(msg *nats.Msg) { + resp := clusterv1.NodeVMsResponse{ Node: nodeName, - VMs: []types.VMInfo{ + VMs: []clusterv1.VMInfo{ {InstanceID: "i-" + nodeName, Status: "running"}, }, } diff --git a/spinifex/handlers/eks/k3s_ha_control_plane.go b/spinifex/handlers/eks/k3s_ha_control_plane.go index bcc328450..b9f5660be 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane.go +++ b/spinifex/handlers/eks/k3s_ha_control_plane.go @@ -11,9 +11,10 @@ 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/admin" - "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors" "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" @@ -576,8 +577,8 @@ func (h *natsHostScheduler) InstanceHosts(ctx context.Context, instanceIDs []str want[id] = true } out := make(map[string]string) - h.fanout(ctx, "spinifex.node.vms", func(data []byte) { - var resp types.NodeVMsResponse + h.fanout(ctx, clusterv1.NodeVMsSubject, func(data []byte) { + var resp clusterv1.NodeVMsResponse if json.Unmarshal(data, &resp) != nil || resp.Node == "" { return } diff --git a/spinifex/types/node.go b/spinifex/types/node.go index 4ff58de9f..346d3bcea 100644 --- a/spinifex/types/node.go +++ b/spinifex/types/node.go @@ -22,15 +22,6 @@ type GPUInfo struct { Slices []GPUSliceInfo `json:"slices,omitempty"` } -// VMGPUInfo describes one GPU attached to a VM. -type VMGPUInfo struct { - Model string `json:"model"` - VRAMMiB int64 `json:"vram_mib"` - PCIAddress string `json:"pci_address,omitempty"` // whole-GPU passthrough only - Profile string `json:"profile,omitempty"` // MIG profile name; empty for whole-GPU - MdevPath string `json:"mdev_path,omitempty"` // MIG only -} - // NodeStatusResponse is returned by the spinifex.node.status NATS topic (fan-out). // Schedulable capacity = TotalVCPU - ReservedVCPU - AllocVCPU (same for memory). type NodeStatusResponse struct { @@ -76,30 +67,3 @@ type InstanceTypeCap struct { MemoryGB float64 `json:"memory_gb"` Available int `json:"available"` } - -// VMInfo describes a single VM for the cluster stats CLI. -type VMInfo struct { - InstanceID string `json:"instance_id"` - Status string `json:"status"` - InstanceType string `json:"instance_type"` - VCPU int `json:"vcpu"` - MemoryGB float64 `json:"memory_gb"` - LaunchTime int64 `json:"launch_time"` - // ManagedBy is the Spinifex platform component that owns this VM - // (e.g. "elbv2"). Empty for customer VMs. The UI uses this to filter - // system-managed resources out of customer-facing listings. - ManagedBy string `json:"managed_by,omitempty"` - GPUs []VMGPUInfo `json:"gpus,omitempty"` - // Health is a display label for instance health: "ok", "impaired", - // "recovering", or "-" for non-running VMs. CrashCount is the lifetime - // crash tally within the current restart window. - Health string `json:"health,omitempty"` - CrashCount int `json:"crash_count,omitempty"` -} - -// NodeVMsResponse is returned by the spinifex.node.vms NATS topic (fan-out). -type NodeVMsResponse struct { - Node string `json:"node"` - Host string `json:"host"` - VMs []VMInfo `json:"vms"` -}