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

refactor: establish node status contract

Browse Source
This commit is contained in:
joshsiv-mulga
2026-10-01 08:22:26 +10:00
parent 4b35a2a33d
commit 08bd335f0a
34 changed files with 173 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/spinifex/cmd/admin_gpu.go
+4 -4
View File
@@ -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
}
cmd/spinifex/cmd/get.go
+4 -5
View File
@@ -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
}
cmd/spinifex/cmd/get_test.go
+1 -1
View File
@@ -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"
)
cmd/spinifex/cmd/top.go
+4 -4
View File
@@ -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
}
contracts/cluster/v1/README.md
+7
View File
@@ -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.
spinifex/types/node.go → contracts/cluster/v1/status.go
+10 -3
View File
@@ -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"`
contracts/cluster/v1/status_test.go
+81
View File
@@ -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)
}
docs/PACKAGE_BOUNDARY_MIGRATION.md
+1
View File
@@ -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
docs/service-interfaces.yaml
+1
View File
@@ -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"
spinifex/daemon/daemon.go
+2 -3
View File
@@ -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"},
spinifex/daemon/daemon_gpu_handlers_test.go
+4 -5
View File
@@ -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
}
spinifex/daemon/daemon_handlers.go
+7 -8
View File
@@ -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])
}
spinifex/daemon/daemon_handlers_test.go
+2 -2
View File
@@ -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)
spinifex/daemon/gpu_admission_test.go
+3 -3
View File
@@ -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")
spinifex/daemon/resource.go
+3 -3
View File
@@ -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,
spinifex/gateway/ec2/capacityreservation/capacityreservation_test.go
+2 -2
View File
@@ -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"
spinifex/gateway/ec2/capacityreservation/fanout.go
+3 -3
View File
@@ -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
spinifex/gateway/ec2/capacityreservation/fanout_test.go
+1 -1
View File
@@ -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"
spinifex/gateway/ec2/instance/placement.go
+3 -3
View File
@@ -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
spinifex/gateway/ec2/instance/placement_spread_test.go
+1 -1
View File
@@ -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"
spinifex/gateway/ec2/instance/placement_test.go
+1 -1
View File
@@ -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"
spinifex/gateway/ec2/spotinstance/quota_gate_order_test.go
+2 -2
View File
@@ -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"
spinifex/gateway/spx/spx.go
+5 -6
View File
@@ -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)
}
spinifex/gateway/spx/spx_test.go
+5 -6
View File
@@ -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)
})
spinifex/handlers/eks/k3s_ha_control_plane.go
+4 -5
View File
@@ -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
spinifex/handlers/eks/k3s_ha_control_plane_test.go
+3 -3
View File
@@ -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"
tests/integration/harness.go
+2 -2
View File
@@ -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)
tests/integration/launchtemplate_test.go
+1 -1
View File
@@ -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"
tests/integration/run_instances_clienttoken_test.go
+1 -1
View File
@@ -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"
)
tests/integration/run_instances_launchtemplate_test.go
+1 -1
View File
@@ -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"
)
tests/integration/run_instances_passrole_test.go
+1 -1
View File
@@ -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"
)
tests/integration/run_instances_placementgroup_test.go
+1 -1
View File
@@ -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"
)
tests/integration/run_instances_test.go
+1 -1
View File
@@ -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"
)
tests/integration/spot_instance_test.go
+1 -1
View File
@@ -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"