GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
f55b3255dfc6ea3aa98b48d39eb15dac86844039

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

421lines · modecode

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