From ad8c97ffc47eac1dae789ce7e6069667e37b9032 Mon Sep 17 00:00:00 2001 From: Charalampos Stratakis Date: Thu, 3 Sep 2026 12:31:13 +0200 Subject: [PATCH 1/3] Add netSoftnetExpose to the VM spec Softnet can forward host ports to a VM, but Orchard had no way to ask for it, so a VM behind Softnet was reachable only through the controller's SSH proxy. --- api/openapi.yaml | 11 ++ internal/command/create/vm.go | 5 + internal/command/get/vm.go | 1 + internal/controller/api_vms.go | 14 ++- internal/tests/spec_update_test.go | 14 ++- internal/worker/vmmanager/tart/tart.go | 3 + internal/worker/worker.go | 2 +- pkg/resource/v1/endpoint.go | 11 ++ pkg/resource/v1/exposed_port.go | 55 +++++++++ pkg/resource/v1/exposed_port_test.go | 45 ++++++++ pkg/resource/v1/v1.go | 52 ++++++++- pkg/resource/v1/vm_validate_test.go | 151 +++++++++++++++++++++++++ pkg/resource/v1/worker.go | 5 +- 13 files changed, 355 insertions(+), 14 deletions(-) create mode 100644 pkg/resource/v1/exposed_port.go create mode 100644 pkg/resource/v1/exposed_port_test.go create mode 100644 pkg/resource/v1/vm_validate_test.go diff --git a/api/openapi.yaml b/api/openapi.yaml index 030ea43..3025119 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -741,6 +741,17 @@ 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. 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..a90d491 100644 --- a/internal/controller/api_vms.go +++ b/internal/controller/api_vms.go @@ -89,9 +89,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 +223,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 } 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 From 714bf00549bbd6d85d2d6ddb59b98215b3f8196e Mon Sep 17 00:00:00 2001 From: Charalampos Stratakis Date: Fri, 4 Sep 2026 00:47:01 +0200 Subject: [PATCH 2/3] scheduler: extract the capacity test helpers Register the workers with the notifier as well, otherwise every placement waits a second for its worker to connect. --- .../scheduler/scheduler_capacity_test.go | 159 ++++++++++++------ 1 file changed, 106 insertions(+), 53 deletions(-) diff --git a/internal/controller/scheduler/scheduler_capacity_test.go b/internal/controller/scheduler/scheduler_capacity_test.go index 0bcf6f4..ef2a229 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, + }} + + return vm +} - pending := newPendingVM("pending-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,75 @@ 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) - 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 scheduler +} - return nil +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) + + currentVM := getVM(t, store, pending.Name) + require.False(t, currentVM.IsScheduled()) + require.Empty(t, currentVM.Worker) } From af18f6cd8511b638403179dc68065c101d887df0 Mon Sep 17 00:00:00 2001 From: Charalampos Stratakis Date: Thu, 3 Sep 2026 12:31:13 +0200 Subject: [PATCH 3/3] scheduler: skip workers with an exposed port conflict Two VMs exposing the same external port on one worker both reach the running state, and the second silently does not get the port. Endpoint listeners take worker ports too, so they count on both sides, a scheduled VM's ports cannot change while the worker still has them bound, and a worker that does not advertise the field is skipped as well. --- api/openapi.yaml | 8 +- internal/controller/api_vms.go | 32 ++ internal/controller/api_vms_test.go | 212 +++++++++++++ internal/controller/scheduler/scheduler.go | 33 +- .../scheduler/scheduler_capacity_test.go | 226 ++++++++++++++ internal/controller/scheduler/workerinfo.go | 282 +++++++++++++++++- .../controller/scheduler/workerinfo_test.go | 281 ++++++++++++++++- 7 files changed, 1057 insertions(+), 17 deletions(-) create mode 100644 internal/controller/api_vms_test.go diff --git a/api/openapi.yaml b/api/openapi.yaml index 3025119..a1708a7 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -747,7 +747,13 @@ components: 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. Enables `netSoftnet`. + 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: diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index a90d491..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" @@ -260,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 ef2a229..3d46943 100644 --- a/internal/controller/scheduler/scheduler_capacity_test.go +++ b/internal/controller/scheduler/scheduler_capacity_test.go @@ -156,3 +156,229 @@ func TestSchedulingLoopSkipsOvercommittedWorker(t *testing.T) { 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) + } + + 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...))) + }) + } }