Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

feat(runner): execute actions jobs with the vendored engine

ZhenYi committed
parent c4a2956 commit 9054d68 Browse files

9 files changed

+388 −30

File tree 9
Added app/runner/src/actions.rs View app/runner/src/actions.rs
@@ -0,0 +1,224 @@
1use ci_protocol::actions::EngineEvent;
2use ci_protocol::runner::{
3 JobFinishRequest, JobFinishState, JobPayload, LogChunk, LogChunksRequest,
4};
5use config::runner::RunnerConfig;
6use serde_json::json;
7use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
8use tokio::process::Command;
9
10use crate::client::RunnerClient;
11
12pub fn available(config: &RunnerConfig) -> bool {
13 use std::io::Write;
14 let Ok(mut child) = std::process::Command::new(&config.actions_engine_binary)
15 .stdin(std::process::Stdio::piped())
16 .stdout(std::process::Stdio::null())
17 .stderr(std::process::Stdio::null())
18 .spawn()
19 else {
20 return false;
21 };
22 if let Some(mut input) = child.stdin.take() {
23 let _ = input.write_all(b"{\"op\":\"version\"}\n");
24 }
25 child.wait().is_ok_and(|status| status.success())
26}
27
28pub async fn execute(
29 client: &RunnerClient,
30 config: &RunnerConfig,
31 payload: &JobPayload,
32) -> JobFinishRequest {
33 let mut finish = match execute_inner(client, config, payload).await {
34 Ok(finish) => finish,
35 Err(error) => JobFinishRequest {
36 state: JobFinishState::Failure,
37 exit_code: Some(1),
38 error: Some(error.to_string()),
39 ..Default::default()
40 },
41 };
42 finish.lease_token = Some(payload.lease.token.clone());
43 finish
44}
45
46async fn execute_inner(
47 client: &RunnerClient,
48 config: &RunnerConfig,
49 payload: &JobPayload,
50) -> anyhow::Result<JobFinishRequest> {
51 let actions = payload
52 .actions
53 .as_ref()
54 .ok_or_else(|| anyhow::anyhow!("Actions execution payload missing"))?;
55 client.start(&payload.job_id).await?;
56 let workspace = std::path::Path::new(&config.workspace_root)
57 .join(payload.run_id.to_string())
58 .join(payload.job_id.to_string());
59 let repo = workspace.join("repo");
60 tokio::fs::create_dir_all(&repo).await?;
61 let request = json!({
62 "op":"execute", "source":actions.workflow.source, "file":actions.workflow.file,
63 "job":actions.job, "matrix":actions.matrix, "github":actions.github, "event":actions.event,
64 "needs":actions.needs, "inputs":actions.inputs, "vars":actions.vars, "secrets":payload.secrets,
65 "env":payload.env, "workdir":repo, "platforms":config.actions_platforms,
66 "node_runtimes":config.actions_node_runtimes, "action_cache_dir":config.actions_cache_dir,
67 "action_server":config.actions_action_server, "action_token":config.actions_action_token,
68 "token":actions.token, "runtime_url":actions.runtime_url, "runtime_token":actions.runtime_token,
69 "timeout_seconds":payload.timeout_secs,
70 });
71 let mut command = Command::new(&config.actions_engine_binary);
72 command
73 .stdin(std::process::Stdio::piped())
74 .stdout(std::process::Stdio::piped())
75 .stderr(std::process::Stdio::piped())
76 .kill_on_drop(true);
77 #[cfg(unix)]
78 command.process_group(0);
79 let mut child = command.spawn()?;
80 let mut input = child
81 .stdin
82 .take()
83 .ok_or_else(|| anyhow::anyhow!("engine stdin missing"))?;
84 input.write_all(&serde_json::to_vec(&request)?).await?;
85 input.write_all(b"\n").await?;
86 drop(input);
87 let stderr = child
88 .stderr
89 .take()
90 .ok_or_else(|| anyhow::anyhow!("engine stderr missing"))?;
91 let stderr_task = tokio::spawn(async move {
92 let mut reader = BufReader::new(stderr).lines();
93 let mut tail = String::new();
94 while let Ok(Some(line)) = reader.next_line().await {
95 if tail.len() < 65536 {
96 tail.push_str(&line);
97 tail.push('\n');
98 }
99 }
100 tail
101 });
102 let renew_client = client.clone();
103 let job_id = payload.job_id;
104 let interval = payload.lease.heartbeat_interval_secs.max(1);
105 let renew = tokio::spawn(async move {
106 let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval));
107 loop {
108 ticker.tick().await;
109 if renew_client.renew(&job_id).await.is_err() {
110 crate::executor::request_cancel(&[job_id]);
111 }
112 }
113 });
114 let mut stdout = BufReader::new(
115 child
116 .stdout
117 .take()
118 .ok_or_else(|| anyhow::anyhow!("engine stdout missing"))?,
119 )
120 .lines();
121 let mut ticker = tokio::time::interval(std::time::Duration::from_millis(250));
122 let mut seq = 0;
123 let mut logs = String::new();
124 let mut result = None;
125 let mut cancelled_at = None;
126 let deadline = tokio::time::Instant::now()
127 + std::time::Duration::from_secs(payload.timeout_secs.saturating_add(90));
128 let processing: anyhow::Result<()> = async {
129 loop {
130 tokio::select! {
131 line = stdout.next_line() => {
132 let Some(line)=line? else {break};
133 let mut event:EngineEvent=serde_json::from_str(&line)?;
134 event.lease_token=Some(payload.lease.token.clone());
135 match event.event.as_str() {
136 "log" => { logs.push_str(event.data["text"].as_str().unwrap_or_default()); if logs.len() >= config.log_flush_bytes { flush(client,payload,&mut logs,&mut seq).await?; } }
137 "result" => { if let Some(error)=event.error { anyhow::bail!("{error}"); } result=Some(event.data); }
138 _ => { flush(client,payload,&mut logs,&mut seq).await?; client.event(payload.job_id,&event).await?; }
139 }
140 }
141 _ = ticker.tick() => {
142 flush(client,payload,&mut logs,&mut seq).await?;
143 if crate::executor::is_cancelled(&payload.job_id) && cancelled_at.is_none() {
144 cancelled_at=Some(tokio::time::Instant::now());
145 #[cfg(unix)]
146 if let Some(pid)=child.id() { unsafe extern "C" {fn kill(pid:i32,signal:i32)->i32;} unsafe {kill(pid as i32,15);} }
147 #[cfg(not(unix))]
148 child.kill().await?;
149 }
150 if tokio::time::Instant::now()>deadline || cancelled_at.is_some_and(|at| at.elapsed().as_secs()>90) { child.kill().await?; anyhow::bail!("Actions engine cleanup deadline exceeded"); }
151 }
152 }
153 }
154 flush(client,payload,&mut logs,&mut seq).await?;
155 Ok(())
156 }.await;
157 renew.abort();
158 if processing.is_err() {
159 let _ = child.kill().await;
160 }
161 let status = child.wait().await?;
162 let stderr = stderr_task.await.unwrap_or_default();
163 if !config.keep_workspace {
164 let _ = tokio::fs::remove_dir_all(&workspace).await;
165 }
166 processing?;
167 if !status.success() {
168 anyhow::bail!("Actions engine exited {status}: {stderr}");
169 }
170 let result =
171 result.ok_or_else(|| anyhow::anyhow!("Actions engine returned no job result: {stderr}"))?;
172 let state = match result["result"].as_str().unwrap_or("failure") {
173 "success" | "skipped" => JobFinishState::Success,
174 "cancelled" => JobFinishState::Cancelled,
175 "timed_out" => JobFinishState::TimedOut,
176 _ => JobFinishState::Failure,
177 };
178 Ok(JobFinishRequest {
179 state,
180 exit_code: Some(if state == JobFinishState::Success {
181 0
182 } else {
183 1
184 }),
185 error: result["error"]
186 .as_str()
187 .filter(|v| !v.is_empty())
188 .map(str::to_owned),
189 outputs: serde_json::from_value(result["outputs"].clone()).unwrap_or_default(),
190 lease_token: Some(payload.lease.token.clone()),
191 })
192}
193
194async fn flush(
195 client: &RunnerClient,
196 payload: &JobPayload,
197 logs: &mut String,
198 seq: &mut i32,
199) -> anyhow::Result<()> {
200 if logs.is_empty() {
201 return Ok(());
202 }
203 let mut text = std::mem::take(logs);
204 for secret in payload.secrets.values().chain(
205 payload
206 .actions
207 .iter()
208 .flat_map(|actions| [&actions.token, &actions.runtime_token]),
209 ) {
210 if !secret.is_empty() {
211 text = text.replace(secret, "***");
212 }
213 }
214 client
215 .upload_logs(
216 payload.job_id,
217 &LogChunksRequest {
218 chunks: vec![LogChunk { seq: *seq, text }],
219 },
220 )
221 .await?;
222 *seq += 1;
223 Ok(())
224}
Modified app/runner/src/client.rs View app/runner/src/client.rs
@@ -107,6 +107,14 @@
107107 self.post(&format!("/jobs/{job_id}/finish"), request).await
108108 }
109109
110 pub async fn event(
111 &self,
112 job_id: Uuid,
113 event: &ci_protocol::actions::EngineEvent,
114 ) -> anyhow::Result<Uuid> {
115 self.post(&format!("/jobs/{job_id}/events"), event).await
116 }
117
110118 pub async fn init_artifact(
111119 &self,
112120 job_id: Uuid,
Modified app/runner/src/executor.rs View app/runner/src/executor.rs
@@ -57,7 +57,7 @@
5757 }
5858}
5959
60fn is_cancelled(job_id: &uuid::Uuid) -> bool {
60pub(crate) fn is_cancelled(job_id: &uuid::Uuid) -> bool {
6161 cancelled_set()
6262 .lock()
6363 .map(|set| set.contains_key(job_id))
@@ -355,7 +355,6 @@
355355 .stdin(Stdio::null());
356356 #[cfg(unix)]
357357 {
358 use std::os::unix::process::CommandExt;
359358 command.process_group(0);
360359 }
361360
Modified app/runner/src/lib.rs View app/runner/src/lib.rs
@@ -1,3 +1,4 @@
1pub mod actions;
12pub mod client;
23pub mod config;
34pub mod executor;
Modified app/runner/src/main.rs View app/runner/src/main.rs
@@ -13,7 +13,12 @@
1313async fn main() -> anyhow::Result<()> {
1414 init_tracing();
1515
16 let cfg = config::runner::RunnerConfig::load().context("loading runner configuration")?;
16 let mut cfg = config::runner::RunnerConfig::load().context("loading runner configuration")?;
17 for (label, platform) in &cfg.actions_platforms {
18 if (cfg.docker_enabled || platform == "-self-hosted") && !cfg.labels.contains(label) {
19 cfg.labels.push(label.clone());
20 }
21 }
1722 if cfg.registration_token.trim().is_empty() && !credential_path(&cfg).exists() {
1823 bail!("runner needs APP_RUNNER__REGISTRATION_TOKEN or an existing credential file");
1924 }
@@ -22,6 +27,25 @@
2227 .with_context(|| format!("creating workspace root {}", cfg.workspace_root))?;
2328
2429 let client = RunnerClient::new(&cfg)?;
30 let actions_available = runner::actions::available(&cfg);
31 let protocol_versions = if actions_available {
32 vec![2, 1]
33 } else {
34 vec![1]
35 };
36 let mut capabilities: Vec<String> = if actions_available {
37 vec!["actions-v2".into()]
38 } else {
39 Vec::new()
40 };
41 if cfg.docker_enabled {
42 capabilities.push("docker".into());
43 }
44 capabilities.extend(
45 cfg.actions_groups
46 .iter()
47 .map(|group| format!("runner-group:{group}")),
48 );
2549 let credential = match load_credential(&cfg)? {
2650 Some(credential) => credential,
2751 None => {
@@ -33,10 +57,12 @@
3357 labels: cfg.labels.clone(),
3458 capacity: cfg.capacity,
3559 version: Some(env!("CARGO_PKG_VERSION").to_string()),
60 protocol_versions: protocol_versions.clone(),
61 capabilities: capabilities.clone(),
3662 })
3763 .await
3864 .context("registering runner")?;
39 if response.protocol_version != ci_protocol::PROTOCOL_VERSION {
65 if !protocol_versions.contains(&response.protocol_version) {
4066 bail!(
4167 "runner protocol mismatch: server {} / runner {}",
4268 response.protocol_version,
@@ -68,6 +94,8 @@
6894 match heartbeat_client
6995 .heartbeat(&HeartbeatRequest {
7096 running_job_ids: running,
97 protocol_versions: protocol_versions.clone(),
98 capabilities: capabilities.clone(),
7199 })
72100 .await
73101 {
@@ -83,7 +111,15 @@
83111 });
84112
85113 info!(server = %cfg.server_url, "runner polling for jobs");
114 let mut jobs = tokio::task::JoinSet::new();
86115 loop {
116 while jobs.len() >= cfg.capacity.max(1) as usize {
117 if let Some(result) = jobs.join_next().await {
118 if let Err(error) = result {
119 warn!(%error,"runner job task failed");
120 }
121 }
122 }
87123 let payload = match client.claim(&ClaimRequest { wait_secs: None }).await {
88124 Ok(response) => match response.job {
89125 Some(job) => job,
@@ -99,44 +135,72 @@
99135 if let Ok(mut set) = active.lock() {
100136 set.insert(payload.job_id);
101137 }
102 let outcome = run_job(&client, &cfg, &payload).await;
103 if let Ok(mut set) = active.lock() {
104 set.remove(&payload.job_id);
138 let job_client = client.clone();
139 let job_config = cfg.clone();
140 let job_active = active.clone();
141 jobs.spawn(async move {
142 execute_claimed_job(&job_client, &job_config, &payload).await;
143 if let Ok(mut set) = job_active.lock() {
144 set.remove(&payload.job_id);
145 }
146 runner::executor::clear_cancel(&payload.job_id);
147 });
148 if cfg.once {
149 info!("runner configured for a single job; exiting");
150 break;
151 }
152 }
153 while let Some(result) = jobs.join_next().await {
154 if let Err(error) = result {
155 warn!(%error,"runner job task failed");
105156 }
106 let finish = match outcome {
157 }
158
159 heartbeat_task.abort();
160 Ok(())
161}
162
163async fn execute_claimed_job(
164 client: &RunnerClient,
165 cfg: &config::runner::RunnerConfig,
166 payload: &JobPayload,
167) {
168 let finish = if payload.actions.is_some() {
169 runner::actions::execute(client, cfg, payload).await
170 } else {
171 match run_job(client, cfg, payload).await {
107172 JobOutcome::Success { exit_code } => JobFinishRequest {
108173 state: JobFinishState::Success,
109174 exit_code,
110 error: None,
175 ..Default::default()
111176 },
112177 JobOutcome::Failure { exit_code, error } => JobFinishRequest {
113178 state: JobFinishState::Failure,
114179 exit_code,
115180 error,
181 ..Default::default()
116182 },
117183 JobOutcome::Cancelled => JobFinishRequest {
118184 state: JobFinishState::Cancelled,
119 exit_code: None,
120 error: Some("cancelled".to_string()),
185 error: Some("cancelled".into()),
186 ..Default::default()
121187 },
122188 JobOutcome::TimedOut => JobFinishRequest {
123189 state: JobFinishState::TimedOut,
124 exit_code: None,
125 error: Some("job timed out".to_string()),
190 error: Some("job timed out".into()),
191 ..Default::default()
126192 },
127 };
128 if let Err(error) = client.finish(&payload.job_id, &finish).await {
129 warn!(%error, "job finish report failed");
130193 }
131 runner::executor::clear_cancel(&payload.job_id);
132 if cfg.once {
133 info!("runner configured for a single job; exiting");
134 break;
194 };
195 for attempt in 0..3 {
196 match client.finish(&payload.job_id, &finish).await {
197 Ok(_) => return,
198 Err(error) => {
199 warn!(%error,attempt,"job finish report failed");
200 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
201 }
135202 }
136203 }
137
138 heartbeat_task.abort();
139 Ok(())
140204}
141205
142206async fn run_job(
@@ -197,6 +261,7 @@
197261 .then(time::OffsetDateTime::now_utc),
198262 finished_at: (signal == runner::executor::StepSignal::Finished)
199263 .then(time::OffsetDateTime::now_utc),
264 ..Default::default()
200265 };
201266 let _ = client.report_step(job_id, index, &report).await;
202267 });
Modified app/webhook/src/main.rs View app/webhook/src/main.rs
@@ -62,13 +62,13 @@
6262 .await
6363 .with_context(|| format!("binding webhook {addr}"))?;
6464 let runner_routes = ci::routes::runner_router().with_state(service.clone());
65 let router: Router =
66 Router::new()
67 .nest("/api/v1", runner_routes)
68 .merge(export::health::router_with(
69 Some(service.state.database.clone()),
70 service.state.redis.clone(),
71 ));
65 let router: Router = Router::new()
66 .nest("/api/v1", runner_routes)
67 .merge(ci::actions_runtime::router().with_state(service.clone()))
68 .merge(export::health::router_with(
69 Some(service.state.database.clone()),
70 service.state.redis.clone(),
71 ));
7272 info!(%addr, "webhook scheduler listening");
7373
7474 let server_cancel = cancel.clone();
@@ -129,6 +129,7 @@
129129 let event_id = match &event {
130130 ci_protocol::events::CiEvent::Push(push) => push.id,
131131 ci_protocol::events::CiEvent::PullRequest(pr) => pr.id,
132 ci_protocol::events::CiEvent::Trigger(event) => event.id,
132133 };
133134 match service.handle_event(&event).await {
134135 Ok(created) => {
@@ -172,6 +173,9 @@
172173}
173174
174175async fn run_scheduler(service: CiService, interval_secs: u64, cancel: CancellationToken) {
176 if let Err(error) = service.discover_actions_schedules().await {
177 warn!(%error,"initial schedule discovery failed");
178 }
175179 let interval = Duration::from_secs(interval_secs.max(1));
176180 let mut ticker = tokio::time::interval(interval);
177181 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
@@ -197,6 +201,9 @@
197201 warn!(%error, "run finalization failed");
198202 }
199203 ticks = ticks.wrapping_add(1);
204 if let Err(error) = service.run_due_actions_schedules().await {
205 warn!(%error,"scheduled workflow execution failed");
206 }
200207 if ticks.is_multiple_of(3600)
201208 && let Err(error) = service.prune_expired().await
202209 {
Modified lib/config/src/ci.rs View lib/config/src/ci.rs
@@ -23,6 +23,12 @@
2323 pub outbound_timeout_secs: u64,
2424 pub scheduler_interval_secs: u64,
2525 pub cancel_grace_secs: u64,
26 pub actions_engine_binary: String,
27 pub actions_engine_timeout_secs: u64,
28 pub runtime_base_url: String,
29 pub cache_retention_days: i64,
30 pub cache_max_bytes: u64,
31 pub oidc_private_key: String,
2632}
2733
2834impl Default for CiConfig {
@@ -48,6 +54,12 @@
4854 outbound_timeout_secs: 10,
4955 scheduler_interval_secs: 2,
5056 cancel_grace_secs: 120,
57 actions_engine_binary: "gitdata-actions-engine".into(),
58 actions_engine_timeout_secs: 30,
59 runtime_base_url: String::new(),
60 cache_retention_days: 7,
61 cache_max_bytes: 10 * 1024 * 1024 * 1024,
62 oidc_private_key: String::new(),
5163 }
5264 }
5365}
Modified lib/config/src/runner.rs View lib/config/src/runner.rs
@@ -31,6 +31,13 @@
3131 pub log_flush_bytes: usize,
3232 pub keep_workspace: bool,
3333 pub once: bool,
34 pub actions_engine_binary: String,
35 pub actions_platforms: std::collections::BTreeMap<String, String>,
36 pub actions_node_runtimes: std::collections::BTreeMap<String, String>,
37 pub actions_cache_dir: String,
38 pub actions_action_server: String,
39 pub actions_action_token: String,
40 pub actions_groups: Vec<String>,
3441}
3542
3643impl Default for RunnerConfig {
@@ -56,6 +63,36 @@
5663 log_flush_bytes: 65536,
5764 keep_workspace: false,
5865 once: false,
66 actions_engine_binary: "gitdata-actions-engine".into(),
67 actions_platforms: [
68 ("self-hosted".into(), "-self-hosted".into()),
69 (
70 "ubuntu-latest".into(),
71 "ghcr.io/catthehacker/ubuntu:act-24.04".into(),
72 ),
73 (
74 "ubuntu-24.04".into(),
75 "ghcr.io/catthehacker/ubuntu:act-24.04".into(),
76 ),
77 (
78 "ubuntu-22.04".into(),
79 "ghcr.io/catthehacker/ubuntu:act-22.04".into(),
80 ),
81 ]
82 .into(),
83 actions_node_runtimes: ["node12", "node16", "node20", "node24"]
84 .into_iter()
85 .map(|version| {
86 (
87 version.into(),
88 format!("/opt/gitdata/node/{version}/bin/node"),
89 )
90 })
91 .collect(),
92 actions_cache_dir: "/var/lib/gitdata-runner/actions".into(),
93 actions_action_server: "https://github.com".into(),
94 actions_action_token: String::new(),
95 actions_groups: Vec::new(),
5996 }
6097 }
6198}
Modified lib/config/src/scoped.rs View lib/config/src/scoped.rs
@@ -185,6 +185,8 @@
185185#[derive(Clone, Debug, Default, Deserialize)]
186186#[serde(default)]
187187pub struct ApihostConfig {
188 pub ci: CiConfig,
189 pub webhook: WebhookConfig,
188190 pub s3: S3Config,
189191 pub redis: RedisConfig,
190192 pub chat: ChatConfig,
@@ -211,6 +213,8 @@
211213impl ApihostConfig {
212214 pub fn into_app_config(self) -> AppConfig {
213215 AppConfig {
216 ci: self.ci,
217 webhook: self.webhook,
214218 s3: self.s3,
215219 redis: self.redis,
216220 chat: self.chat,
@@ -239,6 +243,7 @@
239243
240244impl ScopedConfig for ApihostConfig {
241245 fn validate_scoped(&self) -> Result<(), String> {
246 self.ci.validate()?;
242247 self.redis.validate()?;
243248 self.database.validate()?;
244249 self.pow.validate()?;