From 3341c2e992248e68511a3e499e42d679a4a0d4b2 Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 16 Jul 2026 23:08:18 +0000 Subject: [PATCH] shuttle,spindle/engines/microvm: move cmd exec stdio to a vsock channel Signed-off-by: dawn --- cmd/spindle-microvm-run/main_linux.go | 2 +- shuttle/src/activation.rs | 5 +++-- shuttle/src/command.rs | 15 ++++++++++++++- shuttle/src/exec.rs | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------- shuttle/src/protocol.rs | 4 +--- shuttle/src/session.rs | 5 +++-- spindle/agentproto/protocol.go | 2 +- shuttle/src/gen/file_descriptor_set.bin | 0 spindle/agentproto/gen/agent.pb.go | 191 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------------------------------------- spindle/engines/microvm/README.md | 5 +++-- spindle/engines/microvm/agent.go | 84 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------- spindle/engines/microvm/engine.go | 2 +- spindle/agentproto/spindle/agent/v1/agent.proto | 17 ++++++++--------- shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs | 13 ++++--------- 14 file(s) changed, 255 insertion(s)(+), 179 deletion(s)(-) diff --git a/cmd/spindle-microvm-run/main_linux.go b/cmd/spindle-microvm-run/main_linux.go --- a/cmd/spindle-microvm-run/main_linux.go +++ b/cmd/spindle-microvm-run/main_linux.go @@ -220,7 +220,7 @@ } defer dnsProxy.Close() - session := microvm.NewAgentSession(conn, logger) + session := microvm.NewAgentSession(conn, vm.CID(), logger) initCtx, cancelInit := context.WithTimeout(ctx, 30*time.Second) defer cancelInit() diff --git a/shuttle/src/activation.rs b/shuttle/src/activation.rs --- a/shuttle/src/activation.rs +++ b/shuttle/src/activation.rs @@ -201,11 +201,12 @@ OutKind::Stdout => stdout.extend_from_slice(&event.data), OutKind::Stderr => { stderr.extend_from_slice(&event.data); - let data = String::from_utf8_lossy(&event.data).into_owned(); let _ = out .send(Message { id: id.to_owned(), - exec_stderr: Some(v1::ExecStderr { data }), + exec_stderr: Some(v1::ExecStderr { + data: event.data.into(), + }), ..Default::default() }) .await; diff --git a/shuttle/src/command.rs b/shuttle/src/command.rs --- a/shuttle/src/command.rs +++ b/shuttle/src/command.rs @@ -8,7 +8,7 @@ use std::process::Stdio; use std::time::Duration; use tokio::io::{AsyncRead, AsyncReadExt}; -use tokio::process::{Child, Command}; +use tokio::process::{Child, ChildStdin, Command}; use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio::task::JoinHandle; use tracing::warn; @@ -22,6 +22,7 @@ pub timeout: Option, pub uid: Option, pub gid: Option, + pub piped_stdin: bool, } impl Spec { @@ -34,6 +35,7 @@ timeout: None, uid: None, gid: None, + piped_stdin: false, } } @@ -79,6 +81,11 @@ self.gid = Some(gid); self } + + pub fn piped_stdin(mut self) -> Self { + self.piped_stdin = true; + self + } } #[derive(Clone, Debug)] @@ -120,6 +127,7 @@ } pub struct StreamingCommand { + pub stdin: Option, events: Receiver, exit: JoinHandle>, } @@ -174,6 +182,7 @@ pub fn spawn_streaming(mut spec: Spec) -> Result { let oom_kill_before = read_oom_kill_count(); let mut child = spawn(&mut spec)?; + let stdin = child.stdin.take(); let stdout = child.stdout.take().context("stdout pipe missing")?; let stderr = child.stderr.take().context("stderr pipe missing")?; @@ -194,6 +203,7 @@ }); Ok(StreamingCommand { + stdin, events: events_rx, exit, }) @@ -205,6 +215,9 @@ .envs(spec.env.iter().map(|(key, value)| (key, value))) .stdout(Stdio::piped()) .stderr(Stdio::piped()); + if spec.piped_stdin { + cmd.stdin(Stdio::piped()); + } if let Some(cwd) = &spec.cwd { cmd.current_dir(cwd); diff --git a/shuttle/src/exec.rs b/shuttle/src/exec.rs --- a/shuttle/src/exec.rs +++ b/shuttle/src/exec.rs @@ -4,12 +4,15 @@ use std::ffi::OsString; use std::path::PathBuf; use std::time::Duration; +use tokio::io::AsyncWriteExt; use tokio::sync::mpsc::Sender; +use tokio_vsock::{VsockAddr, VsockStream}; use tracing::{info, warn}; const DEFAULT_USER: &str = "spindle-workflow"; +const VSOCK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); -pub async fn run(id: String, req: v1::ExecStart, out: Sender) { +pub async fn run(id: String, req: v1::ExecStart, out: Sender, host_cid: u32) { let send_exit = async |exit_code: i32, error: Option, timed_out: bool| { let msg = Message { id: id.clone(), @@ -57,6 +60,29 @@ .envs(env) .envs(parse_env(&req.env)) .run_as(run_as.uid, run_as.gid); + + if req.stdio_vsock_port == 0 { + send_exit( + 1, + Some("exec is missing a stdio vsock port".to_owned()), + false, + ) + .await; + return; + } + let addr = VsockAddr::new(host_cid, req.stdio_vsock_port); + let conn = match tokio::time::timeout(VSOCK_CONNECT_TIMEOUT, VsockStream::connect(addr)).await { + Ok(Ok(conn)) => conn, + Ok(Err(error)) => { + send_exit(1, Some(format!("dial host stdio port: {error}")), false).await; + return; + } + Err(_) => { + send_exit(1, Some("dial host stdio port: timed out".to_owned()), false).await; + return; + } + }; + spec = spec.piped_stdin(); if !req.cwd.is_empty() { spec = spec.cwd(req.cwd.clone()); } @@ -76,30 +102,47 @@ "starting exec" ); - let cmd = match command::spawn_streaming(spec) { + let mut cmd = match command::spawn_streaming(spec) { Ok(cmd) => cmd, Err(err) => { send_exit(127, Some(err.to_string()), false).await; return; } }; + + let (mut conn_reader, mut conn_writer) = tokio::io::split(conn); + // dropping this also closes the socket if the exec gets cancelled + let mut child_stdin = cmd.stdin.take().expect("piped_stdin"); + let _stdin_pump = AbortOnDrop(tokio::spawn(async move { + let _ = tokio::io::copy(&mut conn_reader, &mut child_stdin).await; + })); + let (mut events, exit_task) = cmd.into_parts(); + let mut stdio_error = None; while let Some(event) = events.recv().await { - let data = String::from_utf8_lossy(&event.data).into_owned(); - let output = match event.kind { - OutKind::Stdout => Message { - id: id.clone(), - exec_stdout: Some(v1::ExecStdout { data }), - ..Default::default() - }, - OutKind::Stderr => Message { - id: id.clone(), - exec_stderr: Some(v1::ExecStderr { data }), - ..Default::default() - }, - }; - let _ = out.send(output).await; + match event.kind { + OutKind::Stdout => { + if let Err(error) = conn_writer.write_all(&event.data).await { + stdio_error = Some(format!("forward stdout to host: {error}")); + // stop these before their pipes fill and block the child + events.close(); + break; + } + } + OutKind::Stderr => { + let _ = out + .send(Message { + id: id.clone(), + exec_stderr: Some(v1::ExecStderr { + data: event.data.into(), + }), + ..Default::default() + }) + .await; + } + } } + // the child might keep reading stdin after it closes stdout let exit = match exit_task .await .unwrap_or_else(|error| Err(anyhow::anyhow!("command supervisor failed: {error}"))) @@ -111,7 +154,21 @@ } }; + // losing stdout fails the exec even if the child exited cleanly + if let Some(error) = stdio_error { + send_exit(1, Some(error), false).await; + return; + } send_exit(exit.exit_code, exit.error, exit.timed_out).await +} + +// aborts the task when its exec goes away +struct AbortOnDrop(tokio::task::JoinHandle<()>); + +impl Drop for AbortOnDrop { + fn drop(&mut self) { + self.0.abort(); + } } #[derive(Clone, Debug)] diff --git a/shuttle/src/protocol.rs b/shuttle/src/protocol.rs --- a/shuttle/src/protocol.rs +++ b/shuttle/src/protocol.rs @@ -33,7 +33,6 @@ Hello, Init, ExecStart, - ExecStdout, ExecStderr, ExecExit, ActivateConfig, @@ -46,7 +45,7 @@ Message, ); -pub const PROTOCOL_VERSION: u32 = 1; +pub const PROTOCOL_VERSION: u32 = 2; pub const DEFAULT_PORT: u32 = 10240; pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024; @@ -69,7 +68,6 @@ hello => "hello", init => "init", exec_start => "exec_start", - exec_stdout => "exec_stdout", exec_stderr => "exec_stderr", exec_exit => "exec_exit", activate_config => "activate_config", diff --git a/shuttle/src/session.rs b/shuttle/src/session.rs --- a/shuttle/src/session.rs +++ b/shuttle/src/session.rs @@ -61,7 +61,7 @@ let read_result: Result<()> = loop { tokio::select! { read = protocol::read_message(&mut reader) => match read { - Ok(Some(msg)) => spawn_message_task(&mut tasks, msg, &out_tx, uploader.clone()), + Ok(Some(msg)) => spawn_message_task(&mut tasks, host_cid, msg, &out_tx, uploader.clone()), Ok(None) => break Ok(()), Err(error) => break Err(error).context("read message"), }, @@ -84,6 +84,7 @@ fn spawn_message_task( tasks: &mut JoinSet<()>, + host_cid: u32, msg: Message, out_tx: &Sender, uploader: Option, @@ -91,7 +92,7 @@ let kind = protocol::kind(&msg); let handle = on_payload!(msg, { activate_config => tasks.spawn(activation::run(msg.id, activate_config, out_tx.clone())), - exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone())), + exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone(), host_cid)), cache_drain => tasks.spawn(run_cache_drain(msg.id, cache_drain, out_tx.clone(), uploader)), poweroff => tasks.spawn(run_poweroff(msg.id, poweroff, out_tx.clone())), }); diff --git a/spindle/agentproto/protocol.go b/spindle/agentproto/protocol.go --- a/spindle/agentproto/protocol.go +++ b/spindle/agentproto/protocol.go @@ -13,7 +13,7 @@ ) const ( - ProtocolVersion = 1 + ProtocolVersion = 2 DefaultPort = 10240 MaxMessageBytes = 1024 * 1024 ) diff --git a/shuttle/src/gen/file_descriptor_set.bin b/shuttle/src/gen/file_descriptor_set.bin --- a/shuttle/src/gen/file_descriptor_set.bin +++ b/shuttle/src/gen/file_descriptor_set.bin diff --git a/spindle/agentproto/gen/agent.pb.go b/spindle/agentproto/gen/agent.pb.go --- a/spindle/agentproto/gen/agent.pb.go +++ b/spindle/agentproto/gen/agent.pb.go @@ -173,6 +173,7 @@ Cwd string `protobuf:"bytes,3,opt,name=cwd,proto3" json:"cwd,omitempty"` User string `protobuf:"bytes,4,opt,name=user,proto3" json:"user,omitempty"` TimeoutSeconds uint32 `protobuf:"varint,5,opt,name=timeout_seconds,json=timeoutSeconds,proto3" json:"timeout_seconds,omitempty"` + StdioVsockPort uint32 `protobuf:"varint,6,opt,name=stdio_vsock_port,json=stdioVsockPort,proto3" json:"stdio_vsock_port,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -242,60 +243,23 @@ return 0 } -type ExecStdout struct { - state protoimpl.MessageState `protogen:"open.v1"` - Data string `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache -} - -func (x *ExecStdout) Reset() { - *x = ExecStdout{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[3] - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - ms.StoreMessageInfo(mi) -} - -func (x *ExecStdout) String() string { - return protoimpl.X.MessageStringOf(x) -} - -func (*ExecStdout) ProtoMessage() {} - -func (x *ExecStdout) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[3] +func (x *ExecStart) GetStdioVsockPort() uint32 { if x != nil { - ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) - if ms.LoadMessageInfo() == nil { - ms.StoreMessageInfo(mi) - } - return ms + return x.StdioVsockPort } - return mi.MessageOf(x) -} - -// Deprecated: Use ExecStdout.ProtoReflect.Descriptor instead. -func (*ExecStdout) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{3} -} - -func (x *ExecStdout) GetData() string { - if x != nil { - return x.Data - } - return "" + return 0 } type ExecStderr struct { state protoimpl.MessageState `protogen:"open.v1"` - Data string `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } func (x *ExecStderr) Reset() { *x = ExecStderr{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[4] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[3] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -307,7 +271,7 @@ func (*ExecStderr) ProtoMessage() {} func (x *ExecStderr) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[4] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[3] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -320,14 +284,14 @@ // Deprecated: Use ExecStderr.ProtoReflect.Descriptor instead. func (*ExecStderr) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{4} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{3} } -func (x *ExecStderr) GetData() string { +func (x *ExecStderr) GetData() []byte { if x != nil { return x.Data } - return "" + return nil } type ExecExit struct { @@ -343,7 +307,7 @@ func (x *ExecExit) Reset() { *x = ExecExit{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[5] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -355,7 +319,7 @@ func (*ExecExit) ProtoMessage() {} func (x *ExecExit) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[5] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[4] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -368,7 +332,7 @@ // Deprecated: Use ExecExit.ProtoReflect.Descriptor instead. func (*ExecExit) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{5} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{4} } func (x *ExecExit) GetExitCode() int32 { @@ -405,7 +369,7 @@ func (x *ActivateConfig) Reset() { *x = ActivateConfig{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[6] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[5] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -417,7 +381,7 @@ func (*ActivateConfig) ProtoMessage() {} func (x *ActivateConfig) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[6] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[5] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -430,7 +394,7 @@ // Deprecated: Use ActivateConfig.ProtoReflect.Descriptor instead. func (*ActivateConfig) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{6} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{5} } func (x *ActivateConfig) GetConfigKey() string { @@ -479,7 +443,7 @@ func (x *ActivateConfigResult) Reset() { *x = ActivateConfigResult{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[7] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[6] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -491,7 +455,7 @@ func (*ActivateConfigResult) ProtoMessage() {} func (x *ActivateConfigResult) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[7] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[6] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -504,7 +468,7 @@ // Deprecated: Use ActivateConfigResult.ProtoReflect.Descriptor instead. func (*ActivateConfigResult) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{7} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{6} } func (x *ActivateConfigResult) GetConfigKey() string { @@ -538,7 +502,7 @@ func (x *BuiltPaths) Reset() { *x = BuiltPaths{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[8] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[7] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -550,7 +514,7 @@ func (*BuiltPaths) ProtoMessage() {} func (x *BuiltPaths) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[8] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[7] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -563,7 +527,7 @@ // Deprecated: Use BuiltPaths.ProtoReflect.Descriptor instead. func (*BuiltPaths) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{8} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{7} } func (x *BuiltPaths) GetPaths() []string { @@ -589,7 +553,7 @@ func (x *CacheDrain) Reset() { *x = CacheDrain{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[9] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[8] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -601,7 +565,7 @@ func (*CacheDrain) ProtoMessage() {} func (x *CacheDrain) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[9] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[8] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -614,7 +578,7 @@ // Deprecated: Use CacheDrain.ProtoReflect.Descriptor instead. func (*CacheDrain) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{9} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{8} } func (x *CacheDrain) GetTimeoutSeconds() uint32 { @@ -637,7 +601,7 @@ func (x *CacheDrainResult) Reset() { *x = CacheDrainResult{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[10] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[9] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -649,7 +613,7 @@ func (*CacheDrainResult) ProtoMessage() {} func (x *CacheDrainResult) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[10] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[9] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -662,7 +626,7 @@ // Deprecated: Use CacheDrainResult.ProtoReflect.Descriptor instead. func (*CacheDrainResult) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{10} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{9} } func (x *CacheDrainResult) GetError() string { @@ -708,7 +672,7 @@ func (x *Poweroff) Reset() { *x = Poweroff{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[11] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[10] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -720,7 +684,7 @@ func (*Poweroff) ProtoMessage() {} func (x *Poweroff) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[11] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[10] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -733,7 +697,7 @@ // Deprecated: Use Poweroff.ProtoReflect.Descriptor instead. func (*Poweroff) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{11} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{10} } type PoweroffResult struct { @@ -745,7 +709,7 @@ func (x *PoweroffResult) Reset() { *x = PoweroffResult{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[12] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -757,7 +721,7 @@ func (*PoweroffResult) ProtoMessage() {} func (x *PoweroffResult) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[12] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -770,7 +734,7 @@ // Deprecated: Use PoweroffResult.ProtoReflect.Descriptor instead. func (*PoweroffResult) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{12} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{11} } func (x *PoweroffResult) GetError() string { @@ -786,7 +750,6 @@ Hello *Hello `protobuf:"bytes,2,opt,name=hello,proto3" json:"hello,omitempty"` Init *Init `protobuf:"bytes,3,opt,name=init,proto3" json:"init,omitempty"` ExecStart *ExecStart `protobuf:"bytes,4,opt,name=exec_start,json=execStart,proto3" json:"exec_start,omitempty"` - ExecStdout *ExecStdout `protobuf:"bytes,5,opt,name=exec_stdout,json=execStdout,proto3" json:"exec_stdout,omitempty"` ExecStderr *ExecStderr `protobuf:"bytes,6,opt,name=exec_stderr,json=execStderr,proto3" json:"exec_stderr,omitempty"` ExecExit *ExecExit `protobuf:"bytes,7,opt,name=exec_exit,json=execExit,proto3" json:"exec_exit,omitempty"` ActivateConfig *ActivateConfig `protobuf:"bytes,8,opt,name=activate_config,json=activateConfig,proto3" json:"activate_config,omitempty"` @@ -802,7 +765,7 @@ func (x *Message) Reset() { *x = Message{} - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -814,7 +777,7 @@ func (*Message) ProtoMessage() {} func (x *Message) ProtoReflect() protoreflect.Message { - mi := &file_spindle_agent_v1_agent_proto_msgTypes[13] + mi := &file_spindle_agent_v1_agent_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -827,7 +790,7 @@ // Deprecated: Use Message.ProtoReflect.Descriptor instead. func (*Message) Descriptor() ([]byte, []int) { - return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13} + return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{12} } func (x *Message) GetId() string { @@ -854,13 +817,6 @@ func (x *Message) GetExecStart() *ExecStart { if x != nil { return x.ExecStart - } - return nil -} - -func (x *Message) GetExecStdout() *ExecStdout { - if x != nil { - return x.ExecStdout } return nil } @@ -944,19 +900,17 @@ "\x19cache_trusted_public_keys\x18\x02 \x03(\tR\x16cacheTrustedPublicKeys\x121\n" + "\x15cache_read_proxy_port\x18\x03 \x01(\rR\x12cacheReadProxyPort\x125\n" + "\x17cache_upload_proxy_port\x18\x04 \x01(\rR\x14cacheUploadProxyPort\x12$\n" + - "\x0edns_proxy_port\x18\x05 \x01(\rR\fdnsProxyPort\"\x80\x01\n" + + "\x0edns_proxy_port\x18\x05 \x01(\rR\fdnsProxyPort\"\xaa\x01\n" + "\tExecStart\x12\x12\n" + "\x04argv\x18\x01 \x03(\tR\x04argv\x12\x10\n" + "\x03env\x18\x02 \x03(\tR\x03env\x12\x10\n" + "\x03cwd\x18\x03 \x01(\tR\x03cwd\x12\x12\n" + "\x04user\x18\x04 \x01(\tR\x04user\x12'\n" + - "\x0ftimeout_seconds\x18\x05 \x01(\rR\x0etimeoutSeconds\" \n" + - "\n" + - "ExecStdout\x12\x12\n" + - "\x04data\x18\x01 \x01(\tR\x04data\" \n" + + "\x0ftimeout_seconds\x18\x05 \x01(\rR\x0etimeoutSeconds\x12(\n" + + "\x10stdio_vsock_port\x18\x06 \x01(\rR\x0estdioVsockPort\" \n" + "\n" + "ExecStderr\x12\x12\n" + - "\x04data\x18\x01 \x01(\tR\x04data\"Z\n" + + "\x04data\x18\x01 \x01(\fR\x04data\"Z\n" + "\bExecExit\x12\x1b\n" + "\texit_code\x18\x01 \x01(\x05R\bexitCode\x12\x14\n" + "\x05error\x18\x02 \x01(\tR\x05error\x12\x1b\n" + @@ -990,15 +944,13 @@ "\n" + "\bPoweroff\"&\n" + "\x0ePoweroffResult\x12\x14\n" + - "\x05error\x18\x01 \x01(\tR\x05error\"\xa8\b\n" + + "\x05error\x18\x01 \x01(\tR\x05error\"\xe8\a\n" + "\aMessage\x12\x17\n" + "\x02id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x02id\x12-\n" + "\x05hello\x18\x02 \x01(\v2\x17.spindle.agent.v1.HelloR\x05hello\x12*\n" + "\x04init\x18\x03 \x01(\v2\x16.spindle.agent.v1.InitR\x04init\x12:\n" + "\n" + "exec_start\x18\x04 \x01(\v2\x1b.spindle.agent.v1.ExecStartR\texecStart\x12=\n" + - "\vexec_stdout\x18\x05 \x01(\v2\x1c.spindle.agent.v1.ExecStdoutR\n" + - "execStdout\x12=\n" + "\vexec_stderr\x18\x06 \x01(\v2\x1c.spindle.agent.v1.ExecStderrR\n" + "execStderr\x127\n" + "\texec_exit\x18\a \x01(\v2\x1a.spindle.agent.v1.ExecExitR\bexecExit\x12I\n" + @@ -1011,12 +963,11 @@ "cacheDrain\x12P\n" + "\x12cache_drain_result\x18\f \x01(\v2\".spindle.agent.v1.CacheDrainResultR\x10cacheDrainResult\x126\n" + "\bpoweroff\x18\r \x01(\v2\x1a.spindle.agent.v1.PoweroffR\bpoweroff\x12I\n" + - "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xb9\x01\xbaH\xb5\x01\"\xb2\x01\n" + + "\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xac\x01\xbaH\xa8\x01\"\xa5\x01\n" + "\x05hello\n" + "\x04init\n" + "\n" + "exec_start\n" + - "\vexec_stdout\n" + "\vexec_stderr\n" + "\texec_exit\n" + "\x0factivate_config\n" + @@ -1025,7 +976,7 @@ "\vcache_drain\n" + "\x12cache_drain_result\n" + "\bpoweroff\n" + - "\x0fpoweroff_result\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" + "\x0fpoweroff_result\x10\x01J\x04\b\x05\x10\x06J\x04\b\x0f\x10\x10B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3" var ( file_spindle_agent_v1_agent_proto_rawDescOnce sync.Once @@ -1039,42 +990,40 @@ return file_spindle_agent_v1_agent_proto_rawDescData } -var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 14) +var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 13) var file_spindle_agent_v1_agent_proto_goTypes = []any{ (*Hello)(nil), // 0: spindle.agent.v1.Hello (*Init)(nil), // 1: spindle.agent.v1.Init (*ExecStart)(nil), // 2: spindle.agent.v1.ExecStart - (*ExecStdout)(nil), // 3: spindle.agent.v1.ExecStdout - (*ExecStderr)(nil), // 4: spindle.agent.v1.ExecStderr - (*ExecExit)(nil), // 5: spindle.agent.v1.ExecExit - (*ActivateConfig)(nil), // 6: spindle.agent.v1.ActivateConfig - (*ActivateConfigResult)(nil), // 7: spindle.agent.v1.ActivateConfigResult - (*BuiltPaths)(nil), // 8: spindle.agent.v1.BuiltPaths - (*CacheDrain)(nil), // 9: spindle.agent.v1.CacheDrain - (*CacheDrainResult)(nil), // 10: spindle.agent.v1.CacheDrainResult - (*Poweroff)(nil), // 11: spindle.agent.v1.Poweroff - (*PoweroffResult)(nil), // 12: spindle.agent.v1.PoweroffResult - (*Message)(nil), // 13: spindle.agent.v1.Message + (*ExecStderr)(nil), // 3: spindle.agent.v1.ExecStderr + (*ExecExit)(nil), // 4: spindle.agent.v1.ExecExit + (*ActivateConfig)(nil), // 5: spindle.agent.v1.ActivateConfig + (*ActivateConfigResult)(nil), // 6: spindle.agent.v1.ActivateConfigResult + (*BuiltPaths)(nil), // 7: spindle.agent.v1.BuiltPaths + (*CacheDrain)(nil), // 8: spindle.agent.v1.CacheDrain + (*CacheDrainResult)(nil), // 9: spindle.agent.v1.CacheDrainResult + (*Poweroff)(nil), // 10: spindle.agent.v1.Poweroff + (*PoweroffResult)(nil), // 11: spindle.agent.v1.PoweroffResult + (*Message)(nil), // 12: spindle.agent.v1.Message } var file_spindle_agent_v1_agent_proto_depIdxs = []int32{ 0, // 0: spindle.agent.v1.Message.hello:type_name -> spindle.agent.v1.Hello 1, // 1: spindle.agent.v1.Message.init:type_name -> spindle.agent.v1.Init 2, // 2: spindle.agent.v1.Message.exec_start:type_name -> spindle.agent.v1.ExecStart - 3, // 3: spindle.agent.v1.Message.exec_stdout:type_name -> spindle.agent.v1.ExecStdout - 4, // 4: spindle.agent.v1.Message.exec_stderr:type_name -> spindle.agent.v1.ExecStderr - 5, // 5: spindle.agent.v1.Message.exec_exit:type_name -> spindle.agent.v1.ExecExit - 6, // 6: spindle.agent.v1.Message.activate_config:type_name -> spindle.agent.v1.ActivateConfig - 7, // 7: spindle.agent.v1.Message.activate_config_result:type_name -> spindle.agent.v1.ActivateConfigResult - 8, // 8: spindle.agent.v1.Message.built_paths:type_name -> spindle.agent.v1.BuiltPaths - 9, // 9: spindle.agent.v1.Message.cache_drain:type_name -> spindle.agent.v1.CacheDrain - 10, // 10: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult - 11, // 11: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff - 12, // 12: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult - 13, // [13:13] is the sub-list for method output_type - 13, // [13:13] is the sub-list for method input_type - 13, // [13:13] is the sub-list for extension type_name - 13, // [13:13] is the sub-list for extension extendee - 0, // [0:13] is the sub-list for field type_name + 3, // 3: spindle.agent.v1.Message.exec_stderr:type_name -> spindle.agent.v1.ExecStderr + 4, // 4: spindle.agent.v1.Message.exec_exit:type_name -> spindle.agent.v1.ExecExit + 5, // 5: spindle.agent.v1.Message.activate_config:type_name -> spindle.agent.v1.ActivateConfig + 6, // 6: spindle.agent.v1.Message.activate_config_result:type_name -> spindle.agent.v1.ActivateConfigResult + 7, // 7: spindle.agent.v1.Message.built_paths:type_name -> spindle.agent.v1.BuiltPaths + 8, // 8: spindle.agent.v1.Message.cache_drain:type_name -> spindle.agent.v1.CacheDrain + 9, // 9: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult + 10, // 10: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff + 11, // 11: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult + 12, // [12:12] is the sub-list for method output_type + 12, // [12:12] is the sub-list for method input_type + 12, // [12:12] is the sub-list for extension type_name + 12, // [12:12] is the sub-list for extension extendee + 0, // [0:12] is the sub-list for field type_name } func init() { file_spindle_agent_v1_agent_proto_init() } @@ -1088,7 +1037,7 @@ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_agent_v1_agent_proto_rawDesc), len(file_spindle_agent_v1_agent_proto_rawDesc)), NumEnums: 0, - NumMessages: 14, + NumMessages: 13, NumExtensions: 0, NumServices: 0, }, diff --git a/spindle/engines/microvm/README.md b/spindle/engines/microvm/README.md --- a/spindle/engines/microvm/README.md +++ b/spindle/engines/microvm/README.md @@ -179,8 +179,9 @@ builds and activates it before the user steps run. Afterwards, each step is sent as an exec request (`$shell -lc ` as an unprivileged workflow user in `/workspace/repo`, with workflow/step environment and unlocked secrets), and -stdout/stderr stream back as messages until an exit message arrives. Timeouts -are cooperative: we derive a deadline from the workflow timeout and ship it to +stdout streams back raw over a dedicated per-exec vsock connection (the +agent dials it; stdin rides the same socket), while stderr and the exit +status stay on the control channel. Timeouts are cooperative: we derive a deadline from the workflow timeout and ship it to the guest, with a little grace on the host side so the guest gets to report the timeout itself. While a step runs we also watch for the VM crashing, if it does we tail the serial (and qemu) logs into the step's stderr so you get something diff --git a/spindle/engines/microvm/agent.go b/spindle/engines/microvm/agent.go --- a/spindle/engines/microvm/agent.go +++ b/spindle/engines/microvm/agent.go @@ -107,21 +107,24 @@ type AgentExec struct { *agentv1.ExecStart ID string + Stdin io.Reader Stdout io.Writer Stderr io.Writer } type AgentSession struct { conn net.Conn + cid uint32 enc *agentproto.Encoder dec *agentproto.Decoder l *slog.Logger mu sync.Mutex } -func NewAgentSession(conn net.Conn, l *slog.Logger) *AgentSession { +func NewAgentSession(conn net.Conn, cid uint32, l *slog.Logger) *AgentSession { return &AgentSession{ conn: conn, + cid: cid, enc: agentproto.NewEncoder(conn), dec: agentproto.NewDecoder(conn), l: l, @@ -139,6 +142,9 @@ helloPayload := hello.Hello if helloPayload == nil { return fmt.Errorf("expected agent hello, got nil") + } + if helloPayload.ProtocolVersion != agentproto.ProtocolVersion { + return fmt.Errorf("agent protocol version %d, want %d (stale guest image?)", helloPayload.ProtocolVersion, agentproto.ProtocolVersion) } s.l.Info("agent connected", "protocol", helloPayload.ProtocolVersion, "version", helloPayload.AgentVersion, "boot", helloPayload.BootId, "nix", helloPayload.NixVersion) @@ -163,6 +169,13 @@ exec.ExecStart.TimeoutSeconds = timeoutSeconds(ctx, guestTimeoutGrace) } + ln, port, err := listenRandomVsockPort(ctx) + if err != nil { + return 0, fmt.Errorf("listen for exec stdio: %w", err) + } + defer ln.Close() + exec.ExecStart.StdioVsockPort = port + if err := s.enc.Encode(&agentproto.Message{ Id: exec.ID, ExecStart: exec.ExecStart, @@ -170,26 +183,36 @@ return 0, fmt.Errorf("send exec_start: %w", err) } + filtered := &cidFilteredVsockListener{Listener: ln, cid: s.cid, logger: s.l} + stdioDone := make(chan error, 1) + go func() { + stdioDone <- pumpStdio(ctx, filtered, exec.Stdin, exec.Stdout) + }() + for { msg, err := s.decode(ctx) if err != nil { + ln.Close() + <-stdioDone return 0, err } if msg.BuiltPaths == nil && msg.Id != exec.ID { continue } - if p := msg.ExecStdout; p != nil { - _, _ = io.WriteString(exec.Stdout, p.Data) - } else if p := msg.ExecStderr; p != nil { - _, _ = io.WriteString(exec.Stderr, p.Data) - } else if p := msg.BuiltPaths; p != nil { + if p := msg.BuiltPaths; p != nil { // s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths)) + } else if p := msg.ExecStderr; p != nil { + _, _ = exec.Stderr.Write(p.Data) } else if p := msg.ExecExit; p != nil { var err error if p.Error != "" { s.l.Warn("guest exec error", "id", msg.Id, "error", p.Error) err = fmt.Errorf("guest exec error: %s", p.Error) + } + ln.Close() // wake Accept if the guest never dialed + if err := <-stdioDone; err != nil { + return 0, err } if p.TimedOut { return int(p.ExitCode), errGuestTimedOut @@ -197,6 +220,49 @@ return int(p.ExitCode), err } } +} + +func pumpStdio(ctx context.Context, ln net.Listener, stdin io.Reader, stdout io.Writer) error { + conn, err := ln.Accept() + if err != nil { + if errors.Is(err, net.ErrClosed) { + return nil // exec ended before dialing + } + return fmt.Errorf("accept guest stdio connection: %w", err) + } + vsockConn, ok := conn.(*vsock.Conn) + if !ok { + conn.Close() + return fmt.Errorf("guest connection is not a vsock connection") + } + defer vsockConn.Close() + stop := context.AfterFunc(ctx, func() { vsockConn.Close() }) + defer stop() + + stdinDone := make(chan error, 1) + go func() { + stdinDone <- writeStdin(vsockConn, stdin) + }() + + if _, err := io.Copy(stdout, conn); err != nil { + vsockConn.Close() + <-stdinDone + return fmt.Errorf("read guest stdout: %w", err) + } + if err := <-stdinDone; err != nil { + return fmt.Errorf("write guest stdin: %w", err) + } + return nil +} + +// send stdin EOF without closing stdout +func writeStdin(conn *vsock.Conn, r io.Reader) error { + if r != nil { + if _, err := io.Copy(conn, r); err != nil { + return err + } + } + return conn.CloseWrite() } func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agentv1.ActivateConfig, out io.Writer) (*agentv1.ActivateConfigResult, error) { @@ -229,11 +295,7 @@ // s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths)) } else if p := msg.ExecStderr; p != nil { if out != nil { - _, _ = io.WriteString(out, p.Data) - } - } else if p := msg.ExecStdout; p != nil { - if out != nil { - _, _ = io.WriteString(out, p.Data) + _, _ = out.Write(p.Data) } } else if p := msg.ActivateConfigResult; p != nil { if p.Error != "" { diff --git a/spindle/engines/microvm/engine.go b/spindle/engines/microvm/engine.go --- a/spindle/engines/microvm/engine.go +++ b/spindle/engines/microvm/engine.go @@ -310,7 +310,7 @@ return err } - agentSession := NewAgentSession(conn, l) + agentSession := NewAgentSession(conn, cid, l) initCtx, cancelInit := context.WithTimeout(ctx, agentHandshakeTimeout) defer cancelInit() if err := agentSession.Init(initCtx, &agentv1.Init{ diff --git a/spindle/agentproto/spindle/agent/v1/agent.proto b/spindle/agentproto/spindle/agent/v1/agent.proto --- a/spindle/agentproto/spindle/agent/v1/agent.proto +++ b/spindle/agentproto/spindle/agent/v1/agent.proto @@ -27,14 +27,11 @@ string cwd = 3; string user = 4; uint32 timeout_seconds = 5; -} - -message ExecStdout { - string data = 1; + uint32 stdio_vsock_port = 6; } message ExecStderr { - string data = 1; + bytes data = 1; } message ExecExit { @@ -85,9 +82,9 @@ message Message { option (buf.validate.message).oneof = { fields: [ - "hello", "init", "exec_start", "exec_stdout", "exec_stderr", "exec_exit", - "activate_config", "activate_config_result", "built_paths", "cache_drain", - "cache_drain_result", "poweroff", "poweroff_result" + "hello", "init", "exec_start", "exec_stderr", + "exec_exit", "activate_config", "activate_config_result", "built_paths", + "cache_drain", "cache_drain_result", "poweroff", "poweroff_result" ], required: true }; @@ -97,7 +94,6 @@ Hello hello = 2; Init init = 3; ExecStart exec_start = 4; - ExecStdout exec_stdout = 5; ExecStderr exec_stderr = 6; ExecExit exec_exit = 7; ActivateConfig activate_config = 8; @@ -107,4 +103,7 @@ CacheDrainResult cache_drain_result = 12; Poweroff poweroff = 13; PoweroffResult poweroff_result = 14; + + // old framed stdio fields + reserved 5, 15; } diff --git a/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs b/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs --- a/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs +++ b/shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs @@ -36,16 +36,13 @@ pub user: ::prost::alloc::string::String, #[prost(uint32, tag = "5")] pub timeout_seconds: u32, -} -#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] -pub struct ExecStdout { - #[prost(string, tag = "1")] - pub data: ::prost::alloc::string::String, + #[prost(uint32, tag = "6")] + pub stdio_vsock_port: u32, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ExecStderr { - #[prost(string, tag = "1")] - pub data: ::prost::alloc::string::String, + #[prost(bytes = "bytes", tag = "1")] + pub data: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ExecExit { @@ -122,8 +119,6 @@ pub init: ::core::option::Option, #[prost(message, optional, tag = "4")] pub exec_start: ::core::option::Option, - #[prost(message, optional, tag = "5")] - pub exec_stdout: ::core::option::Option, #[prost(message, optional, tag = "6")] pub exec_stderr: ::core::option::Option, #[prost(message, optional, tag = "7")] -- tangled.sh