diff --git a/api/openapi.yaml b/api/openapi.yaml index 030ea43..a1708a7 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -741,6 +741,23 @@ components: - "66.66.0.0/16" items: type: string + netSoftnetExpose: + type: array + description: | + TCP ports to expose when using Softnet isolation, each in the + `EXTERNAL:INTERNAL` format (see `tart run`'s `--net-softnet-expose`), + with traffic to the external port on the host forwarded to the + internal port on the VM. The scheduler does not place a VM on a + worker where another VM already holds one of its external ports or + draws them from an endpoint's port range, and the ports of a + scheduled VM cannot be changed at all, because its worker keeps them + bound until the VM stops. Pick external ports outside the host's + ephemeral range, which the operating system hands to whatever asks + for a port, endpoint listeners included. Enables `netSoftnet`. + example: + - "2222:22" + items: + type: string suspendable: type: boolean description: | diff --git a/internal/command/create/vm.go b/internal/command/create/vm.go index dc5022f..be60386 100644 --- a/internal/command/create/vm.go +++ b/internal/command/create/vm.go @@ -25,6 +25,7 @@ var diskSize uint64 var netSoftnet bool var netSoftnetAllow []string var netSoftnetBlock []string +var netSoftnetExpose []string var netBridged string var headless bool var nested bool @@ -69,6 +70,9 @@ func newCreateVMCommand() *cobra.Command { command.Flags().StringSliceVar(&netSoftnetBlock, "net-softnet-block", []string{}, "comma-separated list of CIDRs to block the traffic to when using Softnet isolation, see "+ "\"tart run\"'s help for \"--net-softnet-block\" for more details; automatically enables --net-softnet") + command.Flags().StringSliceVar(&netSoftnetExpose, "net-softnet-expose", []string{}, + "comma-separated list of TCP ports to expose when using Softnet isolation, in the EXTERNAL:INTERNAL "+ + "format, see \"tart run\"'s help for \"--net-softnet-expose\" for more details; automatically enables --net-softnet") command.Flags().StringVar(&netBridged, "net-bridged", "", "whether to use Bridged network mode") command.Flags().BoolVar(&headless, "headless", true, "whether to run without graphics") command.Flags().BoolVar(&nested, "nested", false, "enable nested virtualization") @@ -155,6 +159,7 @@ func runCreateVM(cmd *cobra.Command, args []string) error { NetSoftnet: netSoftnet, NetSoftnetAllow: netSoftnetAllow, NetSoftnetBlock: netSoftnetBlock, + NetSoftnetExpose: netSoftnetExpose, Suspendable: suspendable, }, NetBridged: netBridged, diff --git a/internal/command/get/vm.go b/internal/command/get/vm.go index dadfbb0..bb5c758 100644 --- a/internal/command/get/vm.go +++ b/internal/command/get/vm.go @@ -95,6 +95,7 @@ func runGetVM(cmd *cobra.Command, args []string) error { table.AddRow("Softnet enabled", vm.NetSoftnetDeprecated || vm.NetSoftnet) table.AddRow("Softnet allowed CIDRs", strings.Join(vm.NetSoftnetAllow, "\n")) table.AddRow("Softnet blocked CIDRs", strings.Join(vm.NetSoftnetBlock, "\n")) + table.AddRow("Softnet exposed ports", strings.Join(vm.NetSoftnetExpose, "\n")) table.AddRow("Bridged networking interface", nonEmptyOrNone(vm.NetBridged)) table.AddRow("Headless mode", vm.Headless) table.AddRow("Nested virtualization", vm.Nested) diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index 823b576..01c2e5b 100644 --- a/internal/controller/api_vms.go +++ b/internal/controller/api_vms.go @@ -6,11 +6,13 @@ import ( "encoding/json" "errors" "net/http" + "slices" "strconv" "strings" "time" "github.com/cirruslabs/orchard/internal/controller/lifecycle" + "github.com/cirruslabs/orchard/internal/controller/scheduler" storepkg "github.com/cirruslabs/orchard/internal/controller/store" "github.com/cirruslabs/orchard/internal/responder" "github.com/cirruslabs/orchard/internal/simplename" @@ -89,9 +91,10 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder { return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("%v", err)) } - // Softnet-specific logic: automatically enable Softnet when NetSoftnetAllow or NetSoftnetBlock are set - // and propagate deprecated and non-deprecated boolean fields into each other - if vm.NetSoftnetDeprecated || vm.NetSoftnet || len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 { + // Softnet-specific logic: automatically enable Softnet when NetSoftnetAllow, NetSoftnetBlock + // or NetSoftnetExpose are set and propagate deprecated and non-deprecated boolean fields into each other + if vm.NetSoftnetDeprecated || vm.NetSoftnet || len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 || + len(vm.NetSoftnetExpose) != 0 { vm.NetSoftnetDeprecated = true vm.NetSoftnet = true } @@ -222,9 +225,10 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid host processes: %v", err)) } - // Softnet-specific logic: automatically enable Softnet when NetSoftnetAllow or NetSoftnetBlock are set - // and propagate deprecated and non-deprecated boolean fields into each other - if userVM.NetSoftnetDeprecated || userVM.NetSoftnet || len(userVM.NetSoftnetAllow) != 0 || len(userVM.NetSoftnetBlock) != 0 { + // Softnet-specific logic: automatically enable Softnet when NetSoftnetAllow, NetSoftnetBlock + // or NetSoftnetExpose are set and propagate deprecated and non-deprecated boolean fields into each other + if userVM.NetSoftnetDeprecated || userVM.NetSoftnet || len(userVM.NetSoftnetAllow) != 0 || + len(userVM.NetSoftnetBlock) != 0 || len(userVM.NetSoftnetExpose) != 0 { userVM.NetSoftnetDeprecated = true userVM.NetSoftnet = true } @@ -258,6 +262,36 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder return responder.JSON(http.StatusOK, dbVM) } + // A worker keeps the old ports bound until it stops the VM to apply the new + // generation, so releasing them in the controller's view straight away would + // let another VM be scheduled onto a port that is still in use + if dbVM.IsScheduled() && !slices.Equal(dbVM.NetSoftnetExpose, userVM.NetSoftnetExpose) { + return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("\"netSoftnetExpose\" "+ + "cannot be changed once the VM is scheduled on worker %q", dbVM.Worker)) + } + + // Endpoints can be changed on a VM that is already scheduled, which never goes + // past the scheduler, so the worker it sits on is checked here instead + if dbVM.IsScheduled() && !v1.SemanticallyEqual(dbVM.Endpoints, userVM.Endpoints) { + vms, err := txn.ListVMs() + if err != nil { + controller.logger.Errorf("failed to list VMs in the DB: %v", err) + + return responder.Code(http.StatusInternalServerError) + } + + updatedVM := *dbVM + updatedVM.VMSpec = userVM.VMSpec + + // the whole specification has to come out clean: a VM being placed is not + // refused a worker for a clash it did not make, but this one owns its own + if scheduler.WorkerPortConflict(vms, dbVM.Worker, dbVM.Name, updatedVM) { + return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse( + "\"endpoints\" cannot take a worker port that another VM on worker %q holds", + dbVM.Worker)) + } + } + // VM specification was changed dbVM.VMSpec = userVM.VMSpec dbVM.Generation++ diff --git a/internal/controller/api_vms_test.go b/internal/controller/api_vms_test.go new file mode 100644 index 0000000..d3bd5ed --- /dev/null +++ b/internal/controller/api_vms_test.go @@ -0,0 +1,212 @@ +//nolint:testpackage // The handler under test is unexported. +package controller + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "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/gin-gonic/gin" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func newScheduledVM(name string, worker string, expose []string) v1.VM { + var vm v1.VM + vm.Name = name + vm.UID = name + "-uid" + vm.Worker = worker + vm.Status = v1.VMStatusRunning + vm.PowerState = v1.PowerStateRunning + vm.OS = v1.OSDarwin + vm.Arch = v1.ArchitectureARM64 + vm.Runtime = v1.RuntimeTart + vm.Image = "ghcr.io/cirruslabs/macos-sequoia-base:latest" + vm.NetSoftnetExpose = expose + vm.Conditions = []v1.Condition{{Type: v1.ConditionTypeScheduled, State: v1.ConditionStateTrue}} + + return vm +} + +func endpointOn(port uint16) []v1.EndpointSpec { + return []v1.EndpointSpec{ + { + Name: "ssh", + Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}, + WorkerPortRange: &v1.PortRange{Min: port, Max: port}, + }, + } +} + +// Endpoints may be added to a VM that is already scheduled, so the update is the only +// place where they can be held against the other VMs on its worker. +func TestUpdateVMSpecEndpointPortConflict(t *testing.T) { + for _, test := range []struct { + name string + otherExpose []string + port uint16 + status int + }{ + {name: "port another VM exposes", port: 2222, status: http.StatusPreconditionFailed}, + {name: "free port", port: 2224, status: http.StatusOK}, + // ports of its own that clash already are no licence to add an endpoint that does + {name: "clash on top of a clash", otherExpose: []string{"2222:22", "2223:22"}, + port: 2222, status: http.StatusPreconditionFailed}, + } { + t.Run(test.name, func(t *testing.T) { + store, err := badger.NewBadgerStore(t.TempDir(), true, zap.NewNop().Sugar()) + require.NoError(t, err) + + otherExpose := test.otherExpose + if otherExpose == nil { + otherExpose = []string{"2222:22"} + } + + other := newScheduledVM("other-vm", "worker-a", otherExpose) + updated := newScheduledVM("updated-vm", "worker-a", []string{"2223:22"}) + + require.NoError(t, store.Update(func(txn storepkg.Transaction) error { + if err := txn.SetVM(other); err != nil { + return err + } + + return txn.SetVM(updated) + })) + + logger := zap.NewNop().Sugar() + workerNotifier := notifier.NewNotifier(logger) + + _, unregister := workerNotifier.Register(t.Context(), updated.Worker) + defer unregister() + + controller := &Controller{ + insecureAuthDisabled: true, + store: store, + workerNotifier: workerNotifier, + logger: logger, + } + + body := updated + body.Endpoints = endpointOn(test.port) + + payload, err := json.Marshal(body) + require.NoError(t, err) + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Params = gin.Params{{Key: "name", Value: updated.Name}} + ctx.Request = httptest.NewRequest(http.MethodPut, "/v1/vms/"+updated.Name, bytes.NewReader(payload)) + ctx.Request.Header.Set("Content-Type", "application/json") + + controller.updateVMSpec(ctx).Respond(ctx) + require.Equal(t, test.status, recorder.Code, recorder.Body.String()) + + var dbVM *v1.VM + + require.NoError(t, store.View(func(txn storepkg.Transaction) (err error) { + dbVM, err = txn.GetVM(updated.Name) + + return + })) + + if test.status == http.StatusOK { + require.Equal(t, body.Endpoints, dbVM.Endpoints) + } else { + require.Empty(t, dbVM.Endpoints) + } + }) + } +} + +func TestUpdateVMSpecExposedPorts(t *testing.T) { + for _, test := range []struct { + name string + scheduled bool + expose []string + status int + }{ + // A scheduled VM keeps its ports bound on the worker until it is stopped to apply a new + // generation, so they cannot be changed, whether or not the new ones are free + {name: "conflicting port", scheduled: true, expose: []string{"02222:80"}, + status: http.StatusPreconditionFailed}, + {name: "free port", scheduled: true, expose: []string{"2224:22"}, status: http.StatusPreconditionFailed}, + {name: "unscheduled", scheduled: false, expose: []string{"2224:22"}, status: http.StatusOK}, + } { + t.Run(test.name, func(t *testing.T) { + store, err := badger.NewBadgerStore(t.TempDir(), true, zap.NewNop().Sugar()) + require.NoError(t, err) + + other := newScheduledVM("other-vm", "worker-a", []string{"2222:22"}) + updated := newScheduledVM("updated-vm", "worker-a", []string{"2223:22"}) + + if !test.scheduled { + updated.Worker = "" + updated.Status = v1.VMStatusPending + updated.Conditions = []v1.Condition{ + {Type: v1.ConditionTypeScheduled, State: v1.ConditionStateFalse}, + } + } + + require.NoError(t, store.Update(func(txn storepkg.Transaction) error { + if err := txn.SetVM(other); err != nil { + return err + } + + return txn.SetVM(updated) + })) + + logger := zap.NewNop().Sugar() + workerNotifier := notifier.NewNotifier(logger) + + // A successful update notifies the VM's worker; register it, otherwise the + // notification waits for the worker to connect before giving up + _, unregister := workerNotifier.Register(t.Context(), updated.Worker) + defer unregister() + + controller := &Controller{ + insecureAuthDisabled: true, + store: store, + workerNotifier: workerNotifier, + logger: logger, + } + + body := updated + body.NetSoftnetExpose = test.expose + + payload, err := json.Marshal(body) + require.NoError(t, err) + + recorder := httptest.NewRecorder() + ctx, _ := gin.CreateTestContext(recorder) + ctx.Params = gin.Params{{Key: "name", Value: updated.Name}} + ctx.Request = httptest.NewRequest(http.MethodPut, "/v1/vms/"+updated.Name, bytes.NewReader(payload)) + ctx.Request.Header.Set("Content-Type", "application/json") + + controller.updateVMSpec(ctx).Respond(ctx) + require.Equal(t, test.status, recorder.Code, recorder.Body.String()) + + var dbVM *v1.VM + + require.NoError(t, store.View(func(txn storepkg.Transaction) (err error) { + dbVM, err = txn.GetVM(updated.Name) + + return + })) + + if test.status == http.StatusOK { + require.Equal(t, test.expose, dbVM.NetSoftnetExpose) + require.Equal(t, uint64(1), dbVM.Generation) + } else { + require.Contains(t, recorder.Body.String(), "cannot be changed once the VM is scheduled") + require.Equal(t, []string{"2223:22"}, dbVM.NetSoftnetExpose) + require.Equal(t, uint64(0), dbVM.Generation) + } + }) + } +} diff --git a/internal/controller/scheduler/scheduler.go b/internal/controller/scheduler/scheduler.go index f8a958a..3a91f3b 100644 --- a/internal/controller/scheduler/scheduler.go +++ b/internal/controller/scheduler/scheduler.go @@ -275,7 +275,8 @@ NextVM: worker.SchedulingPaused || !compatibleArchAndRuntime(unscheduledVM, worker) || !resourcesRemaining.CanFit(unscheduledVM.Resources) || - !worker.Labels.Contains(unscheduledVM.Labels) { + !worker.Labels.Contains(unscheduledVM.Labels) || + workerInfos.PortConflict(worker.Name, unscheduledVM) { continue NextWorker } @@ -285,6 +286,13 @@ NextVM: continue NextWorker } + // An older worker drops the exposed ports it knows nothing about and + // starts the VM without them, so wait for one that can honour them + if len(unscheduledVM.NetSoftnetExpose) != 0 && + !worker.Capabilities.Has(v1.WorkerCapabilityVMExposedPorts) { + continue NextWorker + } + err := scheduler.store.Update(func(txn storepkg.Transaction) error { currentUnscheduledVM, err := txn.GetVM(unscheduledVM.Name) if err != nil { @@ -346,12 +354,31 @@ NextVM: return ErrWorkerSchedulingSkipped } + if len(unscheduledVM.NetSoftnetExpose) != 0 && + !currentWorker.Capabilities.Has(v1.WorkerCapabilityVMExposedPorts) { + return ErrWorkerSchedulingSkipped + } + if currentWorker.MachineID != worker.MachineID || !currentWorker.Resources.Equal(worker.Resources) { // Worker has changed return ErrWorkerSchedulingSkipped } + if len(unscheduledVM.NetSoftnetExpose) != 0 || len(unscheduledVM.Endpoints) != 0 { + // A specification update may have exposed one of the ports + // on this worker since the lagging view was built, so repeat + // the check on the VMs as this transaction sees them + currentVMs, err := txn.ListVMs() + if err != nil { + return err + } + + if WorkerPortConflict(currentVMs, worker.Name, unscheduledVM.Name, unscheduledVM) { + return ErrWorkerSchedulingSkipped + } + } + unscheduledVM.Worker = worker.Name unscheduledVM.ScheduledAt = time.Now() v1.ConditionsSet(&unscheduledVM.Conditions, v1.Condition{ @@ -402,7 +429,7 @@ NextVM: } // Update lagging resource usage - workerInfos.AddVM(worker.Name, unscheduledVM.Resources) + workerInfos.AddVM(worker.Name, unscheduledVM) // Ping the worker afterward for faster VM execution affectedWorkers.Add(worker.Name) @@ -437,7 +464,7 @@ func ProcessVMs(vms []v1.VM) ([]v1.VM, WorkerInfos) { for _, vm := range vms { if vm.IsScheduled() { - workerToResources.AddVM(vm.Worker, vm.Resources) + workerToResources.AddVM(vm.Worker, vm) } else { unscheduledVMs = append(unscheduledVMs, vm) } diff --git a/internal/controller/scheduler/scheduler_capacity_test.go b/internal/controller/scheduler/scheduler_capacity_test.go index 0bcf6f4..3d46943 100644 --- a/internal/controller/scheduler/scheduler_capacity_test.go +++ b/internal/controller/scheduler/scheduler_capacity_test.go @@ -13,66 +13,65 @@ import ( "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) - +func newTestWorker(name string) v1.Worker { var worker v1.Worker - worker.Name = "worker-a" + worker.Name = name worker.LastSeen = time.Now() - worker.MachineID = "machine-a" + worker.MachineID = name + "-machine" 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 + worker.Capabilities = v1.WorkerCapabilities{ + v1.WorkerCapabilityVMEndpoints, + v1.WorkerCapabilityVMExposedPorts, } - 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 worker +} - return vm - } +func newPendingVM(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, + }} - pending := newPendingVM("pending-vm") + return vm +} + +func newAssignedVM(name string, workerName string, status v1.VMStatus) v1.VM { + vm := newPendingVM(name) + vm.Worker = workerName + vm.Status = status + vm.Conditions[0].State = v1.ConditionStateTrue + + return vm +} + +func newTestStore(t *testing.T, profile v1.SchedulerProfile, workers []v1.Worker, vms []v1.VM) storepkg.Store { + store, err := badger.NewBadgerStore(t.TempDir(), true, zap.NewNop().Sugar()) + require.NoError(t, err) var settings v1.ClusterSettings - settings.SchedulerProfile = v1.SchedulerProfileOptimizeUtilization + settings.SchedulerProfile = profile 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 _, worker := range workers { + if err := txn.SetWorker(worker); err != nil { + return err + } } for _, vm := range vms { @@ -85,21 +84,301 @@ func TestSchedulingLoopSkipsOvercommittedWorker(t *testing.T) { }) require.NoError(t, err) - scheduler, err := NewScheduler(store, notifier.NewNotifier(logger), time.Minute, logger) + return store +} + +func newTestScheduler(t *testing.T, store storepkg.Store) *Scheduler { + logger := zap.NewNop().Sugar() + workerNotifier := notifier.NewNotifier(logger) + + // Register the workers with the notifier, otherwise each placement + // waits a second for its worker to connect before giving up + var workers []v1.Worker + + err := store.View(func(txn storepkg.Transaction) (err error) { + workers, err = txn.ListWorkers() + + return + }) require.NoError(t, err) - numWorkers, numVMs, err := scheduler.schedulingLoopIteration() + for _, worker := range workers { + instructionCh, cancel := workerNotifier.Register(t.Context(), worker.Name) + t.Cleanup(cancel) + + go func() { + for { + select { + case <-instructionCh: + case <-t.Context().Done(): + return + } + } + }() + } + + scheduler, err := NewScheduler(store, workerNotifier, time.Minute, logger) + require.NoError(t, err) + + return scheduler +} + +func getVM(t *testing.T, store storepkg.Store, name string) v1.VM { + var vm *v1.VM + + err := store.View(func(txn storepkg.Transaction) (err error) { + vm, err = txn.GetVM(name) + + return + }) + require.NoError(t, err) + + return *vm +} + +func TestSchedulingLoopSkipsOvercommittedWorker(t *testing.T) { + worker := newTestWorker("worker-a") + pending := newPendingVM("pending-vm") + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, []v1.VM{ + newAssignedVM("running-first", worker.Name, v1.VMStatusRunning), + newAssignedVM("running-second", worker.Name, v1.VMStatusRunning), + newAssignedVM("failed-third", worker.Name, v1.VMStatusFailed), + pending, + }) + + numWorkers, numVMs, err := newTestScheduler(t, store).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) + currentVM := getVM(t, store, pending.Name) + require.False(t, currentVM.IsScheduled()) + require.Empty(t, currentVM.Worker) +} + +func TestSchedulingLoopSkipsWorkerWithExposedPortConflict(t *testing.T) { + worker := newTestWorker("worker-a") + // Enough capacity for all VMs, so that an exposed + // port conflict is the only thing preventing scheduling + worker.Resources = v1.Resources{v1.ResourceTartVMs: 3} + + running := newAssignedVM("running-vm", worker.Name, v1.VMStatusRunning) + running.NetSoftnetExpose = []string{"2222:22"} + + conflicting := newPendingVM("conflicting-vm") + conflicting.NetSoftnetExpose = []string{"2222:80"} + + nonConflicting := newPendingVM("non-conflicting-vm") + nonConflicting.NetSoftnetExpose = []string{"2223:22"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, + []v1.VM{running, conflicting, nonConflicting}) + + numWorkers, numVMs, err := newTestScheduler(t, store).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 3, numVMs) + + currentConflicting := getVM(t, store, conflicting.Name) + require.False(t, currentConflicting.IsScheduled()) + require.Empty(t, currentConflicting.Worker) + + currentNonConflicting := getVM(t, store, nonConflicting.Name) + require.True(t, currentNonConflicting.IsScheduled()) + require.Equal(t, worker.Name, currentNonConflicting.Worker) +} + +func TestSchedulingLoopSkipsWorkerWithoutExposedPortSupport(t *testing.T) { + stale := newTestWorker("worker-a") + // A worker from before the field existed advertises everything but the ports + stale.Capabilities = v1.WorkerCapabilities{v1.WorkerCapabilityVMEndpoints} + + capable := newTestWorker("worker-b") + + pending := newPendingVM("pending-vm") + pending.NetSoftnetExpose = []string{"2222:22"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, + []v1.Worker{stale, capable}, []v1.VM{pending}) + + numWorkers, numVMs, err := newTestScheduler(t, store).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 2, numWorkers) + require.Equal(t, 1, numVMs) + + currentPending := getVM(t, store, pending.Name) + require.True(t, currentPending.IsScheduled()) + require.Equal(t, capable.Name, currentPending.Worker) +} + +func TestSchedulingLoopPlacesVMWithExposedPortConflictOnAnotherWorker(t *testing.T) { + for _, profile := range []v1.SchedulerProfile{ + v1.SchedulerProfileOptimizeUtilization, + v1.SchedulerProfileDistributeLoad, + } { + t.Run(string(profile), func(t *testing.T) { + workerA := newTestWorker("worker-a") + workerB := newTestWorker("worker-b") + + running := newAssignedVM("running-vm", workerA.Name, v1.VMStatusRunning) + running.NetSoftnetExpose = []string{"2222:22"} + + pending := newPendingVM("pending-vm") + pending.NetSoftnetExpose = []string{"2222:80"} + + store := newTestStore(t, profile, []v1.Worker{workerA, workerB}, []v1.VM{running, pending}) + + numWorkers, numVMs, err := newTestScheduler(t, store).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 2, numWorkers) + require.Equal(t, 2, numVMs) + + currentPending := getVM(t, store, pending.Name) + require.True(t, currentPending.IsScheduled()) + require.Equal(t, workerB.Name, currentPending.Worker) + }) + } +} + +func TestSchedulingLoopSchedulesOlderVMWithSharedExposedPortFirst(t *testing.T) { + worker := newTestWorker("worker-a") + + // The store lists VMs by name, which puts the newer VM first, + // so the scheduler has to order them by creation time itself + older := newPendingVM("older-vm") + older.NetSoftnetExpose = []string{"2222:22"} + + newer := newPendingVM("newer-vm") + newer.CreatedAt = older.CreatedAt.Add(time.Second) + newer.NetSoftnetExpose = []string{"2222:22"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, []v1.VM{newer, older}) + + numWorkers, numVMs, err := newTestScheduler(t, store).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 2, numVMs) + + currentOlder := getVM(t, store, older.Name) + require.True(t, currentOlder.IsScheduled()) + require.Equal(t, worker.Name, currentOlder.Worker) + + currentNewer := getVM(t, store, newer.Name) + require.False(t, currentNewer.IsScheduled()) + require.Empty(t, currentNewer.Worker) +} + +func TestSchedulingLoopFailedScheduledVMHoldsExposedPort(t *testing.T) { + worker := newTestWorker("worker-a") + + // Failed, but not yet de-scheduled by the health-checking loop + failed := newAssignedVM("failed-vm", worker.Name, v1.VMStatusFailed) + failed.NetSoftnetExpose = []string{"2222:22"} + + pending := newPendingVM("pending-vm") + pending.NetSoftnetExpose = []string{"2222:80"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, []v1.VM{failed, pending}) + + numWorkers, numVMs, err := newTestScheduler(t, store).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 2, numVMs) + + currentPending := getVM(t, store, pending.Name) + require.False(t, currentPending.IsScheduled()) + require.Empty(t, currentPending.Worker) +} + +// hookedStore runs a callback before the first Update, standing in for +// a specification update that lands between the scheduler's snapshot of +// the VMs and its scheduling transaction. +type hookedStore struct { + storepkg.Store + beforeUpdate func() +} + +func (store *hookedStore) Update(cb func(txn storepkg.Transaction) error) error { + if store.beforeUpdate != nil { + beforeUpdate := store.beforeUpdate + store.beforeUpdate = nil + beforeUpdate() + } + + return store.Store.Update(cb) +} + +func TestSchedulingLoopRechecksExposedPortInTransaction(t *testing.T) { + worker := newTestWorker("worker-a") + worker.Resources = v1.Resources{v1.ResourceTartVMs: 3} + + running := newAssignedVM("running-vm", worker.Name, v1.VMStatusRunning) + + pending := newPendingVM("pending-vm") + pending.NetSoftnetExpose = []string{"2222:22"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, + []v1.VM{running, pending}) + + hooked := &hookedStore{Store: store} + hooked.beforeUpdate = func() { + // Expose the pending VM's port on the worker after the scheduler took its snapshot + err := store.Update(func(txn storepkg.Transaction) error { + vm, err := txn.GetVM(running.Name) + if err != nil { + return err + } + + vm.NetSoftnetExpose = []string{"2222:80"} + + return txn.SetVM(*vm) + }) require.NoError(t, err) - require.False(t, currentVM.IsScheduled()) - require.Empty(t, currentVM.Worker) + } - return nil - }) + numWorkers, numVMs, err := newTestScheduler(t, hooked).schedulingLoopIteration() require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 2, numVMs) + + currentPending := getVM(t, store, pending.Name) + require.False(t, currentPending.IsScheduled()) + require.Empty(t, currentPending.Worker) +} + +func TestSchedulingLoopSkipsVMUpdatedSinceSnapshot(t *testing.T) { + worker := newTestWorker("worker-a") + + pending := newPendingVM("pending-vm") + pending.NetSoftnetExpose = []string{"2222:22"} + + store := newTestStore(t, v1.SchedulerProfileOptimizeUtilization, []v1.Worker{worker}, []v1.VM{pending}) + + hooked := &hookedStore{Store: store} + hooked.beforeUpdate = func() { + // Change the pending VM's specification after the scheduler took its snapshot + err := store.Update(func(txn storepkg.Transaction) error { + vm, err := txn.GetVM(pending.Name) + if err != nil { + return err + } + + vm.NetSoftnetExpose = []string{"2223:22"} + vm.Generation++ + + return txn.SetVM(*vm) + }) + require.NoError(t, err) + } + + numWorkers, numVMs, err := newTestScheduler(t, hooked).schedulingLoopIteration() + require.NoError(t, err) + require.Equal(t, 1, numWorkers) + require.Equal(t, 1, numVMs) + + // The update survived and the VM waits for the next iteration + currentPending := getVM(t, store, pending.Name) + require.False(t, currentPending.IsScheduled()) + require.Equal(t, []string{"2223:22"}, currentPending.NetSoftnetExpose) } diff --git a/internal/controller/scheduler/workerinfo.go b/internal/controller/scheduler/workerinfo.go index 00c608a..e3887a3 100644 --- a/internal/controller/scheduler/workerinfo.go +++ b/internal/controller/scheduler/workerinfo.go @@ -1,37 +1,301 @@ package scheduler -import v1 "github.com/cirruslabs/orchard/pkg/resource/v1" +import ( + "cmp" + "maps" + "slices" + + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" +) type WorkerInfo struct { ResourcesUsed v1.Resources NumRunningVMs int + // UsedPorts is a set of worker ports the VMs scheduled on this worker hold + // outright, through Softnet or through an endpoint pinned to a single port. + UsedPorts map[uint16]struct{} + // ListenerClaims is what the endpoints drawing from a range of ports lay claim to. + ListenerClaims []ListenerClaim +} + +// ListenerClaim is the port range of one endpoint listener, together with the port it +// holds today, zero when it holds none. A listener scans its range from the start every +// time the worker creates it, so the range stays with it either way; the port only says +// which one of the range it is on for now, and that it needs no second one. +type ListenerClaim struct { + Range v1.PortRange + Port uint16 } type WorkerInfos map[string]WorkerInfo -func (workerInfos WorkerInfos) AddVM(name string, resourcesUsed v1.Resources) { +func newWorkerInfo() WorkerInfo { + return WorkerInfo{ + ResourcesUsed: v1.Resources{}, + UsedPorts: map[uint16]struct{}{}, + } +} + +func (workerInfos WorkerInfos) AddVM(name string, vm v1.VM) { workerInfo, ok := workerInfos[name] if !ok { - workerInfo = WorkerInfo{ - ResourcesUsed: v1.Resources{}, - } + workerInfo = newWorkerInfo() } - workerInfo.ResourcesUsed.Add(resourcesUsed) + workerInfo.ResourcesUsed.Add(vm.Resources) workerInfo.NumRunningVMs++ + for _, port := range vmPorts(vm) { + workerInfo.UsedPorts[port] = struct{}{} + } + + workerInfo.ListenerClaims = append(workerInfo.ListenerClaims, listenerClaims(vm)...) + workerInfos[name] = workerInfo } func (workerInfos WorkerInfos) Get(name string) WorkerInfo { workerInfo, ok := workerInfos[name] if !ok { - workerInfo = WorkerInfo{ - ResourcesUsed: v1.Resources{}, - } + workerInfo = newWorkerInfo() workerInfos[name] = workerInfo } return workerInfo } + +// PortConflict reports whether the given VM clashes with the VMs already scheduled on +// the given worker: a worker port it holds or pins is taken, or lies in a range an +// endpoint listener draws from, or the listeners, its own and those of the VMs placed +// before it, cannot be given a port each. +func (workerInfos WorkerInfos) PortConflict(name string, vm v1.VM) bool { + workerInfo := workerInfos[name] + ports := vmPorts(vm) + + for _, port := range ports { + if _, ok := workerInfo.UsedPorts[port]; ok { + return true + } + + // a listener takes the lowest port of its range it manages to bind, again from + // the start every time the worker creates it anew, and never learns which port + // the scheduler had in mind, so no port of a range is one this VM can hold + for _, claim := range workerInfo.ListenerClaims { + if rangeContains(claim.Range, port) { + return true + } + } + } + + if !workerInfo.exhaustsListenerClaims(listenerClaims(vm), ports) { + return false + } + + // the listeners already on the worker can outnumber the ports of their ranges, + // which endpoints added to a VM that is already scheduled do without passing here + // at all; that is not this VM's doing and not something it mends by staying away + return !workerInfo.exhaustsListenerClaims(nil, nil) +} + +// exhaustsListenerClaims reports whether the endpoint listeners of this worker, +// together with those of a VM about to join it, are left without a port each once that +// VM takes the given ones. A listener that is on a port keeps it; the rest have to be +// found one, and since one range can hold the port another needs, they are matched +// rather than weighed one by one: serving the range that ends first finds a port for +// every one of them whenever the ports allow it. The question is whether the worker has +// the ports, not which one each listener takes, as two listeners settle between +// themselves: the one that loses a bind moves on. +func (workerInfo WorkerInfo) exhaustsListenerClaims(joining []ListenerClaim, ports []uint16) bool { + claims := slices.Concat(joining, workerInfo.ListenerClaims) + if len(claims) == 0 { + return false + } + + takenPorts := make(map[uint16]struct{}, len(workerInfo.UsedPorts)+len(ports)+len(claims)) + maps.Copy(takenPorts, workerInfo.UsedPorts) + + for _, port := range ports { + takenPorts[port] = struct{}{} + } + + var waiting []v1.PortRange + + for _, claim := range claims { + if claim.Port == 0 { + waiting = append(waiting, claim.Range) + + continue + } + + takenPorts[claim.Port] = struct{}{} + } + + if len(waiting) == 0 { + return false + } + + // a range wider than every taken port and every other range together cannot be + // left empty by them, which is the ordinary case and needs no matching at all + narrowest := slices.MinFunc(waiting, func(a, b v1.PortRange) int { + return cmp.Compare(a.Max-a.Min, b.Max-b.Min) + }) + if int(narrowest.Max)-int(narrowest.Min)+1 > len(takenPorts)+len(waiting) { + return false + } + + slices.SortFunc(waiting, func(a, b v1.PortRange) int { + return cmp.Compare(a.Max, b.Max) + }) + + for _, portRange := range waiting { + port, ok := portRange.FreePort(takenPorts) + if !ok { + return true + } + + takenPorts[port] = struct{}{} + } + + return false +} + +// WorkerPortConflict reports whether any worker port the given VM needs is +// already taken by a VM from vms that is scheduled on the given worker, +// ignoring the VM with the given name. +// +// This is the same check that the scheduling loop performs through the +// WorkerInfos it builds with ProcessVMs, so that the pre-filter and the +// re-check inside the scheduling transaction agree on conflicts. +func WorkerPortConflict(vms []v1.VM, workerName, ignoredVMName string, vm v1.VM) bool { + var otherVMs []v1.VM + + for _, other := range vms { + if other.Worker == workerName && other.Name != ignoredVMName { + otherVMs = append(otherVMs, other) + } + } + + _, workerInfos := ProcessVMs(otherVMs) + + return workerInfos.PortConflict(workerName, vm) +} + +// vmPorts returns the worker ports a VM holds outright, or asks for when it is not +// scheduled yet: the external ports it exposes through Softnet, the ports single-port +// ranges pin its endpoints to, and the port a listener is on outside the range its +// endpoint now asks for, which the worker keeps until it applies the new generation. A +// listener on a port of the range it draws from is left to listenerClaims, which counts +// it once, as a range with a port rather than a port and a range. +func vmPorts(vm v1.VM) []uint16 { + seen := map[uint16]struct{}{} + + var result []uint16 + + add := func(port uint16) { + if port == 0 { + return + } + + if _, ok := seen[port]; ok { + return + } + + seen[port] = struct{}{} + result = append(result, port) + } + + for _, port := range externalPorts(vm.NetSoftnetExpose) { + add(port) + } + + for _, observed := range vm.ObservedEndpoints { + portRange := endpointRange(vm, observed.Name) + + if portRange != nil && portRange.Min != portRange.Max && + rangeContains(*portRange, observed.WorkerPort) { + continue + } + + add(observed.WorkerPort) + } + + for _, endpoint := range vm.Endpoints { + if portRange := endpoint.WorkerPortRange; portRange != nil && portRange.Min == portRange.Max { + add(portRange.Min) + } + } + + return result +} + +// listenerClaims returns what the VM's endpoints lay claim to. The port a listener is +// on is no substitute for its range: the worker tears every listener of a VM down when +// its endpoints change or the VM restarts, and the one it creates in their place scans +// the range again from the start. A range narrowed down to a single port is left out, +// since vmPorts holds that port outright. +func listenerClaims(vm v1.VM) []ListenerClaim { + var result []ListenerClaim + + for _, endpoint := range vm.Endpoints { + portRange := endpoint.WorkerPortRange + + if portRange == nil || portRange.Min == portRange.Max { + continue + } + + claim := ListenerClaim{Range: *portRange} + + if port, ok := observedPort(vm, endpoint.Name); ok && rangeContains(*portRange, port) { + claim.Port = port + } + + result = append(result, claim) + } + + return result +} + +// endpointRange returns the range the VM's endpoint of the given name draws from. +func endpointRange(vm v1.VM, name string) *v1.PortRange { + for _, endpoint := range vm.Endpoints { + if endpoint.Name == name { + return endpoint.WorkerPortRange + } + } + + return nil +} + +// observedPort returns the port the VM's endpoint of the given name is on. +func observedPort(vm v1.VM, name string) (uint16, bool) { + for _, observed := range vm.ObservedEndpoints { + if observed.Name == name && observed.WorkerPort != 0 { + return observed.WorkerPort, true + } + } + + return 0, false +} + +func rangeContains(portRange v1.PortRange, port uint16) bool { + return port >= portRange.Min && port <= portRange.Max +} + +// externalPorts extracts the external ports from a list of +// EXTERNAL:INTERNAL entries, skipping the ones that fail to parse +// (the API validates them, so this only guards data written by other means). +func externalPorts(netSoftnetExpose []string) []uint16 { + var result []uint16 + + for _, entry := range netSoftnetExpose { + exposedPort, err := v1.NewExposedPortFromString(entry) + if err != nil { + continue + } + + result = append(result, exposedPort.External) + } + + return result +} diff --git a/internal/controller/scheduler/workerinfo_test.go b/internal/controller/scheduler/workerinfo_test.go index 4539f1f..4a03554 100644 --- a/internal/controller/scheduler/workerinfo_test.go +++ b/internal/controller/scheduler/workerinfo_test.go @@ -11,19 +11,292 @@ func TestWorkerInfos(t *testing.T) { workerInfos := make(scheduler.WorkerInfos) require.Len(t, workerInfos, 0) - workerInfos.AddVM("worker-name", v1.Resources{ + var firstVM v1.VM + firstVM.Resources = v1.Resources{ "tart-vms": 1, - }) + } + firstVM.NetSoftnetExpose = []string{"2222:22"} + + workerInfos.AddVM("worker-name", firstVM) require.Len(t, workerInfos, 1) - workerInfos.AddVM("worker-name", v1.Resources{ + var secondVM v1.VM + secondVM.Resources = v1.Resources{ "tart-vms": 1, - }) + } + secondVM.NetSoftnetExpose = []string{"2223:22", "08080:80", "bogus"} + + workerInfos.AddVM("worker-name", secondVM) require.Len(t, workerInfos, 1) require.Equal(t, scheduler.WorkerInfo{ ResourcesUsed: map[string]uint64{ "tart-vms": 2, }, NumRunningVMs: 2, + UsedPorts: map[uint16]struct{}{ + 2222: {}, + 2223: {}, + 8080: {}, + }, }, workerInfos.Get("worker-name")) + + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2222:80"))) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2224:22", "8080:80"))) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("002222:80"))) + require.False(t, workerInfos.PortConflict("worker-name", exposingVM("2224:22", "22", ""))) + require.False(t, workerInfos.PortConflict("worker-name", v1.VM{})) + require.False(t, workerInfos.PortConflict("other-worker-name", exposingVM("2222:22"))) +} + +func exposingVM(netSoftnetExpose ...string) v1.VM { + var vm v1.VM + vm.NetSoftnetExpose = netSoftnetExpose + + return vm +} + +func endpointVM(name string, min uint16, max uint16) v1.VM { + var vm v1.VM + vm.Endpoints = []v1.EndpointSpec{ + { + Name: name, + Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}, + WorkerPortRange: &v1.PortRange{Min: min, Max: max}, + }, + } + + return vm +} + +// Endpoint listeners bind worker ports too, so a Softnet-exposing VM must not be +// scheduled onto a port an endpoint already holds, or is pinned to by a single-port range. +func TestWorkerInfosEndpointPorts(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + var boundVM v1.VM + boundVM.Endpoints = []v1.EndpointSpec{ + {Name: "ssh", Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}, + WorkerPortRange: &v1.PortRange{Min: 2200, Max: 2299}}, + } + boundVM.ObservedEndpoints = []v1.EndpointStatus{ + {Name: "ssh", WorkerPort: 2240, State: v1.EndpointStateListening}, + } + + var pinnedVM v1.VM + pinnedVM.Endpoints = []v1.EndpointSpec{ + // not bound yet, but the range leaves the worker no choice + {Name: "ssh", Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}, + WorkerPortRange: &v1.PortRange{Min: 2245, Max: 2245}}, + // a wider range picks a port only at bind time, so nothing can be reserved for it + {Name: "http", Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 80}}, + WorkerPortRange: &v1.PortRange{Min: 2250, Max: 2260}}, + } + + workerInfos.AddVM("worker-name", boundVM) + workerInfos.AddVM("worker-name", pinnedVM) + + // the listener on 2240 is counted by its claim, not as a port of its own + require.Equal(t, map[uint16]struct{}{2245: {}}, workerInfos.Get("worker-name").UsedPorts) + + // a Softnet VM must not be placed on a port an endpoint holds, and neither must a VM + // whose own endpoint is pinned to it + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2240:22"))) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2245:22"))) + require.True(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2240, 2240))) + // the ports of the range the other endpoint is still waiting for are spoken for too + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2250:22"))) + require.False(t, workerInfos.PortConflict("worker-name", exposingVM("2300:22"))) + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2250, 2260))) +} + +// A listener binds any port of its range, so only a worker that leaves the range no +// port at all is a conflict. +func TestWorkerInfosEndpointRangeTaken(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + workerInfos.AddVM("worker-name", exposingVM("2222:22", "2223:23")) + + require.True(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2222, 2223))) + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2222, 2224))) + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2224, 2225))) +} + +// A listener scans its range from the start every time the worker creates it, so the +// whole range stays with the listeners whether one has bound or not. +func TestWorkerInfosEndpointRangeIsReserved(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + boundVM := endpointVM("ssh", 2222, 2224) + boundVM.ObservedEndpoints = []v1.EndpointStatus{ + {Name: "ssh", WorkerPort: 2224, State: v1.EndpointStateListening}, + } + + workerInfos.AddVM("worker-name", boundVM) + + require.Equal(t, []scheduler.ListenerClaim{{Range: v1.PortRange{Min: 2222, Max: 2224}, Port: 2224}}, + workerInfos.Get("worker-name").ListenerClaims) + + // the port the listener holds now, and the ones it would scan again after a + // specification change or a restart + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2224:22"))) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2222:22"))) + require.False(t, workerInfos.PortConflict("worker-name", exposingVM("2225:22"))) + + // another listener can still join, since the two settle between themselves + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("http", 2222, 2224))) +} + +// A listener that is on a port of its range needs no second one, so an endpoint with +// the same range still fits beside it. +func TestWorkerInfosEndpointRangeCountsOnce(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + boundVM := endpointVM("ssh", 2222, 2223) + boundVM.ObservedEndpoints = []v1.EndpointStatus{ + {Name: "ssh", WorkerPort: 2222, State: v1.EndpointStateListening}, + } + + workerInfos.AddVM("worker-name", boundVM) + + require.Empty(t, workerInfos.Get("worker-name").UsedPorts) + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("http", 2222, 2223))) + + // the two of them fill the range, and a third has nowhere to go + workerInfos.AddVM("worker-name", endpointVM("http", 2222, 2223)) + require.True(t, workerInfos.PortConflict("worker-name", endpointVM("other", 2222, 2223))) +} + +// Listeners already on a worker can outnumber the ports of their ranges, since +// endpoints added to a scheduled VM never pass the scheduler. The VMs that come after +// them are not the cause and are let through. +func TestWorkerInfosEndpointRangesAlreadyExhausted(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + workerInfos.AddVM("worker-name", endpointVM("ssh", 2222, 2223)) + workerInfos.AddVM("worker-name", endpointVM("ssh", 2222, 2223)) + workerInfos.AddVM("worker-name", endpointVM("ssh", 2222, 2223)) + + require.False(t, workerInfos.PortConflict("worker-name", v1.VM{})) + require.False(t, workerInfos.PortConflict("worker-name", exposingVM("2224:22"))) + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("http", 2222, 2223))) + + // the ports those listeners draw from are still theirs + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2223:22"))) +} + +// Endpoints need a port each, and the range that ends first has to be +// served first, or a wider one takes the single port it could have had. +func TestWorkerInfosEndpointRangesCompete(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + workerInfos.AddVM("worker-name", endpointVM("ssh", 2228, 2229)) + workerInfos.AddVM("worker-name", endpointVM("ssh", 2229, 2230)) + + // the three of them fit in 2228-2230 + require.False(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2228, 2230))) + + // a third listener waiting leaves the fourth nowhere to go + workerInfos.AddVM("worker-name", endpointVM("ssh", 2228, 2230)) + require.True(t, workerInfos.PortConflict("worker-name", endpointVM("ssh", 2229, 2230))) +} + +// An endpoint moved from one fixed port to another keeps the port it bound until the +// worker applies the new generation, so both are taken meanwhile. +func TestWorkerInfosEndpointUpdatePending(t *testing.T) { + workerInfos := make(scheduler.WorkerInfos) + + movedVM := endpointVM("ssh", 3333, 3333) + movedVM.ObservedEndpoints = []v1.EndpointStatus{ + {Name: "ssh", WorkerPort: 2222, State: v1.EndpointStateListening}, + } + + workerInfos.AddVM("worker-name", movedVM) + + require.Equal(t, map[uint16]struct{}{2222: {}, 3333: {}}, workerInfos.Get("worker-name").UsedPorts) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("2222:22"))) + require.True(t, workerInfos.PortConflict("worker-name", exposingVM("3333:22"))) +} + +func TestWorkerPortConflict(t *testing.T) { + newVM := func(name string, worker string, scheduled v1.ConditionState, netSoftnetExpose ...string) v1.VM { + var vm v1.VM + vm.Name = name + vm.Worker = worker + vm.NetSoftnetExpose = netSoftnetExpose + vm.Conditions = []v1.Condition{{ + Type: v1.ConditionTypeScheduled, + State: scheduled, + }} + + return vm + } + + vms := []v1.VM{ + newVM("first", "worker-a", v1.ConditionStateTrue, "2222:22", "8080:80"), + newVM("second", "worker-b", v1.ConditionStateTrue, "2223:22"), + // De-scheduled after a failure, so its port is free again + newVM("third", "worker-a", v1.ConditionStateFalse, "9090:90"), + } + + for _, test := range []struct { + name string + worker string + ignoredVM string + netSoftnetExpose []string + conflict bool + }{ + { + name: "port held by another VM on the worker", + worker: "worker-a", + ignoredVM: "new", + netSoftnetExpose: []string{"2222:80"}, + conflict: true, + }, + { + name: "one of the ports held by another VM on the worker", + worker: "worker-a", + ignoredVM: "new", + netSoftnetExpose: []string{"2224:22", "8080:80"}, + conflict: true, + }, + { + name: "ports held by the ignored VM itself", + worker: "worker-a", + ignoredVM: "first", + netSoftnetExpose: []string{"2222:80", "8080:80"}, + conflict: false, + }, + { + name: "port held on another worker", + worker: "worker-a", + ignoredVM: "new", + netSoftnetExpose: []string{"2223:22"}, + conflict: false, + }, + { + name: "port held by a de-scheduled VM", + worker: "worker-a", + ignoredVM: "new", + netSoftnetExpose: []string{"9090:90"}, + conflict: false, + }, + { + name: "nothing exposed", + worker: "worker-a", + ignoredVM: "new", + conflict: false, + }, + { + name: "unknown worker", + worker: "worker-c", + ignoredVM: "new", + netSoftnetExpose: []string{"2222:22"}, + conflict: false, + }, + } { + t.Run(test.name, func(t *testing.T) { + require.Equal(t, test.conflict, scheduler.WorkerPortConflict(vms, test.worker, test.ignoredVM, + exposingVM(test.netSoftnetExpose...))) + }) + } } diff --git a/internal/tests/spec_update_test.go b/internal/tests/spec_update_test.go index 6de28cb..97b507f 100644 --- a/internal/tests/spec_update_test.go +++ b/internal/tests/spec_update_test.go @@ -119,7 +119,7 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { []worker.Option{worker.WithSoftnetPolicyUpdates(false)}, ) - // Create a suspendable VM with Softnet enabled + // Create a suspendable VM with Softnet enabled and two ports exposed vmName := "test" err := devClient.VMs().Create(t.Context(), &v1.VM{ @@ -131,8 +131,9 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { Memory: 8 * 1024, Headless: true, VMSpec: v1.VMSpec{ - Suspendable: true, - NetSoftnet: true, + Suspendable: true, + NetSoftnet: true, + NetSoftnetExpose: []string{"2222:22", "8080:80"}, }, }) require.NoError(t, err) @@ -149,13 +150,14 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { return vm.Status == v1.VMStatusRunning }), "failed to start a VM") - // Ensure that the VM is using "--suspendable" and "--net-softnet" + // Ensure that the VM is using "--suspendable", "--net-softnet" and "--net-softnet-expose" tartVMName := ondiskname.New(vmName, vm.UID, vm.RestartCount).String() tartRunCmdline, err := tartRunProcessCmdline(tartVMName) require.NoError(t, err) require.Contains(t, tartRunCmdline, "--suspendable") require.Contains(t, tartRunCmdline, "--net-softnet") + require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-expose", "2222:22,8080:80"})) // Update the VM's specification and tighten the Softnet restrictions vm.NetSoftnetAllow = []string{"10.0.0.0/16"} @@ -175,13 +177,15 @@ func TestSpecUpdateSoftnetSuspendable(t *testing.T) { return vm.ObservedGeneration == 1 }), "failed to wait for the VM's observed generation to be updated") - // Ensure that the VM is using "--suspendable", "--net-softnet" and "--net-softnet-{allow,block}" + // Ensure that the VM is using "--suspendable", "--net-softnet", + // "--net-softnet-{allow,block}" and still "--net-softnet-expose" tartRunCmdline, err = tartRunProcessCmdline(tartVMName) require.NoError(t, err) require.Contains(t, tartRunCmdline, "--suspendable") require.Contains(t, tartRunCmdline, "--net-softnet") require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-allow", "10.0.0.0/16"})) require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-block", "0.0.0.0/0"})) + require.True(t, sliceContainsAnotherSlice(tartRunCmdline, []string{"--net-softnet-expose", "2222:22,8080:80"})) } //nolint:gosec,modernize,perfsprint,staticcheck // preserve the original integration test diff --git a/internal/worker/vmmanager/tart/tart.go b/internal/worker/vmmanager/tart/tart.go index b0e157c..5d0462c 100644 --- a/internal/worker/vmmanager/tart/tart.go +++ b/internal/worker/vmmanager/tart/tart.go @@ -352,6 +352,9 @@ func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { if len(resource.NetSoftnetBlock) != 0 { runArgs = append(runArgs, "--net-softnet-block", strings.Join(resource.NetSoftnetBlock, ",")) } + if len(resource.NetSoftnetExpose) != 0 { + runArgs = append(runArgs, "--net-softnet-expose", strings.Join(resource.NetSoftnetExpose, ",")) + } if resource.NetBridged != "" { runArgs = append(runArgs, fmt.Sprintf("--net-bridged=%s", resource.NetBridged)) } diff --git a/internal/worker/worker.go b/internal/worker/worker.go index a212648..8389758 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -455,7 +455,7 @@ func (worker *Worker) registerWorker(ctx context.Context) error { DefaultCPU: worker.defaultCPU, DefaultMemory: worker.defaultMemory, Capabilities: lo.Ternary(worker.runtime.ID() == v1.RuntimeTart || worker.runtime.Synthetic(), - v1.WorkerCapabilities{v1.WorkerCapabilityVMEndpoints}, nil), + v1.WorkerCapabilities{v1.WorkerCapabilityVMEndpoints, v1.WorkerCapabilityVMExposedPorts}, nil), }) if err != nil { return err diff --git a/pkg/resource/v1/endpoint.go b/pkg/resource/v1/endpoint.go index 3ab2b87..a7ca3fc 100644 --- a/pkg/resource/v1/endpoint.go +++ b/pkg/resource/v1/endpoint.go @@ -55,6 +55,17 @@ func (portRange PortRange) Validate() error { return nil } +// FreePort returns the lowest port of the range that is not in taken. +func (portRange PortRange) FreePort(taken map[uint16]struct{}) (uint16, bool) { + for port := int(portRange.Min); port <= int(portRange.Max); port++ { + if _, ok := taken[uint16(port)]; !ok { + return uint16(port), true + } + } + + return 0, false +} + type EndpointState string const ( diff --git a/pkg/resource/v1/exposed_port.go b/pkg/resource/v1/exposed_port.go new file mode 100644 index 0000000..db86e46 --- /dev/null +++ b/pkg/resource/v1/exposed_port.go @@ -0,0 +1,55 @@ +package v1 + +import ( + "errors" + "fmt" + "strconv" + "strings" +) + +var ( + ErrInvalidExposedPort = errors.New("invalid exposed port specification") + errPortOutOfRange = errors.New("port must be in the range 1-65535") +) + +// ExposedPort is a parsed netSoftnetExpose entry. +type ExposedPort struct { + External uint16 + Internal uint16 +} + +// NewExposedPortFromString parses an EXTERNAL:INTERNAL entry. +// +// Orchard is stricter than Softnet here: it rejects port 0, +// which cannot be exposed, and a leading sign. +func NewExposedPortFromString(s string) (ExposedPort, error) { + splits := strings.Split(s, ":") + if len(splits) != 2 { + return ExposedPort{}, fmt.Errorf("%w %q: the format should be EXTERNAL:INTERNAL", ErrInvalidExposedPort, s) + } + + external, err := parsePort(splits[0]) + if err != nil { + return ExposedPort{}, fmt.Errorf("%w %q: invalid external port: %w", ErrInvalidExposedPort, s, err) + } + + internal, err := parsePort(splits[1]) + if err != nil { + return ExposedPort{}, fmt.Errorf("%w %q: invalid internal port: %w", ErrInvalidExposedPort, s, err) + } + + return ExposedPort{External: external, Internal: internal}, nil +} + +func parsePort(s string) (uint16, error) { + port, err := strconv.ParseUint(s, 10, 16) + if err != nil { + return 0, err + } + + if port == 0 { + return 0, errPortOutOfRange + } + + return uint16(port), nil +} diff --git a/pkg/resource/v1/exposed_port_test.go b/pkg/resource/v1/exposed_port_test.go new file mode 100644 index 0000000..ef7dcf0 --- /dev/null +++ b/pkg/resource/v1/exposed_port_test.go @@ -0,0 +1,45 @@ +package v1_test + +import ( + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" + "testing" +) + +func TestNewExposedPortFromString(t *testing.T) { + valid := map[string]v1.ExposedPort{ + "1:1": {External: 1, Internal: 1}, + "2222:22": {External: 2222, Internal: 22}, + "08080:80": {External: 8080, Internal: 80}, + "65535:1": {External: 65535, Internal: 1}, + "65535:65535": {External: 65535, Internal: 65535}, + } + + for entry, expected := range valid { + exposedPort, err := v1.NewExposedPortFromString(entry) + require.NoError(t, err, entry) + require.Equal(t, expected, exposedPort, entry) + } + + invalid := []string{ + "", + "2222", + "2222:22:1", + "2222:", + ":22", + "a:22", + "0:22", + "22:0", + "65536:22", + "-1:22", + "+1:22", + " 2222:22", + "2_222:22", + "0x8ae:22", + } + + for _, entry := range invalid { + _, err := v1.NewExposedPortFromString(entry) + require.ErrorIs(t, err, v1.ErrInvalidExposedPort, entry) + } +} diff --git a/pkg/resource/v1/v1.go b/pkg/resource/v1/v1.go index fd6d111..1431fdf 100644 --- a/pkg/resource/v1/v1.go +++ b/pkg/resource/v1/v1.go @@ -3,6 +3,7 @@ package v1 import ( "encoding/json" "fmt" + "maps" "slices" "time" @@ -162,6 +163,9 @@ func (vm *VM) Validate() error { if len(vm.NetSoftnetBlock) != 0 { return unsupportedFieldError("netSoftnetBlock") } + if len(vm.NetSoftnetExpose) != 0 { + return unsupportedFieldError("netSoftnetExpose") + } if len(vm.HostDirs) != 0 { return unsupportedFieldError("hostDirs") } @@ -170,6 +174,50 @@ func (vm *VM) Validate() error { } } + seenExternalPorts := map[uint16]struct{}{} + + for _, entry := range vm.NetSoftnetExpose { + exposedPort, err := NewExposedPortFromString(entry) + if err != nil { + return fmt.Errorf("netSoftnetExpose: %w", err) + } + + if _, ok := seenExternalPorts[exposedPort.External]; ok { + return fmt.Errorf("netSoftnetExpose: %w %q: external port %d is exposed more than once", + ErrInvalidExposedPort, entry, exposedPort.External) + } + + seenExternalPorts[exposedPort.External] = struct{}{} + } + + // Tart binds the Softnet ports and every endpoint listener binds one of its own. + // The worker creates the listeners in the order they are declared, each taking the + // lowest port of its range that is still free, so walking them the same way says + // what it will arrive at, and which endpoint it will leave with nothing. + takenPorts := maps.Clone(seenExternalPorts) + + for _, endpoint := range vm.Endpoints { + portRange := endpoint.WorkerPortRange + + if portRange == nil { + continue + } + + port, ok := portRange.FreePort(takenPorts) + if !ok { + if portRange.Min == portRange.Max { + return fmt.Errorf("endpoint %q: %w: worker port %d is taken by netSoftnetExpose "+ + "or another endpoint", endpoint.Name, ErrInvalidExposedPort, portRange.Min) + } + + return fmt.Errorf("endpoint %q: %w: every worker port in %d-%d is taken by "+ + "netSoftnetExpose or the other endpoints", endpoint.Name, ErrInvalidExposedPort, + portRange.Min, portRange.Max) + } + + takenPorts[port] = struct{}{} + } + return nil } @@ -198,6 +246,7 @@ type VMSpec struct { NetSoftnet bool `json:"netSoftnet,omitempty"` NetSoftnetAllow []string `json:"netSoftnetAllow,omitempty"` NetSoftnetBlock []string `json:"netSoftnetBlock,omitempty"` + NetSoftnetExpose []string `json:"netSoftnetExpose,omitempty"` Suspendable bool `json:"suspendable,omitempty"` PowerState PowerState `json:"powerState,omitempty"` } @@ -214,7 +263,8 @@ func (vm VMSpec) HostProcessesEqual(other VMSpec) bool { func (vm VMSpec) SoftnetEnabled() bool { return vm.NetSoftnetDeprecated || vm.NetSoftnet || - len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 + len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 || + len(vm.NetSoftnetExpose) != 0 } func (vm VMSpec) SoftnetPolicyChanged(other VMSpec) bool { diff --git a/pkg/resource/v1/vm_validate_test.go b/pkg/resource/v1/vm_validate_test.go new file mode 100644 index 0000000..539710a --- /dev/null +++ b/pkg/resource/v1/vm_validate_test.go @@ -0,0 +1,151 @@ +package v1_test + +import ( + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" + "testing" +) + +func TestVMValidateNetSoftnetExpose(t *testing.T) { + for _, test := range []struct { + name string + runtime v1.Runtime + expose []string + errIs error + errContains string + }{ + {name: "tart valid", runtime: v1.RuntimeTart, expose: []string{"2222:22", "08080:80"}}, + {name: "tart nil", runtime: v1.RuntimeTart}, + {name: "tart same internal port", runtime: v1.RuntimeTart, expose: []string{"2222:22", "2223:22"}}, + {name: "empty runtime valid", expose: []string{"2222:22"}}, + { + name: "tart malformed", + runtime: v1.RuntimeTart, + expose: []string{"2222:22", "2222"}, + errIs: v1.ErrInvalidExposedPort, + errContains: "netSoftnetExpose", + }, + { + name: "tart duplicate external port", + runtime: v1.RuntimeTart, + expose: []string{"2222:22", "02222:80"}, + errIs: v1.ErrInvalidExposedPort, + errContains: "external port 2222 is exposed more than once", + }, + { + name: "vetu unsupported", + runtime: v1.RuntimeVetu, + expose: []string{"2222:22"}, + errContains: "does not support field \"netSoftnetExpose\"", + }, + { + name: "vetu malformed reports unsupported field", + runtime: v1.RuntimeVetu, + expose: []string{"2222"}, + errContains: "does not support field \"netSoftnetExpose\"", + }, + {name: "vetu nil", runtime: v1.RuntimeVetu}, + } { + t.Run(test.name, func(t *testing.T) { + var vm v1.VM + vm.Runtime = test.runtime + vm.NetSoftnetExpose = test.expose + + err := vm.Validate() + + if test.errContains == "" { + require.NoError(t, err) + + return + } + + if test.errIs != nil { + require.ErrorIs(t, err, test.errIs) + } else { + // Unsupported field errors take precedence over parsing the entries + require.NotErrorIs(t, err, v1.ErrInvalidExposedPort) + } + + require.ErrorContains(t, err, test.errContains) + }) + } +} + +// Tart binds the Softnet ports while the worker binds one port per endpoint listener, +// so a VM cannot ask for more of them than its ranges hold. +func TestVMValidateEndpointReusesSoftnetPort(t *testing.T) { + endpoint := func(name string, min uint16, max uint16) v1.EndpointSpec { + return v1.EndpointSpec{ + Name: name, + Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}, + WorkerPortRange: &v1.PortRange{Min: min, Max: max}, + } + } + + for _, test := range []struct { + name string + expose []string + endpoints []v1.EndpointSpec + errContains string + }{ + {name: "same port", endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2222)}, + errContains: "worker port 2222 is taken by netSoftnetExpose or another endpoint"}, + {name: "other port", endpoints: []v1.EndpointSpec{endpoint("ssh", 2223, 2223)}}, + // a range with a port left over lets the worker avoid the Softnet ones + {name: "range", endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2299)}}, + { + name: "range fully exposed", + expose: []string{"2222:22", "2223:23"}, + endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2223)}, + errContains: "every worker port in 2222-2223 is taken by netSoftnetExpose " + + "or the other endpoints", + }, + {name: "range with one port left", expose: []string{"2222:22", "2223:23"}, + endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2224)}}, + // two listeners need two ports between them, wherever their ranges overlap + { + name: "two endpoints, one port left", + endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2223), endpoint("http", 2222, 2223)}, + errContains: "is taken by netSoftnetExpose or the other endpoints", + }, + {name: "two endpoints, two ports left", + endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2224), endpoint("http", 2222, 2224)}}, + {name: "two endpoints, one pinned", expose: []string{"2299:22"}, + endpoints: []v1.EndpointSpec{endpoint("ssh", 2222, 2222), endpoint("http", 2222, 2223)}}, + // the worker works down the endpoints in the order they are declared, so a wide + // range ahead of a pinned one takes the port that pinned one has to have + { + name: "wide range ahead of a pinned one", + endpoints: []v1.EndpointSpec{endpoint("http", 2222, 2224), + endpoint("ssh", 2223, 2223)}, + errContains: "worker port 2223 is taken by netSoftnetExpose or another endpoint", + }, + {name: "pinned range ahead of a wide one", endpoints: []v1.EndpointSpec{ + endpoint("ssh", 2223, 2223), endpoint("http", 2222, 2224)}}, + {name: "no range", endpoints: []v1.EndpointSpec{{Name: "ssh", + Target: v1.ConnectionTarget{VM: &v1.ConnectionTargetVM{Port: 22}}}}}, + } { + t.Run(test.name, func(t *testing.T) { + expose := test.expose + if expose == nil { + expose = []string{"2222:22"} + } + + var vm v1.VM + vm.Runtime = v1.RuntimeTart + vm.NetSoftnetExpose = expose + vm.Endpoints = test.endpoints + + err := vm.Validate() + + if test.errContains == "" { + require.NoError(t, err) + + return + } + + require.ErrorIs(t, err, v1.ErrInvalidExposedPort) + require.ErrorContains(t, err, test.errContains) + }) + } +} diff --git a/pkg/resource/v1/worker.go b/pkg/resource/v1/worker.go index 622a46a..cbe579d 100644 --- a/pkg/resource/v1/worker.go +++ b/pkg/resource/v1/worker.go @@ -43,7 +43,10 @@ func (worker *Worker) SetVersion(_ uint64) {} type WorkerCapability string -const WorkerCapabilityVMEndpoints WorkerCapability = "vm-endpoints" +const ( + WorkerCapabilityVMEndpoints WorkerCapability = "vm-endpoints" + WorkerCapabilityVMExposedPorts WorkerCapability = "vm-exposed-ports" +) type WorkerCapabilities []WorkerCapability