GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
release/2411-fork

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

openvmm/flowey/flowey_cli/src/pipeline_resolver

flowey/flowey_cli/src/pipeline_resolver/direct_run.rs

363lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4use crate::cli::exec_snippet::VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR;
5use crate::flow_resolver::stage1_dag::OutputGraphEntry;
6use crate::flow_resolver::stage1_dag::Step;
7use crate::pipeline_resolver::generic::ResolvedJobArtifact;
8use crate::pipeline_resolver::generic::ResolvedJobUseParameter;
9use crate::pipeline_resolver::generic::ResolvedPipeline;
10use crate::pipeline_resolver::generic::ResolvedPipelineJob;
11use flowey_core::node::steps::rust::RustRuntimeServices;
12use flowey_core::node::FlowArch;
13use flowey_core::node::FlowBackend;
14use flowey_core::node::FlowPlatform;
15use flowey_core::node::NodeHandle;
16use flowey_core::node::RuntimeVarDb;
17use flowey_core::pipeline::internal::Parameter;
18use petgraph::prelude::NodeIndex;
19use petgraph::visit::EdgeRef;
20use std::collections::BTreeSet;
21use std::path::Path;
22use std::path::PathBuf;
23
24pub struct ResolvedRunnableNode {
25 pub node_handle: NodeHandle,
26 pub steps: Vec<(
27 usize,
28 String,
29 Box<dyn for<'a> FnOnce(&'a mut RustRuntimeServices<'_>) -> anyhow::Result<()> + 'static>,
30 )>,
31}
32
33/// Directly run the pipeline using flowey
34pub fn direct_run(
35 pipeline: ResolvedPipeline,
36 windows_as_wsl: bool,
37 out_dir: PathBuf,
38 persist_dir: PathBuf,
39) -> anyhow::Result<()> {
40 direct_run_do_work(pipeline, windows_as_wsl, out_dir.clone(), persist_dir)?;
41
42 // cleanup
43 if out_dir.join(".job_artifacts").exists() {
44 fs_err::remove_dir_all(out_dir.join(".job_artifacts"))?;
45 }
46 if out_dir.join(".work").exists() {
47 fs_err::remove_dir_all(out_dir.join(".work"))?;
48 }
49
50 Ok(())
51}
52
53fn direct_run_do_work(
54 pipeline: ResolvedPipeline,
55 windows_as_wsl: bool,
56 out_dir: PathBuf,
57 persist_dir: PathBuf,
58) -> anyhow::Result<()> {
59 fs_err::create_dir_all(&out_dir)?;
60 let out_dir = std::path::absolute(out_dir)?;
61
62 fs_err::create_dir_all(&persist_dir)?;
63 let persist_dir = std::path::absolute(persist_dir)?;
64
65 let ResolvedPipeline {
66 graph,
67 order,
68 parameters,
69 ado_name: _,
70 ado_schedule_triggers: _,
71 ado_ci_triggers: _,
72 ado_pr_triggers: _,
73 ado_bootstrap_template: _,
74 ado_resources_repository: _,
75 ado_post_process_yaml_cb: _,
76 ado_variables: _,
77 gh_name: _,
78 gh_schedule_triggers: _,
79 gh_ci_triggers: _,
80 gh_pr_triggers: _,
81 gh_bootstrap_template: _,
82 } = pipeline;
83
84 let mut skipped_jobs = BTreeSet::new();
85
86 for idx in order {
87 let ResolvedPipelineJob {
88 ref root_nodes,
89 ref patches,
90 ref label,
91 platform,
92 arch,
93 cond_param_idx,
94 ado_pool: _,
95 ado_variables: _,
96 gh_override_if: _,
97 gh_global_env: _,
98 gh_pool: _,
99 gh_permissions: _,
100 ref external_read_vars,
101 ref parameters_used,
102 ref artifacts_used,
103 ref artifacts_published,
104 } = graph[idx];
105
106 // orange color
107 log::info!("\x1B[0;33m### job: {label} ###\x1B[0m");
108 log::info!("");
109
110 if graph
111 .edges_directed(idx, petgraph::Direction::Incoming)
112 .any(|e| skipped_jobs.contains(&NodeIndex::from(e.source().index() as u32)))
113 {
114 log::error!("job depends on job that was skipped. skipping job...");
115 log::info!("");
116 skipped_jobs.insert(idx);
117 continue;
118 }
119
120 // xtask-fmt allow-target-arch oneoff-flowey
121 let flow_arch = if cfg!(target_arch = "x86_64") {
122 FlowArch::X86_64
123 // xtask-fmt allow-target-arch oneoff-flowey
124 } else if cfg!(target_arch = "aarch64") {
125 FlowArch::Aarch64
126 } else {
127 unreachable!("flowey only runs on X86_64 or Aarch64 at the moment")
128 };
129
130 match (arch, flow_arch) {
131 (FlowArch::X86_64, FlowArch::X86_64) | (FlowArch::Aarch64, FlowArch::Aarch64) => (),
132 _ => {
133 log::error!("mismatch between job arch and local arch. skipping job...");
134 continue;
135 }
136 }
137
138 let platform_ok = match platform {
139 FlowPlatform::Windows => cfg!(windows) || (cfg!(target_os = "linux") && windows_as_wsl),
140 FlowPlatform::Linux(_) => cfg!(target_os = "linux"),
141 FlowPlatform::MacOs => cfg!(target_os = "macos"),
142 platform => panic!("unknown platform {platform}"),
143 };
144
145 if !platform_ok {
146 log::error!("mismatch between job platform and local platform. skipping job...");
147 log::info!("");
148 if crate::running_in_wsl() && matches!(platform, FlowPlatform::Windows) {
149 log::warn!("###");
150 log::warn!("### NOTE: detected that you're running in WSL2");
151 log::warn!(
152 "### if the the pipeline supports it, you can try passing --windows-as-wsl"
153 );
154 log::warn!("###");
155 log::info!("");
156 }
157 skipped_jobs.insert(idx);
158 continue;
159 }
160
161 let nodes = {
162 let mut resolved_local_nodes = Vec::new();
163
164 let (mut output_graph, _, err_unreachable_nodes) =
165 crate::flow_resolver::stage1_dag::stage1_dag(
166 FlowBackend::Local,
167 platform,
168 flow_arch,
169 patches.clone(),
170 root_nodes
171 .clone()
172 .into_iter()
173 .map(|(node, requests)| (node, (true, requests)))
174 .collect(),
175 external_read_vars.clone(),
176 Some(VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR.into()),
177 )?;
178
179 if err_unreachable_nodes.is_some() {
180 anyhow::bail!("detected unreachable nodes")
181 }
182
183 let output_order = petgraph::algo::toposort(&output_graph, None)
184 .map_err(|e| {
185 format!(
186 "includes node {}",
187 output_graph[e.node_id()].0.node.modpath()
188 )
189 })
190 .expect("runtime variables cannot introduce a DAG cycle");
191
192 for idx in output_order.into_iter().rev() {
193 let OutputGraphEntry { node_handle, step } = output_graph[idx].1.take().unwrap();
194
195 let mut steps = Vec::new();
196 let (label, code, idx) = match step {
197 Step::Anchor { .. } => continue,
198 Step::Rust { label, code, idx } => (label, code, idx),
199 Step::AdoYaml { .. } => {
200 anyhow::bail!("{} emitted ADO YAML. Fix the node by checking `ctx.backend()` appropriately", node_handle.modpath())
201 }
202 Step::GitHubYaml { .. } => {
203 anyhow::bail!("{} emitted GitHub YAML. Fix the node by checking `ctx.backend()` appropriately", node_handle.modpath())
204 }
205 };
206 steps.push((idx, label, code.lock().take().unwrap()));
207
208 resolved_local_nodes.push(ResolvedRunnableNode { node_handle, steps })
209 }
210
211 resolved_local_nodes
212 };
213
214 let mut in_mem_var_db = crate::var_db::in_memory::InMemoryVarDb::new();
215
216 for ResolvedJobUseParameter {
217 flowey_var,
218 pipeline_param_idx,
219 } in parameters_used
220 {
221 let (desc, value) = match &parameters[*pipeline_param_idx] {
222 Parameter::Bool {
223 description,
224 default,
225 } => (
226 description,
227 default.as_ref().map(|v| serde_json::to_vec(v).unwrap()),
228 ),
229 Parameter::String {
230 description,
231 default,
232 possible_values: _,
233 } => (
234 description,
235 default.as_ref().map(|v| serde_json::to_vec(v).unwrap()),
236 ),
237 Parameter::Num {
238 description,
239 default,
240 possible_values: _,
241 } => (
242 description,
243 default.as_ref().map(|v| serde_json::to_vec(v).unwrap()),
244 ),
245 };
246
247 let Some(value) = value else {
248 anyhow::bail!(
249 "pipeline must specify default value for params when running locally. missing default for '{desc}'"
250 )
251 };
252
253 in_mem_var_db.set_var(flowey_var, false, value);
254 }
255
256 in_mem_var_db.set_var(
257 VAR_DB_SEEDVAR_FLOWEY_PERSISTENT_STORAGE_DIR,
258 false,
259 serde_json::to_string(&persist_dir).unwrap().into(),
260 );
261
262 for ResolvedJobArtifact { flowey_var, name } in artifacts_published {
263 let path = out_dir.join("artifacts").join(name);
264 fs_err::create_dir_all(&path)?;
265
266 in_mem_var_db.set_var(
267 flowey_var,
268 false,
269 serde_json::to_string(&path).unwrap().into(),
270 );
271 }
272
273 if out_dir.join(".job_artifacts").exists() {
274 fs_err::remove_dir_all(out_dir.join(".job_artifacts"))?;
275 }
276 fs_err::create_dir_all(out_dir.join(".job_artifacts"))?;
277
278 for ResolvedJobArtifact { flowey_var, name } in artifacts_used {
279 let path = out_dir.join(".job_artifacts").join(name);
280 fs_err::create_dir_all(&path)?;
281 copy_dir_all(out_dir.join("artifacts").join(name), &path)?;
282
283 in_mem_var_db.set_var(
284 flowey_var,
285 false,
286 serde_json::to_string(&path).unwrap().into(),
287 );
288 }
289
290 if out_dir.join(".work").exists() {
291 fs_err::remove_dir_all(out_dir.join(".work"))?;
292 }
293 fs_err::create_dir_all(out_dir.join(".work"))?;
294
295 let mut runtime_services = flowey_core::node::steps::rust::new_rust_runtime_services(
296 &mut in_mem_var_db,
297 FlowBackend::Local,
298 platform,
299 flow_arch,
300 );
301
302 if let Some(cond_param_idx) = cond_param_idx {
303 let Parameter::Bool {
304 description: _,
305 default,
306 } = &parameters[cond_param_idx]
307 else {
308 panic!("cond param is guaranteed to be bool by type system")
309 };
310
311 let Some(should_run) = default else {
312 anyhow::bail!(
313 "when running locally, job condition parameter must include a default value"
314 )
315 };
316
317 if !should_run {
318 log::warn!("job condition was false - skipping job...");
319 continue;
320 }
321 }
322
323 for ResolvedRunnableNode { node_handle, steps } in nodes {
324 for (idx, label, code) in steps {
325 let node_working_dir = out_dir.join(".work").join(format!(
326 "{}_{}",
327 node_handle.modpath().replace("::", "__"),
328 idx
329 ));
330 if !node_working_dir.exists() {
331 fs_err::create_dir(&node_working_dir)?;
332 }
333 std::env::set_current_dir(node_working_dir)?;
334
335 log::info!(
336 // green color
337 "\x1B[0;32m=== {} ({}) ===\x1B[0m",
338 label,
339 node_handle.modpath(),
340 );
341 code(&mut runtime_services)?;
342 log::info!("\x1B[0;32m=== done! ===\x1B[0m");
343 log::info!(""); // log a newline, for the pretty
344 }
345 }
346 }
347
348 Ok(())
349}
350
351fn copy_dir_all(src: impl AsRef<Path>, dst: impl AsRef<Path>) -> std::io::Result<()> {
352 fs_err::create_dir_all(&dst)?;
353 for entry in fs_err::read_dir(src.as_ref())? {
354 let entry = entry?;
355 let ty = entry.file_type()?;
356 if ty.is_dir() {
357 copy_dir_all(entry.path(), dst.as_ref().join(entry.file_name()))?;
358 } else {
359 fs_err::copy(entry.path(), dst.as_ref().join(entry.file_name()))?;
360 }
361 }
362 Ok(())
363}
364