GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

microsoft/openvmm

Public

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

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
e8acaefba35969bbca0ffae96452cd05b47aa621

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

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