Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 51 additions & 3 deletions api/openapi.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,13 @@ paths:
type: integer
minimum: 1
maximum: 65535
required: true
required: false
- in: query
name: hostProcess
description: Name of a host process declared in the VM's `hostProcesses` field; mutually exclusive with `port`.
schema:
type: string
required: false
- in: query
name: wait
description: Duration in seconds to wait for the VM to transition into "running" state if not already running.
Expand All @@ -414,9 +420,9 @@ paths:
type: string
responses:
'400':
description: Invalid port specified
description: Invalid or ambiguous port-forward target specified
'404':
description: VM resource with the given name doesn't exist
description: VM or requested host process doesn't exist
'503':
description: Failed to establish connection with the worker responsible for the specified VM
/vms/{name}/exec:
Expand Down Expand Up @@ -832,6 +838,14 @@ components:
- path: /path/on/host/to/sources
ro: true
- path: /path/on/host/to/builds
hostProcesses:
type: array
description: |
Generic long-running processes run by the Orchard worker alongside this VM.
Their lifecycle is tied to the VM. Updating this field restarts the associated
host processes without restarting the VM.
items:
$ref: '#/components/schemas/HostProcess'
powerState:
type: string
description: |
Expand Down Expand Up @@ -958,6 +972,40 @@ components:
message:
type: string
description: Human-readable detail, primarily populated when `state` is `error`.
HostProcess:
title: VM-associated host process
type: object
description: A long-running process run by the Orchard worker alongside a VM.
required:
- name
- program
properties:
name:
type: string
description: Process name, unique within the VM.
program:
type: string
description: Executable path or name resolved using the worker's `PATH`.
args:
type: array
description: |
Argument vector passed directly to the configured program without shell interpretation.

The worker only expands these placeholders in each argument:

- `${ORCHARD_WORKER_NAME}`: name of this Orchard Worker
- `${ORCHARD_VM_NAME}`: name of the associated VM
- `${ORCHARD_VM_CONTROL_SOCKET}`: path to the VM's control socket
- `${ORCHARD_PROCESS_SOCKET}`: Unix socket path that the process must listen on to receive incoming port-forward connections
items:
type: string
env:
type: object
description: |
Additional environment variables passed to the process. `PATH` and the worker-provided
`ORCHARD_*` variables take precedence over values supplied here.
additionalProperties:
type: string
VMState:
title: Virtual Machine State
type: object
Expand Down
20 changes: 17 additions & 3 deletions internal/controller/api_rpc_watch.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,16 @@ import (
"encoding/json"
"errors"
"fmt"
"time"

"github.com/cirruslabs/orchard/internal/responder"
v1 "github.com/cirruslabs/orchard/pkg/resource/v1"
"github.com/cirruslabs/orchard/rpc"
"github.com/coder/websocket"
"github.com/gin-gonic/gin"
"time"
)

