Add endpoint support

api/openapi.yaml

@@ -237,8 +237,10 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/VM'
+        '400':
+          description: Invalid VM specification
         '409':
-          description: VM resource with with the same name already exists
+          description: VM resource with the same name already exists
     get:
       summary: "List VMs"
       tags:
@@ -311,6 +313,8 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/VM'
+        '400':
+          description: Invalid VM specification
         '404':
           description: VM resource with the given name doesn't exist
     delete:
@@ -682,6 +686,19 @@ components:
           type: number
           description: Disk size for this VM
           example: 100
+        endpoints:
+          type: array
+          description: |
+            TCP services inside the VM to expose through TCP ports on the Orchard
+            Worker. The worker ports assigned to the endpoints are reported in
+            `observedEndpoints`.
+            Endpoint ports listen on all worker network interfaces. Use firewall
+            rules to restrict access.
+            VMs with endpoints are scheduled only on workers that support endpoint
+            forwarding. Adding endpoints to a VM already assigned to an unsupported
+            worker succeeds, but those endpoints are reported in the `error` state.
+          items:
+            $ref: '#/components/schemas/EndpointSpec'
         net-softnet:
           type: boolean
           description: Please use `netSoftnet` instead
@@ -854,6 +871,93 @@ components:
             Deprecated alias for `localName`.
           readOnly: true
           deprecated: true
+    ConnectionTarget:
+      title: Connection target
+      type: object
+      description: A connection destination relative to the containing resource or action.
+      required:
+        - vm
+      properties:
+        vm:
+          $ref: '#/components/schemas/ConnectionTargetVM'
+    ConnectionTargetVM:
+      title: VM connection target
+      type: object
+      required:
+        - port
+      properties:
+        port:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: TCP port inside the contextual VM.
+    PortRange:
+      title: Port range
+      type: object
+      description: |
+        Inclusive bounds for selecting a network port. `min` must be less than or
+        equal to `max`; equal bounds identify one port.
+      required:
+        - min
+        - max
+      properties:
+        min:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: Inclusive lower bound for port selection.
+        max:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: Inclusive upper bound for port selection.
+    EndpointSpec:
+      title: Endpoint specification
+      type: object
+      description: A desired worker TCP endpoint backed by a connection target.
+      required:
+        - name
+        - target
+      properties:
+        name:
+          type: string
+          minLength: 1
+          description: Stable endpoint identifier, unique within `endpoints`.
+        target:
+          $ref: '#/components/schemas/ConnectionTarget'
+        workerPortRange:
+          description: |
+            Optional inclusive bounds for selecting the worker port. When omitted,
+            the operating system selects an available port. Equal bounds request
+            that exact port.
+          allOf:
+            - $ref: '#/components/schemas/PortRange'
+    EndpointStatus:
+      title: Endpoint status
+      type: object
+      description: Current observation for a desired endpoint.
+      required:
+        - name
+        - state
+      properties:
+        name:
+          type: string
+          minLength: 1
+          description: Stable endpoint identifier matching the desired endpoint.
+        workerPort:
+          type: integer
+          minimum: 1
+          maximum: 65535
+          description: TCP port assigned to this endpoint on the Orchard Worker.
+        state:
+          type: string
+          enum: [ listening, error ]
+          description: |
+            Exposure state. `listening` means that the worker TCP port is bound; it
+            does not imply that the service inside the VM is healthy or accepting connections.
+        message:
+          type: string
+          description: Human-readable detail, primarily populated when `state` is `error`.
     VMState:
       title: Virtual Machine State
       type: object
@@ -871,6 +975,13 @@ components:
         observedGeneration:
           type: number
           description: Corresponds to the `Generation` value on which the worker had acted upon
+        observedEndpoints:
+          type: array
+          readOnly: true
+          description: |
+            Current observations for the endpoints in `endpoints`.
+          items:
+            $ref: '#/components/schemas/EndpointStatus'
     Events:
       title: Events
       type: object

internal/controller/api_vms.go

