Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(ci-protocol): add the actions workflow protocol and trigger events

ZhenYi committed
parent 561835a commit 7d6321e Browse files

6 files changed

+314 −129

File tree 6
Added lib/ci-protocol/src/actions.rs View lib/ci-protocol/src/actions.rs
@@ -0,0 +1,236 @@
1use std::collections::BTreeMap;
2
3use serde::{Deserialize, Serialize};
4use serde_json::{Value, json};
5
6use crate::dsl::{Concurrency, Job, MarkedPermissions, Step, WorkflowDefinition, WorkflowTriggers};
7
8pub const ACTIONS_WORKFLOW_DIR: &str = ".github/workflows";
9pub const SNAPSHOT_KEY: &str = "_gitdata_actions";
10
11#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
12pub struct ActionsWorkflow {
13 pub file: String,
14 pub source: String,
15 pub document: Value,
16}
17
18#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
19pub struct InputDefinition {
20 #[serde(default)]
21 pub description: Option<String>,
22 #[serde(default)]
23 pub required: bool,
24 #[serde(default, rename = "type")]
25 pub kind: Option<String>,
26 #[serde(default)]
27 pub default: Option<Value>,
28 #[serde(default)]
29 pub options: Vec<String>,
30}
31
32#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
33pub struct NeedResult {
34 pub result: String,
35 #[serde(default)]
36 pub outputs: BTreeMap<String, String>,
37}
38
39#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
40pub struct EngineInstance {
41 pub index: usize,
42 pub matrix: Value,
43 pub name: String,
44 pub runs_on: Vec<String>,
45 pub environment: String,
46 pub timeout_seconds: u64,
47 pub continue_on_error: bool,
48 pub fail_fast: bool,
49 pub max_parallel: usize,
50 pub concurrency: Value,
51 pub permissions: Value,
52 #[serde(rename = "with")]
53 pub inputs: Value,
54 pub uses: String,
55 pub group: Option<String>,
56}
57
58#[derive(Debug, Clone, Default, Serialize, Deserialize)]
59pub struct EngineExpansion {
60 pub skipped: bool,
61 pub instances: Vec<EngineInstance>,
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
65pub struct ActionsExecution {
66 pub workflow: ActionsWorkflow,
67 pub job: String,
68 pub matrix: Value,
69 pub github: Value,
70 pub event: Value,
71 pub needs: BTreeMap<String, NeedResult>,
72 pub inputs: BTreeMap<String, Value>,
73 pub vars: BTreeMap<String, String>,
74 pub token: String,
75 pub runtime_url: String,
76 pub runtime_token: String,
77}
78
79#[derive(Debug, Clone, Default, Serialize, Deserialize, utoipa::ToSchema)]
80pub struct EngineEvent {
81 pub event: String,
82 #[serde(default)]
83 pub data: Value,
84 #[serde(default)]
85 pub error: Option<String>,
86 #[serde(default)]
87 pub lease_token: Option<String>,
88}
89
90impl ActionsWorkflow {
91 pub fn name(&self) -> Option<String> {
92 self.document
93 .get("name")
94 .and_then(Value::as_str)
95 .map(str::to_owned)
96 }
97
98 pub fn triggers(&self) -> &Value {
99 self.document.get("on").unwrap_or(&Value::Null)
100 }
101
102 pub fn inputs(&self) -> BTreeMap<String, InputDefinition> {
103 self.triggers()
104 .get("workflow_dispatch")
105 .and_then(|v| v.get("inputs"))
106 .and_then(|v| serde_json::from_value(v.clone()).ok())
107 .unwrap_or_default()
108 }
109
110 pub fn projection(&self) -> WorkflowDefinition {
111 let on = self.triggers();
112 let triggers = WorkflowTriggers {
113 push: on.get("push").map(|value| crate::dsl::PushTrigger {
114 branches: strings(value.get("branches")),
115 tags: strings(value.get("tags")),
116 }),
117 pull_request: on
118 .get("pull_request")
119 .map(|value| crate::dsl::PullRequestTrigger {
120 branches: strings(value.get("branches")),
121 types: strings(value.get("types"))
122 .into_iter()
123 .filter_map(|s| serde_json::from_value(json!(s)).ok())
124 .collect(),
125 }),
126 workflow_dispatch: on
127 .get("workflow_dispatch")
128 .map(|_| crate::dsl::WorkflowDispatchTrigger::default()),
129 };
130 let jobs = self
131 .document
132 .get("jobs")
133 .and_then(Value::as_object)
134 .into_iter()
135 .flatten()
136 .map(|(key, value)| {
137 (
138 key.clone(),
139 Job {
140 runs_on: strings(value.get("runs-on")),
141 condition: None,
142 needs: strings(value.get("needs")),
143 environment: value
144 .get("environment")
145 .and_then(|v| {
146 v.as_str().or_else(|| v.get("name").and_then(Value::as_str))
147 })
148 .map(str::to_owned),
149 permissions: MarkedPermissions::default(),
150 container: None,
151 timeout_minutes: value.get("timeout-minutes").and_then(Value::as_u64),
152 env: env_map(value.get("env")),
153 artifacts: Vec::new(),
154 steps: value
155 .get("steps")
156 .and_then(Value::as_array)
157 .into_iter()
158 .flatten()
159 .map(|step| Step {
160 name: step.get("name").and_then(Value::as_str).map(str::to_owned),
161 run: step
162 .get("run")
163 .or_else(|| step.get("uses"))
164 .and_then(Value::as_str)
165 .unwrap_or_default()
166 .to_owned(),
167 shell: step.get("shell").and_then(Value::as_str).map(str::to_owned),
168 working_directory: step
169 .get("working-directory")
170 .and_then(Value::as_str)
171 .map(str::to_owned),
172 env: env_map(step.get("env")),
173 timeout_minutes: step
174 .get("timeout-minutes")
175 .and_then(Value::as_u64),
176 })
177 .collect(),
178 },
179 )
180 })
181 .collect();
182 WorkflowDefinition {
183 name: self.name(),
184 on: triggers,
185 env: env_map(self.document.get("env")),
186 concurrency: self
187 .document
188 .get("concurrency")
189 .and_then(|value| {
190 value
191 .as_str()
192 .or_else(|| value.get("group").and_then(Value::as_str))
193 })
194 .map(|group| Concurrency {
195 group: group.to_owned(),
196 cancel_in_progress: self.document["concurrency"]["cancel-in-progress"]
197 .as_bool()
198 .unwrap_or(false),
199 }),
200 jobs,
201 }
202 }
203}
204
205pub fn strings(value: Option<&Value>) -> Vec<String> {
206 match value {
207 Some(Value::String(value)) => vec![value.clone()],
208 Some(Value::Array(values)) => values
209 .iter()
210 .filter_map(Value::as_str)
211 .map(str::to_owned)
212 .collect(),
213 _ => Vec::new(),
214 }
215}
216
217pub fn env_map(value: Option<&Value>) -> BTreeMap<String, String> {
218 value
219 .and_then(Value::as_object)
220 .into_iter()
221 .flatten()
222 .map(|(name, value)| {
223 (
224 name.clone(),
225 value
226 .as_str()
227 .map(str::to_owned)
228 .unwrap_or_else(|| value.to_string()),
229 )
230 })
231 .collect()
232}
233
234pub fn from_snapshot(value: &Value) -> Option<ActionsWorkflow> {
235 serde_json::from_value(value.get(SNAPSHOT_KEY)?.clone()).ok()
236}
Modified lib/ci-protocol/src/dsl.rs View lib/ci-protocol/src/dsl.rs
@@ -3,10 +3,7 @@
33use globset::{Glob, GlobSet, GlobSetBuilder};
44use serde::{Deserialize, Serialize};
55
6pub const WORKFLOW_DIR: &str = ".workflows";
7pub const MAX_JOBS: usize = 50;
8pub const MAX_STEPS: usize = 50;
9pub const DEFAULT_JOB_TIMEOUT_MINUTES: u64 = 60;
6pub const WORKFLOW_DIR: &str = ".github/workflows";
107
118#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
129#[serde(deny_unknown_fields)]
@@ -173,130 +170,6 @@
173170 pub timeout_minutes: Option<u64>,
174171}
175172
176#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
177#[error("{0}")]
178pub struct WorkflowParseError(pub String);
179
180pub fn parse_workflow_str(source: &str) -> Result<WorkflowDefinition, WorkflowParseError> {
181 let workflow: WorkflowDefinition =
182 serde_yaml_ng::from_str(source).map_err(|e| WorkflowParseError(e.to_string()))?;
183 validate_workflow(&workflow)?;
184 Ok(workflow)
185}
186
187pub fn validate_workflow(workflow: &WorkflowDefinition) -> Result<(), WorkflowParseError> {
188 if workflow.on.is_empty() {
189 return Err(WorkflowParseError(
190 "`on` must declare at least one trigger".to_string(),
191 ));
192 }
193 if workflow.jobs.is_empty() {
194 return Err(WorkflowParseError(
195 "`jobs` must declare at least one job".to_string(),
196 ));
197 }
198 if workflow.jobs.len() > MAX_JOBS {
199 return Err(WorkflowParseError(format!(
200 "too many jobs: {} (max {MAX_JOBS})",
201 workflow.jobs.len()
202 )));
203 }
204 for (name, job) in &workflow.jobs {
205 if name.trim().is_empty() || name.chars().count() > 128 {
206 return Err(WorkflowParseError(
207 "job name must be 1-128 characters".to_string(),
208 ));
209 }
210 if job.steps.is_empty() {
211 return Err(WorkflowParseError(format!(
212 "job `{name}` must declare at least one step"
213 )));
214 }
215 if job.steps.len() > MAX_STEPS {
216 return Err(WorkflowParseError(format!(
217 "job `{name}` has too many steps: {} (max {MAX_STEPS})",
218 job.steps.len()
219 )));
220 }
221 if job.runs_on.is_empty() || job.runs_on.iter().any(|l| l.trim().is_empty()) {
222 return Err(WorkflowParseError(format!(
223 "job `{name}` runs-on label must not be empty"
224 )));
225 }
226 if let Some(container) = &job.container
227 && container.trim().is_empty()
228 {
229 return Err(WorkflowParseError(format!(
230 "job `{name}` container must not be empty when set"
231 )));
232 }
233 for need in &job.needs {
234 if !workflow.jobs.contains_key(need) {
235 return Err(WorkflowParseError(format!(
236 "job `{name}` needs unknown job `{need}`"
237 )));
238 }
239 if need == name {
240 return Err(WorkflowParseError(format!(
241 "job `{name}` must not depend on itself"
242 )));
243 }
244 }
245 for step in &job.steps {
246 if step.run.trim().is_empty() {
247 return Err(WorkflowParseError(format!(
248 "job `{name}` has an empty run step"
249 )));
250 }
251 }
252 }
253 ensure_acyclic(workflow)?;
254 Ok(())
255}
256
257#[derive(Clone, Copy, PartialEq, Eq)]
258enum VisitMark {
259 Visiting,
260 Done,
261}
262
263fn visit_job<'a>(
264 nodes: &'a BTreeMap<String, Job>,
265 name: &'a str,
266 marks: &mut BTreeMap<&'a str, VisitMark>,
267) -> Result<(), WorkflowParseError> {
268 match marks.get(name) {
269 Some(VisitMark::Done) => return Ok(()),
270 Some(VisitMark::Visiting) => {
271 return Err(WorkflowParseError(format!(
272 "dependency cycle detected at job `{name}`"
273 )));
274 }
275 None => {}
276 }
277 marks.insert(name, VisitMark::Visiting);
278 let job = nodes
279 .get(name)
280 .ok_or_else(|| WorkflowParseError(format!("unknown job `{name}`")))?;
281 for need in &job.needs {
282 let key = nodes
283 .get_key_value(need.as_str())
284 .map(|(key, _)| key.as_str())
285 .ok_or_else(|| WorkflowParseError(format!("unknown job `{need}`")))?;
286 visit_job(nodes, key, marks)?;
287 }
288 marks.insert(name, VisitMark::Done);
289 Ok(())
290}
291
292fn ensure_acyclic(workflow: &WorkflowDefinition) -> Result<(), WorkflowParseError> {
293 let mut marks: BTreeMap<&str, VisitMark> = BTreeMap::new();
294 for name in workflow.jobs.keys() {
295 visit_job(&workflow.jobs, name.as_str(), &mut marks)?;
296 }
297 Ok(())
298}
299
300173#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
301174pub struct WorkflowSummary {
302175 pub file: String,
@@ -304,6 +177,10 @@
304177 pub valid: bool,
305178 pub errors: Vec<String>,
306179 pub triggers: TriggerSummary,
180 #[serde(default)]
181 pub inputs: BTreeMap<String, crate::actions::InputDefinition>,
182 #[serde(default)]
183 pub engine: String,
307184}
308185
309186#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
Modified lib/ci-protocol/src/events.rs View lib/ci-protocol/src/events.rs
@@ -62,6 +62,7 @@
6262pub enum CiEvent {
6363 Push(PushEvent),
6464 PullRequest(PullRequestEvent),
65 Trigger(TriggerEvent),
6566}
6667
6768#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -85,4 +86,16 @@
8586 pub head_ref: String,
8687 pub base_sha: String,
8788 pub base_ref: String,
89 #[serde(default)]
90 pub head_repo_id: Option<Uuid>,
91}
92
93#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
94pub struct TriggerEvent {
95 pub id: Uuid,
96 pub repo_id: Uuid,
97 pub space_id: Uuid,
98 pub actor_id: Option<Uuid>,
99 pub event: String,
100 pub payload: serde_json::Value,
88101}
Modified lib/ci-protocol/src/lib.rs View lib/ci-protocol/src/lib.rs
@@ -1,8 +1,9 @@
1pub mod actions;
12pub mod ai;
23pub mod ai_languages;
34pub mod dsl;
45pub mod events;
56pub mod runner;
67
7pub const PROTOCOL_VERSION: u32 = 1;
8pub const PROTOCOL_VERSION: u32 = 2;
89pub const LOG_CHUNK_MAX_BYTES: usize = 262_144;
Modified lib/ci-protocol/src/runner.rs View lib/ci-protocol/src/runner.rs
@@ -14,6 +14,10 @@
1414 pub capacity: i32,
1515 #[serde(default)]
1616 pub version: Option<String>,
17 #[serde(default)]
18 pub protocol_versions: Vec<u32>,
19 #[serde(default)]
20 pub capabilities: Vec<String>,
1721}
1822
1923fn default_capacity() -> i32 {
@@ -33,6 +37,10 @@
3337pub struct HeartbeatRequest {
3438 #[serde(default)]
3539 pub running_job_ids: Vec<Uuid>,
40 #[serde(default)]
41 pub protocol_versions: Vec<u32>,
42 #[serde(default)]
43 pub capabilities: Vec<String>,
3644}
3745
3846#[derive(Debug, Clone, Default, Serialize, Deserialize, utoipa::ToSchema)]
@@ -127,6 +135,8 @@
127135 pub artifacts: Vec<ArtifactSpecDto>,
128136 pub needs_artifacts: Vec<NeedsArtifactDto>,
129137 pub lease: LeaseDto,
138 #[serde(default, skip_serializing_if = "Option::is_none")]
139 pub actions: Option<crate::actions::ActionsExecution>,
130140}
131141
132142#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
@@ -155,6 +165,16 @@
155165 pub started_at: Option<OffsetDateTime>,
156166 #[serde(default)]
157167 pub finished_at: Option<OffsetDateTime>,
168 #[serde(default)]
169 pub step_key: Option<String>,
170 #[serde(default)]
171 pub stage: Option<String>,
172 #[serde(default)]
173 pub outcome: Option<String>,
174 #[serde(default)]
175 pub conclusion: Option<String>,
176 #[serde(default)]
177 pub outputs: BTreeMap<String, String>,
158178}
159179
160180#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
@@ -184,6 +204,43 @@
184204 pub exit_code: Option<i32>,
185205 #[serde(default)]
186206 pub error: Option<String>,
207 #[serde(default)]
208 pub outputs: BTreeMap<String, String>,
209 #[serde(default)]
210 pub lease_token: Option<String>,
211}
212
213impl Default for JobFinishRequest {
214 fn default() -> Self {
215 Self {
216 state: JobFinishState::Failure,
217 exit_code: None,
218 error: None,
219 outputs: Default::default(),
220 lease_token: None,
221 }
222 }
223}
224
225impl Default for StepReport {
226 fn default() -> Self {
227 Self {
228 state: StepReportState::Started,
229 exit_code: None,
230 name: None,
231 shell: None,
232 working_directory: None,
233 script: None,
234 timeout_secs: None,
235 started_at: None,
236 finished_at: None,
237 step_key: None,
238 stage: None,
239 outcome: None,
240 conclusion: None,
241 outputs: Default::default(),
242 }
243 }
187244}
188245
189246#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
Modified lib/pulls/src/services/events.rs View lib/pulls/src/services/events.rs
@@ -141,6 +141,7 @@
141141 head_ref: pull.head_branch.clone(),
142142 base_sha: pull.base_sha.clone(),
143143 base_ref: pull.base_branch.clone(),
144 head_repo_id: pull.head_repo_id,
144145 });
145146 let payload = match serde_json::to_vec(&event) {
146147 Ok(payload) => payload,