//nolint:protogetter // Preserve the original host-process wire conversion.
func (controller *Controller) rpcWatch(ctx *gin.Context) responder.Responder {
if responder := controller.authorize(ctx, v1.ServiceAccountRoleComputeRead); responder != nil {
return responder
Expand Down Expand Up @@ -61,8 +63,20 @@ func (controller *Controller) rpcWatch(ctx *gin.Context) responder.Responder {
case *rpc.WatchInstruction_PortForwardAction:
watchInstruction.PortForwardAction = &v1.PortForwardAction{
Session: typedAction.PortForwardAction.Session,
VMUID: typedAction.PortForwardAction.VmUid,
Port: uint16(typedAction.PortForwardAction.Port),
}

if target := typedAction.PortForwardAction.GetTarget(); target != nil {
watchInstruction.PortForwardAction.Target = &v1.PortForwardTarget{}

if hostProcess := target.GetHostProcess(); hostProcess != nil {
watchInstruction.PortForwardAction.Target.HostProcess = &v1.PortForwardTargetHostProcess{
VMUID: hostProcess.VmUid,
Name: hostProcess.Name,
}
}
} else {
watchInstruction.PortForwardAction.VMUID = typedAction.PortForwardAction.VmUid
watchInstruction.PortForwardAction.Port = uint16(typedAction.PortForwardAction.Port)
}
case *rpc.WatchInstruction_SyncVmsAction:
watchInstruction.SyncVMsAction = &v1.SyncVMsAction{}
Expand Down
25 changes: 25 additions & 0 deletions internal/controller/api_vms.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,13 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid JSON was provided"))
}

// Host processes require an additional role
if len(vm.HostProcesses) != 0 {
if responder := controller.authorize(ctx, v1.ServiceAccountRoleHostProcessWrite); responder != nil {
return responder
}
}

if vm.Name == "" {
return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("VM name is empty"))
} else if err := simplename.Validate(vm.Name); err != nil {
Expand Down Expand Up @@ -122,6 +129,11 @@ func (controller *Controller) createVM(ctx *gin.Context) responder.Responder {
vm.RestartPolicy = v1.RestartPolicyNever
}

// Validate hostProcesses
if err := v1.ValidateHostProcesses(vm.HostProcesses); err != nil {
return responder.JSON(http.StatusBadRequest, NewErrorResponse("invalid host processes: %v", err))
}

// Validate hostDirs
if responder := controller.validateHostDirs(vm.HostDirs); responder != nil {
return responder
Expand Down Expand Up @@ -182,6 +194,14 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
return responder.Error(err)
}

// Changes to host processes require an additional role
//nolint:staticcheck // Preserve the original explicit VMSpec comparison.
if !dbVM.VMSpec.HostProcessesEqual(userVM.VMSpec) {
if responder := controller.authorize(ctx, v1.ServiceAccountRoleHostProcessWrite); responder != nil {
return responder
}
}

if dbVM.TerminalState() {
return responder.JSON(http.StatusPreconditionFailed,
NewErrorResponse("cannot update VM in a terminal state"))
Expand All @@ -197,6 +217,11 @@ func (controller *Controller) updateVMSpec(ctx *gin.Context) responder.Responder
return responder.JSON(http.StatusPreconditionFailed, NewErrorResponse("%v", err))
}

// Validate hostProcesses
if err := v1.ValidateHostProcesses(userVM.HostProcesses); err != nil {
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 {
Expand Down
1 change: 1 addition & 0 deletions internal/controller/api_vms_exec.go
Original file line number Diff line number Diff line change
Expand Up @@ -211,6 +211,7 @@ func (controller *Controller) newSSHExecSession(
vm.Worker,
vm.UID,
22,
"",
)
if err != nil {
return nil, err
Expand Down
66 changes: 52 additions & 14 deletions internal/controller/api_vms_portforward.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/pkg/errors"
"github.com/samber/lo"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
Expand All @@ -33,14 +34,29 @@ func (controller *Controller) portForwardVM(ctx *gin.Context) responder.Responde

// Retrieve and parse path and query parameters
name := ctx.Param("name")

portRaw := ctx.Query("port")
port, err := strconv.ParseUint(portRaw, 10, 16)
if err != nil {
return responder.Code(http.StatusBadRequest)
}
if port < 1 || port > 65535 {
return responder.Code(http.StatusBadRequest)
hostProcess := ctx.Query("hostProcess")

// Host process connections require an additional role
var port uint64
var err error

if hostProcess != "" {
if responder := controller.authorizeAny(ctx, v1.ServiceAccountRoleHostProcessWrite,
v1.ServiceAccountRoleHostProcessConnect); responder != nil {
return responder
}

// Host process forwarding cannot also target a VM port
if portRaw != "" {
return responder.Code(http.StatusBadRequest)
}
} else {
// VM port forwarding requires a valid non-zero TCP port
port, err = strconv.ParseUint(portRaw, 10, 16)
if err != nil || port < 1 || port > 65535 {
return responder.Code(http.StatusBadRequest)
}
}

waitRaw := ctx.DefaultQuery("wait", "10")
Expand All @@ -58,8 +74,16 @@ func (controller *Controller) portForwardVM(ctx *gin.Context) responder.Responde
return responderImpl
}

// Verify that the requested host process is declared on the VM.
if hostProcess != "" && !lo.SomeBy(vm.HostProcesses, func(process v1.HostProcess) bool {
return process.Name == hostProcess
}) {
return responder.JSON(http.StatusNotFound,
NewErrorResponse("host process %q is not declared on VM %q", hostProcess, vm.Name))
}

// Commence port forwarding
return controller.portForward(ctx, waitContext, vm.Worker, vm.UID, uint32(port))
return controller.portForward(ctx, waitContext, vm.Worker, vm.UID, uint32(port), hostProcess)
}

func (controller *Controller) portForward(
Expand All @@ -68,6 +92,7 @@ func (controller *Controller) portForward(
workerName string,
vmUID string,
port uint32,
hostProcess string,
) responder.Responder {
// Request and wait for a connection with a worker
rendezvousConn, err := retry.NewWithData[net.Conn](
Expand All @@ -77,7 +102,7 @@ func (controller *Controller) portForward(
retry.Attempts(0),
retry.LastErrorOnly(true),
).Do(func() (net.Conn, error) {
return controller.portForwardConnection(ctx, notifyContext, workerName, vmUID, port)
return controller.portForwardConnection(ctx, notifyContext, workerName, vmUID, port, hostProcess)
})
if err != nil {
if errors.Is(err, errPortForwardRequest) {
Expand Down Expand Up @@ -200,6 +225,7 @@ func (controller *Controller) portForwardConnection(
workerName string,
vmUID string,
port uint32,
hostProcess string,
) (net.Conn, error) {
// Create a rendezvous connection point
rendezvousCtx, rendezvousCtxCancel := context.WithCancel(ctx)
Expand All @@ -213,13 +239,25 @@ func (controller *Controller) portForwardConnection(
}

// Send request to a worker to initiate a port forwarding connection back to us
portForwardAction := &rpc.WatchInstruction_PortForward{
Session: session,
}
if hostProcess != "" {
portForwardAction.Target = &rpc.WatchInstruction_PortForward_Target{
Value: &rpc.WatchInstruction_PortForward_Target_HostProcess_{
HostProcess: &rpc.WatchInstruction_PortForward_Target_HostProcess{
VmUid: vmUID,
Name: hostProcess,
},
},
}
} else {
portForwardAction.VmUid = vmUID
portForwardAction.Port = port
}
err := controller.workerNotifier.Notify(waitContext, workerName, &rpc.WatchInstruction{
Action: &rpc.WatchInstruction_PortForwardAction{
PortForwardAction: &rpc.WatchInstruction_PortForward{
Session: session,
VmUid: vmUID,
Port: port,
},
PortForwardAction: portForwardAction,
},
})
if err != nil {
Expand Down
2 changes: 1 addition & 1 deletion internal/controller/api_workers_portforward.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,5 +50,5 @@ func (controller *Controller) portForwardWorker(ctx *gin.Context) responder.Resp
}

// Commence port-forwarding
return controller.portForward(ctx, waitContext, worker.Name, "", uint32(port))
return controller.portForward(ctx, waitContext, worker.Name, "", uint32(port), "")
}
Loading