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/fix-code-for-review-comment

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

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