GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
b8910f97c258ca39c8f2e535fbc241ea1e1be3ff

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

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