mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 09:01:31 +00:00
refactor: establish node VM inventory contract
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"},
|
||||
},
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"},
|
||||
},
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user