diff --git a/api/openapi.yaml b/api/openapi.yaml index a76d06a7..030ea432 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -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. @@ -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: @@ -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: | @@ -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 diff --git a/internal/controller/api_rpc_watch.go b/internal/controller/api_rpc_watch.go index d7d0d180..8fa3c991 100644 --- a/internal/controller/api_rpc_watch.go +++ b/internal/controller/api_rpc_watch.go @@ -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 @@ -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{} diff --git a/internal/controller/api_vms.go b/internal/controller/api_vms.go index 51352fb2..823b5767 100644 --- a/internal/controller/api_vms.go +++ b/internal/controller/api_vms.go @@ -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 { @@ -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 @@ -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")) @@ -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 { diff --git a/internal/controller/api_vms_exec.go b/internal/controller/api_vms_exec.go index a3ebf0df..bae7b972 100644 --- a/internal/controller/api_vms_exec.go +++ b/internal/controller/api_vms_exec.go @@ -211,6 +211,7 @@ func (controller *Controller) newSSHExecSession( vm.Worker, vm.UID, 22, + "", ) if err != nil { return nil, err diff --git a/internal/controller/api_vms_portforward.go b/internal/controller/api_vms_portforward.go index bf14be11..fc44112b 100644 --- a/internal/controller/api_vms_portforward.go +++ b/internal/controller/api_vms_portforward.go @@ -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" ) @@ -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") @@ -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( @@ -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]( @@ -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) { @@ -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) @@ -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 { diff --git a/internal/controller/api_workers_portforward.go b/internal/controller/api_workers_portforward.go index 1ef62b16..71ac1c77 100644 --- a/internal/controller/api_workers_portforward.go +++ b/internal/controller/api_workers_portforward.go @@ -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), "") } diff --git a/internal/tests/hostprocess_test.go b/internal/tests/hostprocess_test.go new file mode 100644 index 00000000..7fee0647 --- /dev/null +++ b/internal/tests/hostprocess_test.go @@ -0,0 +1,148 @@ +//nolint:modernize,noctx,testpackage // Preserve the original integration-test setup and process helper. +package tests + +import ( + "context" + "io" + "net" + "os" + "testing" + "time" + + "github.com/cirruslabs/orchard/internal/controller" + "github.com/cirruslabs/orchard/internal/tests/devcontroller" + "github.com/cirruslabs/orchard/internal/tests/wait" + "github.com/cirruslabs/orchard/internal/worker" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" +) + +const integrationHelperArg = "--orchard-host-process-integration-test-helper" + +func TestMain(m *testing.M) { + if len(os.Args) == 2 && os.Args[1] == integrationHelperArg { + os.Exit(runIntegrationEchoHelper()) + } + + os.Exit(m.Run()) +} + +func TestHostProcessIntegration(t *testing.T) { + // Use this test binary as an echo host process + executable, err := os.Executable() + require.NoError(t, err) + + const startupMarker = "synthetic VM started canary (from startup script)" + + // Create a development environment with a synthetic Controller and Worker + devClient, _, _ := devcontroller.StartIntegrationTestEnvironmentWithAdditionalOpts( + t, + false, + []controller.Option{controller.WithSynthetic()}, + false, + []worker.Option{worker.WithSynthetic()}, + ) + + // Advertise the original test's Tart platform even on a Linux synthetic worker. + workers, err := devClient.Workers().List(t.Context()) + require.NoError(t, err) + require.Len(t, workers, 1) + workers[0].Arch = v1.ArchitectureARM64 + workers[0].Runtime = v1.RuntimeTart + workers[0].Resources[v1.ResourceTartVMs] = 1 + _, err = devClient.Workers().Create(t.Context(), workers[0]) + require.NoError(t, err) + + // Create a VM without any host processes + const vmName = "test-vm" + + err = devClient.VMs().Create(t.Context(), &v1.VM{ + Meta: v1.Meta{Name: vmName}, + Image: "synthetic", + StartupScript: &v1.VMScript{ScriptContent: startupMarker}, + }) + require.NoError(t, err) + + // Wait for the VM to start + var vm *v1.VM + + require.True(t, wait.Wait(time.Minute, func() bool { + vm, err = devClient.VMs().Get(t.Context(), vmName) + require.NoError(t, err) + + t.Logf("Waiting for the VM to start. Current status: %s", vm.Status) + + return vm.Status == v1.VMStatusRunning + }), "failed to wait for the VM to start") + + // Add an echo host process to the running VM + const hostProcessName = "echo" + + vm.HostProcesses = []v1.HostProcess{{ + Name: hostProcessName, + Program: executable, + Args: []string{integrationHelperArg}, + }} + + vm, err = devClient.VMs().Update(t.Context(), *vm) + require.NoError(t, err) + require.EqualValues(t, 1, vm.Generation) + require.EqualValues(t, 0, vm.ObservedGeneration) + + // Wait for the Worker to start the host process without restarting the VM + require.True(t, wait.Wait(time.Minute, func() bool { + vm, err = devClient.VMs().Get(t.Context(), vmName) + require.NoError(t, err) + + t.Logf("Waiting for the host process to start. Current observed generation: %d", vm.ObservedGeneration) + + return vm.Status == v1.VMStatusRunning && + vm.ObservedGeneration == 1 && + v1.ConditionIsTrue(vm.Conditions, v1.ConditionTypeHostProcessesReady) + }), "failed to wait for the host process to start") + + // Connect to the host process through the Controller + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + connection, err := devClient.VMs().PortForwardHostProcess(ctx, vmName, hostProcessName, 30) + require.NoError(t, err) + defer connection.Close() + + require.NoError(t, connection.SetDeadline(time.Now().Add(10*time.Second))) + + message := []byte("Hello, World!") + + _, err = connection.Write(message) + require.NoError(t, err) + + response := make([]byte, len(message)) + + _, err = io.ReadFull(connection, response) + require.NoError(t, err) + require.Equal(t, message, response) + + // The startup script would run again if applying the update restarted the VM + logLines, err := devClient.VMs().Logs(t.Context(), vmName) + require.NoError(t, err) + require.Equal(t, []string{startupMarker}, logLines) +} + +func runIntegrationEchoHelper() int { + listener, err := net.Listen("unix", os.Getenv("ORCHARD_PROCESS_SOCKET")) + if err != nil { + return 2 + } + defer listener.Close() + + for { + connection, err := listener.Accept() + if err != nil { + return 0 + } + go func() { + defer connection.Close() + _, _ = io.Copy(connection, connection) + }() + } +} diff --git a/internal/tests/spec_update_test.go b/internal/tests/spec_update_test.go index 95bac22a..6de28cbb 100644 --- a/internal/tests/spec_update_test.go +++ b/internal/tests/spec_update_test.go @@ -351,7 +351,14 @@ func TestSpecUpdatePowerStateSuspend(t *testing.T) { _, err = tartRunProcessCmdline(tartVMName) require.NoError(t, err) - // Update the VM's specification and change it's power state + // Include a host process that would fail to start to ensure that host-process + // reconciliation cannot block the terminal power-state transition. + vm.HostProcesses = []v1.HostProcess{{ + Name: "unavailable", + Program: "/does/not/exist", + }} + + // Update the VM's specification and change its power state vm.PowerState = v1.PowerStateSuspended vm, err = devClient.VMs().Update(t.Context(), *vm) @@ -426,7 +433,14 @@ func TestSpecUpdatePowerStateStopped(t *testing.T) { _, err = tartRunProcessCmdline(tartVMName) require.NoError(t, err) - // Update the VM's specification and change it's power state + // Include a host process that would fail to start to ensure that host-process + // reconciliation cannot block the terminal power-state transition. + vm.HostProcesses = []v1.HostProcess{{ + Name: "unavailable", + Program: "/does/not/exist", + }} + + // Update the VM's specification and change its power state vm.PowerState = v1.PowerStateStopped vm, err = devClient.VMs().Update(t.Context(), *vm) diff --git a/internal/worker/hostprocess/process.go b/internal/worker/hostprocess/process.go new file mode 100644 index 00000000..b003db5f --- /dev/null +++ b/internal/worker/hostprocess/process.go @@ -0,0 +1,170 @@ +//nolint:err113,mnd // Preserve the original host-process errors and retry policy. +package hostprocess + +import ( + "context" + "errors" + "maps" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "time" + + "github.com/avast/retry-go/v4" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" +) + +type Process struct { + cancel context.CancelFunc + socketPath string + done chan struct{} +} + +func NewProcess( + spec v1.HostProcess, + workerName string, + vmName string, + controlSocket string, +) (*Process, error) { + // Create a directory where the host process will create its Unix socket + // + // To keep the Unix socket path short enough and fit the platform limit, + // we're specifically requesting "/var/tmp" instead of "/var/folders/.../T/". + // + // And unlike "/tmp", "/var/tmp" is not subject to macOS's nightly cleanup. + runtimeDir, err := os.MkdirTemp("/var/tmp", "orchard-hp-") + if err != nil { + return nil, err + } + + // Use a fixed socket name within the process-specific runtime directory + socketPath := filepath.Join(runtimeDir, "process.sock") + + // Route the Tart control socket through the runtime directory as well + // to work around macOS's 104-byte Unix-domain socket limit + controlSocket, err = shortenControlSocketPath(runtimeDir, controlSocket) + if err != nil { + return nil, errors.Join(err, os.RemoveAll(runtimeDir)) + } + + // Give the process an independently cancellable lifetime + cmdCtx, cmdCtxCancel := context.WithCancel(context.Background()) + + //nolint:gosec // Executing the configured host process is the purpose of this API + command := exec.CommandContext( + cmdCtx, + spec.Program, + expandArgs(spec.Args, workerName, vmName, controlSocket, socketPath)..., + ) + command.Dir = runtimeDir + + // Ensure that we terminate the host process and any children it spawns + command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + command.Cancel = func() error { + return syscall.Kill(-command.Process.Pid, syscall.SIGKILL) + } + + // Craft the environment starting from the host process specification + env := make(map[string]string) + + maps.Copy(env, spec.Env) + + // Preserve only PATH from the worker and make Orchard-managed variables authoritative + env["PATH"] = os.Getenv("PATH") + env["ORCHARD_WORKER_NAME"] = workerName + env["ORCHARD_VM_NAME"] = vmName + env["ORCHARD_VM_CONTROL_SOCKET"] = controlSocket + env["ORCHARD_PROCESS_SOCKET"] = socketPath + + // Convert the environment to the format expected by exec.Cmd + for name, value := range env { + command.Env = append(command.Env, name+"="+value) + } + + if err := command.Start(); err != nil { + cmdCtxCancel() + + return nil, errors.Join(err, os.RemoveAll(runtimeDir)) + } + + process := &Process{ + cancel: cmdCtxCancel, + socketPath: socketPath, + done: make(chan struct{}), + } + + go func() { + defer close(process.done) + + // Reap the process; callers observe termination through done regardless of exit status + _ = command.Wait() + _ = command.Cancel() + _ = os.RemoveAll(runtimeDir) + }() + + return process, nil +} + +func shortenControlSocketPath(runtimeDir string, controlSocketPath string) (string, error) { + // Make the symlink target absolute because a relative TART_HOME would + // otherwise be resolved from runtimeDir, breaking the symlink + absoluteControlSocketPath, err := filepath.Abs(controlSocketPath) + if err != nil { + return "", err + } + + // Place the VM's control socket alias in the runtime directory + aliasControlSocketPath := filepath.Join(runtimeDir, "vm.sock") + + if err := os.Symlink(absoluteControlSocketPath, aliasControlSocketPath); err != nil { + return "", err + } + + return aliasControlSocketPath, nil +} + +func (process *Process) Dial(ctx context.Context) (net.Conn, error) { + var dialer net.Dialer + + return retry.DoWithData(func() (net.Conn, error) { + select { + case <-process.done: + return nil, retry.Unrecoverable(errors.New("process exited before accepting connections")) + default: + // Continue + } + + return dialer.DialContext(ctx, "unix", process.socketPath) + }, + retry.Context(ctx), + retry.Attempts(200), + retry.Delay(50*time.Millisecond), + retry.DelayType(retry.FixedDelay), + retry.LastErrorOnly(true), + ) +} + +func (process *Process) Close() { + process.cancel() + <-process.done +} + +func expandArgs(args []string, workerName string, vmName string, controlSocket string, socketPath string) []string { + replacer := strings.NewReplacer( + "${ORCHARD_WORKER_NAME}", workerName, + "${ORCHARD_VM_NAME}", vmName, + "${ORCHARD_VM_CONTROL_SOCKET}", controlSocket, + "${ORCHARD_PROCESS_SOCKET}", socketPath, + ) + + var expanded []string + + for _, arg := range args { + expanded = append(expanded, replacer.Replace(arg)) + } + + return expanded +} diff --git a/internal/worker/hostprocess/process_test.go b/internal/worker/hostprocess/process_test.go new file mode 100644 index 00000000..d570c305 --- /dev/null +++ b/internal/worker/hostprocess/process_test.go @@ -0,0 +1,58 @@ +//nolint:noctx,testpackage,usetesting // Preserve the socket-path tests and their required short /var/tmp paths. +package hostprocess + +import ( + "net" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestShortenControlSocketPath(t *testing.T) { + // Create a control socket path that exceeds macOS's 104-byte limit + const controlSocketName = "control.sock" + + vmDir := filepath.Join(t.TempDir(), strings.Repeat("v", 104)) + require.NoError(t, os.MkdirAll(vmDir, 0o700)) + + controlSocketPath := filepath.Join(vmDir, controlSocketName) + require.Greater(t, len(controlSocketPath), 104) + + // Create Orchard's short per-process runtime directory + runtimeDir, err := os.MkdirTemp("/var/tmp", "orchard-hp-test-") + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, os.RemoveAll(runtimeDir)) + }) + + // Shorten the control socket path and verify the resulting alias + shortControlSocketPath, err := shortenControlSocketPath(runtimeDir, controlSocketPath) + require.NoError(t, err) + require.Equal(t, filepath.Join(runtimeDir, "vm.sock"), shortControlSocketPath) + require.Less(t, len(shortControlSocketPath), 104) + + // Model Tart binding the socket relative to the long VM directory + t.Chdir(vmDir) + + listener, err := net.Listen("unix", controlSocketName) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, listener.Close()) + }) + + // Model the host process running from Orchard's runtime directory + t.Chdir(runtimeDir) + + // Verify connecting through the long absolute path fails + connection, err := net.Dial("unix", controlSocketPath) + require.Error(t, err) + require.Nil(t, connection) + + // Connect to Tart's control socket through Orchard's short alias + connection, err = net.Dial("unix", shortControlSocketPath) + require.NoError(t, err) + require.NoError(t, connection.Close()) +} diff --git a/internal/worker/hostprocess/set.go b/internal/worker/hostprocess/set.go new file mode 100644 index 00000000..e6fab265 --- /dev/null +++ b/internal/worker/hostprocess/set.go @@ -0,0 +1,164 @@ +//nolint:contextcheck,err113 // Host processes have independent lifetimes; preserve the original errors. +package hostprocess + +import ( + "context" + "fmt" + "net" + "sync" + + "github.com/cirruslabs/orchard/internal/worker/ondiskname" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" +) + +// Set owns the host processes associated with one VM. +type Set struct { + workerName string + vmName string + onDiskName ondiskname.OnDiskName + processes map[string]*Process + started bool + mtx sync.Mutex +} + +func NewSet( + workerName string, + vmName string, + onDiskName ondiskname.OnDiskName, +) *Set { + return &Set{ + workerName: workerName, + vmName: vmName, + onDiskName: onDiskName, + processes: make(map[string]*Process), + } +} + +func (set *Set) Start(ctx context.Context, specs []v1.HostProcess) error { + set.mtx.Lock() + defer set.mtx.Unlock() + + set.stopLocked() + + return set.startLocked(ctx, specs) +} + +func (set *Set) Replace(ctx context.Context, specs []v1.HostProcess) error { + set.mtx.Lock() + defer set.mtx.Unlock() + + set.stopLocked() + + return set.startLocked(ctx, specs) +} + +func (set *Set) Dial(ctx context.Context, name string) (net.Conn, error) { + process := set.Lookup(name) + + if process == nil { + return nil, fmt.Errorf("host process %q is not running", name) + } + + connection, err := process.Dial(ctx) + if err != nil { + return nil, fmt.Errorf("failed to connect to host process %q: %w", name, err) + } + + return connection, nil +} + +func (set *Set) Lookup(name string) *Process { + set.mtx.Lock() + defer set.mtx.Unlock() + + return set.processes[name] +} + +func (set *Set) Ready() bool { + set.mtx.Lock() + defer set.mtx.Unlock() + + if !set.started { + return false + } + + for _, process := range set.processes { + select { + case <-process.done: + return false + default: + } + } + + return true +} + +func (set *Set) Stop() { + set.mtx.Lock() + defer set.mtx.Unlock() + + set.stopLocked() +} + +func (set *Set) startLocked(ctx context.Context, specs []v1.HostProcess) error { + set.started = false + + if len(specs) == 0 { + set.started = true + + return nil + } + + controlSocket, err := set.onDiskName.ControlSocketPath() + if err != nil { + return err + } + + // Start and register every process + // + // If any process fails to start, stop the entire set + // rather than leaving it partially running. + for _, spec := range specs { + process, err := NewProcess( + spec, + set.workerName, + set.vmName, + controlSocket, + ) + if err != nil { + set.stopLocked() + + return fmt.Errorf("failed to start host process %q: %w", spec.Name, err) + } + + set.processes[spec.Name] = process + } + + // A host process is ready once it accepts connections on its socket + for _, spec := range specs { + process := set.processes[spec.Name] + + connection, err := process.Dial(ctx) + if err != nil { + set.stopLocked() + + return fmt.Errorf("failed to connect to host process %q: %w", spec.Name, err) + } + + _ = connection.Close() + } + + set.started = true + + return nil +} + +func (set *Set) stopLocked() { + set.started = false + + for name, process := range set.processes { + delete(set.processes, name) + + process.Close() + } +} diff --git a/internal/worker/hostprocess/set_test.go b/internal/worker/hostprocess/set_test.go new file mode 100644 index 00000000..a07967e8 --- /dev/null +++ b/internal/worker/hostprocess/set_test.go @@ -0,0 +1,170 @@ +//nolint:goconst,noctx,testpackage // Preserve the original helper-process fixtures. +package hostprocess + +import ( + "net" + "os" + "testing" + + "github.com/cirruslabs/orchard/internal/worker/ondiskname" + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "github.com/stretchr/testify/require" +) + +const ( + testHelperArg = "--orchard-host-process-test-helper" + testHelperEnv = "ORCHARD_HOST_PROCESS_TEST_ENV" + testHelperInheritedEnv = "ORCHARD_HOST_PROCESS_TEST_INHERITED_ENV" +) + +func TestMain(m *testing.M) { + if len(os.Args) == 2 && os.Args[1] == testHelperArg { + os.Exit(runTestHelper()) + } + + os.Exit(m.Run()) +} + +func TestSetStartAndStop(t *testing.T) { + t.Setenv(testHelperInheritedEnv, "must-not-be-inherited") + + // Use this test binary as a host process + executable, err := os.Executable() + require.NoError(t, err) + + // Create an empty host process set + set := NewSet("", "", ondiskname.OnDiskName{}) + t.Cleanup(set.Stop) + + // Ensure that a new set is not ready until started + require.False(t, set.Ready()) + + // Start the first host process + require.NoError(t, set.Start(t.Context(), []v1.HostProcess{{ + Name: "first", + Program: executable, + Args: []string{testHelperArg}, + Env: map[string]string{ + testHelperEnv: "set", + "PATH": "must-not-win", + "ORCHARD_PROCESS_SOCKET": "/must-not-win", + }, + }})) + require.True(t, set.Ready()) + + // Ensure that the first host process is reachable + connection, err := set.Dial(t.Context(), "first") + require.NoError(t, err) + require.NoError(t, connection.Close()) + + // A failed replacement leaves the set not ready + require.Error(t, set.Replace(t.Context(), []v1.HostProcess{{ + Name: "invalid", + Program: "/does/not/exist", + }})) + require.False(t, set.Ready()) + _, err = set.Dial(t.Context(), "first") + require.Error(t, err) + + // Replace the first host process + require.NoError(t, set.Replace(t.Context(), []v1.HostProcess{{ + Name: "replacement", + Program: executable, + Args: []string{testHelperArg}, + Env: map[string]string{ + testHelperEnv: "set", + "PATH": "must-not-win", + "ORCHARD_PROCESS_SOCKET": "/must-not-win", + }, + }})) + require.True(t, set.Ready()) + + // Ensure that only the replacement host process is reachable + _, err = set.Dial(t.Context(), "first") + require.Error(t, err) + + connection, err = set.Dial(t.Context(), "replacement") + require.NoError(t, err) + require.NoError(t, connection.Close()) + + // Stop all host processes + set.Stop() + require.False(t, set.Ready()) + + // Ensure that the stopped host process is no longer reachable + _, err = set.Dial(t.Context(), "replacement") + require.Error(t, err) + + // Ensure that an explicitly started empty set is ready + require.NoError(t, set.Start(t.Context(), nil)) + require.True(t, set.Ready()) +} + +func TestSetStartReplacesExistingProcesses(t *testing.T) { + // Use this test binary as a host process + executable, err := os.Executable() + require.NoError(t, err) + + // Create a host process set and start a new host process + set := NewSet("", "", ondiskname.OnDiskName{}) + defer set.Stop() + + hostProcesses := []v1.HostProcess{{ + Name: "process", + Program: executable, + Args: []string{testHelperArg}, + Env: map[string]string{ + testHelperEnv: "set", + }, + }} + + require.NoError(t, set.Start(t.Context(), hostProcesses)) + + // Keep track of the original process to verify it is stopped + original := set.Lookup("process") + require.NotNil(t, original) + defer original.Close() + + // Start a new host process + require.NoError(t, set.Start(t.Context(), hostProcesses)) + + // Ensure that the original process was stopped + select { + case <-original.done: + // It was stopped, nice + default: + require.FailNow(t, "the original process is still running") + } + + // Ensure that a new ready process replaced the original + replacement := set.Lookup("process") + require.NotNil(t, replacement) + require.NotSame(t, original, replacement) + require.True(t, set.Ready()) +} + +func runTestHelper() int { + if os.Getenv(testHelperEnv) != "set" { + return 3 + } + if os.Getenv(testHelperInheritedEnv) != "" { + return 4 + } + if os.Getenv("PATH") == "must-not-win" { + return 5 + } + + listener, err := net.Listen("unix", os.Getenv("ORCHARD_PROCESS_SOCKET")) + if err != nil { + return 2 + } + defer listener.Close() + + for { + connection, err := listener.Accept() + if err != nil { + return 0 + } + _ = connection.Close() + } +} diff --git a/internal/worker/ondiskname/ondiskname.go b/internal/worker/ondiskname/ondiskname.go index a458d5b4..c7623716 100644 --- a/internal/worker/ondiskname/ondiskname.go +++ b/internal/worker/ondiskname/ondiskname.go @@ -3,9 +3,12 @@ package ondiskname import ( "errors" "fmt" - v1 "github.com/cirruslabs/orchard/pkg/resource/v1" + "os" + "path/filepath" "strconv" "strings" + + v1 "github.com/cirruslabs/orchard/pkg/resource/v1" ) var ( @@ -81,3 +84,19 @@ func Parse(s string) (OnDiskName, error) { func (odn OnDiskName) String() string { return fmt.Sprintf("%s-%s-%s-%d", prefix, odn.Name, odn.UID, odn.RestartCount) } + +func (odn OnDiskName) ControlSocketPath() (string, error) { + // Try user-overridden TART_HOME first + tartHome := os.Getenv("TART_HOME") + if tartHome == "" { + // Fall back to default TART_HOME + homeDir, err := os.UserHomeDir() + if err != nil { + return "", fmt.Errorf("failed to determine user home directory: %w", err) + } + + tartHome = filepath.Join(homeDir, ".tart") + } + + return filepath.Join(tartHome, "vms", odn.String(), "control.sock"), nil +} diff --git a/internal/worker/rpc.go b/internal/worker/rpc.go index f3a87ba6..c259bd38 100644 --- a/internal/worker/rpc.go +++ b/internal/worker/rpc.go @@ -67,6 +67,7 @@ func (worker *Worker) watchRPC(ctx context.Context, operationCtx context.Context } } +//nolint:nestif,protogetter // Preserve the original host-process forwarding implementation. func (worker *Worker) handlePortForward( ctx context.Context, client rpc.ControllerClient, @@ -87,47 +88,71 @@ func (worker *Worker) handlePortForward( return } - var host string + var targetConn net.Conn - if portForwardAction.VmUid == "" { - // Port-forwarding request to a worker - host = "localhost" - } else { - // Port-forwarding request to a VM, find that VM - vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool { - return item.Resource().UID == portForwardAction.VmUid - }) - if !ok { - worker.logger.Warnf("port forwarding failed: failed to get the VM: %v", err) + if target := portForwardAction.Target; target != nil { + // Sanity check + if portForwardAction.VmUid != "" || portForwardAction.Port != 0 { + worker.logger.Warn("port forwarding failed: target and legacy fields are mutually exclusive") return } - // Obtain VM's IP address - host, err = vm.IP(ctx) - if err != nil { - worker.logger.Warnf("port forwarding failed: failed to get VM's IP: %v", err) - + // Retrieve the typed host process target, it's the only possible target right now + hostProcess := target.GetHostProcess() + if hostProcess == nil || hostProcess.VmUid == "" || hostProcess.Name == "" { + worker.logger.Warn("port forwarding failed: invalid or unsupported target") return } - } - - // Connect to the VM's port - var vmConn net.Conn - if worker.dialer != nil { - vmConn, err = worker.dialer.DialContext(ctx, "tcp", - fmt.Sprintf("%s:%d", host, portForwardAction.Port)) + // Dial host process + targetConn, err = worker.dialHostProcess(ctx, hostProcess.VmUid, hostProcess.Name) + if err != nil { + worker.logger.Warnf("port forwarding failed: failed to connect to host process: %v", err) + return + } } else { - dialer := net.Dialer{} + var host string + + if portForwardAction.VmUid == "" { + // Port-forwarding request to a worker + host = "localhost" + } else { + // Port-forwarding request to a VM, find that VM + vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool { + return item.Resource().UID == portForwardAction.VmUid + }) + if !ok { + worker.logger.Warnf("port forwarding failed: failed to get VM with UID %q", + portForwardAction.VmUid) + + return + } + + // Obtain VM's IP address + host, err = vm.IP(ctx) + if err != nil { + worker.logger.Warnf("port forwarding failed: failed to get VM's IP: %v", err) + + return + } + } - vmConn, err = dialer.DialContext(ctx, "tcp", - fmt.Sprintf("%s:%d", host, portForwardAction.Port)) - } - if err != nil { - worker.logger.Warnf("port forwarding failed: failed to connect to the VM: %v", err) + // Connect to the VM's port + if worker.dialer != nil { + targetConn, err = worker.dialer.DialContext(ctx, "tcp", + fmt.Sprintf("%s:%d", host, portForwardAction.Port)) + } else { + dialer := net.Dialer{} - return + targetConn, err = dialer.DialContext(ctx, "tcp", + fmt.Sprintf("%s:%d", host, portForwardAction.Port)) + } + if err != nil { + worker.logger.Warnf("port forwarding failed: failed to connect to the VM: %v", err) + + return + } } // Proxy bytes @@ -143,7 +168,7 @@ func (worker *Worker) handlePortForward( }), } - _ = proxy.Connections(vmConn, grpcConn) + _ = proxy.Connections(targetConn, grpcConn) } func (worker *Worker) handleGetIP( diff --git a/internal/worker/rpcv2.go b/internal/worker/rpcv2.go index 7ddee44d..0f6cd516 100644 --- a/internal/worker/rpcv2.go +++ b/internal/worker/rpcv2.go @@ -71,10 +71,26 @@ func (worker *Worker) handlePortForwardV2(ctx context.Context, portForward *v1.P } } +//nolint:err113,perfsprint // Preserve the original host-process forwarding errors. func (worker *Worker) handlePortForwardV2Inner( ctx context.Context, portForward *v1.PortForwardAction, ) (net.Conn, error) { + if target := portForward.Target; target != nil { + // Sanity check + if portForward.VMUID != "" || portForward.Port != 0 { + return nil, fmt.Errorf("target and legacy fields are mutually exclusive") + } + + // Retrieve the typed host process target, it's the only possible target right now + if target.HostProcess == nil || target.HostProcess.VMUID == "" || target.HostProcess.Name == "" { + return nil, fmt.Errorf("invalid or unsupported target") + } + + // Dial host process + return worker.dialHostProcess(ctx, target.HostProcess.VMUID, target.HostProcess.Name) + } + var host string var err error @@ -87,7 +103,7 @@ func (worker *Worker) handlePortForwardV2Inner( return item.Resource().UID == portForward.VMUID }) if !ok { - return nil, fmt.Errorf("failed to get the VM: %v", err) + return nil, fmt.Errorf("failed to get VM with UID %q", portForward.VMUID) } // Obtain VM's IP address @@ -159,3 +175,28 @@ func (worker *Worker) handleGetIPV2Inner( return ip, nil } + +//nolint:err113,ireturn // Preserve the original VM lookup helper required by host processes. +func (worker *Worker) findVMByUID(uid string) (vmmanager.VM, error) { + vm, ok := lo.Find(worker.vmm.List(), func(item vmmanager.VM) bool { + return item.Resource().UID == uid + }) + if !ok { + return nil, fmt.Errorf("VM with UID %q not found", uid) + } + + if !vm.Started() { + return nil, fmt.Errorf("VM with UID %q is not running", uid) + } + + return vm, nil +} + +func (worker *Worker) dialHostProcess(ctx context.Context, vmUID string, name string) (net.Conn, error) { + vm, err := worker.findVMByUID(vmUID) + if err != nil { + return nil, err + } + + return vm.HostProcessSet().Dial(ctx, name) +} diff --git a/internal/worker/vmmanager/base/base.go b/internal/worker/vmmanager/base/base.go index bd276645..e896edfb 100644 --- a/internal/worker/vmmanager/base/base.go +++ b/internal/worker/vmmanager/base/base.go @@ -15,9 +15,12 @@ import ( "github.com/avast/retry-go/v4" "github.com/cirruslabs/orchard/internal/dialer" "github.com/cirruslabs/orchard/internal/worker/endpoint" + "github.com/cirruslabs/orchard/internal/worker/hostprocess" + "github.com/cirruslabs/orchard/internal/worker/ondiskname" "github.com/cirruslabs/orchard/pkg/client" v1 "github.com/cirruslabs/orchard/pkg/resource/v1" mapset "github.com/deckarep/golang-set/v2" + "github.com/samber/lo" "go.uber.org/zap" "golang.org/x/crypto/ssh" ) @@ -43,16 +46,18 @@ type VM struct { statusMessage atomic.Pointer[string] err atomic.Pointer[error] + hostProcesses *hostprocess.Set endpoints *endpoint.Set logger *zap.SugaredLogger } -func NewVM(logger *zap.SugaredLogger) *VM { +func NewVM(vmResource v1.VM, onDiskName ondiskname.OnDiskName, logger *zap.SugaredLogger) *VM { return &VM{ - conditions: mapset.NewSet(v1.ConditionTypeCloning), - endpoints: endpoint.NewSet(logger), - logger: logger, + conditions: mapset.NewSet(v1.ConditionTypeCloning), + hostProcesses: hostprocess.NewSet(vmResource.Worker, vmResource.Name, onDiskName), + endpoints: endpoint.NewSet(logger), + logger: logger, } } @@ -60,10 +65,18 @@ func (vm *VM) EndpointSet() *endpoint.Set { return vm.endpoints } +func (vm *VM) HostProcessSet() *hostprocess.Set { + return vm.hostProcesses +} + func (vm *VM) SetStarted(val bool) { vm.started.Store(val) } +func (vm *VM) Started() bool { + return vm.started.Load() +} + func (vm *VM) Status() v1.VMStatus { if vm.Err() != nil { return v1.VMStatusFailed @@ -113,6 +126,11 @@ func (vm *VM) Conditions() []v1.Condition { // The worker must observe transitions before applying a new specification. return []v1.Condition{ vm.conditionTypeToCondition(v1.ConditionTypeRunning), + { + Type: v1.ConditionTypeHostProcessesReady, + State: lo.Ternary(vm.HostProcessSet().Ready(), + v1.ConditionStateTrue, v1.ConditionStateFalse), + }, vm.conditionTypeToCondition(v1.ConditionTypeSuspending), vm.conditionTypeToCondition(v1.ConditionTypeStopping), } diff --git a/internal/worker/vmmanager/synthetic/synthetic.go b/internal/worker/vmmanager/synthetic/synthetic.go index e30d8148..dcd021a2 100644 --- a/internal/worker/vmmanager/synthetic/synthetic.go +++ b/internal/worker/vmmanager/synthetic/synthetic.go @@ -2,6 +2,7 @@ package synthetic import ( "context" + "fmt" "math/rand" "strings" "sync" @@ -41,13 +42,15 @@ func NewVM( "vm_restart_count", vmResource.RestartCount, ) + onDiskName := ondiskname.NewFromResource(vmResource) vm := &VM{ - onDiskName: ondiskname.NewFromResource(vmResource), + onDiskName: onDiskName, resource: vmResource, ctx: ctx, cancel: cancel, logger: logger, - VM: base.NewVM(logger), + + VM: base.NewVM(vmResource, onDiskName, logger), } vm.wg.Add(1) @@ -121,6 +124,8 @@ func (vm *VM) Start(eventStreamer *client.EventStreamer) { } func (vm *VM) Suspend() <-chan error { + vm.HostProcessSet().Stop() + vm.EndpointSet().Stop() errChan := make(chan error, 1) @@ -136,6 +141,8 @@ func (vm *VM) IP(ctx context.Context) (string, error) { } func (vm *VM) Stop() <-chan error { + vm.HostProcessSet().Stop() + vm.EndpointSet().Stop() errChan := make(chan error, 1) @@ -160,6 +167,13 @@ func (vm *VM) Delete() error { func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending, v1.ConditionTypeStopping) + if err := vm.HostProcessSet().Start(ctx, vm.resource.HostProcesses); err != nil { + vm.SetErr(fmt.Errorf("failed to start host processes: %w", err)) + + return + } + defer vm.HostProcessSet().Stop() + vm.EndpointSet().Start() defer vm.EndpointSet().Stop() diff --git a/internal/worker/vmmanager/tart/delete_test.go b/internal/worker/vmmanager/tart/delete_test.go index 5c59e309..a6da45b9 100644 --- a/internal/worker/vmmanager/tart/delete_test.go +++ b/internal/worker/vmmanager/tart/delete_test.go @@ -98,7 +98,7 @@ func newVMForDelete(t *testing.T, script string) (*VM, string) { logger: logger, ctx: ctx, cancel: cancel, - VM: base.NewVM(logger), + VM: base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, logger), } vm.ConditionsSet().Remove(v1.ConditionTypeCloning) diff --git a/internal/worker/vmmanager/tart/tart.go b/internal/worker/vmmanager/tart/tart.go index c6cc55d8..b0e157ce 100644 --- a/internal/worker/vmmanager/tart/tart.go +++ b/internal/worker/vmmanager/tart/tart.go @@ -66,8 +66,9 @@ func NewVM( "vm_restart_count", vmResource.RestartCount, ) + onDiskName := ondiskname.NewFromResource(vmResource) vm := &VM{ - onDiskName: ondiskname.NewFromResource(vmResource), + onDiskName: onDiskName, resource: vmResource, logger: logger, @@ -79,7 +80,7 @@ func NewVM( dialer: dialer, softnetPolicyUpdates: softnetPolicyUpdates, - VM: base.NewVM(logger), + VM: base.NewVM(vmResource, onDiskName, logger), } vm.wg.Add(1) @@ -304,6 +305,12 @@ func (vm *VM) run(ctx context.Context, eventStreamer *client.EventStreamer) { defer vm.ConditionsSet().RemoveAll(v1.ConditionTypeRunning, v1.ConditionTypeSuspending) resource := vm.Resource() + if err := vm.HostProcessSet().Start(ctx, resource.HostProcesses); err != nil { + vm.SetErr(fmt.Errorf("failed to start host processes: %w", err)) + + return + } + defer vm.HostProcessSet().Stop() vm.EndpointSet().Start() defer vm.EndpointSet().Stop() @@ -426,6 +433,8 @@ func (vm *VM) IP(ctx context.Context) (string, error) { } func (vm *VM) Suspend() <-chan error { + vm.HostProcessSet().Stop() + vm.EndpointSet().Stop() errCh := make(chan error, 1) @@ -459,6 +468,8 @@ func (vm *VM) Suspend() <-chan error { } func (vm *VM) Stop() <-chan error { + vm.HostProcessSet().Stop() + vm.EndpointSet().Stop() vm.stopMtx.Lock() defer vm.stopMtx.Unlock() diff --git a/internal/worker/vmmanager/tart/tart_test.go b/internal/worker/vmmanager/tart/tart_test.go index 2aad25e3..87688aa8 100644 --- a/internal/worker/vmmanager/tart/tart_test.go +++ b/internal/worker/vmmanager/tart/tart_test.go @@ -172,7 +172,7 @@ func newCloneTestVM(resource v1.VM) *VM { onDiskName: ondiskname.NewFromResource(resource), resource: resource, logger: logger, - VM: base.NewVM(logger), + VM: base.NewVM(resource, ondiskname.NewFromResource(resource), logger), } } diff --git a/internal/worker/vmmanager/vetu/vetu.go b/internal/worker/vmmanager/vetu/vetu.go index 59542368..afe7f038 100644 --- a/internal/worker/vmmanager/vetu/vetu.go +++ b/internal/worker/vmmanager/vetu/vetu.go @@ -66,7 +66,7 @@ func NewVM( dialer: dialer, - VM: base.NewVM(logger), + VM: base.NewVM(vmResource, ondiskname.NewFromResource(vmResource), logger), } vm.wg.Add(1) diff --git a/internal/worker/vmmanager/vmmanager.go b/internal/worker/vmmanager/vmmanager.go index 9aae59e0..a2ed664e 100644 --- a/internal/worker/vmmanager/vmmanager.go +++ b/internal/worker/vmmanager/vmmanager.go @@ -4,6 +4,7 @@ import ( "context" "github.com/cirruslabs/orchard/internal/worker/endpoint" + "github.com/cirruslabs/orchard/internal/worker/hostprocess" "github.com/cirruslabs/orchard/internal/worker/ondiskname" "github.com/cirruslabs/orchard/pkg/client" v1 "github.com/cirruslabs/orchard/pkg/resource/v1" @@ -17,6 +18,8 @@ type VM interface { OnDiskName() ondiskname.OnDiskName ImageFQN() *string EndpointSet() *endpoint.Set + HostProcessSet() *hostprocess.Set + Started() bool Status() v1.VMStatus StatusMessage() string Err() error diff --git a/internal/worker/worker.go b/internal/worker/worker.go index 1fcf9a07..98106b6b 100644 --- a/internal/worker/worker.go +++ b/internal/worker/worker.go @@ -663,6 +663,7 @@ func (worker *Worker) syncVMs( return nil } +//nolint:staticcheck // Preserve the original host-process specification comparison. func (worker *Worker) monitorRunningVM( ctx context.Context, vmResource *v1.VM, @@ -674,6 +675,26 @@ func (worker *Worker) monitorRunningVM( // Tracks whether any specification changes were applied without restarting the VM appliedInPlace := false + // When set, retries during the next sync instead of restarting the VM + deferSpecReconciliation := false + + // Try to apply the host process updates in-place or recover processes that exited unexpectedly + hostProcessesChanged := !currentVMResource.VMSpec.HostProcessesEqual(vmResource.VMSpec) + hostProcessesNeedRestart := currentVMResource.Generation == vmResource.Generation && + v1.ConditionIsTrue(vm.Conditions(), v1.ConditionTypeRunning) && + !vm.HostProcessSet().Ready() + + if vmResource.PowerState == v1.PowerStateRunning && (hostProcessesChanged || hostProcessesNeedRestart) { + if err := vm.HostProcessSet().Replace(ctx, vmResource.HostProcesses); err != nil { + worker.logger.Warnf("failed to update host processes for VM %q: %v", + vmResource.Name, err) + deferSpecReconciliation = true + } else { + currentVMResource.HostProcesses = vmResource.HostProcesses + appliedInPlace = true + } + } + // Endpoint specification changes do not require restarting the VM; // reconciliation happens separately below endpointsChanged := !v1.SemanticallyEqual( @@ -699,7 +720,10 @@ func (worker *Worker) monitorRunningVM( vm.SetResource(currentVMResource) } - worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions. + // Reconcile specification changes that could not be applied in-place + if !deferSpecReconciliation { + worker.reconcileRunningVM(vmResource, vm) //nolint:contextcheck // Event streams outlive sync sessions. + } var updateNeeded bool diff --git a/internal/worker/worker_stop_test.go b/internal/worker/worker_stop_test.go index 727e7d0a..58ba2ebf 100644 --- a/internal/worker/worker_stop_test.go +++ b/internal/worker/worker_stop_test.go @@ -87,7 +87,7 @@ func TestMonitorWaitsForStopBeforeApplyingGeneration(t *testing.T) { require.NoError(t, err) worker := &Worker{client: apiClient, runtime: runtime.NewSynthetic()} vm := &delayedStopVM{ - VM: &synthetic.VM{VM: base.NewVM(zap.NewNop().Sugar())}, + VM: &synthetic.VM{VM: base.NewVM(v1.VM{}, ondiskname.OnDiskName{}, zap.NewNop().Sugar())}, resource: v1.VM{Meta: v1.Meta{Name: "test-vm"}}, stopStarted: make(chan struct{}), stopResult: make(chan error), diff --git a/pkg/client/vms.go b/pkg/client/vms.go index a8190c91..096e8b93 100644 --- a/pkg/client/vms.go +++ b/pkg/client/vms.go @@ -169,6 +169,19 @@ func (service *VMsService) PortForward( }) } +func (service *VMsService) PortForwardHostProcess( + ctx context.Context, + name string, + hostProcessName string, + waitSeconds uint16, +) (net.Conn, error) { + return service.client.wsRequest(ctx, fmt.Sprintf("vms/%s/port-forward", url.PathEscape(name)), + map[string]string{ + "hostProcess": hostProcessName, + "wait": strconv.FormatUint(uint64(waitSeconds), 10), + }) +} + func (service *VMsService) Exec( ctx context.Context, name string, diff --git a/pkg/resource/v1/cmp_test.go b/pkg/resource/v1/cmp_test.go index a54c68ab..317d55b2 100644 --- a/pkg/resource/v1/cmp_test.go +++ b/pkg/resource/v1/cmp_test.go @@ -22,3 +22,12 @@ func TestSemanticallyEqualEquatesEmptySlices(t *testing.T) { } require.True(t, v1.SemanticallyEqual(nilSlicesSpec, emptySlicesSpec)) } + +func TestVMSpecHostProcessesEqualEquatesEmptySlices(t *testing.T) { + nilHostProcessesSpec := v1.VMSpec{} + emptyHostProcessesSpec := v1.VMSpec{ + HostProcesses: []v1.HostProcess{}, + } + + require.True(t, nilHostProcessesSpec.HostProcessesEqual(emptyHostProcessesSpec)) +} diff --git a/pkg/resource/v1/hostprocess.go b/pkg/resource/v1/hostprocess.go new file mode 100644 index 00000000..2c05efe6 --- /dev/null +++ b/pkg/resource/v1/hostprocess.go @@ -0,0 +1,36 @@ +//nolint:err113 // Preserve host-process validation messages. +package v1 + +import ( + "fmt" + + "github.com/cirruslabs/orchard/internal/simplename" + mapset "github.com/deckarep/golang-set/v2" +) + +// HostProcess describes a process run by the worker alongside a VM. Program is +// an executable path or a name resolved using the worker's PATH. +type HostProcess struct { + Name string `json:"name"` + Program string `json:"program"` + Args []string `json:"args,omitempty"` + Env map[string]string `json:"env,omitempty"` +} + +func ValidateHostProcesses(processes []HostProcess) error { + seenNames := mapset.NewSetWithSize[string](len(processes)) + + for _, process := range processes { + if process.Name == "" || simplename.Validate(process.Name) != nil { + return fmt.Errorf("host process %q is invalid", process.Name) + } + if !seenNames.Add(process.Name) { + return fmt.Errorf("host process %q is duplicated", process.Name) + } + if process.Program == "" { + return fmt.Errorf("host process %q has an empty program", process.Name) + } + } + + return nil +} diff --git a/pkg/resource/v1/service_account_role.go b/pkg/resource/v1/service_account_role.go index ecae894a..17be749e 100644 --- a/pkg/resource/v1/service_account_role.go +++ b/pkg/resource/v1/service_account_role.go @@ -10,11 +10,13 @@ var ErrUnsupportedServiceAccountRole = errors.New("unsupported service account r type ServiceAccountRole string const ( - ServiceAccountRoleComputeRead ServiceAccountRole = "compute:read" - ServiceAccountRoleComputeWrite ServiceAccountRole = "compute:write" - ServiceAccountRoleComputeConnect ServiceAccountRole = "compute:connect" - ServiceAccountRoleAdminRead ServiceAccountRole = "admin:read" - ServiceAccountRoleAdminWrite ServiceAccountRole = "admin:write" + ServiceAccountRoleComputeRead ServiceAccountRole = "compute:read" + ServiceAccountRoleComputeWrite ServiceAccountRole = "compute:write" + ServiceAccountRoleComputeConnect ServiceAccountRole = "compute:connect" + ServiceAccountRoleHostProcessWrite ServiceAccountRole = "host-process:write" + ServiceAccountRoleHostProcessConnect ServiceAccountRole = "host-process:connect" + ServiceAccountRoleAdminRead ServiceAccountRole = "admin:read" + ServiceAccountRoleAdminWrite ServiceAccountRole = "admin:write" ) func NewServiceAccountRole(name string) (ServiceAccountRole, error) { @@ -25,6 +27,10 @@ func NewServiceAccountRole(name string) (ServiceAccountRole, error) { return ServiceAccountRoleComputeWrite, nil case string(ServiceAccountRoleComputeConnect): return ServiceAccountRoleComputeConnect, nil + case string(ServiceAccountRoleHostProcessWrite): + return ServiceAccountRoleHostProcessWrite, nil + case string(ServiceAccountRoleHostProcessConnect): + return ServiceAccountRoleHostProcessConnect, nil case string(ServiceAccountRoleAdminRead): return ServiceAccountRoleAdminRead, nil case string(ServiceAccountRoleAdminWrite): @@ -39,6 +45,8 @@ func AllServiceAccountRoles() []ServiceAccountRole { ServiceAccountRoleComputeRead, ServiceAccountRoleComputeWrite, ServiceAccountRoleComputeConnect, + ServiceAccountRoleHostProcessWrite, + ServiceAccountRoleHostProcessConnect, ServiceAccountRoleAdminRead, ServiceAccountRoleAdminWrite, } diff --git a/pkg/resource/v1/v1.go b/pkg/resource/v1/v1.go index 2dc764bb..fd6d1114 100644 --- a/pkg/resource/v1/v1.go +++ b/pkg/resource/v1/v1.go @@ -150,6 +150,9 @@ func (vm *VM) Validate() error { switch vm.Runtime { case RuntimeVetu: + if len(vm.HostProcesses) != 0 { + return unsupportedFieldError("hostProcesses") + } if vm.NetSoftnetDeprecated || vm.NetSoftnet { return unsupportedFieldError("netSoftnet") } @@ -189,6 +192,7 @@ type VMSpec struct { // so this field defaults to that when not set. Runtime Runtime `json:"runtime,omitempty"` + HostProcesses []HostProcess `json:"hostProcesses,omitempty"` Endpoints []EndpointSpec `json:"endpoints,omitempty"` NetSoftnetDeprecated bool `json:"net-softnet,omitempty"` //nolint:tagliatelle // legacy JSON key NetSoftnet bool `json:"netSoftnet,omitempty"` @@ -203,6 +207,11 @@ func SemanticallyEqual[T any](current, desired T) bool { return cmp.Equal(current, desired, cmpopts.EquateEmpty()) } +func (vm VMSpec) HostProcessesEqual(other VMSpec) bool { + // Treat omitted and explicitly empty host process collections as the same specification + return cmp.Equal(vm.HostProcesses, other.HostProcesses, cmpopts.EquateEmpty()) +} + func (vm VMSpec) SoftnetEnabled() bool { return vm.NetSoftnetDeprecated || vm.NetSoftnet || len(vm.NetSoftnetAllow) != 0 || len(vm.NetSoftnetBlock) != 0 diff --git a/pkg/resource/v1/vm_condition.go b/pkg/resource/v1/vm_condition.go index fe818e6f..59223110 100644 --- a/pkg/resource/v1/vm_condition.go +++ b/pkg/resource/v1/vm_condition.go @@ -13,8 +13,9 @@ type Condition struct { type ConditionType string const ( - ConditionTypeScheduled ConditionType = "scheduled" - ConditionTypeRunning ConditionType = "running" + ConditionTypeScheduled ConditionType = "scheduled" + ConditionTypeRunning ConditionType = "running" + ConditionTypeHostProcessesReady ConditionType = "host-processes-ready" ConditionTypeCloning ConditionType = "cloning" ConditionTypeSuspending ConditionType = "suspending" diff --git a/pkg/resource/v1/watch_instruction.go b/pkg/resource/v1/watch_instruction.go index f7b08fd8..0c3ca7f1 100644 --- a/pkg/resource/v1/watch_instruction.go +++ b/pkg/resource/v1/watch_instruction.go @@ -1,3 +1,4 @@ +//nolint:tagliatelle // Preserve the original vmUID keys for wire compatibility. package v1 type WatchInstruction struct { @@ -7,9 +8,19 @@ type WatchInstruction struct { } type PortForwardAction struct { - Session string `json:"session"` - VMUID string `json:"vmUID"` - Port uint16 `json:"port"` + Session string `json:"session"` + VMUID string `json:"vmUID"` + Port uint16 `json:"port"` + Target *PortForwardTarget `json:"target,omitempty"` +} + +type PortForwardTarget struct { + HostProcess *PortForwardTargetHostProcess `json:"hostProcess,omitempty"` +} + +type PortForwardTargetHostProcess struct { + VMUID string `json:"vmUID"` + Name string `json:"name"` } type SyncVMsAction struct { diff --git a/rpc/orchard.pb.go b/rpc/orchard.pb.go index 6f7cbef0..0b544537 100644 --- a/rpc/orchard.pb.go +++ b/rpc/orchard.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.34.2 +// protoc-gen-go v1.36.11 // protoc (unknown) // source: orchard.proto @@ -12,6 +12,7 @@ import ( emptypb "google.golang.org/protobuf/types/known/emptypb" reflect "reflect" sync "sync" + unsafe "unsafe" ) const ( @@ -22,25 +23,22 @@ const ( ) type WatchInstruction struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache - unknownFields protoimpl.UnknownFields - - // Types that are assignable to Action: + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Action: // // *WatchInstruction_PortForwardAction // *WatchInstruction_SyncVmsAction // *WatchInstruction_ResolveIpAction - Action isWatchInstruction_Action `protobuf_oneof:"action"` + Action isWatchInstruction_Action `protobuf_oneof:"action"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WatchInstruction) Reset() { *x = WatchInstruction{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[0] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *WatchInstruction) String() string { @@ -51,7 +49,7 @@ func (*WatchInstruction) ProtoMessage() {} func (x *WatchInstruction) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[0] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -66,30 +64,36 @@ func (*WatchInstruction) Descriptor() ([]byte, []int) { return file_orchard_proto_rawDescGZIP(), []int{0} } -func (m *WatchInstruction) GetAction() isWatchInstruction_Action { - if m != nil { - return m.Action +func (x *WatchInstruction) GetAction() isWatchInstruction_Action { + if x != nil { + return x.Action } return nil } func (x *WatchInstruction) GetPortForwardAction() *WatchInstruction_PortForward { - if x, ok := x.GetAction().(*WatchInstruction_PortForwardAction); ok { - return x.PortForwardAction + if x != nil { + if x, ok := x.Action.(*WatchInstruction_PortForwardAction); ok { + return x.PortForwardAction + } } return nil } func (x *WatchInstruction) GetSyncVmsAction() *WatchInstruction_SyncVMs { - if x, ok := x.GetAction().(*WatchInstruction_SyncVmsAction); ok { - return x.SyncVmsAction + if x != nil { + if x, ok := x.Action.(*WatchInstruction_SyncVmsAction); ok { + return x.SyncVmsAction + } } return nil } func (x *WatchInstruction) GetResolveIpAction() *WatchInstruction_ResolveIP { - if x, ok := x.GetAction().(*WatchInstruction_ResolveIpAction); ok { - return x.ResolveIpAction + if x != nil { + if x, ok := x.Action.(*WatchInstruction_ResolveIpAction); ok { + return x.ResolveIpAction + } } return nil } @@ -117,20 +121,17 @@ func (*WatchInstruction_SyncVmsAction) isWatchInstruction_Action() {} func (*WatchInstruction_ResolveIpAction) isWatchInstruction_Action() {} type PortForwardData struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` unknownFields protoimpl.UnknownFields - - Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + sizeCache protoimpl.SizeCache } func (x *PortForwardData) Reset() { *x = PortForwardData{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[1] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *PortForwardData) String() string { @@ -141,7 +142,7 @@ func (*PortForwardData) ProtoMessage() {} func (x *PortForwardData) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[1] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -164,21 +165,18 @@ func (x *PortForwardData) GetData() []byte { } type ResolveIPResult struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Session string `protobuf:"bytes,1,opt,name=session,proto3" json:"session,omitempty"` + Ip string `protobuf:"bytes,2,opt,name=ip,proto3" json:"ip,omitempty"` unknownFields protoimpl.UnknownFields - - Session string `protobuf:"bytes,1,opt,name=session,proto3" json:"session,omitempty"` - Ip string `protobuf:"bytes,2,opt,name=ip,proto3" json:"ip,omitempty"` + sizeCache protoimpl.SizeCache } func (x *ResolveIPResult) Reset() { *x = ResolveIPResult{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[2] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *ResolveIPResult) String() string { @@ -189,7 +187,7 @@ func (*ResolveIPResult) ProtoMessage() {} func (x *ResolveIPResult) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[2] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -219,25 +217,24 @@ func (x *ResolveIPResult) GetIp() string { } type WatchInstruction_PortForward struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache - unknownFields protoimpl.UnknownFields - - // we can have multiple port forwards for the same vm/port pair - // let's distinguish them by a unique session + state protoimpl.MessageState `protogen:"open.v1"` + // Distinguish multiple port forwards to the same VM/port pair or a target Session string `protobuf:"bytes,1,opt,name=session,proto3" json:"session,omitempty"` - // can be empty to request port-forwarding to the worker itself + // Legacy port-forwarding destination that supports workers and VMs VmUid string `protobuf:"bytes,2,opt,name=vm_uid,json=vmUid,proto3" json:"vm_uid,omitempty"` Port uint32 `protobuf:"varint,3,opt,name=port,proto3" json:"port,omitempty"` + // Typed alternative to the legacy port-forwarding destination, + // adding support for host processes + Target *WatchInstruction_PortForward_Target `protobuf:"bytes,4,opt,name=target,proto3" json:"target,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WatchInstruction_PortForward) Reset() { *x = WatchInstruction_PortForward{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[3] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *WatchInstruction_PortForward) String() string { @@ -248,7 +245,7 @@ func (*WatchInstruction_PortForward) ProtoMessage() {} func (x *WatchInstruction_PortForward) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[3] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -284,19 +281,24 @@ func (x *WatchInstruction_PortForward) GetPort() uint32 { return 0 } +func (x *WatchInstruction_PortForward) GetTarget() *WatchInstruction_PortForward_Target { + if x != nil { + return x.Target + } + return nil +} + type WatchInstruction_SyncVMs struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WatchInstruction_SyncVMs) Reset() { *x = WatchInstruction_SyncVMs{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[4] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *WatchInstruction_SyncVMs) String() string { @@ -307,7 +309,7 @@ func (*WatchInstruction_SyncVMs) ProtoMessage() {} func (x *WatchInstruction_SyncVMs) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[4] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -323,23 +325,20 @@ func (*WatchInstruction_SyncVMs) Descriptor() ([]byte, []int) { } type WatchInstruction_ResolveIP struct { - state protoimpl.MessageState - sizeCache protoimpl.SizeCache - unknownFields protoimpl.UnknownFields - + state protoimpl.MessageState `protogen:"open.v1"` // we can have multiple IP resolution requests for the same vm // let's distinguish them by a unique session - Session string `protobuf:"bytes,1,opt,name=session,proto3" json:"session,omitempty"` - VmUid string `protobuf:"bytes,2,opt,name=vm_uid,json=vmUid,proto3" json:"vm_uid,omitempty"` + Session string `protobuf:"bytes,1,opt,name=session,proto3" json:"session,omitempty"` + VmUid string `protobuf:"bytes,2,opt,name=vm_uid,json=vmUid,proto3" json:"vm_uid,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *WatchInstruction_ResolveIP) Reset() { *x = WatchInstruction_ResolveIP{} - if protoimpl.UnsafeEnabled { - mi := &file_orchard_proto_msgTypes[5] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) - } + mi := &file_orchard_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) } func (x *WatchInstruction_ResolveIP) String() string { @@ -350,7 +349,7 @@ func (*WatchInstruction_ResolveIP) ProtoMessage() {} func (x *WatchInstruction_ResolveIP) ProtoReflect() protoreflect.Message { mi := &file_orchard_proto_msgTypes[5] - if protoimpl.UnsafeEnabled && x != nil { + if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { ms.StoreMessageInfo(mi) @@ -379,97 +378,205 @@ func (x *WatchInstruction_ResolveIP) GetVmUid() string { return "" } -var File_orchard_proto protoreflect.FileDescriptor +// Target identifies where the worker forwards the byte stream +type WatchInstruction_PortForward_Target struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Value: + // + // *WatchInstruction_PortForward_Target_HostProcess_ + Value isWatchInstruction_PortForward_Target_Value `protobuf_oneof:"value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *WatchInstruction_PortForward_Target) Reset() { + *x = WatchInstruction_PortForward_Target{} + mi := &file_orchard_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *WatchInstruction_PortForward_Target) String() string { + return protoimpl.X.MessageStringOf(x) +} -var file_orchard_proto_rawDesc = []byte{ - 0x0a, 0x0d, 0x6f, 0x72, 0x63, 0x68, 0x61, 0x72, 0x64, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, - 0x1b, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, - 0x2f, 0x65, 0x6d, 0x70, 0x74, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x9a, 0x03, 0x0a, - 0x10, 0x57, 0x61, 0x74, 0x63, 0x68, 0x49, 0x6e, 0x73, 0x74, 0x72, 0x75, 0x63, 0x74, 0x69, 0x6f, - 0x6e, 0x12, 0x4f, 0x0a, 0x13, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x66, 0x6f, 0x72, 0x77, 0x61, 0x72, - 0x64, 0x5f, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, - 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x49, 0x6e, 0x73, 0x74, 0x72, 0x75, 0x63, 0x74, 0x69, 0x6f, - 0x6e, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x48, 0x00, 0x52, - 0x11, 0x70, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x41, 0x63, 0x74, 0x69, - 0x6f, 0x6e, 0x12, 0x43, 0x0a, 0x0f, 0x73, 0x79, 0x6e, 0x63, 0x5f, 0x76, 0x6d, 0x73, 0x5f, 0x61, - 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x57, 0x61, - 0x74, 0x63, 0x68, 0x49, 0x6e, 0x73, 0x74, 0x72, 0x75, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x53, - 0x79, 0x6e, 0x63, 0x56, 0x4d, 0x73, 0x48, 0x00, 0x52, 0x0d, 0x73, 0x79, 0x6e, 0x63, 0x56, 0x6d, - 0x73, 0x41, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x49, 0x0a, 0x11, 0x72, 0x65, 0x73, 0x6f, 0x6c, - 0x76, 0x65, 0x5f, 0x69, 0x70, 0x5f, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, - 0x28, 0x0b, 0x32, 0x1b, 0x2e, 0x57, 0x61, 0x74, 0x63, 0x68, 0x49, 0x6e, 0x73, 0x74, 0x72, 0x75, - 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x76, 0x65, 0x49, 0x50, 0x48, - 0x00, 0x52, 0x0f, 0x72, 0x65, 0x73, 0x6f, 0x6c, 0x76, 0x65, 0x49, 0x70, 0x41, 0x63, 0x74, 0x69, - 0x6f, 0x6e, 0x1a, 0x52, 0x0a, 0x0b, 0x50, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, - 0x64, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x0a, 0x06, 0x76, - 0x6d, 0x5f, 0x75, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x6d, 0x55, - 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d, - 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x1a, 0x09, 0x0a, 0x07, 0x53, 0x79, 0x6e, 0x63, 0x56, 0x4d, - 0x73, 0x1a, 0x3c, 0x0a, 0x09, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x76, 0x65, 0x49, 0x50, 0x12, 0x18, - 0x0a, 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x0a, 0x06, 0x76, 0x6d, 0x5f, 0x75, - 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x76, 0x6d, 0x55, 0x69, 0x64, 0x42, - 0x08, 0x0a, 0x06, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x25, 0x0a, 0x0f, 0x50, 0x6f, 0x72, - 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x44, 0x61, 0x74, 0x61, 0x12, 0x12, 0x0a, 0x04, - 0x64, 0x61, 0x74, 0x61, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61, - 0x22, 0x3b, 0x0a, 0x0f, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x76, 0x65, 0x49, 0x50, 0x52, 0x65, 0x73, - 0x75, 0x6c, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x01, - 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x73, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x12, 0x0e, 0x0a, - 0x02, 0x69, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x70, 0x32, 0xb0, 0x01, - 0x0a, 0x0a, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x12, 0x34, 0x0a, 0x05, - 0x57, 0x61, 0x74, 0x63, 0x68, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x11, 0x2e, - 0x57, 0x61, 0x74, 0x63, 0x68, 0x49, 0x6e, 0x73, 0x74, 0x72, 0x75, 0x63, 0x74, 0x69, 0x6f, 0x6e, - 0x30, 0x01, 0x12, 0x35, 0x0a, 0x0b, 0x50, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, - 0x64, 0x12, 0x10, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, 0x64, 0x44, - 0x61, 0x74, 0x61, 0x1a, 0x10, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x46, 0x6f, 0x72, 0x77, 0x61, 0x72, - 0x64, 0x44, 0x61, 0x74, 0x61, 0x28, 0x01, 0x30, 0x01, 0x12, 0x35, 0x0a, 0x09, 0x52, 0x65, 0x73, - 0x6f, 0x6c, 0x76, 0x65, 0x49, 0x50, 0x12, 0x10, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x6c, 0x76, 0x65, - 0x49, 0x50, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, - 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, - 0x42, 0x23, 0x5a, 0x21, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x63, - 0x69, 0x72, 0x72, 0x75, 0x73, 0x6c, 0x61, 0x62, 0x73, 0x2f, 0x6f, 0x72, 0x63, 0x68, 0x61, 0x72, - 0x64, 0x2f, 0x72, 0x70, 0x63, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, +func (*WatchInstruction_PortForward_Target) ProtoMessage() {} + +func (x *WatchInstruction_PortForward_Target) ProtoReflect() protoreflect.Message { + mi := &file_orchard_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use WatchInstruction_PortForward_Target.ProtoReflect.Descriptor instead. +func (*WatchInstruction_PortForward_Target) Descriptor() ([]byte, []int) { + return file_orchard_proto_rawDescGZIP(), []int{0, 0, 0} +} + +func (x *WatchInstruction_PortForward_Target) GetValue() isWatchInstruction_PortForward_Target_Value { + if x != nil { + return x.Value + } + return nil +} + +func (x *WatchInstruction_PortForward_Target) GetHostProcess() *WatchInstruction_PortForward_Target_HostProcess { + if x != nil { + if x, ok := x.Value.(*WatchInstruction_PortForward_Target_HostProcess_); ok { + return x.HostProcess + } + } + return nil +} + +type isWatchInstruction_PortForward_Target_Value interface { + isWatchInstruction_PortForward_Target_Value() +} + +type WatchInstruction_PortForward_Target_HostProcess_ struct { + HostProcess *WatchInstruction_PortForward_Target_HostProcess `protobuf:"bytes,1,opt,name=host_process,json=hostProcess,proto3,oneof"` +} + +func (*WatchInstruction_PortForward_Target_HostProcess_) isWatchInstruction_PortForward_Target_Value() { +} + +// Forward the byte stream to a host process, identified by +// a VM UID and a host process name +type WatchInstruction_PortForward_Target_HostProcess struct { + state protoimpl.MessageState `protogen:"open.v1"` + VmUid string `protobuf:"bytes,1,opt,name=vm_uid,json=vmUid,proto3" json:"vm_uid,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *WatchInstruction_PortForward_Target_HostProcess) Reset() { + *x = WatchInstruction_PortForward_Target_HostProcess{} + mi := &file_orchard_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *WatchInstruction_PortForward_Target_HostProcess) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*WatchInstruction_PortForward_Target_HostProcess) ProtoMessage() {} + +func (x *WatchInstruction_PortForward_Target_HostProcess) ProtoReflect() protoreflect.Message { + mi := &file_orchard_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use WatchInstruction_PortForward_Target_HostProcess.ProtoReflect.Descriptor instead. +func (*WatchInstruction_PortForward_Target_HostProcess) Descriptor() ([]byte, []int) { + return file_orchard_proto_rawDescGZIP(), []int{0, 0, 0, 0} +} + +func (x *WatchInstruction_PortForward_Target_HostProcess) GetVmUid() string { + if x != nil { + return x.VmUid + } + return "" } +func (x *WatchInstruction_PortForward_Target_HostProcess) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +var File_orchard_proto protoreflect.FileDescriptor + +const file_orchard_proto_rawDesc = "" + + "\n" + + "\rorchard.proto\x1a\x1bgoogle/protobuf/empty.proto\"\xfe\x04\n" + + "\x10WatchInstruction\x12O\n" + + "\x13port_forward_action\x18\x01 \x01(\v2\x1d.WatchInstruction.PortForwardH\x00R\x11portForwardAction\x12C\n" + + "\x0fsync_vms_action\x18\x02 \x01(\v2\x19.WatchInstruction.SyncVMsH\x00R\rsyncVmsAction\x12I\n" + + "\x11resolve_ip_action\x18\x03 \x01(\v2\x1b.WatchInstruction.ResolveIPH\x00R\x0fresolveIpAction\x1a\xb5\x02\n" + + "\vPortForward\x12\x18\n" + + "\asession\x18\x01 \x01(\tR\asession\x12\x15\n" + + "\x06vm_uid\x18\x02 \x01(\tR\x05vmUid\x12\x12\n" + + "\x04port\x18\x03 \x01(\rR\x04port\x12<\n" + + "\x06target\x18\x04 \x01(\v2$.WatchInstruction.PortForward.TargetR\x06target\x1a\xa2\x01\n" + + "\x06Target\x12U\n" + + "\fhost_process\x18\x01 \x01(\v20.WatchInstruction.PortForward.Target.HostProcessH\x00R\vhostProcess\x1a8\n" + + "\vHostProcess\x12\x15\n" + + "\x06vm_uid\x18\x01 \x01(\tR\x05vmUid\x12\x12\n" + + "\x04name\x18\x02 \x01(\tR\x04nameB\a\n" + + "\x05value\x1a\t\n" + + "\aSyncVMs\x1a<\n" + + "\tResolveIP\x12\x18\n" + + "\asession\x18\x01 \x01(\tR\asession\x12\x15\n" + + "\x06vm_uid\x18\x02 \x01(\tR\x05vmUidB\b\n" + + "\x06action\"%\n" + + "\x0fPortForwardData\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\";\n" + + "\x0fResolveIPResult\x12\x18\n" + + "\asession\x18\x01 \x01(\tR\asession\x12\x0e\n" + + "\x02ip\x18\x02 \x01(\tR\x02ip2\xb0\x01\n" + + "\n" + + "Controller\x124\n" + + "\x05Watch\x12\x16.google.protobuf.Empty\x1a\x11.WatchInstruction0\x01\x125\n" + + "\vPortForward\x12\x10.PortForwardData\x1a\x10.PortForwardData(\x010\x01\x125\n" + + "\tResolveIP\x12\x10.ResolveIPResult\x1a\x16.google.protobuf.EmptyB#Z!github.com/cirruslabs/orchard/rpcb\x06proto3" + var ( file_orchard_proto_rawDescOnce sync.Once - file_orchard_proto_rawDescData = file_orchard_proto_rawDesc + file_orchard_proto_rawDescData []byte ) func file_orchard_proto_rawDescGZIP() []byte { file_orchard_proto_rawDescOnce.Do(func() { - file_orchard_proto_rawDescData = protoimpl.X.CompressGZIP(file_orchard_proto_rawDescData) + file_orchard_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_orchard_proto_rawDesc), len(file_orchard_proto_rawDesc))) }) return file_orchard_proto_rawDescData } -var file_orchard_proto_msgTypes = make([]protoimpl.MessageInfo, 6) +var file_orchard_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_orchard_proto_goTypes = []any{ - (*WatchInstruction)(nil), // 0: WatchInstruction - (*PortForwardData)(nil), // 1: PortForwardData - (*ResolveIPResult)(nil), // 2: ResolveIPResult - (*WatchInstruction_PortForward)(nil), // 3: WatchInstruction.PortForward - (*WatchInstruction_SyncVMs)(nil), // 4: WatchInstruction.SyncVMs - (*WatchInstruction_ResolveIP)(nil), // 5: WatchInstruction.ResolveIP - (*emptypb.Empty)(nil), // 6: google.protobuf.Empty + (*WatchInstruction)(nil), // 0: WatchInstruction + (*PortForwardData)(nil), // 1: PortForwardData + (*ResolveIPResult)(nil), // 2: ResolveIPResult + (*WatchInstruction_PortForward)(nil), // 3: WatchInstruction.PortForward + (*WatchInstruction_SyncVMs)(nil), // 4: WatchInstruction.SyncVMs + (*WatchInstruction_ResolveIP)(nil), // 5: WatchInstruction.ResolveIP + (*WatchInstruction_PortForward_Target)(nil), // 6: WatchInstruction.PortForward.Target + (*WatchInstruction_PortForward_Target_HostProcess)(nil), // 7: WatchInstruction.PortForward.Target.HostProcess + (*emptypb.Empty)(nil), // 8: google.protobuf.Empty } var file_orchard_proto_depIdxs = []int32{ 3, // 0: WatchInstruction.port_forward_action:type_name -> WatchInstruction.PortForward 4, // 1: WatchInstruction.sync_vms_action:type_name -> WatchInstruction.SyncVMs 5, // 2: WatchInstruction.resolve_ip_action:type_name -> WatchInstruction.ResolveIP - 6, // 3: Controller.Watch:input_type -> google.protobuf.Empty - 1, // 4: Controller.PortForward:input_type -> PortForwardData - 2, // 5: Controller.ResolveIP:input_type -> ResolveIPResult - 0, // 6: Controller.Watch:output_type -> WatchInstruction - 1, // 7: Controller.PortForward:output_type -> PortForwardData - 6, // 8: Controller.ResolveIP:output_type -> google.protobuf.Empty - 6, // [6:9] is the sub-list for method output_type - 3, // [3:6] is the sub-list for method input_type - 3, // [3:3] is the sub-list for extension type_name - 3, // [3:3] is the sub-list for extension extendee - 0, // [0:3] is the sub-list for field type_name + 6, // 3: WatchInstruction.PortForward.target:type_name -> WatchInstruction.PortForward.Target + 7, // 4: WatchInstruction.PortForward.Target.host_process:type_name -> WatchInstruction.PortForward.Target.HostProcess + 8, // 5: Controller.Watch:input_type -> google.protobuf.Empty + 1, // 6: Controller.PortForward:input_type -> PortForwardData + 2, // 7: Controller.ResolveIP:input_type -> ResolveIPResult + 0, // 8: Controller.Watch:output_type -> WatchInstruction + 1, // 9: Controller.PortForward:output_type -> PortForwardData + 8, // 10: Controller.ResolveIP:output_type -> google.protobuf.Empty + 8, // [8:11] is the sub-list for method output_type + 5, // [5:8] is the sub-list for method input_type + 5, // [5:5] is the sub-list for extension type_name + 5, // [5:5] is the sub-list for extension extendee + 0, // [0:5] is the sub-list for field type_name } func init() { file_orchard_proto_init() } @@ -477,92 +584,21 @@ func file_orchard_proto_init() { if File_orchard_proto != nil { return } - if !protoimpl.UnsafeEnabled { - file_orchard_proto_msgTypes[0].Exporter = func(v any, i int) any { - switch v := v.(*WatchInstruction); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_orchard_proto_msgTypes[1].Exporter = func(v any, i int) any { - switch v := v.(*PortForwardData); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_orchard_proto_msgTypes[2].Exporter = func(v any, i int) any { - switch v := v.(*ResolveIPResult); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_orchard_proto_msgTypes[3].Exporter = func(v any, i int) any { - switch v := v.(*WatchInstruction_PortForward); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_orchard_proto_msgTypes[4].Exporter = func(v any, i int) any { - switch v := v.(*WatchInstruction_SyncVMs); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - file_orchard_proto_msgTypes[5].Exporter = func(v any, i int) any { - switch v := v.(*WatchInstruction_ResolveIP); i { - case 0: - return &v.state - case 1: - return &v.sizeCache - case 2: - return &v.unknownFields - default: - return nil - } - } - } file_orchard_proto_msgTypes[0].OneofWrappers = []any{ (*WatchInstruction_PortForwardAction)(nil), (*WatchInstruction_SyncVmsAction)(nil), (*WatchInstruction_ResolveIpAction)(nil), } + file_orchard_proto_msgTypes[6].OneofWrappers = []any{ + (*WatchInstruction_PortForward_Target_HostProcess_)(nil), + } type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), - RawDescriptor: file_orchard_proto_rawDesc, + RawDescriptor: unsafe.Slice(unsafe.StringData(file_orchard_proto_rawDesc), len(file_orchard_proto_rawDesc)), NumEnums: 0, - NumMessages: 6, + NumMessages: 8, NumExtensions: 0, NumServices: 1, }, @@ -571,7 +607,6 @@ func file_orchard_proto_init() { MessageInfos: file_orchard_proto_msgTypes, }.Build() File_orchard_proto = out.File - file_orchard_proto_rawDesc = nil file_orchard_proto_goTypes = nil file_orchard_proto_depIdxs = nil } diff --git a/rpc/orchard.proto b/rpc/orchard.proto index 7328e7aa..ca11658c 100644 --- a/rpc/orchard.proto +++ b/rpc/orchard.proto @@ -18,12 +18,30 @@ service Controller { message WatchInstruction { message PortForward { - // we can have multiple port forwards for the same vm/port pair - // let's distinguish them by a unique session + // Target identifies where the worker forwards the byte stream + message Target { + // Forward the byte stream to a host process, identified by + // a VM UID and a host process name + message HostProcess { + string vm_uid = 1; + string name = 2; + } + + oneof value { + HostProcess host_process = 1; + } + } + + // Distinguish multiple port forwards to the same VM/port pair or a target string session = 1; - // can be empty to request port-forwarding to the worker itself + + // Legacy port-forwarding destination that supports workers and VMs string vm_uid = 2; uint32 port = 3; + + // Typed alternative to the legacy port-forwarding destination, + // adding support for host processes + Target target = 4; } message SyncVMs { // nothing for now