GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
f98aebb2e0ca8cbef1204344b8537eecdc2869f6

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

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