GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

mirrored fromhttps://github.com/microsoft/openvmmAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
fe2bbc9829558a07c5e06d2b6ececc383b6593bf

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