gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-22 20:26:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: reserve compute for spinifex services (#170)

Browse Source
This commit is contained in:
Julian Sommer
2026-04-28 16:11:49 +10:00
committed by GitHub
parent 5911a878c6
commit 3dad452b09
12 changed files with 379 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/DESIGN.md
+5 -3
View File
@@ -158,11 +158,13 @@ Tracks available and allocated CPU/memory to prevent overcommit, and drives the
```go
type ResourceManager struct {
mu sync.RWMutex
availableVCPU int
availableMem float64
hostVCPU int // raw runtime.NumCPU
hostMemGB float64 // raw /proc/meminfo MemTotal
reservedVCPU int // held back for spinifex services
reservedMem float64 // held back for spinifex services
allocatedVCPU int
allocatedMem float64
instanceTypes map[string]*ec2.InstanceTypeInfo // t3.micro, t3.small, etc.
instanceTypes map[string]*ec2.InstanceTypeInfo // t3.micro, t3.small, etc.
}
```
spinifex/daemon/daemon.go
+53 -17
View File
@@ -81,9 +81,17 @@ type EBS struct {
// is available for a type, the node subscribes to ec2.RunInstances.{type};
// when full, it unsubscribes so NATS routes requests to other nodes.
type ResourceManager struct {
mu sync.RWMutex
availableVCPU int
availableMem float64
mu sync.RWMutex
// hostVCPU / hostMemGB are the raw figures reported by the host
// (runtime.NumCPU, /proc/meminfo). Schedulable capacity for guest VMs
// is host - reserved - allocated.
hostVCPU int
hostMemGB float64
// reservedVCPU / reservedMem are held back from guest scheduling for
// the spinifex daemon and co-located services (NATS, predastore,
// viperblock, vpcd, awsgw, ui). See hostReserve / defaultHostReserve.
reservedVCPU int
reservedMem float64
allocatedVCPU int
allocatedMem float64
instanceTypes map[string]*ec2.InstanceTypeInfo
@@ -216,6 +224,9 @@ func getSystemMemory() (float64, error) {
// NewResourceManager creates a new resource manager with system capabilities.
// Returns an error if system memory cannot be detected, since an incorrect
// default would either under-provision (large servers) or over-commit (small devices).
// Also returns an error if the host is too small to satisfy the daemon's
// reserve — clamping silently would defeat the reserve and look like a
// runtime bug.
func NewResourceManager() (*ResourceManager, error) {
// Get system CPU cores
numCPU := runtime.NumCPU()
@@ -226,6 +237,14 @@ func NewResourceManager() (*ResourceManager, error) {
return nil, fmt.Errorf("detect system memory: %w", err)
}
reservedVCPU, reservedMem, err := applyHostReserve(defaultHostReserve, numCPU, totalMemGB)
if err != nil {
slog.Error("host below minimum reserve — daemon refuses to start",
"err", err, "hostVCPU", numCPU, "hostMemGB", totalMemGB,
"reserveVCPU", defaultHostReserve.vCPU, "reserveMemGB", defaultHostReserve.memGB)
return nil, fmt.Errorf("validate host reserve: %w", err)
}
// Determine architecture
arch := "x86_64"
if runtime.GOARCH == "arm64" {
@@ -236,12 +255,16 @@ func NewResourceManager() (*ResourceManager, error) {
instanceTypes := instancetypes.DetectAndGenerate(instancetypes.HostCPU{}, arch)
slog.Info("System resources detected",
"vCPUs", numCPU, "memGB", totalMemGB,
"hostVCPU", numCPU, "hostMemGB", totalMemGB,
"reservedVCPU", reservedVCPU, "reservedMemGB", reservedMem,
"schedulableVCPU", numCPU-reservedVCPU, "schedulableMemGB", totalMemGB-reservedMem,
"instanceTypes", len(instanceTypes))
return &ResourceManager{
availableVCPU: numCPU,
availableMem: totalMemGB,
hostVCPU: numCPU,
hostMemGB: totalMemGB,
reservedVCPU: reservedVCPU,
reservedMem: reservedMem,
instanceTypes: instanceTypes,
}, nil
}
@@ -299,8 +322,8 @@ func (rm *ResourceManager) GetAvailableInstanceTypeInfos(showCapacity bool) []*e
}
count := canAllocateCount(
rm.availableVCPU, rm.allocatedVCPU,
rm.availableMem, rm.allocatedMem,
rm.hostVCPU-rm.reservedVCPU, rm.allocatedVCPU,
rm.hostMemGB-rm.reservedMem, rm.allocatedMem,
vCPUs, memMiB,
1<<30, // effectively unlimited — let resources be the constraint
)
@@ -315,23 +338,36 @@ func (rm *ResourceManager) GetAvailableInstanceTypeInfos(showCapacity bool) []*e
}
slog.Info("GetAvailableInstanceTypeInfos", "total_types", len(rm.instanceTypes), "total_available_slots", len(infos),
"hostVCPU", rm.availableVCPU, "hostMem", rm.availableMem, "showCapacity", showCapacity)
"hostVCPU", rm.hostVCPU, "hostMem", rm.hostMemGB,
"reservedVCPU", rm.reservedVCPU, "reservedMem", rm.reservedMem,
"showCapacity", showCapacity)
return infos
}
// GetResourceStats returns current resource allocation stats for the node status response.
func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64, allocVCPU int, allocMemGB float64, caps []types.InstanceTypeCap) {
// totalVCPU / totalMemGB are the raw host figures; reservedVCPU / reservedMemGB are
// held back from guest scheduling. Per-type caps reflect host - reserved - allocated,
// matching what the admission path will actually permit.
func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64, reservedVCPU int, reservedMemGB float64, allocVCPU int, allocMemGB float64, caps []types.InstanceTypeCap) {
rm.mu.RLock()
defer rm.mu.RUnlock()
totalVCPU = rm.availableVCPU
totalMemGB = rm.availableMem
totalVCPU = rm.hostVCPU
totalMemGB = rm.hostMemGB
reservedVCPU = rm.reservedVCPU
reservedMemGB = rm.reservedMem
allocVCPU = rm.allocatedVCPU
allocMemGB = rm.allocatedMem
remainingVCPU := rm.availableVCPU - rm.allocatedVCPU
remainingMem := rm.availableMem - rm.allocatedMem
remainingVCPU := rm.hostVCPU - rm.reservedVCPU - rm.allocatedVCPU
remainingMem := rm.hostMemGB - rm.reservedMem - rm.allocatedMem
if remainingVCPU < 0 || remainingMem < 0 {
slog.Error("schedulable capacity negative — reserve misconfigured or allocation drift",
"hostVCPU", rm.hostVCPU, "reservedVCPU", rm.reservedVCPU, "allocatedVCPU", rm.allocatedVCPU,
"hostMemGB", rm.hostMemGB, "reservedMem", rm.reservedMem, "allocatedMem", rm.allocatedMem,
"remainingVCPU", remainingVCPU, "remainingMem", remainingMem)
}
for name, it := range rm.instanceTypes {
if instancetypes.IsSystemType(name) {
@@ -343,7 +379,7 @@ func (rm *ResourceManager) GetResourceStats() (totalVCPU int, totalMemGB float64
}
caps = append(caps, typeCap)
}
return totalVCPU, totalMemGB, allocVCPU, allocMemGB, caps
return totalVCPU, totalMemGB, reservedVCPU, reservedMemGB, allocVCPU, allocMemGB, caps
}
// SetConfigPath sets the configuration file path for cluster management
@@ -2686,8 +2722,8 @@ func (rm *ResourceManager) canAllocate(instanceType *ec2.InstanceTypeInfo, count
defer rm.mu.RUnlock()
return canAllocateCount(
rm.availableVCPU, rm.allocatedVCPU,
rm.availableMem, rm.allocatedMem,
rm.hostVCPU-rm.reservedVCPU, rm.allocatedVCPU,
rm.hostMemGB-rm.reservedMem, rm.allocatedMem,
instanceTypeVCPUs(instanceType),
instanceTypeMemoryMiB(instanceType),
count,
spinifex/daemon/daemon_handlers.go
+3 -1
View File
@@ -156,7 +156,7 @@ func (d *Daemon) daemonIP() string {
// handleNodeStatus responds with this node's status and resource stats.
// Used by the CLI: spx get nodes, spx top nodes.
func (d *Daemon) handleNodeStatus(msg *nats.Msg) {
totalVCPU, totalMemGB, allocVCPU, allocMemGB, caps := d.resourceMgr.GetResourceStats()
totalVCPU, totalMemGB, reservedVCPU, reservedMemGB, allocVCPU, allocMemGB, caps := d.resourceMgr.GetResourceStats()
d.Instances.Mu.Lock()
vmCount := 0
@@ -177,6 +177,8 @@ func (d *Daemon) handleNodeStatus(msg *nats.Msg) {
Services: d.config.GetServices(),
TotalVCPU: totalVCPU,
TotalMemGB: totalMemGB,
ReservedVCPU: reservedVCPU,
ReservedMemGB: reservedMemGB,
AllocVCPU: allocVCPU,
AllocMemGB: allocMemGB,
VMCount: vmCount,
spinifex/daemon/daemon_handlers_test.go
+6
View File
@@ -2484,6 +2484,10 @@ func TestHandleNodeStatus(t *testing.T) {
assert.Equal(t, 2, resp.VMCount)
assert.Greater(t, resp.TotalVCPU, 0)
assert.Greater(t, resp.TotalMemGB, 0.0)
assert.Equal(t, daemon.resourceMgr.reservedVCPU, resp.ReservedVCPU, "ReservedVCPU must be populated on wire")
assert.InDelta(t, daemon.resourceMgr.reservedMem, resp.ReservedMemGB, 0.001, "ReservedMemGB must be populated on wire")
assert.Greater(t, resp.ReservedVCPU, 0, "default reserve is non-zero")
assert.Greater(t, resp.ReservedMemGB, 0.0, "default reserve is non-zero")
}
func TestHandleNodeStatus_NoVMs(t *testing.T) {
@@ -2503,6 +2507,8 @@ func TestHandleNodeStatus_NoVMs(t *testing.T) {
assert.Equal(t, 0, resp.VMCount)
assert.Equal(t, "Ready", resp.Status)
assert.Greater(t, resp.ReservedVCPU, 0, "default reserve is exposed even when no VMs")
assert.Greater(t, resp.ReservedMemGB, 0.0, "default reserve is exposed even when no VMs")
}
// --- handleNodeVMs tests ---
spinifex/daemon/daemon_test.go
+144 -112
View File
@@ -403,8 +403,8 @@ func TestResourceManager(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, rm)
assert.Greater(t, rm.availableVCPU, 0)
assert.Greater(t, rm.availableMem, float64(0))
assert.Greater(t, rm.hostVCPU, 0)
assert.Greater(t, rm.hostMemGB, float64(0))
// Test allocation using the first available instance type (dynamic based on CPU)
require.NotEmpty(t, rm.instanceTypes, "Should have at least one instance type")
@@ -521,24 +521,28 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) {
allTypes := rm.GetInstanceTypeInfos()
initialAvailable := rm.GetAvailableInstanceTypeInfos(false)
t.Logf("System has %d vCPUs, %.2f GB RAM", rm.availableVCPU, rm.availableMem)
t.Logf("System has %d vCPUs, %.2f GB RAM (reserved: %d vCPU, %.2f GB)",
rm.hostVCPU, rm.hostMemGB, rm.reservedVCPU, rm.reservedMem)
t.Logf("All instance types: %d, Initially available: %d", len(allTypes), len(initialAvailable))
// Initially available types should only include those that fit system resources
// (on small machines, xlarge/2xlarge may already be filtered out)
// Initially available types should only include those that fit schedulable
// capacity (host - reserved). On small machines, xlarge/2xlarge may already
// be filtered out.
assert.LessOrEqual(t, len(initialAvailable), len(allTypes),
"Available types should be <= total types")
assert.Greater(t, len(initialAvailable), 0, "Should have at least one available type")
// Verify all initially available types fit within system resources
// Verify all initially available types fit within schedulable resources.
schedulableVCPU := rm.hostVCPU - rm.reservedVCPU
schedulableMem := rm.hostMemGB - rm.reservedMem
for _, info := range initialAvailable {
vcpus := int(*info.VCpuInfo.DefaultVCpus)
memGB := float64(*info.MemoryInfo.SizeInMiB) / 1024
assert.LessOrEqual(t, vcpus, rm.availableVCPU,
"Instance type %s vCPUs should fit system", *info.InstanceType)
assert.LessOrEqual(t, memGB, rm.availableMem,
"Instance type %s memory should fit system", *info.InstanceType)
assert.LessOrEqual(t, vcpus, schedulableVCPU,
"Instance type %s vCPUs should fit schedulable capacity", *info.InstanceType)
assert.LessOrEqual(t, memGB, schedulableMem,
"Instance type %s memory should fit schedulable capacity", *info.InstanceType)
}
// Allocate the smallest instance type (nano) to consume some resources
@@ -565,9 +569,10 @@ func TestGetAvailableInstanceTypeInfos_ResourceFiltering(t *testing.T) {
afterAllocation := rm.GetAvailableInstanceTypeInfos(false)
t.Logf("Available after allocation: %d", len(afterAllocation))
// Verify all returned types fit within REMAINING resources
remainingVCPU := rm.availableVCPU - rm.allocatedVCPU
remainingMem := rm.availableMem - rm.allocatedMem
// Verify all returned types fit within REMAINING schedulable resources
// (host - reserved - allocated).
remainingVCPU := rm.hostVCPU - rm.reservedVCPU - rm.allocatedVCPU
remainingMem := rm.hostMemGB - rm.reservedMem - rm.allocatedMem
for _, info := range afterAllocation {
typeName := *info.InstanceType
@@ -737,9 +742,9 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
assert.LessOrEqual(t, afterAllocationCount, initialCount,
"Should have fewer or equal instance types after allocation")
// Calculate remaining resources
remainingVCPU := daemon.resourceMgr.availableVCPU - daemon.resourceMgr.allocatedVCPU
remainingMem := daemon.resourceMgr.availableMem - daemon.resourceMgr.allocatedMem
// Calculate remaining schedulable resources (host - reserved - allocated).
remainingVCPU := daemon.resourceMgr.hostVCPU - daemon.resourceMgr.reservedVCPU - daemon.resourceMgr.allocatedVCPU
remainingMem := daemon.resourceMgr.hostMemGB - daemon.resourceMgr.reservedMem - daemon.resourceMgr.allocatedMem
t.Logf("Remaining resources: %d vCPUs, %.2f GB RAM", remainingVCPU, remainingMem)
@@ -801,18 +806,25 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
// Test 4: Verify "capacity" filter returns duplicates
t.Run("VerifyCapacityFilter_Duplicates", func(t *testing.T) {
// Force resources to a predictable state
// Force schedulable capacity to a predictable state by zeroing the
// reserve and pinning host figures directly.
daemon.resourceMgr.mu.Lock()
oldAvailableVCPU := daemon.resourceMgr.availableVCPU
oldAvailableMem := daemon.resourceMgr.availableMem
daemon.resourceMgr.availableVCPU = 2
daemon.resourceMgr.availableMem = 16.0
oldHostVCPU := daemon.resourceMgr.hostVCPU
oldHostMem := daemon.resourceMgr.hostMemGB
oldReservedVCPU := daemon.resourceMgr.reservedVCPU
oldReservedMem := daemon.resourceMgr.reservedMem
daemon.resourceMgr.hostVCPU = 2
daemon.resourceMgr.hostMemGB = 16.0
daemon.resourceMgr.reservedVCPU = 0
daemon.resourceMgr.reservedMem = 0
daemon.resourceMgr.mu.Unlock()
defer func() {
daemon.resourceMgr.mu.Lock()
daemon.resourceMgr.availableVCPU = oldAvailableVCPU
daemon.resourceMgr.availableMem = oldAvailableMem
daemon.resourceMgr.hostVCPU = oldHostVCPU
daemon.resourceMgr.hostMemGB = oldHostMem
daemon.resourceMgr.reservedVCPU = oldReservedVCPU
daemon.resourceMgr.reservedMem = oldReservedMem
daemon.resourceMgr.mu.Unlock()
}()
@@ -851,8 +863,8 @@ func TestHandleEC2DescribeInstanceTypes(t *testing.T) {
// Now increase capacity to test duplicate slots
daemon.resourceMgr.mu.Lock()
daemon.resourceMgr.availableVCPU = 4
daemon.resourceMgr.availableMem = 15.0
daemon.resourceMgr.hostVCPU = 4
daemon.resourceMgr.hostMemGB = 15.0
daemon.resourceMgr.mu.Unlock()
reply, err = daemon.natsConn.Request("ec2.DescribeInstanceTypes", msgData, 5*time.Second)
@@ -1065,8 +1077,20 @@ func TestCanAllocate_CountEdgeCases(t *testing.T) {
}
require.NotNil(t, microType)
// Pin host capacity so the test is independent of the runner's
// schedulable headroom (host - reserved). CI runners have only
// 4 vCPU, leaving 2 schedulable after the default reserve — not
// enough to fit two micro instances (2 vCPU each).
rm.mu.Lock()
rm.hostVCPU = 16
rm.hostMemGB = 32.0
rm.reservedVCPU = 0
rm.reservedMem = 0
rm.mu.Unlock()
initial := rm.canAllocate(microType, 100)
t.Logf("Initial capacity: %d micro instances", initial)
require.GreaterOrEqual(t, initial, 2, "test needs at least 2 micro slots")
// Allocate one
err = rm.allocate(microType)
@@ -1468,8 +1492,8 @@ func TestInstanceTypeSubscriptions(t *testing.T) {
// Allocate all resources so nothing fits
rm.mu.Lock()
rm.allocatedVCPU = rm.availableVCPU
rm.allocatedMem = rm.availableMem
rm.allocatedVCPU = rm.hostVCPU - rm.reservedVCPU
rm.allocatedMem = rm.hostMemGB - rm.reservedMem
rm.mu.Unlock()
rm.updateInstanceSubscriptions()
@@ -1492,8 +1516,8 @@ func TestInstanceTypeSubscriptions(t *testing.T) {
// Fill all resources
rm.mu.Lock()
rm.allocatedVCPU = rm.availableVCPU
rm.allocatedMem = rm.availableMem
rm.allocatedVCPU = rm.hostVCPU - rm.reservedVCPU
rm.allocatedMem = rm.hostMemGB - rm.reservedMem
rm.mu.Unlock()
rm.updateInstanceSubscriptions()
assert.Equal(t, 0, len(rm.instanceSubs))
@@ -1519,10 +1543,10 @@ func TestInstanceTypeSubscriptions(t *testing.T) {
handler := func(msg *nats.Msg) {}
rm.initSubscriptions(nc, handler, "test-node")
// Leave only 2 vCPUs and 1 GB free — enough for nano/micro but not larger types
// Leave only 2 vCPUs and 1 GB schedulable — enough for nano/micro but not larger types.
rm.mu.Lock()
rm.allocatedVCPU = rm.availableVCPU - 2
rm.allocatedMem = rm.availableMem - 1.0
rm.allocatedVCPU = (rm.hostVCPU - rm.reservedVCPU) - 2
rm.allocatedMem = (rm.hostMemGB - rm.reservedMem) - 1.0
rm.mu.Unlock()
rm.updateInstanceSubscriptions()
@@ -1594,8 +1618,8 @@ func TestInstanceTypeSubscriptions(t *testing.T) {
// Fill the node completely
rm.mu.Lock()
rm.allocatedVCPU = rm.availableVCPU
rm.allocatedMem = rm.availableMem
rm.allocatedVCPU = rm.hostVCPU - rm.reservedVCPU
rm.allocatedMem = rm.hostMemGB - rm.reservedMem
rm.mu.Unlock()
rm.updateInstanceSubscriptions()
assert.Equal(t, 0, len(rm.instanceSubs))
@@ -3158,8 +3182,8 @@ func TestDaemon_LoadState_NilJSManager(t *testing.T) {
func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) {
rm := &ResourceManager{
availableVCPU: 4,
availableMem: 8.0,
hostVCPU: 4,
hostMemGB: 8.0,
allocatedVCPU: 8, // Over-committed
allocatedMem: 16.0,
instanceTypes: map[string]*ec2.InstanceTypeInfo{
@@ -3180,8 +3204,8 @@ func TestGetAvailableInstanceTypeInfos_Overcommitted(t *testing.T) {
func TestGetAvailableInstanceTypeInfos_ShowCapacity(t *testing.T) {
rm := &ResourceManager{
availableVCPU: 8,
availableMem: 16.0,
hostVCPU: 8,
hostMemGB: 16.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceTypes: map[string]*ec2.InstanceTypeInfo{
@@ -3614,8 +3638,10 @@ func TestCanAllocate(t *testing.T) {
tests := []struct {
name string
availableVCPU int
availableMem float64
hostVCPU int
hostMemGB float64
reservedVCPU int
reservedMem float64
allocatedVCPU int
allocatedMem float64
instanceType *ec2.InstanceTypeInfo
@@ -3623,39 +3649,33 @@ func TestCanAllocate(t *testing.T) {
want int
}{
{
name: "plenty of resources",
availableVCPU: 16,
availableMem: 32.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(2, 2048), // 2 vCPU, 2 GiB
count: 3,
want: 3,
name: "plenty of resources",
hostVCPU: 16,
hostMemGB: 32.0,
instanceType: makeInstanceType(2, 2048), // 2 vCPU, 2 GiB
count: 3,
want: 3,
},
{
name: "CPU limited",
availableVCPU: 4,
availableMem: 32.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(2, 2048),
count: 5,
want: 2,
name: "CPU limited",
hostVCPU: 4,
hostMemGB: 32.0,
instanceType: makeInstanceType(2, 2048),
count: 5,
want: 2,
},
{
name: "memory limited",
availableVCPU: 16,
availableMem: 4.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(1, 2048), // 1 vCPU, 2 GiB
count: 5,
want: 2,
name: "memory limited",
hostVCPU: 16,
hostMemGB: 4.0,
instanceType: makeInstanceType(1, 2048), // 1 vCPU, 2 GiB
count: 5,
want: 2,
},
{
name: "no resources returns 0",
availableVCPU: 4,
availableMem: 4.0,
hostVCPU: 4,
hostMemGB: 4.0,
allocatedVCPU: 4,
allocatedMem: 4.0,
instanceType: makeInstanceType(2, 2048),
@@ -3663,39 +3683,33 @@ func TestCanAllocate(t *testing.T) {
want: 0,
},
{
name: "zero vCPU instance type returns requested count",
availableVCPU: 8,
availableMem: 16.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(0, 2048),
count: 4,
want: 4,
name: "zero vCPU instance type returns requested count",
hostVCPU: 8,
hostMemGB: 16.0,
instanceType: makeInstanceType(0, 2048),
count: 4,
want: 4,
},
{
name: "zero memory instance type returns requested count",
availableVCPU: 8,
availableMem: 16.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(2, 0),
count: 4,
want: 4,
name: "zero memory instance type returns requested count",
hostVCPU: 8,
hostMemGB: 16.0,
instanceType: makeInstanceType(2, 0),
count: 4,
want: 4,
},
{
name: "nil VCpuInfo and MemoryInfo returns requested count",
availableVCPU: 8,
availableMem: 16.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: &ec2.InstanceTypeInfo{},
count: 3,
want: 3,
name: "nil VCpuInfo and MemoryInfo returns requested count",
hostVCPU: 8,
hostMemGB: 16.0,
instanceType: &ec2.InstanceTypeInfo{},
count: 3,
want: 3,
},
{
name: "partial allocation reduces available",
availableVCPU: 8,
availableMem: 8.0,
hostVCPU: 8,
hostMemGB: 8.0,
allocatedVCPU: 4,
allocatedMem: 4.0,
instanceType: makeInstanceType(2, 2048),
@@ -3703,42 +3717,60 @@ func TestCanAllocate(t *testing.T) {
want: 2,
},
{
name: "exact fit",
availableVCPU: 4,
availableMem: 4.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(2, 2048),
count: 2,
want: 2,
name: "exact fit",
hostVCPU: 4,
hostMemGB: 4.0,
instanceType: makeInstanceType(2, 2048),
count: 2,
want: 2,
},
{
name: "count zero returns 0",
availableVCPU: 8,
availableMem: 16.0,
allocatedVCPU: 0,
allocatedMem: 0,
instanceType: makeInstanceType(2, 2048),
count: 0,
want: 0,
name: "count zero returns 0",
hostVCPU: 8,
hostMemGB: 16.0,
instanceType: makeInstanceType(2, 2048),
count: 0,
want: 0,
},
{
name: "negative available returns 0",
availableVCPU: 2,
availableMem: 2.0,
hostVCPU: 2,
hostMemGB: 2.0,
allocatedVCPU: 4,
allocatedMem: 4.0,
instanceType: makeInstanceType(2, 2048),
count: 1,
want: 0,
},
{
name: "reserve subtracts from schedulable",
hostVCPU: 16,
hostMemGB: 32.0,
reservedVCPU: 2,
reservedMem: 4.0,
instanceType: makeInstanceType(2, 2048), // 2 vCPU, 2 GiB
count: 100,
want: 7, // CPU: (16-2)/2=7, mem: (32-4)/2=14 → min=7
},
{
name: "reserve consumes all CPU",
hostVCPU: 4,
hostMemGB: 32.0,
reservedVCPU: 4,
reservedMem: 4.0,
instanceType: makeInstanceType(2, 2048),
count: 5,
want: 0,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
rm := &ResourceManager{
availableVCPU: tt.availableVCPU,
availableMem: tt.availableMem,
hostVCPU: tt.hostVCPU,
hostMemGB: tt.hostMemGB,
reservedVCPU: tt.reservedVCPU,
reservedMem: tt.reservedMem,
allocatedVCPU: tt.allocatedVCPU,
allocatedMem: tt.allocatedMem,
instanceTypes: map[string]*ec2.InstanceTypeInfo{},
spinifex/daemon/health_test.go
+4 -3
View File
@@ -422,10 +422,11 @@ func TestMaybeRestart_InsufficientResources(t *testing.T) {
allocType := smallestAllocType(t, d.resourceMgr)
// Exhaust all resources by setting allocated = available
// Exhaust all schedulable resources (host - reserved) so canAllocate
// returns 0, matching what's actually achievable at runtime.
d.resourceMgr.mu.Lock()
d.resourceMgr.allocatedVCPU = d.resourceMgr.availableVCPU
d.resourceMgr.allocatedMem = d.resourceMgr.availableMem
d.resourceMgr.allocatedVCPU = d.resourceMgr.hostVCPU - d.resourceMgr.reservedVCPU
d.resourceMgr.allocatedMem = d.resourceMgr.hostMemGB - d.resourceMgr.reservedMem
d.resourceMgr.mu.Unlock()
instance := &vm.VM{
spinifex/daemon/heartbeat.go
+6 -1
View File
@@ -48,8 +48,11 @@ func (d *Daemon) publishHeartbeat() {
}
// buildHeartbeat constructs a Heartbeat from the daemon's current state.
// AvailableVCPU / AvailableMem retain "host - allocated" semantics for
// observability continuity; the daemon's reserve is surfaced separately
// so consumers can compute schedulable capacity.
func (d *Daemon) buildHeartbeat() *Heartbeat {
totalVCPU, totalMem, allocVCPU, allocMem, _ := d.resourceMgr.GetResourceStats()
totalVCPU, totalMem, reservedVCPU, reservedMem, allocVCPU, allocMem, _ := d.resourceMgr.GetResourceStats()
d.Instances.Mu.Lock()
vmCount := len(d.Instances.VMS)
@@ -65,5 +68,7 @@ func (d *Daemon) buildHeartbeat() *Heartbeat {
AvailableVCPU: totalVCPU - allocVCPU,
AllocatedMem: allocMem,
AvailableMem: totalMem - allocMem,
ReservedVCPU: reservedVCPU,
ReservedMem: reservedMem,
}
}
spinifex/daemon/heartbeat_test.go
+8
View File
@@ -38,6 +38,10 @@ func TestBuildHeartbeat(t *testing.T) {
assert.Equal(t, 0, h.AllocatedVCPU)
assert.Greater(t, h.AvailableVCPU, 0)
assert.Greater(t, h.AvailableMem, 0.0)
assert.Equal(t, rm.reservedVCPU, h.ReservedVCPU, "ReservedVCPU must be populated from ResourceManager")
assert.InDelta(t, rm.reservedMem, h.ReservedMem, 0.001, "ReservedMem must be populated from ResourceManager")
assert.Greater(t, h.ReservedVCPU, 0, "default reserve is non-zero")
assert.Greater(t, h.ReservedMem, 0.0, "default reserve is non-zero")
}
// TestHeartbeatReflectsAllocation verifies that allocating resources changes the heartbeat values.
@@ -114,6 +118,8 @@ func TestHeartbeatKVContract(t *testing.T) {
AvailableVCPU: 12,
AllocatedMem: 8.0,
AvailableMem: 24.0,
ReservedVCPU: 2,
ReservedMem: 4.0,
}
require.NoError(t, jsm.WriteHeartbeat(h))
@@ -134,6 +140,8 @@ func TestHeartbeatKVContract(t *testing.T) {
assert.Equal(t, float64(12), raw["available_vcpu"])
assert.Equal(t, 8.0, raw["allocated_mem_gb"])
assert.Equal(t, 24.0, raw["available_mem_gb"])
assert.Equal(t, float64(2), raw["reserved_vcpu"])
assert.Equal(t, 4.0, raw["reserved_mem_gb"])
// And ReadHeartbeat decodes the same value.
loaded, err := jsm.ReadHeartbeat(h.Node)
spinifex/daemon/jetstream.go
+8
View File
@@ -226,6 +226,12 @@ func (m *JetStreamManager) recoverTerminatedKVBucket() (nats.KeyValue, error) {
}
// Heartbeat represents a daemon's periodic health status published to cluster KV.
//
// AvailableVCPU / AvailableMem are observability-only (host - allocated,
// raw). Scheduling routing happens at the local daemon via admission
// control, which already accounts for the reserve; ReservedVCPU /
// ReservedMem are exposed purely for operator dashboards and capacity
// reporting.
type Heartbeat struct {
Node string `json:"node"`
Epoch uint64 `json:"epoch"`
@@ -236,6 +242,8 @@ type Heartbeat struct {
AvailableVCPU int `json:"available_vcpu"`
AllocatedMem float64 `json:"allocated_mem_gb"`
AvailableMem float64 `json:"available_mem_gb"`
ReservedVCPU int `json:"reserved_vcpu"`
ReservedMem float64 `json:"reserved_mem_gb"`
}
// WriteHeartbeat writes a heartbeat entry for the given node to the cluster-state KV.
spinifex/daemon/resource.go
+57 -2
View File
@@ -2,6 +2,8 @@ package daemon
import (
"errors"
"fmt"
"log/slog"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/types"
@@ -11,13 +13,63 @@ import (
// cannot be satisfied.
var errInsufficientCapacity = errors.New("insufficient capacity to satisfy MinCount")
// hostReserve is the fixed amount of host CPU and RAM held back from guest
// scheduling so the spinifex daemon and co-located services (NATS,
// predastore, viperblock, vpcd, awsgw, ui) cannot be starved by guest VMs
// at maximum density. A future `capacity` command will lift this into
// operator-tunable config.
type hostReserve struct {
vCPU int
memGB float64
}
// defaultHostReserve is sized to cover predastore + viperblock under load;
// the daemon itself needs little.
var defaultHostReserve = hostReserve{vCPU: 2, memGB: 4.0}
// minHostMemHeadroomGB is the minimum schedulable memory we require above
// the reserve, so a host that just meets the reserve still has a small
// amount left to launch the smallest guest type.
const minHostMemHeadroomGB = 0.5
// applyHostReserve validates that the host meets the minimum size for the
// given reserve and returns the reserve to apply. Pure function — no locks
// or side effects. Exists as a helper for unit-testability of the
// validation bounds.
//
// vCPU and memory are checked separately so the returned error names the
// failing dimension, letting alerting/log filters distinguish a CPU
// shortfall from a memory shortfall.
func applyHostReserve(host hostReserve, totalVCPU int, totalMemGB float64) (vcpu int, mem float64, err error) {
if totalVCPU <= host.vCPU {
return 0, 0, fmt.Errorf(
"host vCPU below required minimum: have %d, need at least %d (reserve %d + 1 schedulable)",
totalVCPU, host.vCPU+1, host.vCPU,
)
}
if totalMemGB < host.memGB+minHostMemHeadroomGB {
return 0, 0, fmt.Errorf(
"host memory below required minimum: have %.2f GB, need at least %.2f GB (reserve %.1f + %.1f headroom)",
totalMemGB, host.memGB+minHostMemHeadroomGB, host.memGB, minHostMemHeadroomGB,
)
}
return host.vCPU, host.memGB, nil
}
// canAllocateCount returns how many instances of the given type can fit
// in the remaining capacity, capped at maxCount. Pure function — no locks
// or side effects.
// in the remaining capacity, capped at maxCount. Pure aside from a single
// slog.Error when remaining capacity is negative — that condition would
// otherwise be silently clamped to zero, hiding a misconfigured reserve
// or allocation accounting drift.
func canAllocateCount(availVCPU, allocVCPU int, availMem, allocMem float64,
vCPUs int64, memMiB int64, maxCount int) int {
remainingVCPU := availVCPU - allocVCPU
remainingMem := availMem - allocMem
if remainingVCPU < 0 || remainingMem < 0 {
slog.Error("schedulable capacity negative — reserve misconfigured or allocation drift",
"availVCPU", availVCPU, "allocVCPU", allocVCPU, "remainingVCPU", remainingVCPU,
"availMem", availMem, "allocMem", allocMem, "remainingMem", remainingMem)
}
memoryGB := float64(memMiB) / 1024.0
countByCPU := maxCount
@@ -37,6 +89,9 @@ func canAllocateCount(availVCPU, allocVCPU int, availMem, allocMem float64,
// resourceStatsForType computes the InstanceTypeCap for a single instance type
// given the remaining host resources. Pure function — no locks or side effects.
// Callers are responsible for alarming on negative remainVCPU/remainMem
// (see canAllocateCount); resourceStatsForType silently clamps to zero
// because it's invoked in a per-type loop and would otherwise log N times.
func resourceStatsForType(remainVCPU int, remainMem float64, it *ec2.InstanceTypeInfo) types.InstanceTypeCap {
vCPUs := instanceTypeVCPUs(it)
memGB := float64(instanceTypeMemoryMiB(it)) / 1024.0
spinifex/daemon/resource_test.go
+78
View File
@@ -210,6 +210,84 @@ func TestResourceStatsForType(t *testing.T) {
}
}
func TestApplyHostReserve(t *testing.T) {
tests := []struct {
name string
host hostReserve
totalVCPU int
totalMemGB float64
wantVCPU int
wantMem float64
wantErr bool
}{
{
name: "normal host with default reserve",
host: defaultHostReserve,
totalVCPU: 16,
totalMemGB: 64.0,
wantVCPU: 2,
wantMem: 4.0,
},
{
name: "boundary host (3 vCPU, 4.5 GB)",
host: defaultHostReserve,
totalVCPU: 3,
totalMemGB: 4.5,
wantVCPU: 2,
wantMem: 4.0,
},
{
name: "too small: vCPU equals reserve",
host: defaultHostReserve,
totalVCPU: 2,
totalMemGB: 8.0,
wantErr: true,
},
{
name: "too small: vCPU below reserve",
host: defaultHostReserve,
totalVCPU: 1,
totalMemGB: 8.0,
wantErr: true,
},
{
name: "too small: mem at reserve threshold (no headroom)",
host: defaultHostReserve,
totalVCPU: 8,
totalMemGB: 4.0,
wantErr: true,
},
{
name: "too small: mem just under reserve+headroom",
host: defaultHostReserve,
totalVCPU: 8,
totalMemGB: 4.49,
wantErr: true,
},
{
name: "custom reserve passed through",
host: hostReserve{vCPU: 4, memGB: 8.0},
totalVCPU: 16,
totalMemGB: 64.0,
wantVCPU: 4,
wantMem: 8.0,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
gotVCPU, gotMem, err := applyHostReserve(tc.host, tc.totalVCPU, tc.totalMemGB)
if tc.wantErr {
assert.Error(t, err)
return
}
assert.NoError(t, err)
assert.Equal(t, tc.wantVCPU, gotVCPU)
assert.InDelta(t, tc.wantMem, gotMem, 0.001)
})
}
}
func TestAllocateForLaunch(t *testing.T) {
tests := []struct {
name string
spinifex/types/node.go
+7
View File
@@ -6,6 +6,11 @@ type NodeDiscoverResponse struct {
}
// NodeStatusResponse is returned by the spinifex.node.status NATS topic (fan-out).
//
// ReservedVCPU / ReservedMemGB are held back from guest scheduling for the
// spinifex daemon and co-located services. Schedulable capacity is
// TotalVCPU - ReservedVCPU - AllocVCPU (and the equivalent for memory);
// per-type counts in InstanceTypes already account for the reserve.
type NodeStatusResponse struct {
Node string `json:"node"`
Status string `json:"status"`
@@ -16,6 +21,8 @@ type NodeStatusResponse struct {
Services []string `json:"services"`
TotalVCPU int `json:"total_vcpu"`
TotalMemGB float64 `json:"total_mem_gb"`
ReservedVCPU int `json:"reserved_vcpu"`
ReservedMemGB float64 `json:"reserved_mem_gb"`
AllocVCPU int `json:"alloc_vcpu"`
AllocMemGB float64 `json:"alloc_mem_gb"`
VMCount int `json:"vm_count"`