Monorepo for Tangled
spindle/microvm: stream realize / build output in nixos activation step
Signed-off-by: dawn <dawn@tangled.org>
+72
-12
+1
-1
cmd/spindle-microvm-run/main_linux.go
+1
-1
cmd/spindle-microvm-run/main_linux.go
+61
-9
shuttle/src/activation.rs
+61
-9
shuttle/src/activation.rs
···
1
-
use crate::command::{self, Spec, run_capture};
1
+
use crate::command::{CaptureOutput, OutKind, Spec, run_capture, spawn_streaming};
2
2
use crate::nix_config::{SPINDLE_RUN_DIR, nix_executable};
3
3
use crate::protocol::{self, Message, v1};
4
4
use anyhow::{Context, Result};
···
14
14
15
15
pub async fn run(id: String, req: v1::ActivateConfig, out: Sender<Message>) {
16
16
let config_key = req.config_key.clone();
17
-
let result = activate(&req).await;
17
+
let result = activate(&id, &req, &out).await;
18
18
let msg = Message {
19
19
id,
20
20
activate_config_result: Some(v1::ActivateConfigResult {
···
29
29
let _ = out.send(msg).await;
30
30
}
31
31
32
-
async fn activate(req: &v1::ActivateConfig) -> Result<PathBuf> {
32
+
async fn activate(id: &str, req: &v1::ActivateConfig, out: &Sender<Message>) -> Result<PathBuf> {
33
33
let need_build = req.toplevel.is_empty();
34
34
let timeout = (req.timeout_seconds > 0)
35
35
.then(|| Duration::from_secs(u64::from(req.timeout_seconds)))
···
37
37
.unwrap_or(Duration::from_secs(2 * 60));
38
38
39
39
let toplevel = if need_build {
40
-
build_toplevel(req, timeout).await?
40
+
build_toplevel(id, req, timeout, out).await?
41
41
} else {
42
-
realise_toplevel(&req.toplevel, timeout).await?
42
+
realise_toplevel(id, &req.toplevel, timeout, out).await?
43
43
};
44
44
45
45
if !toplevel.starts_with("/nix/store/") {
···
97
97
Ok(())
98
98
}
99
99
100
-
async fn build_toplevel(req: &v1::ActivateConfig, timeout: Duration) -> Result<PathBuf> {
100
+
async fn build_toplevel(
101
+
id: &str,
102
+
req: &v1::ActivateConfig,
103
+
timeout: Duration,
104
+
out: &Sender<Message>,
105
+
) -> Result<PathBuf> {
101
106
let user_config = (req.user_config.is_empty())
102
107
.then_some("{}")
103
108
.unwrap_or_else(|| &req.user_config);
···
106
111
write_user_config(user_config).context("write user config")?;
107
112
108
113
info!("running nix build command for user config toplevel...");
109
-
let output = run_capture(
114
+
let output = run_streaming_stderr(
110
115
Spec::new(nix_executable())
111
116
.args([
112
117
"build",
···
118
123
])
119
124
.cwd(SPINDLE_RUN_DIR)
120
125
.timeout(timeout),
126
+
id,
127
+
out,
121
128
)
122
129
.await?;
123
130
···
151
158
Ok(())
152
159
}
153
160
154
-
async fn realise_toplevel(toplevel: &str, timeout: Duration) -> Result<PathBuf> {
161
+
async fn realise_toplevel(
162
+
id: &str,
163
+
toplevel: &str,
164
+
timeout: Duration,
165
+
out: &Sender<Message>,
166
+
) -> Result<PathBuf> {
155
167
if !toplevel.starts_with("/nix/store/") {
156
168
anyhow::bail!("cached config toplevel {toplevel:?} is not a nix store path");
157
169
}
158
-
let output = command::run_capture(
170
+
let output = run_streaming_stderr(
159
171
Spec::new(nix_executable())
160
172
.args(["build", "--no-link", "--show-trace", toplevel])
161
173
.timeout(timeout),
174
+
id,
175
+
out,
162
176
)
163
177
.await?;
164
178
if !output.success() {
···
171
185
}
172
186
173
187
Ok(PathBuf::from(toplevel))
188
+
}
189
+
190
+
// streams stderr but captures stdout
191
+
async fn run_streaming_stderr(
192
+
spec: Spec,
193
+
id: &str,
194
+
out: &Sender<Message>,
195
+
) -> Result<CaptureOutput> {
196
+
let running = spawn_streaming(spec)?;
197
+
let mut stdout = Vec::new();
198
+
let mut stderr = Vec::new();
199
+
let (mut events, exit_task) = running.into_parts();
200
+
201
+
while let Some(event) = events.recv().await {
202
+
match event.kind {
203
+
OutKind::Stdout => stdout.extend_from_slice(&event.data),
204
+
OutKind::Stderr => {
205
+
stderr.extend_from_slice(&event.data);
206
+
let data = String::from_utf8_lossy(&event.data).into_owned();
207
+
let _ = out
208
+
.send(Message {
209
+
id: id.to_owned(),
210
+
exec_stderr: Some(v1::ExecStderr { data }),
211
+
..Default::default()
212
+
})
213
+
.await;
214
+
}
215
+
}
216
+
}
217
+
218
+
let exit = exit_task
219
+
.await
220
+
.unwrap_or_else(|error| Err(anyhow::anyhow!("command supervisor failed: {error}")))?;
221
+
Ok(CaptureOutput {
222
+
exit,
223
+
stdout,
224
+
stderr,
225
+
})
174
226
}
175
227
176
228
async fn switch_to_configuration(toplevel: &Path, timeout: Duration) -> Result<()> {
+9
-1
spindle/engines/microvm/agent.go
+9
-1
spindle/engines/microvm/agent.go
···
197
197
}
198
198
}
199
199
200
-
func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agentv1.ActivateConfig) (*agentv1.ActivateConfigResult, error) {
200
+
func (s *AgentSession) ActivateConfig(ctx context.Context, id string, req *agentv1.ActivateConfig, out io.Writer) (*agentv1.ActivateConfigResult, error) {
201
201
s.mu.Lock()
202
202
defer s.mu.Unlock()
203
203
···
225
225
226
226
if p := msg.BuiltPaths; p != nil {
227
227
// s.l.Debug("guest built paths", "reason", p.Reason, "count", len(p.Paths))
228
+
} else if p := msg.ExecStderr; p != nil {
229
+
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)
235
+
}
228
236
} else if p := msg.ActivateConfigResult; p != nil {
229
237
if p.Error != "" {
230
238
return nil, fmt.Errorf("activate config failed: %s", p.Error)