Monorepo for Tangled
shuttle,spindle/engines/microvm: move cmd exec stdio to a vsock channel
Signed-off-by: dawn <dawn@tangled.org>
+255
-179
+1
-1
cmd/spindle-microvm-run/main_linux.go
+1
-1
cmd/spindle-microvm-run/main_linux.go
+3
-2
shuttle/src/activation.rs
+3
-2
shuttle/src/activation.rs
···
201
201
OutKind::Stdout => stdout.extend_from_slice(&event.data),
202
202
OutKind::Stderr => {
203
203
stderr.extend_from_slice(&event.data);
204
-
let data = String::from_utf8_lossy(&event.data).into_owned();
205
204
let _ = out
206
205
.send(Message {
207
206
id: id.to_owned(),
208
-
exec_stderr: Some(v1::ExecStderr { data }),
207
+
exec_stderr: Some(v1::ExecStderr {
208
+
data: event.data.into(),
209
+
}),
209
210
..Default::default()
210
211
})
211
212
.await;
+14
-1
shuttle/src/command.rs
+14
-1
shuttle/src/command.rs
···
8
8
use std::process::Stdio;
9
9
use std::time::Duration;
10
10
use tokio::io::{AsyncRead, AsyncReadExt};
11
-
use tokio::process::{Child, Command};
11
+
use tokio::process::{Child, ChildStdin, Command};
12
12
use tokio::sync::mpsc::{self, Receiver, Sender};
13
13
use tokio::task::JoinHandle;
14
14
use tracing::warn;
···
22
22
pub timeout: Option<Duration>,
23
23
pub uid: Option<u32>,
24
24
pub gid: Option<u32>,
25
+
pub piped_stdin: bool,
25
26
}
26
27
27
28
impl Spec {
···
34
35
timeout: None,
35
36
uid: None,
36
37
gid: None,
38
+
piped_stdin: false,
37
39
}
38
40
}
39
41
···
79
81
self.gid = Some(gid);
80
82
self
81
83
}
84
+
85
+
pub fn piped_stdin(mut self) -> Self {
86
+
self.piped_stdin = true;
87
+
self
88
+
}
82
89
}
83
90
84
91
#[derive(Clone, Debug)]
···
120
127
}
121
128
122
129
pub struct StreamingCommand {
130
+
pub stdin: Option<ChildStdin>,
123
131
events: Receiver<OutData>,
124
132
exit: JoinHandle<Result<ExitResult>>,
125
133
}
···
155
163
156
164
pub fn spawn_streaming(mut spec: Spec) -> Result<StreamingCommand> {
157
165
let mut child = spawn(&mut spec)?;
166
+
let stdin = child.stdin.take();
158
167
let stdout = child.stdout.take().context("stdout pipe missing")?;
159
168
let stderr = child.stderr.take().context("stderr pipe missing")?;
160
169
···
175
184
});
176
185
177
186
Ok(StreamingCommand {
187
+
stdin,
178
188
events: events_rx,
179
189
exit,
180
190
})
···
186
196
.envs(spec.env.iter().map(|(key, value)| (key, value)))
187
197
.stdout(Stdio::piped())
188
198
.stderr(Stdio::piped());
199
+
if spec.piped_stdin {
200
+
cmd.stdin(Stdio::piped());
201
+
}
189
202
190
203
if let Some(cwd) = &spec.cwd {
191
204
cmd.current_dir(cwd);
+73
-16
shuttle/src/exec.rs
+73
-16
shuttle/src/exec.rs
···
4
4
use std::ffi::OsString;
5
5
use std::path::PathBuf;
6
6
use std::time::Duration;
7
+
use tokio::io::AsyncWriteExt;
7
8
use tokio::sync::mpsc::Sender;
9
+
use tokio_vsock::{VsockAddr, VsockStream};
8
10
use tracing::{info, warn};
9
11
10
12
const DEFAULT_USER: &str = "spindle-workflow";
13
+
const VSOCK_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
11
14
12
-
pub async fn run(id: String, req: v1::ExecStart, out: Sender<Message>) {
15
+
pub async fn run(id: String, req: v1::ExecStart, out: Sender<Message>, host_cid: u32) {
13
16
let send_exit = async |exit_code: i32, error: Option<String>, timed_out: bool| {
14
17
let msg = Message {
15
18
id: id.clone(),
···
57
60
.envs(env)
58
61
.envs(parse_env(&req.env))
59
62
.run_as(run_as.uid, run_as.gid);
63
+
64
+
if req.stdio_vsock_port == 0 {
65
+
send_exit(
66
+
1,
67
+
Some("exec is missing a stdio vsock port".to_owned()),
68
+
false,
69
+
)
70
+
.await;
71
+
return;
72
+
}
73
+
let addr = VsockAddr::new(host_cid, req.stdio_vsock_port);
74
+
let conn = match tokio::time::timeout(VSOCK_CONNECT_TIMEOUT, VsockStream::connect(addr)).await {
75
+
Ok(Ok(conn)) => conn,
76
+
Ok(Err(error)) => {
77
+
send_exit(1, Some(format!("dial host stdio port: {error}")), false).await;
78
+
return;
79
+
}
80
+
Err(_) => {
81
+
send_exit(1, Some("dial host stdio port: timed out".to_owned()), false).await;
82
+
return;
83
+
}
84
+
};
85
+
spec = spec.piped_stdin();
60
86
if !req.cwd.is_empty() {
61
87
spec = spec.cwd(req.cwd.clone());
62
88
}
···
76
102
"starting exec"
77
103
);
78
104
79
-
let cmd = match command::spawn_streaming(spec) {
105
+
let mut cmd = match command::spawn_streaming(spec) {
80
106
Ok(cmd) => cmd,
81
107
Err(err) => {
82
108
send_exit(127, Some(err.to_string()), false).await;
83
109
return;
84
110
}
85
111
};
112
+
113
+
let (mut conn_reader, mut conn_writer) = tokio::io::split(conn);
114
+
// dropping this also closes the socket if the exec gets cancelled
115
+
let mut child_stdin = cmd.stdin.take().expect("piped_stdin");
116
+
let _stdin_pump = AbortOnDrop(tokio::spawn(async move {
117
+
let _ = tokio::io::copy(&mut conn_reader, &mut child_stdin).await;
118
+
}));
119
+
86
120
let (mut events, exit_task) = cmd.into_parts();
121
+
let mut stdio_error = None;
87
122
while let Some(event) = events.recv().await {
88
-
let data = String::from_utf8_lossy(&event.data).into_owned();
89
-
let output = match event.kind {
90
-
OutKind::Stdout => Message {
91
-
id: id.clone(),
92
-
exec_stdout: Some(v1::ExecStdout { data }),
93
-
..Default::default()
94
-
},
95
-
OutKind::Stderr => Message {
96
-
id: id.clone(),
97
-
exec_stderr: Some(v1::ExecStderr { data }),
98
-
..Default::default()
99
-
},
100
-
};
101
-
let _ = out.send(output).await;
123
+
match event.kind {
124
+
OutKind::Stdout => {
125
+
if let Err(error) = conn_writer.write_all(&event.data).await {
126
+
stdio_error = Some(format!("forward stdout to host: {error}"));
127
+
// stop these before their pipes fill and block the child
128
+
events.close();
129
+
break;
130
+
}
131
+
}
132
+
OutKind::Stderr => {
133
+
let _ = out
134
+
.send(Message {
135
+
id: id.clone(),
136
+
exec_stderr: Some(v1::ExecStderr {
137
+
data: event.data.into(),
138
+
}),
139
+
..Default::default()
140
+
})
141
+
.await;
142
+
}
143
+
}
102
144
}
145
+
// the child might keep reading stdin after it closes stdout
103
146
let exit = match exit_task
104
147
.await
105
148
.unwrap_or_else(|error| Err(anyhow::anyhow!("command supervisor failed: {error}")))
···
111
154
}
112
155
};
113
156
157
+
// losing stdout fails the exec even if the child exited cleanly
158
+
if let Some(error) = stdio_error {
159
+
send_exit(1, Some(error), false).await;
160
+
return;
161
+
}
114
162
send_exit(exit.exit_code, exit.error, exit.timed_out).await
163
+
}
164
+
165
+
// aborts the task when its exec goes away
166
+
struct AbortOnDrop(tokio::task::JoinHandle<()>);
167
+
168
+
impl Drop for AbortOnDrop {
169
+
fn drop(&mut self) {
170
+
self.0.abort();
171
+
}
115
172
}
116
173
117
174
#[derive(Clone, Debug)]
shuttle/src/gen/file_descriptor_set.bin
shuttle/src/gen/file_descriptor_set.bin
This is a binary file and will not be displayed.
+4
-9
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
+4
-9
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
···
36
36
pub user: ::prost::alloc::string::String,
37
37
#[prost(uint32, tag = "5")]
38
38
pub timeout_seconds: u32,
39
-
}
40
-
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
41
-
pub struct ExecStdout {
42
-
#[prost(string, tag = "1")]
43
-
pub data: ::prost::alloc::string::String,
39
+
#[prost(uint32, tag = "6")]
40
+
pub stdio_vsock_port: u32,
44
41
}
45
42
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
46
43
pub struct ExecStderr {
47
-
#[prost(string, tag = "1")]
48
-
pub data: ::prost::alloc::string::String,
44
+
#[prost(bytes = "bytes", tag = "1")]
45
+
pub data: ::prost::bytes::Bytes,
49
46
}
50
47
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
51
48
pub struct ExecExit {
···
122
119
pub init: ::core::option::Option<Init>,
123
120
#[prost(message, optional, tag = "4")]
124
121
pub exec_start: ::core::option::Option<ExecStart>,
125
-
#[prost(message, optional, tag = "5")]
126
-
pub exec_stdout: ::core::option::Option<ExecStdout>,
127
122
#[prost(message, optional, tag = "6")]
128
123
pub exec_stderr: ::core::option::Option<ExecStderr>,
129
124
#[prost(message, optional, tag = "7")]
+1
-3
shuttle/src/protocol.rs
+1
-3
shuttle/src/protocol.rs
···
33
33
Hello,
34
34
Init,
35
35
ExecStart,
36
-
ExecStdout,
37
36
ExecStderr,
38
37
ExecExit,
39
38
ActivateConfig,
···
46
45
Message,
47
46
);
48
47
49
-
pub const PROTOCOL_VERSION: u32 = 1;
48
+
pub const PROTOCOL_VERSION: u32 = 2;
50
49
pub const DEFAULT_PORT: u32 = 10240;
51
50
pub const MAX_MESSAGE_BYTES: usize = 1024 * 1024;
52
51
···
69
68
hello => "hello",
70
69
init => "init",
71
70
exec_start => "exec_start",
72
-
exec_stdout => "exec_stdout",
73
71
exec_stderr => "exec_stderr",
74
72
exec_exit => "exec_exit",
75
73
activate_config => "activate_config",
+3
-2
shuttle/src/session.rs
+3
-2
shuttle/src/session.rs
···
61
61
let read_result: Result<()> = loop {
62
62
tokio::select! {
63
63
read = protocol::read_message(&mut reader) => match read {
64
-
Ok(Some(msg)) => spawn_message_task(&mut tasks, msg, &out_tx, uploader.clone()),
64
+
Ok(Some(msg)) => spawn_message_task(&mut tasks, host_cid, msg, &out_tx, uploader.clone()),
65
65
Ok(None) => break Ok(()),
66
66
Err(error) => break Err(error).context("read message"),
67
67
},
···
84
84
85
85
fn spawn_message_task(
86
86
tasks: &mut JoinSet<()>,
87
+
host_cid: u32,
87
88
msg: Message,
88
89
out_tx: &Sender<Message>,
89
90
uploader: Option<CacheUploadManager>,
···
91
92
let kind = protocol::kind(&msg);
92
93
let handle = on_payload!(msg, {
93
94
activate_config => tasks.spawn(activation::run(msg.id, activate_config, out_tx.clone())),
94
-
exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone())),
95
+
exec_start => tasks.spawn(exec::run(msg.id, exec_start, out_tx.clone(), host_cid)),
95
96
cache_drain => tasks.spawn(run_cache_drain(msg.id, cache_drain, out_tx.clone(), uploader)),
96
97
poweroff => tasks.spawn(run_poweroff(msg.id, poweroff, out_tx.clone())),
97
98
});
+70
-121
spindle/agentproto/gen/agent.pb.go
+70
-121
spindle/agentproto/gen/agent.pb.go
···
173
173
Cwd string `protobuf:"bytes,3,opt,name=cwd,proto3" json:"cwd,omitempty"`
174
174
User string `protobuf:"bytes,4,opt,name=user,proto3" json:"user,omitempty"`
175
175
TimeoutSeconds uint32 `protobuf:"varint,5,opt,name=timeout_seconds,json=timeoutSeconds,proto3" json:"timeout_seconds,omitempty"`
176
+
StdioVsockPort uint32 `protobuf:"varint,6,opt,name=stdio_vsock_port,json=stdioVsockPort,proto3" json:"stdio_vsock_port,omitempty"`
176
177
unknownFields protoimpl.UnknownFields
177
178
sizeCache protoimpl.SizeCache
178
179
}
···
242
243
return 0
243
244
}
244
245
245
-
type ExecStdout struct {
246
-
state protoimpl.MessageState `protogen:"open.v1"`
247
-
Data string `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
248
-
unknownFields protoimpl.UnknownFields
249
-
sizeCache protoimpl.SizeCache
250
-
}
251
-
252
-
func (x *ExecStdout) Reset() {
253
-
*x = ExecStdout{}
254
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[3]
255
-
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
256
-
ms.StoreMessageInfo(mi)
257
-
}
258
-
259
-
func (x *ExecStdout) String() string {
260
-
return protoimpl.X.MessageStringOf(x)
261
-
}
262
-
263
-
func (*ExecStdout) ProtoMessage() {}
264
-
265
-
func (x *ExecStdout) ProtoReflect() protoreflect.Message {
266
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[3]
246
+
func (x *ExecStart) GetStdioVsockPort() uint32 {
267
247
if x != nil {
268
-
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
269
-
if ms.LoadMessageInfo() == nil {
270
-
ms.StoreMessageInfo(mi)
271
-
}
272
-
return ms
248
+
return x.StdioVsockPort
273
249
}
274
-
return mi.MessageOf(x)
275
-
}
276
-
277
-
// Deprecated: Use ExecStdout.ProtoReflect.Descriptor instead.
278
-
func (*ExecStdout) Descriptor() ([]byte, []int) {
279
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{3}
280
-
}
281
-
282
-
func (x *ExecStdout) GetData() string {
283
-
if x != nil {
284
-
return x.Data
285
-
}
286
-
return ""
250
+
return 0
287
251
}
288
252
289
253
type ExecStderr struct {
290
254
state protoimpl.MessageState `protogen:"open.v1"`
291
-
Data string `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
255
+
Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"`
292
256
unknownFields protoimpl.UnknownFields
293
257
sizeCache protoimpl.SizeCache
294
258
}
295
259
296
260
func (x *ExecStderr) Reset() {
297
261
*x = ExecStderr{}
298
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[4]
262
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[3]
299
263
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
300
264
ms.StoreMessageInfo(mi)
301
265
}
···
307
271
func (*ExecStderr) ProtoMessage() {}
308
272
309
273
func (x *ExecStderr) ProtoReflect() protoreflect.Message {
310
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[4]
274
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[3]
311
275
if x != nil {
312
276
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
313
277
if ms.LoadMessageInfo() == nil {
···
320
284
321
285
// Deprecated: Use ExecStderr.ProtoReflect.Descriptor instead.
322
286
func (*ExecStderr) Descriptor() ([]byte, []int) {
323
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{4}
287
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{3}
324
288
}
325
289
326
-
func (x *ExecStderr) GetData() string {
290
+
func (x *ExecStderr) GetData() []byte {
327
291
if x != nil {
328
292
return x.Data
329
293
}
330
-
return ""
294
+
return nil
331
295
}
332
296
333
297
type ExecExit struct {
···
343
307
344
308
func (x *ExecExit) Reset() {
345
309
*x = ExecExit{}
346
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[5]
310
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[4]
347
311
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
348
312
ms.StoreMessageInfo(mi)
349
313
}
···
355
319
func (*ExecExit) ProtoMessage() {}
356
320
357
321
func (x *ExecExit) ProtoReflect() protoreflect.Message {
358
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[5]
322
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[4]
359
323
if x != nil {
360
324
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
361
325
if ms.LoadMessageInfo() == nil {
···
368
332
369
333
// Deprecated: Use ExecExit.ProtoReflect.Descriptor instead.
370
334
func (*ExecExit) Descriptor() ([]byte, []int) {
371
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{5}
335
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{4}
372
336
}
373
337
374
338
func (x *ExecExit) GetExitCode() int32 {
···
405
369
406
370
func (x *ActivateConfig) Reset() {
407
371
*x = ActivateConfig{}
408
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[6]
372
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[5]
409
373
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
410
374
ms.StoreMessageInfo(mi)
411
375
}
···
417
381
func (*ActivateConfig) ProtoMessage() {}
418
382
419
383
func (x *ActivateConfig) ProtoReflect() protoreflect.Message {
420
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[6]
384
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[5]
421
385
if x != nil {
422
386
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
423
387
if ms.LoadMessageInfo() == nil {
···
430
394
431
395
// Deprecated: Use ActivateConfig.ProtoReflect.Descriptor instead.
432
396
func (*ActivateConfig) Descriptor() ([]byte, []int) {
433
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{6}
397
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{5}
434
398
}
435
399
436
400
func (x *ActivateConfig) GetConfigKey() string {
···
479
443
480
444
func (x *ActivateConfigResult) Reset() {
481
445
*x = ActivateConfigResult{}
482
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[7]
446
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[6]
483
447
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
484
448
ms.StoreMessageInfo(mi)
485
449
}
···
491
455
func (*ActivateConfigResult) ProtoMessage() {}
492
456
493
457
func (x *ActivateConfigResult) ProtoReflect() protoreflect.Message {
494
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[7]
458
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[6]
495
459
if x != nil {
496
460
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
497
461
if ms.LoadMessageInfo() == nil {
···
504
468
505
469
// Deprecated: Use ActivateConfigResult.ProtoReflect.Descriptor instead.
506
470
func (*ActivateConfigResult) Descriptor() ([]byte, []int) {
507
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{7}
471
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{6}
508
472
}
509
473
510
474
func (x *ActivateConfigResult) GetConfigKey() string {
···
538
502
539
503
func (x *BuiltPaths) Reset() {
540
504
*x = BuiltPaths{}
541
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[8]
505
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[7]
542
506
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
543
507
ms.StoreMessageInfo(mi)
544
508
}
···
550
514
func (*BuiltPaths) ProtoMessage() {}
551
515
552
516
func (x *BuiltPaths) ProtoReflect() protoreflect.Message {
553
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[8]
517
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[7]
554
518
if x != nil {
555
519
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
556
520
if ms.LoadMessageInfo() == nil {
···
563
527
564
528
// Deprecated: Use BuiltPaths.ProtoReflect.Descriptor instead.
565
529
func (*BuiltPaths) Descriptor() ([]byte, []int) {
566
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{8}
530
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{7}
567
531
}
568
532
569
533
func (x *BuiltPaths) GetPaths() []string {
···
589
553
590
554
func (x *CacheDrain) Reset() {
591
555
*x = CacheDrain{}
592
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[9]
556
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[8]
593
557
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
594
558
ms.StoreMessageInfo(mi)
595
559
}
···
601
565
func (*CacheDrain) ProtoMessage() {}
602
566
603
567
func (x *CacheDrain) ProtoReflect() protoreflect.Message {
604
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[9]
568
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[8]
605
569
if x != nil {
606
570
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
607
571
if ms.LoadMessageInfo() == nil {
···
614
578
615
579
// Deprecated: Use CacheDrain.ProtoReflect.Descriptor instead.
616
580
func (*CacheDrain) Descriptor() ([]byte, []int) {
617
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{9}
581
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{8}
618
582
}
619
583
620
584
func (x *CacheDrain) GetTimeoutSeconds() uint32 {
···
637
601
638
602
func (x *CacheDrainResult) Reset() {
639
603
*x = CacheDrainResult{}
640
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[10]
604
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[9]
641
605
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
642
606
ms.StoreMessageInfo(mi)
643
607
}
···
649
613
func (*CacheDrainResult) ProtoMessage() {}
650
614
651
615
func (x *CacheDrainResult) ProtoReflect() protoreflect.Message {
652
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[10]
616
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[9]
653
617
if x != nil {
654
618
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
655
619
if ms.LoadMessageInfo() == nil {
···
662
626
663
627
// Deprecated: Use CacheDrainResult.ProtoReflect.Descriptor instead.
664
628
func (*CacheDrainResult) Descriptor() ([]byte, []int) {
665
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{10}
629
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{9}
666
630
}
667
631
668
632
func (x *CacheDrainResult) GetError() string {
···
708
672
709
673
func (x *Poweroff) Reset() {
710
674
*x = Poweroff{}
711
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[11]
675
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[10]
712
676
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
713
677
ms.StoreMessageInfo(mi)
714
678
}
···
720
684
func (*Poweroff) ProtoMessage() {}
721
685
722
686
func (x *Poweroff) ProtoReflect() protoreflect.Message {
723
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[11]
687
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[10]
724
688
if x != nil {
725
689
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
726
690
if ms.LoadMessageInfo() == nil {
···
733
697
734
698
// Deprecated: Use Poweroff.ProtoReflect.Descriptor instead.
735
699
func (*Poweroff) Descriptor() ([]byte, []int) {
736
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{11}
700
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{10}
737
701
}
738
702
739
703
type PoweroffResult struct {
···
745
709
746
710
func (x *PoweroffResult) Reset() {
747
711
*x = PoweroffResult{}
748
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[12]
712
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[11]
749
713
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
750
714
ms.StoreMessageInfo(mi)
751
715
}
···
757
721
func (*PoweroffResult) ProtoMessage() {}
758
722
759
723
func (x *PoweroffResult) ProtoReflect() protoreflect.Message {
760
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[12]
724
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[11]
761
725
if x != nil {
762
726
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
763
727
if ms.LoadMessageInfo() == nil {
···
770
734
771
735
// Deprecated: Use PoweroffResult.ProtoReflect.Descriptor instead.
772
736
func (*PoweroffResult) Descriptor() ([]byte, []int) {
773
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{12}
737
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{11}
774
738
}
775
739
776
740
func (x *PoweroffResult) GetError() string {
···
786
750
Hello *Hello `protobuf:"bytes,2,opt,name=hello,proto3" json:"hello,omitempty"`
787
751
Init *Init `protobuf:"bytes,3,opt,name=init,proto3" json:"init,omitempty"`
788
752
ExecStart *ExecStart `protobuf:"bytes,4,opt,name=exec_start,json=execStart,proto3" json:"exec_start,omitempty"`
789
-
ExecStdout *ExecStdout `protobuf:"bytes,5,opt,name=exec_stdout,json=execStdout,proto3" json:"exec_stdout,omitempty"`
790
753
ExecStderr *ExecStderr `protobuf:"bytes,6,opt,name=exec_stderr,json=execStderr,proto3" json:"exec_stderr,omitempty"`
791
754
ExecExit *ExecExit `protobuf:"bytes,7,opt,name=exec_exit,json=execExit,proto3" json:"exec_exit,omitempty"`
792
755
ActivateConfig *ActivateConfig `protobuf:"bytes,8,opt,name=activate_config,json=activateConfig,proto3" json:"activate_config,omitempty"`
···
802
765
803
766
func (x *Message) Reset() {
804
767
*x = Message{}
805
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[13]
768
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[12]
806
769
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
807
770
ms.StoreMessageInfo(mi)
808
771
}
···
814
777
func (*Message) ProtoMessage() {}
815
778
816
779
func (x *Message) ProtoReflect() protoreflect.Message {
817
-
mi := &file_spindle_agent_v1_agent_proto_msgTypes[13]
780
+
mi := &file_spindle_agent_v1_agent_proto_msgTypes[12]
818
781
if x != nil {
819
782
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
820
783
if ms.LoadMessageInfo() == nil {
···
827
790
828
791
// Deprecated: Use Message.ProtoReflect.Descriptor instead.
829
792
func (*Message) Descriptor() ([]byte, []int) {
830
-
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{13}
793
+
return file_spindle_agent_v1_agent_proto_rawDescGZIP(), []int{12}
831
794
}
832
795
833
796
func (x *Message) GetId() string {
···
854
817
func (x *Message) GetExecStart() *ExecStart {
855
818
if x != nil {
856
819
return x.ExecStart
857
-
}
858
-
return nil
859
-
}
860
-
861
-
func (x *Message) GetExecStdout() *ExecStdout {
862
-
if x != nil {
863
-
return x.ExecStdout
864
820
}
865
821
return nil
866
822
}
···
944
900
"\x19cache_trusted_public_keys\x18\x02 \x03(\tR\x16cacheTrustedPublicKeys\x121\n" +
945
901
"\x15cache_read_proxy_port\x18\x03 \x01(\rR\x12cacheReadProxyPort\x125\n" +
946
902
"\x17cache_upload_proxy_port\x18\x04 \x01(\rR\x14cacheUploadProxyPort\x12$\n" +
947
-
"\x0edns_proxy_port\x18\x05 \x01(\rR\fdnsProxyPort\"\x80\x01\n" +
903
+
"\x0edns_proxy_port\x18\x05 \x01(\rR\fdnsProxyPort\"\xaa\x01\n" +
948
904
"\tExecStart\x12\x12\n" +
949
905
"\x04argv\x18\x01 \x03(\tR\x04argv\x12\x10\n" +
950
906
"\x03env\x18\x02 \x03(\tR\x03env\x12\x10\n" +
951
907
"\x03cwd\x18\x03 \x01(\tR\x03cwd\x12\x12\n" +
952
908
"\x04user\x18\x04 \x01(\tR\x04user\x12'\n" +
953
-
"\x0ftimeout_seconds\x18\x05 \x01(\rR\x0etimeoutSeconds\" \n" +
954
-
"\n" +
955
-
"ExecStdout\x12\x12\n" +
956
-
"\x04data\x18\x01 \x01(\tR\x04data\" \n" +
909
+
"\x0ftimeout_seconds\x18\x05 \x01(\rR\x0etimeoutSeconds\x12(\n" +
910
+
"\x10stdio_vsock_port\x18\x06 \x01(\rR\x0estdioVsockPort\" \n" +
957
911
"\n" +
958
912
"ExecStderr\x12\x12\n" +
959
-
"\x04data\x18\x01 \x01(\tR\x04data\"Z\n" +
913
+
"\x04data\x18\x01 \x01(\fR\x04data\"Z\n" +
960
914
"\bExecExit\x12\x1b\n" +
961
915
"\texit_code\x18\x01 \x01(\x05R\bexitCode\x12\x14\n" +
962
916
"\x05error\x18\x02 \x01(\tR\x05error\x12\x1b\n" +
···
990
944
"\n" +
991
945
"\bPoweroff\"&\n" +
992
946
"\x0ePoweroffResult\x12\x14\n" +
993
-
"\x05error\x18\x01 \x01(\tR\x05error\"\xa8\b\n" +
947
+
"\x05error\x18\x01 \x01(\tR\x05error\"\xe8\a\n" +
994
948
"\aMessage\x12\x17\n" +
995
949
"\x02id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x02id\x12-\n" +
996
950
"\x05hello\x18\x02 \x01(\v2\x17.spindle.agent.v1.HelloR\x05hello\x12*\n" +
997
951
"\x04init\x18\x03 \x01(\v2\x16.spindle.agent.v1.InitR\x04init\x12:\n" +
998
952
"\n" +
999
953
"exec_start\x18\x04 \x01(\v2\x1b.spindle.agent.v1.ExecStartR\texecStart\x12=\n" +
1000
-
"\vexec_stdout\x18\x05 \x01(\v2\x1c.spindle.agent.v1.ExecStdoutR\n" +
1001
-
"execStdout\x12=\n" +
1002
954
"\vexec_stderr\x18\x06 \x01(\v2\x1c.spindle.agent.v1.ExecStderrR\n" +
1003
955
"execStderr\x127\n" +
1004
956
"\texec_exit\x18\a \x01(\v2\x1a.spindle.agent.v1.ExecExitR\bexecExit\x12I\n" +
···
1011
963
"cacheDrain\x12P\n" +
1012
964
"\x12cache_drain_result\x18\f \x01(\v2\".spindle.agent.v1.CacheDrainResultR\x10cacheDrainResult\x126\n" +
1013
965
"\bpoweroff\x18\r \x01(\v2\x1a.spindle.agent.v1.PoweroffR\bpoweroff\x12I\n" +
1014
-
"\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xb9\x01\xbaH\xb5\x01\"\xb2\x01\n" +
966
+
"\x0fpoweroff_result\x18\x0e \x01(\v2 .spindle.agent.v1.PoweroffResultR\x0epoweroffResult:\xac\x01\xbaH\xa8\x01\"\xa5\x01\n" +
1015
967
"\x05hello\n" +
1016
968
"\x04init\n" +
1017
969
"\n" +
1018
970
"exec_start\n" +
1019
-
"\vexec_stdout\n" +
1020
971
"\vexec_stderr\n" +
1021
972
"\texec_exit\n" +
1022
973
"\x0factivate_config\n" +
···
1025
976
"\vcache_drain\n" +
1026
977
"\x12cache_drain_result\n" +
1027
978
"\bpoweroff\n" +
1028
-
"\x0fpoweroff_result\x10\x01B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3"
979
+
"\x0fpoweroff_result\x10\x01J\x04\b\x05\x10\x06J\x04\b\x0f\x10\x10B1Z/tangled.org/core/spindle/agentproto/gen;agentv1b\x06proto3"
1029
980
1030
981
var (
1031
982
file_spindle_agent_v1_agent_proto_rawDescOnce sync.Once
···
1039
990
return file_spindle_agent_v1_agent_proto_rawDescData
1040
991
}
1041
992
1042
-
var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 14)
993
+
var file_spindle_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 13)
1043
994
var file_spindle_agent_v1_agent_proto_goTypes = []any{
1044
995
(*Hello)(nil), // 0: spindle.agent.v1.Hello
1045
996
(*Init)(nil), // 1: spindle.agent.v1.Init
1046
997
(*ExecStart)(nil), // 2: spindle.agent.v1.ExecStart
1047
-
(*ExecStdout)(nil), // 3: spindle.agent.v1.ExecStdout
1048
-
(*ExecStderr)(nil), // 4: spindle.agent.v1.ExecStderr
1049
-
(*ExecExit)(nil), // 5: spindle.agent.v1.ExecExit
1050
-
(*ActivateConfig)(nil), // 6: spindle.agent.v1.ActivateConfig
1051
-
(*ActivateConfigResult)(nil), // 7: spindle.agent.v1.ActivateConfigResult
1052
-
(*BuiltPaths)(nil), // 8: spindle.agent.v1.BuiltPaths
1053
-
(*CacheDrain)(nil), // 9: spindle.agent.v1.CacheDrain
1054
-
(*CacheDrainResult)(nil), // 10: spindle.agent.v1.CacheDrainResult
1055
-
(*Poweroff)(nil), // 11: spindle.agent.v1.Poweroff
1056
-
(*PoweroffResult)(nil), // 12: spindle.agent.v1.PoweroffResult
1057
-
(*Message)(nil), // 13: spindle.agent.v1.Message
998
+
(*ExecStderr)(nil), // 3: spindle.agent.v1.ExecStderr
999
+
(*ExecExit)(nil), // 4: spindle.agent.v1.ExecExit
1000
+
(*ActivateConfig)(nil), // 5: spindle.agent.v1.ActivateConfig
1001
+
(*ActivateConfigResult)(nil), // 6: spindle.agent.v1.ActivateConfigResult
1002
+
(*BuiltPaths)(nil), // 7: spindle.agent.v1.BuiltPaths
1003
+
(*CacheDrain)(nil), // 8: spindle.agent.v1.CacheDrain
1004
+
(*CacheDrainResult)(nil), // 9: spindle.agent.v1.CacheDrainResult
1005
+
(*Poweroff)(nil), // 10: spindle.agent.v1.Poweroff
1006
+
(*PoweroffResult)(nil), // 11: spindle.agent.v1.PoweroffResult
1007
+
(*Message)(nil), // 12: spindle.agent.v1.Message
1058
1008
}
1059
1009
var file_spindle_agent_v1_agent_proto_depIdxs = []int32{
1060
1010
0, // 0: spindle.agent.v1.Message.hello:type_name -> spindle.agent.v1.Hello
1061
1011
1, // 1: spindle.agent.v1.Message.init:type_name -> spindle.agent.v1.Init
1062
1012
2, // 2: spindle.agent.v1.Message.exec_start:type_name -> spindle.agent.v1.ExecStart
1063
-
3, // 3: spindle.agent.v1.Message.exec_stdout:type_name -> spindle.agent.v1.ExecStdout
1064
-
4, // 4: spindle.agent.v1.Message.exec_stderr:type_name -> spindle.agent.v1.ExecStderr
1065
-
5, // 5: spindle.agent.v1.Message.exec_exit:type_name -> spindle.agent.v1.ExecExit
1066
-
6, // 6: spindle.agent.v1.Message.activate_config:type_name -> spindle.agent.v1.ActivateConfig
1067
-
7, // 7: spindle.agent.v1.Message.activate_config_result:type_name -> spindle.agent.v1.ActivateConfigResult
1068
-
8, // 8: spindle.agent.v1.Message.built_paths:type_name -> spindle.agent.v1.BuiltPaths
1069
-
9, // 9: spindle.agent.v1.Message.cache_drain:type_name -> spindle.agent.v1.CacheDrain
1070
-
10, // 10: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult
1071
-
11, // 11: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff
1072
-
12, // 12: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult
1073
-
13, // [13:13] is the sub-list for method output_type
1074
-
13, // [13:13] is the sub-list for method input_type
1075
-
13, // [13:13] is the sub-list for extension type_name
1076
-
13, // [13:13] is the sub-list for extension extendee
1077
-
0, // [0:13] is the sub-list for field type_name
1013
+
3, // 3: spindle.agent.v1.Message.exec_stderr:type_name -> spindle.agent.v1.ExecStderr
1014
+
4, // 4: spindle.agent.v1.Message.exec_exit:type_name -> spindle.agent.v1.ExecExit
1015
+
5, // 5: spindle.agent.v1.Message.activate_config:type_name -> spindle.agent.v1.ActivateConfig
1016
+
6, // 6: spindle.agent.v1.Message.activate_config_result:type_name -> spindle.agent.v1.ActivateConfigResult
1017
+
7, // 7: spindle.agent.v1.Message.built_paths:type_name -> spindle.agent.v1.BuiltPaths
1018
+
8, // 8: spindle.agent.v1.Message.cache_drain:type_name -> spindle.agent.v1.CacheDrain
1019
+
9, // 9: spindle.agent.v1.Message.cache_drain_result:type_name -> spindle.agent.v1.CacheDrainResult
1020
+
10, // 10: spindle.agent.v1.Message.poweroff:type_name -> spindle.agent.v1.Poweroff
1021
+
11, // 11: spindle.agent.v1.Message.poweroff_result:type_name -> spindle.agent.v1.PoweroffResult
1022
+
12, // [12:12] is the sub-list for method output_type
1023
+
12, // [12:12] is the sub-list for method input_type
1024
+
12, // [12:12] is the sub-list for extension type_name
1025
+
12, // [12:12] is the sub-list for extension extendee
1026
+
0, // [0:12] is the sub-list for field type_name
1078
1027
}
1079
1028
1080
1029
func init() { file_spindle_agent_v1_agent_proto_init() }
···
1088
1037
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
1089
1038
RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_agent_v1_agent_proto_rawDesc), len(file_spindle_agent_v1_agent_proto_rawDesc)),
1090
1039
NumEnums: 0,
1091
-
NumMessages: 14,
1040
+
NumMessages: 13,
1092
1041
NumExtensions: 0,
1093
1042
NumServices: 0,
1094
1043
},
+1
-1
spindle/agentproto/protocol.go
+1
-1
spindle/agentproto/protocol.go
+8
-9
spindle/agentproto/spindle/agent/v1/agent.proto
+8
-9
spindle/agentproto/spindle/agent/v1/agent.proto
···
27
27
string cwd = 3;
28
28
string user = 4;
29
29
uint32 timeout_seconds = 5;
30
-
}
31
-
32
-
message ExecStdout {
33
-
string data = 1;
30
+
uint32 stdio_vsock_port = 6;
34
31
}
35
32
36
33
message ExecStderr {
37
-
string data = 1;
34
+
bytes data = 1;
38
35
}
39
36
40
37
message ExecExit {
···
85
82
message Message {
86
83
option (buf.validate.message).oneof = {
87
84
fields: [
88
-
"hello", "init", "exec_start", "exec_stdout", "exec_stderr", "exec_exit",
89
-
"activate_config", "activate_config_result", "built_paths", "cache_drain",
90
-
"cache_drain_result", "poweroff", "poweroff_result"
85
+
"hello", "init", "exec_start", "exec_stderr",
86
+
"exec_exit", "activate_config", "activate_config_result", "built_paths",
87
+
"cache_drain", "cache_drain_result", "poweroff", "poweroff_result"
91
88
],
92
89
required: true
93
90
};
···
97
94
Hello hello = 2;
98
95
Init init = 3;
99
96
ExecStart exec_start = 4;
100
-
ExecStdout exec_stdout = 5;
101
97
ExecStderr exec_stderr = 6;
102
98
ExecExit exec_exit = 7;
103
99
ActivateConfig activate_config = 8;
···
107
103
CacheDrainResult cache_drain_result = 12;
108
104
Poweroff poweroff = 13;
109
105
PoweroffResult poweroff_result = 14;
106
+
107
+
// old framed stdio fields
108
+
reserved 5, 15;
110
109
}
+3
-2
spindle/engines/microvm/README.md
+3
-2
spindle/engines/microvm/README.md
···
179
179
builds and activates it before the user steps run. Afterwards, each step is sent
180
180
as an exec request (`$shell -lc <command>` as an unprivileged workflow user in
181
181
`/workspace/repo`, with workflow/step environment and unlocked secrets), and
182
-
stdout/stderr stream back as messages until an exit message arrives. Timeouts
183
-
are cooperative: we derive a deadline from the workflow timeout and ship it to
182
+
stdout streams back raw over a dedicated per-exec vsock connection (the
183
+
agent dials it; stdin rides the same socket), while stderr and the exit
184
+
status stay on the control channel. Timeouts are cooperative: we derive a deadline from the workflow timeout and ship it to
184
185
the guest, with a little grace on the host side so the guest gets to report the
185
186
timeout itself. While a step runs we also watch for the VM crashing, if it does
186
187
we tail the serial (and qemu) logs into the step's stderr so you get something
+73
-11
spindle/engines/microvm/agent.go
+73
-11
spindle/engines/microvm/agent.go
···
107
107
type AgentExec struct {
108
108
*agentv1.ExecStart
109
109
ID string
110
+
Stdin io.Reader
110
111
Stdout io.Writer
111
112
Stderr io.Writer
112
113
}
113
114
114
115
type AgentSession struct {
115
116
conn net.Conn
117
+
cid uint32
116
118
enc *agentproto.Encoder
117
119
dec *agentproto.Decoder
118
120
l *slog.Logger
119
121
mu sync.Mutex
120
122
}
121
123
122
-
func NewAgentSession(conn net.Conn, l *slog.Logger) *AgentSession {
124
+
func NewAgentSession(conn net.Conn, cid uint32, l *slog.Logger) *AgentSession {
123
125
return &AgentSession{
124
126
conn: conn,
127
+
cid: cid,
125
128
enc: agentproto.NewEncoder(conn),
126
129
dec: agentproto.NewDecoder(conn),
127
130
l: l,
···
140
143
if helloPayload == nil {
141
144
return fmt.Errorf("expected agent hello, got nil")
142
145
}
146
+
if helloPayload.ProtocolVersion != agentproto.ProtocolVersion {
147
+
return fmt.Errorf("agent protocol version %d, want %d (stale guest image?)", helloPayload.ProtocolVersion, agentproto.ProtocolVersion)
148
+
}
143
149
s.l.Info("agent connected", "protocol", helloPayload.ProtocolVersion, "version", helloPayload.AgentVersion, "boot", helloPayload.BootId, "nix", helloPayload.NixVersion)
144
150
145
151
if err := s.enc.Encode(&agentproto.Message{
···
163
169
exec.ExecStart.TimeoutSeconds = timeoutSeconds(ctx, guestTimeoutGrace)
164
170
}
165
171
172
+
ln, port, err := listenRandomVsockPort(ctx)
173
+
if err != nil {
174
+
return 0, fmt.Errorf("listen for exec stdio: %w", err)
175
+
}
176
+
defer ln.Close()
177
+
exec.ExecStart.StdioVsockPort = port
178
+
166
179
if err := s.enc.Encode(&agentproto.Message{
167
180
Id: exec.ID,
168
181
ExecStart: exec.ExecStart,
169
182
}); err != nil {
170
183
return 0, fmt.Errorf("send exec_start: %w", err)
171
184
}
185
+
186
+
filtered := &cidFilteredVsockListener{Listener: ln, cid: s.cid, logger: s.l}
187
+
stdioDone := make(chan error, 1)
188
+
go func() {
189
+
stdioDone <- pumpStdio(ctx, filtered, exec.Stdin, exec.Stdout)
190
+
}()
172
191
173
192
for {
174
193
msg, err := s.decode(ctx)
175
194
if err != nil {
195
+
ln.Close()
196
+
<-stdioDone
176
197
return 0, err
177
198
}
178
199
if msg.BuiltPaths == nil && msg.Id != exec.ID {
179
200
continue
180
201
}
181
202
182
-
if p := msg.ExecStdout; p != nil {
183
-
_, _ = io.WriteString(exec.Stdout, p.Data)
203
+
if p := msg.BuiltPaths; p != nil {
204
+
// s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths))
184
205
} else if p := msg.ExecStderr; p != nil {
185
-
_, _ = io.WriteString(exec.Stderr, p.Data)
186
-
} else if p := msg.BuiltPaths; p != nil {
187
-
// s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths))
206
+
_, _ = exec.Stderr.Write(p.Data)
188
207
} else if p := msg.ExecExit; p != nil {
189
208
if p.Error != "" {
190
209
s.l.Warn("guest exec error", "id", msg.Id, "error", p.Error)
191
210
}
211
+
ln.Close() // wake Accept if the guest never dialed
212
+
if err := <-stdioDone; err != nil {
213
+
return 0, err
214
+
}
192
215
if p.TimedOut {
193
216
return int(p.ExitCode), errGuestTimedOut
194
217
}
···
197
220
}
198
221
}
199
222
223
+
func pumpStdio(ctx context.Context, ln net.Listener, stdin io.Reader, stdout io.Writer) error {
224
+
conn, err := ln.Accept()
225
+
if err != nil {
226
+
if errors.Is(err, net.ErrClosed) {
227
+
return nil // exec ended before dialing
228
+
}
229
+
return fmt.Errorf("accept guest stdio connection: %w", err)
230
+
}
231
+
vsockConn, ok := conn.(*vsock.Conn)
232
+
if !ok {
233
+
conn.Close()
234
+
return fmt.Errorf("guest connection is not a vsock connection")
235
+
}
236
+
defer vsockConn.Close()
237
+
stop := context.AfterFunc(ctx, func() { vsockConn.Close() })
238
+
defer stop()
239
+
240
+
stdinDone := make(chan error, 1)
241
+
go func() {
242
+
stdinDone <- writeStdin(vsockConn, stdin)
243
+
}()
244
+
245
+
if _, err := io.Copy(stdout, conn); err != nil {
246
+
vsockConn.Close()
247
+
<-stdinDone
248
+
return fmt.Errorf("read guest stdout: %w", err)
249
+
}
250
+
if err := <-stdinDone; err != nil {
251
+
return fmt.Errorf("write guest stdin: %w", err)
252
+
}
253
+
return nil
254
+
}
255
+
256
+
// send stdin EOF without closing stdout
257
+
func writeStdin(conn *vsock.Conn, r io.Reader) error {
258
+
if r != nil {
259
+
if _, err := io.Copy(conn, r); err != nil {
260
+
return err
261
+
}
262
+
}
263
+
return conn.CloseWrite()
264
+
}
265
+
200
266
func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agentv1.ActivateConfig, out io.Writer) (*agentv1.ActivateConfigResult, error) {
201
267
s.mu.Lock()
202
268
defer s.mu.Unlock()
···
227
293
// s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths))
228
294
} else if p := msg.ExecStderr; p != nil {
229
295
if out != nil {
230
-
_, _ = io.WriteString(out, p.Data)
231
-
}
232
-
} else if p := msg.ExecStdout; p != nil {
233
-
if out != nil {
234
-
_, _ = io.WriteString(out, p.Data)
296
+
_, _ = out.Write(p.Data)
235
297
}
236
298
} else if p := msg.ActivateConfigResult; p != nil {
237
299
if p.Error != "" {
+1
-1
spindle/engines/microvm/engine.go
+1
-1
spindle/engines/microvm/engine.go
···
309
309
return err
310
310
}
311
311
312
-
agentSession := NewAgentSession(conn, l)
312
+
agentSession := NewAgentSession(conn, cid, l)
313
313
initCtx, cancelInit := context.WithTimeout(ctx, agentHandshakeTimeout)
314
314
defer cancelInit()
315
315
if err := agentSession.Init(initCtx, &agentv1.Init{