diff --git a/internal/controller/scheduler/scheduler_capacity_test.go b/internal/controller/scheduler/scheduler_capacity_test.go new file mode 100644 index 0000000..0bcf6f4 --- /dev/null +++ b/internal/controller/scheduler/scheduler_capacity_test.go @@ -0,0 +1,105 @@ +//nolint:testpackage // The regression exercises the unexported scheduler reconciliation loop. +package scheduler + +import ( + "testing" + "time" + + "github.com/cirruslabs/orchard/internal/controller/notifier" + storepkg "github.com/cirruslabs/orchard/internal/controller/store" + "github.com/cirruslabs/orchard/internal/controller/store/badger" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func TestSchedulingLoopSkipsOvercommittedWorker(t *testing.T) { + logger := zap.NewNop().Sugar() + + store, err := badger.NewBadgerStore(t.TempDir(), true, logger) + require.NoError(t, err) + + var worker v1.Worker + worker.Name = "worker-a" + worker.LastSeen = time.Now() + worker.MachineID = "machine-a" + worker.Resources = v1.Resources{v1.ResourceTartVMs: 2} + worker.Arch = v1.ArchitectureARM64 + worker.Runtime = v1.RuntimeTart + + newPendingVM := func(name string) v1.VM { + var vm v1.VM + vm.Name = name + vm.CreatedAt = time.Now() + vm.UID = name + "-uid" + vm.Status = v1.VMStatusPending + vm.Resources = v1.Resources{v1.ResourceTartVMs: 1} + vm.Arch = v1.ArchitectureARM64 + vm.Runtime = v1.RuntimeTart + vm.PowerState = v1.PowerStateRunning + vm.Conditions = []v1.Condition{{ + Type: v1.ConditionTypeScheduled, + State: v1.ConditionStateFalse, + }} + + return vm + } + + assignedVM := func(name string, status v1.VMStatus) v1.VM { + vm := newPendingVM(name) + vm.Worker = worker.Name + vm.Status = status + vm.Conditions[0].State = v1.ConditionStateTrue + + return vm + } + + pending := newPendingVM("pending-vm") + + var settings v1.ClusterSettings + settings.SchedulerProfile = v1.SchedulerProfileOptimizeUtilization + + err = store.Update(func(txn storepkg.Transaction) error { + if err := txn.SetClusterSettings(settings); err != nil { + return err + } + + if err := txn.SetWorker(worker); err != nil { + return err + } + + vms := []v1.VM{ + assignedVM("running-first", v1.VMStatusRunning), + assignedVM("running-second", v1.VMStatusRunning), + assignedVM("failed-third", v1.VMStatusFailed), + pending, + } + + for _, vm := range vms { + if err := txn.SetVM(vm); err != nil { + return err + } + } + + return nil + }) + require.NoError(t, err) + + scheduler, err := NewScheduler(store, notifier.NewNotifier(logger), time.Minute, logger) + require.NoError(t, err) + + numWorkers, numVMs, err := scheduler.schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 4, numVMs) + + err = store.View(func(txn storepkg.Transaction) error { + currentVM, err := txn.GetVM(pending.Name) + require.NoError(t, err) + require.False(t, currentVM.IsScheduled()) + require.Empty(t, currentVM.Worker) + + return nil + }) + require.NoError(t, err) +} diff --git a/pkg/resource/v1/resources.go b/pkg/resource/v1/resources.go index b0c3601..bad5609 100644 --- a/pkg/resource/v1/resources.go +++ b/pkg/resource/v1/resources.go @@ -62,16 +62,18 @@ func (resources Resources) Added(other Resources) Resources { func (resources Resources) Subtract(other Resources) { for otherKey, otherValue := range other { + if otherValue >= resources[otherKey] { + resources[otherKey] = 0 + continue + } + resources[otherKey] -= otherValue } } func (resources Resources) Subtracted(other Resources) Resources { result := resources.Copy() - - for otherKey, otherValue := range other { - result[otherKey] -= otherValue - } + result.Subtract(other) return result } diff --git a/pkg/resource/v1/resources_test.go b/pkg/resource/v1/resources_test.go index 03212ef..2b540dc 100644 --- a/pkg/resource/v1/resources_test.go +++ b/pkg/resource/v1/resources_test.go @@ -55,6 +55,50 @@ func TestResourcesSubtracted(t *testing.T) { })) } +func TestResourcesSubtractSaturatesAtZero(t *testing.T) { + resources := v1.Resources{ + v1.ResourceTartVMs: 2, + v1.ResourceLogicalCores: 8, + } + + resources.Subtract(v1.Resources{ + v1.ResourceTartVMs: 3, + v1.ResourceLogicalCores: 4, + v1.ResourceMemoryMiB: 1, + }) + + require.Equal(t, v1.Resources{ + v1.ResourceTartVMs: 0, + v1.ResourceLogicalCores: 4, + v1.ResourceMemoryMiB: 0, + }, resources) + require.False(t, resources.CanFit(v1.Resources{v1.ResourceTartVMs: 1})) +} + +func TestResourcesSubtractedSaturatesAtZeroWithoutMutatingOriginal(t *testing.T) { + resources := v1.Resources{ + v1.ResourceTartVMs: 2, + v1.ResourceLogicalCores: 8, + } + + remaining := resources.Subtracted(v1.Resources{ + v1.ResourceTartVMs: 3, + v1.ResourceLogicalCores: 4, + v1.ResourceMemoryMiB: 1, + }) + + require.Equal(t, v1.Resources{ + v1.ResourceTartVMs: 0, + v1.ResourceLogicalCores: 4, + v1.ResourceMemoryMiB: 0, + }, remaining) + require.Equal(t, v1.Resources{ + v1.ResourceTartVMs: 2, + v1.ResourceLogicalCores: 8, + }, resources) + require.False(t, remaining.CanFit(v1.Resources{v1.ResourceTartVMs: 1})) +} + func TestResourcesCanFit(t *testing.T) { resources := v1.Resources{ v1.ResourceTartVMs: 2,