GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
15e07a8246fa7bad80ef06f167641da58f4e021b

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

433lines · modecode

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