GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

mirrored from https://github.com/microsoft/openvmmAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
f102fc3006b833e3c6ff5c3dfd0d325f652ad2be

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

openvmm/flowey/flowey_cli/src/cli

flowey/flowey_cli/src/cli/exec_snippet.rs

415lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4use crate::cli::FlowBackendCli;
5use anyhow::Context;
6use flowey_core::node::FlowArch;
7use flowey_core::node::FlowBackend;
8use flowey_core::node::FlowPlatform;
9use flowey_core::node::GhOutput;
10use flowey_core::node::GhToRust;
11use flowey_core::node::NodeHandle;
12use flowey_core::node::RustToGh;
13use flowey_core::node::steps::rust::RustRuntimeServices;
14use flowey_core::node::user_facing::ClaimedGhParam;
15use flowey_core::node::user_facing::GhPermission;
16use flowey_core::node::user_facing::GhPermissionValue;
17use flowey_core::pipeline::HostExt;
18use flowey_core::pipeline::PipelineBackendHint;
19use serde::Deserialize;
20use serde::Serialize;
21use std::collections::BTreeMap;
22use std::path::PathBuf;
23
24pub fn construct_exec_snippet_cli(
25 flowey_bin: &str,
26 node_modpath: &str,
27 snippet_idx: usize,
28 job_idx: usize,
29) -> String {
30 format!(r#"{flowey_bin} e {job_idx} {node_modpath} {snippet_idx}"#)
31}
32
33/// (internal) execute an inline code snippet from the given node.
34#[derive(clap::Args)]
35pub struct ExecSnippet {
36 /// Job idx to query `pipeline_static_db` with
37 pub(crate) job_idx: usize,
38
39 node_modpath_and_snippet_idx: Vec<String>,
40
41 /// (debug) If true, the snippet will not actually be run
42 #[clap(long)]
43 dry_run: bool,
44}
45
46pub const VAR_DB_SEEDVAR_FLOWEY_WORKING_DIR: &str = "_internal_WORKING_DIR";
47pub const VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR: &str = "_internal_PERSISTENT_STORAGE_DIR";
48
49impl ExecSnippet {
50 pub fn run(self) -> anyhow::Result<()> {
51 let Self {
52 node_modpath_and_snippet_idx,
53 job_idx,
54 dry_run,
55 } = self;
56
57 let flow_platform = FlowPlatform::host(PipelineBackendHint::Local);
58 let flow_arch = FlowArch::host(PipelineBackendHint::Local);
59
60 let mut runtime_var_db = super::var_db::open_var_db(job_idx)?;
61
62 let working_dir: PathBuf = {
63 let Some((working_dir, _)) =
64 runtime_var_db.try_get_var(VAR_DB_SEEDVAR_FLOWEY_WORKING_DIR)
65 else {
66 anyhow::bail!("var db was not seeded with {VAR_DB_SEEDVAR_FLOWEY_WORKING_DIR}");
67 };
68 serde_json::from_slice::<String>(&working_dir)
69 .context(format!(
70 "found {VAR_DB_SEEDVAR_FLOWEY_WORKING_DIR} in db, but it wasn't a json string!"
71 ))?
72 .into()
73 };
74
75 let FloweyPipelineStaticDb {
76 flow_backend,
77 var_db_backend_kind: _,
78 job_reqs,
79 job_command_wrappers,
80 } = {
81 let current_exe = std::env::current_exe()
82 .context("failed to get path to current flowey executable")?;
83 let pipeline_static_db =
84 fs_err::File::open(current_exe.with_file_name("pipeline.json"))?;
85 serde_json::from_reader(pipeline_static_db)?
86 };
87
88 let command_wrapper = job_command_wrappers.get(&job_idx).cloned();
89
90 for [node_modpath, snippet_idx] in node_modpath_and_snippet_idx
91 .chunks_exact(2)
92 .map(|x| -> [String; 2] { x.to_vec().try_into().unwrap() })
93 {
94 let snippet_idx = snippet_idx.parse::<usize>().unwrap();
95
96 let raw_json_reqs: Vec<Box<[u8]>> = job_reqs
97 .get(&job_idx)
98 .context("invalid job_idx")?
99 .get(&node_modpath)
100 .context("pipeline db did not include data for specified node")?
101 .iter()
102 .map(|v| v.0.clone())
103 .collect::<Vec<_>>();
104
105 let Some(node_handle) = NodeHandle::try_from_modpath(&node_modpath) else {
106 anyhow::bail!("could not find node with that name")
107 };
108
109 let mut node = node_handle.new_erased_node();
110
111 // each snippet gets its own isolated working dir
112 {
113 let snippet_working_dir = working_dir.join(format!(
114 "{}_{}",
115 node_handle.modpath().replace("::", "__"),
116 snippet_idx
117 ));
118 if !snippet_working_dir.exists() {
119 fs_err::create_dir_all(&snippet_working_dir)?;
120 }
121 log::trace!(
122 "Setting current working directory from {:?} to {:?}",
123 std::env::current_dir()?,
124 snippet_working_dir
125 );
126 std::env::set_current_dir(snippet_working_dir)?;
127 }
128
129 // not all backends support a persistent storage dir, therefore it is optional
130 let persistent_storage_dir_var = runtime_var_db
131 .try_get_var(VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR)
132 .is_some()
133 .then_some(VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR.to_owned());
134
135 let mut rust_runtime_services =
136 flowey_core::node::steps::rust::new_rust_runtime_services(
137 &mut runtime_var_db,
138 flow_backend.into(),
139 flow_platform,
140 flow_arch,
141 )?;
142
143 if let Some(ref wrapper) = command_wrapper {
144 rust_runtime_services.sh.set_wrapper(Some(wrapper.clone()));
145 }
146
147 let mut ctx_backend = ExecSnippetCtx::new(
148 flow_backend.into(),
149 flow_platform,
150 flow_arch,
151 node_handle,
152 snippet_idx,
153 dry_run,
154 persistent_storage_dir_var,
155 &mut rust_runtime_services,
156 );
157
158 let mut ctx = flowey_core::node::new_node_ctx(&mut ctx_backend);
159 node.emit(raw_json_reqs.clone(), &mut ctx)?;
160
161 match ctx_backend.into_result() {
162 Some(res) => res?,
163 None => {
164 if dry_run {
165 // all good, expected
166 } else {
167 anyhow::bail!("snippet wasn't run (invalid index)")
168 }
169 }
170 }
171 }
172
173 // Leave the last snippet's working dir so it can be deleted by later steps
174 std::env::set_current_dir(working_dir)?;
175
176 Ok(())
177 }
178}
179
180pub struct ExecSnippetCtx<'a, 'b> {
181 flow_backend: FlowBackend,
182 flow_platform: FlowPlatform,
183 flow_arch: FlowArch,
184 node_handle: NodeHandle,
185 rust_runtime_services: &'a mut RustRuntimeServices<'b>,
186 idx_tracker: usize,
187 var_tracker: usize,
188 target_idx: usize,
189 dry_run: bool,
190 persistent_storage_dir_var: Option<String>,
191 result: Option<anyhow::Result<()>>,
192}
193
194impl<'a, 'b> ExecSnippetCtx<'a, 'b> {
195 pub fn new(
196 flow_backend: FlowBackend,
197 flow_platform: FlowPlatform,
198 flow_arch: FlowArch,
199 node_handle: NodeHandle,
200 target_idx: usize,
201 dry_run: bool,
202 persistent_storage_dir_var: Option<String>,
203 rust_runtime_services: &'a mut RustRuntimeServices<'b>,
204 ) -> Self {
205 Self {
206 flow_backend,
207 flow_platform,
208 flow_arch,
209 node_handle,
210 rust_runtime_services,
211 var_tracker: 0,
212 idx_tracker: 0,
213 target_idx,
214 dry_run,
215 persistent_storage_dir_var,
216 result: None,
217 }
218 }
219
220 pub fn into_result(self) -> Option<anyhow::Result<()>> {
221 self.result
222 }
223}
224
225impl flowey_core::node::NodeCtxBackend for ExecSnippetCtx<'_, '_> {
226 fn on_request(&mut self, _node_handle: NodeHandle, _req: anyhow::Result<Box<[u8]>>) {
227 // nothing to do - filing requests only matters pre-exec
228 }
229
230 fn on_new_var(&mut self) -> String {
231 let v = self.var_tracker;
232 self.var_tracker += 1;
233 format!("{}:{}", self.node_handle.modpath(), v)
234 }
235
236 fn on_claimed_runtime_var(&mut self, _var: &str, _is_read: bool) {
237 // nothing to do - variable claims only matter pre-exec
238 }
239
240 fn on_emit_rust_step(
241 &mut self,
242 label: &str,
243 _can_merge: bool,
244 code: Box<
245 dyn for<'a> FnOnce(&'a mut RustRuntimeServices<'_>) -> anyhow::Result<()> + 'static,
246 >,
247 ) {
248 if self.idx_tracker == self.target_idx {
249 let label = if !label.is_empty() {
250 label
251 } else {
252 "<unnamed snippet>"
253 };
254
255 self.result = Some(run_code(
256 self.flow_backend,
257 format!("{} ({})", label, self.node_handle.modpath()),
258 self.dry_run,
259 || code(self.rust_runtime_services),
260 ));
261 }
262 self.idx_tracker += 1;
263 }
264
265 fn on_emit_ado_step(
266 &mut self,
267 label: &str,
268 _yaml_snippet: Box<
269 dyn for<'a> FnOnce(
270 &'a mut flowey_core::node::user_facing::AdoStepServices<'_>,
271 ) -> String,
272 >,
273 code: Option<
274 Box<
275 dyn for<'a> FnOnce(&'a mut RustRuntimeServices<'_>) -> anyhow::Result<()> + 'static,
276 >,
277 >,
278 _condvar: Option<String>,
279 ) {
280 // don't need to care about condvar, since we wouldn't have been called
281 // if the YAML resolved the condvar to false.
282 if self.idx_tracker == self.target_idx {
283 if let Some(code) = code {
284 self.result = Some(run_code(
285 self.flow_backend,
286 format!(
287 "(inline snippet) {} ({})",
288 label,
289 self.node_handle.modpath()
290 ),
291 self.dry_run,
292 || code(self.rust_runtime_services),
293 ));
294 }
295 }
296
297 self.idx_tracker += 1;
298 }
299
300 fn on_emit_gh_step(
301 &mut self,
302 _label: &str,
303 _uses: &str,
304 _with: BTreeMap<String, ClaimedGhParam>,
305 _condvar: Option<String>,
306 _outputs: BTreeMap<String, Vec<GhOutput>>,
307 _permissions: BTreeMap<GhPermission, GhPermissionValue>,
308 _gh_to_rust: Vec<GhToRust>,
309 _rust_to_gh: Vec<RustToGh>,
310 ) {
311 self.idx_tracker += 1;
312 }
313
314 fn on_emit_side_effect_step(&mut self) {
315 // not executable, we simply skip
316 }
317
318 fn backend(&mut self) -> FlowBackend {
319 self.flow_backend
320 }
321
322 fn platform(&mut self) -> FlowPlatform {
323 self.flow_platform
324 }
325
326 fn arch(&mut self) -> FlowArch {
327 self.flow_arch
328 }
329
330 fn current_node(&self) -> NodeHandle {
331 self.node_handle
332 }
333
334 fn persistent_dir_path_var(&mut self) -> Option<String> {
335 self.persistent_storage_dir_var.clone()
336 }
337
338 fn on_unused_read_var(&mut self, _var: &str) {
339 // not relevant at runtime
340 }
341}
342
343#[derive(Serialize, Deserialize)]
344pub(crate) enum VarDbBackendKind {
345 Json,
346}
347
348#[derive(Serialize, Deserialize)]
349pub(crate) struct FloweyPipelineStaticDb {
350 pub flow_backend: FlowBackendCli,
351 pub var_db_backend_kind: VarDbBackendKind,
352 pub job_reqs: BTreeMap<usize, BTreeMap<String, Vec<SerializedRequest>>>,
353 pub job_command_wrappers: BTreeMap<usize, flowey_core::shell::CommandWrapperKind>,
354}
355
356// encode requests as JSON stored in a JSON string (to make human inspection
357// easier).
358#[derive(Serialize, Deserialize)]
359#[serde(transparent)]
360pub(crate) struct SerializedRequest(#[serde(with = "serialized_request")] pub Box<[u8]>);
361
362pub(crate) mod serialized_request {
363 use serde::Deserialize;
364 use serde::Deserializer;
365 use serde::Serializer;
366
367 #[expect(clippy::borrowed_box, reason = "required by serde")]
368 pub fn serialize<S: Serializer>(v: &Box<[u8]>, ser: S) -> Result<S::Ok, S::Error> {
369 ser.serialize_str(
370 &serde_json::to_string(&serde_json::from_slice::<serde_json::Value>(v).unwrap())
371 .unwrap(),
372 )
373 }
374
375 pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<Box<[u8]>, D::Error> {
376 let s: String = Deserialize::deserialize(d)?;
377 Ok(
378 serde_json::to_vec(&serde_json::from_str::<serde_json::Value>(&s).unwrap())
379 .unwrap()
380 .into(),
381 )
382 }
383}
384
385fn run_code(
386 flow_backend: FlowBackend,
387 label: impl std::fmt::Display,
388 dry_run: bool,
389 code: impl FnOnce() -> anyhow::Result<()>,
390) -> anyhow::Result<()> {
391 if matches!(flow_backend, FlowBackend::Ado) {
392 println!("##[group]=== {} ===", label)
393 } else {
394 // green color
395 log::info!("\x1B[0;32m=== {} ===\x1B[0m", label);
396 }
397
398 let result = if dry_run {
399 log::info!("...but not actually, because of --dry-run");
400 Ok(())
401 } else {
402 code()
403 };
404
405 // green color
406 log::info!("\x1B[0;32m=== done! ===\x1B[0m");
407
408 if matches!(flow_backend, FlowBackend::Ado) {
409 println!("##[endgroup]")
410 } else {
411 log::info!(""); // log a newline, for the pretty
412 }
413
414 result
415}
416