GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
copilot/remove-redundant-test-disks

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/generic.rs

325lines · modecode

RawDownload
Latest commit unavailable.
1// Copyright (c) Microsoft Corporation.
2// Licensed under the MIT License.
3
4use anyhow::Context;
5use flowey_core::node::FlowArch;
6use flowey_core::node::FlowPlatform;
7use flowey_core::node::NodeHandle;
8use flowey_core::node::user_facing::GhPermission;
9use flowey_core::node::user_facing::GhPermissionValue;
10use flowey_core::patch::ResolvedPatches;
11use flowey_core::pipeline::AdoCiTriggers;
12use flowey_core::pipeline::AdoPrTriggers;
13use flowey_core::pipeline::AdoScheduleTriggers;
14use flowey_core::pipeline::GhCiTriggers;
15use flowey_core::pipeline::GhPrTriggers;
16use flowey_core::pipeline::GhRunner;
17use flowey_core::pipeline::GhScheduleTriggers;
18use flowey_core::pipeline::Pipeline;
19use flowey_core::pipeline::internal::AdoPool;
20use flowey_core::pipeline::internal::ArtifactMeta;
21use flowey_core::pipeline::internal::InternalAdoResourcesRepository;
22use flowey_core::pipeline::internal::Parameter;
23use flowey_core::pipeline::internal::ParameterMeta;
24use flowey_core::pipeline::internal::PipelineFinalized;
25use flowey_core::pipeline::internal::PipelineJobMetadata;
26use std::collections::BTreeMap;
27use std::collections::BTreeSet;
28
29pub struct ResolvedPipeline {
30 pub graph: petgraph::Graph<ResolvedPipelineJob, ()>,
31 pub order: Vec<petgraph::prelude::NodeIndex>,
32 pub parameters: Vec<Parameter>,
33 pub ado_schedule_triggers: Vec<AdoScheduleTriggers>,
34 pub ado_name: Option<String>,
35 pub ado_ci_triggers: Option<AdoCiTriggers>,
36 pub ado_pr_triggers: Option<AdoPrTriggers>,
37 pub ado_bootstrap_template: String,
38 pub ado_resources_repository: Vec<InternalAdoResourcesRepository>,
39 pub ado_post_process_yaml_cb: Option<Box<dyn FnOnce(serde_yaml::Value) -> serde_yaml::Value>>,
40 pub ado_variables: BTreeMap<String, String>,
41 pub ado_job_id_overrides: BTreeMap<usize, String>,
42 pub gh_name: Option<String>,
43 pub gh_schedule_triggers: Vec<GhScheduleTriggers>,
44 pub gh_ci_triggers: Option<GhCiTriggers>,
45 pub gh_pr_triggers: Option<GhPrTriggers>,
46 pub gh_bootstrap_template: String,
47}
48
49#[derive(Debug, Clone)]
50pub struct ResolvedJobArtifact {
51 pub flowey_var: String,
52 pub name: String,
53}
54
55#[derive(Debug, Clone)]
56pub struct ResolvedJobUseParameter {
57 pub flowey_var: String,
58 pub pipeline_param_idx: usize,
59}
60
61#[derive(Debug, Clone)] // Clone is because of shoddy viz code
62pub struct ResolvedPipelineJob {
63 pub root_nodes: BTreeMap<NodeHandle, Vec<Box<[u8]>>>,
64 pub patches: ResolvedPatches,
65 pub label: String,
66 pub platform: FlowPlatform,
67 pub arch: FlowArch,
68 pub ado_pool: Option<AdoPool>,
69 pub timeout_minutes: Option<u32>,
70 pub command_wrapper: Option<flowey_core::shell::CommandWrapperKind>,
71 pub ado_variables: BTreeMap<String, String>,
72 pub gh_override_if: Option<String>,
73 pub gh_global_env: BTreeMap<String, String>,
74 pub gh_pool: Option<GhRunner>,
75 pub gh_permissions: BTreeMap<NodeHandle, BTreeMap<GhPermission, GhPermissionValue>>,
76 pub external_read_vars: BTreeSet<String>,
77 pub cond_param_idx: Option<usize>,
78
79 pub parameters_used: Vec<ResolvedJobUseParameter>,
80 // correspond to injected download nodes at the start of the job
81 pub artifacts_used: Vec<ResolvedJobArtifact>,
82 // correspond to injected publish nodes at the end of the job
83 pub artifacts_published: Vec<ResolvedJobArtifact>,
84}
85
86pub fn resolve_pipeline(pipeline: Pipeline) -> anyhow::Result<ResolvedPipeline> {
87 let PipelineFinalized {
88 jobs,
89 artifacts,
90 parameters,
91 extra_deps,
92 ado_name,
93 ado_schedule_triggers,
94 ado_ci_triggers,
95 ado_pr_triggers,
96 ado_bootstrap_template,
97 ado_resources_repository,
98 ado_post_process_yaml_cb,
99 ado_variables,
100 ado_job_id_overrides,
101 gh_name,
102 gh_schedule_triggers,
103 gh_ci_triggers,
104 gh_pr_triggers,
105 gh_bootstrap_template,
106 } = PipelineFinalized::from_pipeline(pipeline);
107
108 let mut graph = petgraph::Graph::new();
109
110 let mut job_to_artifacts = {
111 let mut m = BTreeMap::<usize, (BTreeSet<String>, BTreeSet<String>)>::new();
112
113 for ArtifactMeta {
114 name,
115 published_by_job,
116 used_by_jobs,
117 } in &artifacts
118 {
119 let no_existing = m
120 .entry(
121 published_by_job
122 .context(format!("artifact '{name}' is not published by any job"))?,
123 )
124 .or_default()
125 .0
126 .insert(name.clone());
127 assert!(no_existing);
128
129 for job_idx in used_by_jobs {
130 let no_existing = m.entry(*job_idx).or_default().1.insert(name.clone());
131 assert!(no_existing);
132 }
133 }
134
135 m
136 };
137
138 let (parameters, mut job_to_params) = {
139 let mut params = Vec::new();
140 let mut m = BTreeMap::<usize, BTreeSet<usize>>::new();
141
142 for (
143 param_idx,
144 ParameterMeta {
145 parameter,
146 used_by_jobs,
147 },
148 ) in parameters.into_iter().enumerate()
149 {
150 params.push(parameter);
151 for job_idx in used_by_jobs {
152 let no_existing = m.entry(job_idx).or_default().insert(param_idx);
153 assert!(no_existing);
154 }
155 }
156
157 (params, m)
158 };
159
160 let mut flowey_bootstrap_platforms = BTreeSet::new();
161
162 // first things first: spin up graph nodes for each job
163 let mut job_graph_idx = Vec::new();
164 for (
165 job_idx,
166 PipelineJobMetadata {
167 root_nodes,
168 patches,
169 label,
170 platform,
171 arch,
172 cond_param_idx,
173 timeout_minutes,
174 command_wrapper,
175 ado_pool,
176 ado_variables,
177 gh_override_if,
178 gh_global_env,
179 gh_pool,
180 gh_permissions,
181 },
182 ) in jobs.into_iter().enumerate()
183 {
184 let (artifacts_published, artifacts_used) =
185 job_to_artifacts.remove(&job_idx).unwrap_or_default();
186 let parameters_used = job_to_params.remove(&job_idx).unwrap_or_default();
187
188 let artifacts_published: Vec<_> = artifacts_published
189 .into_iter()
190 .map(|a| ResolvedJobArtifact {
191 flowey_var: flowey_core::pipeline::internal::consistent_artifact_runtime_var_name(
192 &a, false,
193 ),
194 name: a,
195 })
196 .collect();
197 let artifacts_used: Vec<_> = artifacts_used
198 .into_iter()
199 .map(|a| ResolvedJobArtifact {
200 flowey_var: flowey_core::pipeline::internal::consistent_artifact_runtime_var_name(
201 &a, true,
202 ),
203 name: a,
204 })
205 .collect();
206 let parameters_used: Vec<_> = parameters_used
207 .into_iter()
208 .map(|param_idx| ResolvedJobUseParameter {
209 flowey_var: parameters[param_idx].name().to_string(),
210 pipeline_param_idx: param_idx,
211 })
212 .collect();
213
214 // individual pipeline resolvers still need to ensure that the var is in
215 // the var-db at job start time, but this external-var reporting code
216 // can be shared across all impls
217 let mut external_read_vars = BTreeSet::new();
218 external_read_vars.extend(artifacts_used.iter().map(|a| a.flowey_var.clone()));
219 external_read_vars.extend(artifacts_published.iter().map(|a| a.flowey_var.clone()));
220 external_read_vars.extend(parameters_used.iter().map(|p| p.flowey_var.clone()));
221
222 let idx = graph.add_node(ResolvedPipelineJob {
223 root_nodes,
224 patches: patches.finalize(),
225 label,
226 timeout_minutes,
227 command_wrapper,
228 ado_pool,
229 ado_variables,
230 gh_override_if,
231 gh_global_env,
232 gh_pool,
233 gh_permissions,
234 platform,
235 arch,
236 cond_param_idx,
237 external_read_vars,
238 parameters_used,
239 artifacts_used,
240 artifacts_published,
241 });
242
243 // ...also using this opportunity to keep track of what flowey bins we need to bootstrap
244 flowey_bootstrap_platforms.insert(platform);
245
246 job_graph_idx.push(idx);
247 }
248
249 // next, add node edges based on artifact flow
250 for ArtifactMeta {
251 name: _,
252 published_by_job,
253 used_by_jobs,
254 } in artifacts
255 {
256 let published_idx = job_graph_idx[published_by_job.expect("checked in loop above")];
257 for job in used_by_jobs {
258 let used_idx = job_graph_idx[job];
259 graph.add_edge(published_idx, used_idx, ());
260 }
261 }
262
263 // lastly, add node edges based on any additional explicit dependencies
264 for (from, to) in extra_deps {
265 graph.add_edge(job_graph_idx[from], job_graph_idx[to], ());
266 }
267
268 // TODO: better error handling
269 let order = petgraph::algo::toposort(&graph, None)
270 .map_err(|_| anyhow::anyhow!("detected cycle in pipeline"))?;
271
272 Ok(ResolvedPipeline {
273 graph,
274 order,
275 parameters,
276 ado_name,
277 ado_variables,
278 ado_schedule_triggers,
279 ado_ci_triggers,
280 ado_pr_triggers,
281 ado_bootstrap_template,
282 ado_resources_repository,
283 ado_post_process_yaml_cb,
284 ado_job_id_overrides,
285 gh_name,
286 gh_schedule_triggers,
287 gh_ci_triggers,
288 gh_pr_triggers,
289 gh_bootstrap_template,
290 })
291}
292
293impl ResolvedPipeline {
294 /// Trim the pipeline graph to only include the specified jobs (taking care
295 /// to also preserve any dependant jobs they rely on).
296 pub fn trim_pipeline_graph(&mut self, preserve_jobs: Vec<petgraph::prelude::NodeIndex>) {
297 // DEVNOTE: this is a horribly suboptimal way to implement this, but it
298 // works fine with the graph-sizes we currently have, so we can optimize
299 // this later...
300
301 let mut jobs_to_delete: BTreeSet<_> = self.graph.node_indices().collect();
302 for idx in preserve_jobs {
303 let g = petgraph::visit::Reversed(&self.graph);
304
305 let mut dfs = petgraph::visit::Dfs::new(g, idx);
306 while let Some(save_idx) = dfs.next(g) {
307 jobs_to_delete.remove(&save_idx);
308 }
309 }
310
311 let mut jobs_to_delete = jobs_to_delete.into_iter().collect::<Vec<_>>();
312 jobs_to_delete.sort();
313
314 // in petgraph, when you remove a node, it invalidates the node idx of
315 // all subsequent nodes.
316 //
317 // I'm sure there's a better way to do this filtering, but just removing
318 // nodes in reverse order seems to work fine.
319 for idx in jobs_to_delete.into_iter().rev() {
320 self.graph.remove_node(idx).unwrap();
321 }
322
323 self.order = petgraph::algo::toposort(&self.graph, None).unwrap();
324 }
325}
326