diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 69a9ac4e3..1cace99cd 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -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. } ``` diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index f198c19e3..fd32e7389 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -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, diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index 558db5633..4a42d97b2 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -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, diff --git a/spinifex/daemon/daemon_handlers_test.go b/spinifex/daemon/daemon_handlers_test.go index b3a4a4943..227f1511c 100644 --- a/spinifex/daemon/daemon_handlers_test.go +++ b/spinifex/daemon/daemon_handlers_test.go @@ -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 --- diff --git a/spinifex/daemon/daemon_test.go b/spinifex/daemon/daemon_test.go index 850203efc..c3deba6b1 100644 --- a/spinifex/daemon/daemon_test.go +++ b/spinifex/daemon/daemon_test.go @@ -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{}, diff --git a/spinifex/daemon/health_test.go b/spinifex/daemon/health_test.go index a550c5408..f4f2c3fc6 100644 --- a/spinifex/daemon/health_test.go +++ b/spinifex/daemon/health_test.go @@ -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{ diff --git a/spinifex/daemon/heartbeat.go b/spinifex/daemon/heartbeat.go index 9a5de0cbd..7d7bb7945 100644 --- a/spinifex/daemon/heartbeat.go +++ b/spinifex/daemon/heartbeat.go @@ -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, } } diff --git a/spinifex/daemon/heartbeat_test.go b/spinifex/daemon/heartbeat_test.go index e0d7ccd13..3d0cbfa2c 100644 --- a/spinifex/daemon/heartbeat_test.go +++ b/spinifex/daemon/heartbeat_test.go @@ -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) diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 838947f58..b7c3d4095 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -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. diff --git a/spinifex/daemon/resource.go b/spinifex/daemon/resource.go index d016df346..440ca80c1 100644 --- a/spinifex/daemon/resource.go +++ b/spinifex/daemon/resource.go @@ -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 diff --git a/spinifex/daemon/resource_test.go b/spinifex/daemon/resource_test.go index 6cb962a5d..56a59d08d 100644 --- a/spinifex/daemon/resource_test.go +++ b/spinifex/daemon/resource_test.go @@ -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 diff --git a/spinifex/types/node.go b/spinifex/types/node.go index 6136850fa..de267e382 100644 --- a/spinifex/types/node.go +++ b/spinifex/types/node.go @@ -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"`