@@ -43,6 +43,9 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
 	if vm.Image == "" {
 		return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("VM image is empty"))
 	}
+	if err := v1.ValidateEndpoints(vm.Endpoints); err != nil {
+		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid endpoints: %v", err))
+	}
 
 	// Provide defaults
 	vm.Status = v1.VMStatusPending
@@ -56,6 +59,7 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
 	vm.TartName = vm.LocalName
 	vm.Generation = 0
 	vm.ObservedGeneration = 0
+	vm.ObservedEndpoints = nil
 	vm.Conditions = []v1.Condition{
 		{
 			Type:  v1.ConditionTypeScheduled,
@@ -158,6 +162,9 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 	if err := ctx.ShouldBindJSON(&userVM); err != nil {
 		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid JSON was provided"))
 	}
+	if err := v1.ValidateEndpoints(userVM.Endpoints); err != nil {
+		return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid endpoints: %v", err))
+	}
 
 	if responder := controller.validateHostDirs(userVM.HostDirs); responder != nil {
 		return responder
@@ -221,7 +228,7 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
 				"transition: only suspendable VMs can be suspended"))
 		}
 
-		if dbVM.SemanticallyEqual(userVM.VMSpec) {
+		if v1.SemanticallyEqual(dbVM.VMSpec, userVM.VMSpec) {
 			// Nothing was changed
 			return responder.JSON(http.StatusOK, dbVM)
 		}

internal/controller/api_workers.go

@@ -78,6 +78,7 @@ func (controller *Controller) createWorker(ctx *gin.Context) responder.Responder
 		dbWorker.Labels = worker.Labels
 		dbWorker.DefaultCPU = worker.DefaultCPU
 		dbWorker.DefaultMemory = worker.DefaultMemory
+		dbWorker.Capabilities = worker.Capabilities
 
 		if err := txn.SetWorker(*dbWorker); err != nil {
 			return responder.Error(err)

internal/controller/controller.go

@@ -332,6 +332,7 @@ func (controller *Controller) failVMsWithHostDirs() error {
 			vm.Status = v1.VMStatusFailed
 			vm.StatusMessage = "host directories are used, but host directory sharing is disabled"
 			vm.RestartPolicy = v1.RestartPolicyNever
+			vm.ObservedEndpoints = nil
 
 			if err := txn.SetVM(vm); err != nil {
 				return err

internal/controller/scheduler/scheduler.go

@@ -279,6 +279,12 @@ NextVM:
 				continue NextWorker
 			}
 
+			// Don't schedule VMs with endpoints on workers that don't support them
+			if len(unscheduledVM.Endpoints) != 0 &&
+				!worker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+				continue NextWorker
+			}
+
 			err := scheduler.store.Update(func(txn storepkg.Transaction) error {
 				currentUnscheduledVM, err := txn.GetVM(unscheduledVM.Name)
 				if err != nil {
@@ -291,7 +297,8 @@ NextVM:
 					return err
 				}
 
-				if currentUnscheduledVM.UID != unscheduledVM.UID {
+				if currentUnscheduledVM.UID != unscheduledVM.UID ||
+					currentUnscheduledVM.Generation != unscheduledVM.Generation {
 					// The unscheduled VM had changed, so we'll re-evaluate a new
 					// version of it in the next scheduling loop iteration
 					return ErrVMSchedulingSkipped
@@ -333,6 +340,12 @@ NextVM:
 					return ErrWorkerSchedulingSkipped
 				}
 
+				// Don't schedule VMs with endpoints on workers that don't support them
+				if len(unscheduledVM.Endpoints) != 0 &&
+					!currentWorker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+					return ErrWorkerSchedulingSkipped
+				}
+
 				if currentWorker.MachineID != worker.MachineID ||
 					!currentWorker.Resources.Equal(worker.Resources) {
 					// Worker has changed
@@ -504,6 +517,11 @@ func (scheduler *Scheduler) healthCheckingLoopIteration() (int, error) {
 func (scheduler *Scheduler) healthCheckVM(txn storepkg.Transaction, vm v1.VM) error {
 	logger := scheduler.logger.With("vm_name", vm.Name, "vm_uid", vm.UID, "vm_restart_count", vm.RestartCount)
 
+	// Preserve the current endpoint observations for comparison below and reset them,
+	// so every VM transition persisted below discards the stale endpoint observations
+	savedObservedEndpoints := vm.ObservedEndpoints
+	vm.ObservedEndpoints = nil
+
 	// Schedule a VM restart if the restart policy mandates it
 	needsRestart := vm.RestartPolicy == v1.RestartPolicyOnFailure &&
 		vm.Status == v1.VMStatusFailed &&
@@ -586,5 +604,24 @@ func (scheduler *Scheduler) healthCheckVM(txn storepkg.Transaction, vm v1.VM) er
 		return txn.SetVM(vm)
 	}
 
+	// Report unsupported endpoints to clients without failing the VM
+	if !worker.Capabilities.Has(v1.WorkerCapabilityVMEndpoints) {
+		vm.ObservedEndpoints = make([]v1.EndpointStatus, 0, len(vm.Endpoints))
+
+		for _, endpoint := range vm.Endpoints {
+			vm.ObservedEndpoints = append(vm.ObservedEndpoints, v1.EndpointStatus{
+				Name:    endpoint.Name,
+				State:   v1.EndpointStateError,
+				Message: "worker doesn't support VM endpoints",
+			})
+		}
+
+		if slices.Equal(savedObservedEndpoints, vm.ObservedEndpoints) {
+			return nil
+		}
+
+		return txn.SetVM(vm)
+	}
+
 	return nil
 }

internal/proxy/proxy.go

@@ -1,11 +1,28 @@
 package proxy
 
 import (
+	"errors"
 	"io"
+	"net"
 	"strings"
 )
 
+// halfCloseConnection can stop writing without interrupting concurrent reads.
+type halfCloseConnection interface {
+	net.Conn
+	CloseWrite() error
+}
+
+// Connections copies bytes in both directions. When both connections support
+// write-side shutdown, EOF is propagated independently in each direction.
+// Other connection pairs retain the historical close-on-first-completion behavior.
 func Connections(left io.ReadWriteCloser, right io.ReadWriteCloser) (finalErr error) {
+	if left, ok := left.(halfCloseConnection); ok {
+		if right, ok := right.(halfCloseConnection); ok {
+			return connectionsWithHalfClose(left, right)
+		}
+	}
+
 	leftErrCh := make(chan error, 1)
 	rightErrCh := make(chan error, 1)
 
@@ -44,3 +61,51 @@ func Connections(left io.ReadWriteCloser, right io.ReadWriteCloser) (finalErr er
 
 	return finalErr
 }
+
+// connectionsWithHalfClose keeps the reverse direction open after a clean EOF.
+func connectionsWithHalfClose(left, right halfCloseConnection) error {
+	//nolint:mnd // one result from each copy direction
+	results := make(chan error, 2)
+
+	copyHalf := func(destination, source halfCloseConnection) {
+		_, err := io.Copy(destination, source)
+		if err == nil {
+			err = destination.CloseWrite()
+		}
+
+		results <- err
+	}
+
+	closeBoth := func() {
+		_ = left.Close()
+		_ = right.Close()
+	}
+
+	meaningfulError := func(err error) error {
+		if err == nil ||
+			errors.Is(err, net.ErrClosed) ||
+			strings.Contains(err.Error(), "use of closed network connection") {
+			return nil
+		}
+
+		return err
+	}
+
+	go copyHalf(right, left)
+	go copyHalf(left, right)
+
+	firstErr := <-results
+	if firstErr != nil {
+		// An I/O or CloseWrite failure must unblock both copy operations.
+		closeBoth()
+	}
+
+	secondErr := <-results
+	closeBoth()
+
+	if err := meaningfulError(firstErr); err != nil {
+		return err
+	}
+
+	return meaningfulError(secondErr)
+}