runnerd_core/
github.rs

1//! The GitHub REST API as the scheduler and the history need it: queued jobs, a job's steps, just-in-time runner
2//! configs, repository runners and cancelling a run. Owns the token, conditional requests and rate-limit backoff. It
3//! must never put the token in a log line, a `Debug` output or an error, and it decides nothing about scheduling.
4
5use std::collections::HashMap;
6use std::fmt;
7use std::sync::{Arc, Mutex};
8use std::time::{Duration, SystemTime};
9
10use reqwest::header::{self, HeaderMap, HeaderValue};
11use reqwest::{Method, StatusCode};
12use serde::Deserialize;
13use serde::de::DeserializeOwned;
14
15use crate::docker::JitConfig;
16use crate::sync::lock;
17use crate::time;
18
19/// The public API's address.
20pub const API_URL: &str = "https://api.github.com";
21
22/// How long one request may take, connecting included. GitHub answers in well under a second; a request that hangs
23/// holds back the tick that made it.
24const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
25
26/// A GitHub token. Its `Debug` output never shows the value.
27#[derive(Clone)]
28pub struct Token(String);
29
30impl fmt::Debug for Token {
31    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
32        f.write_str("Token(<redacted>)")
33    }
34}
35
36/// Where the client gets a token: at start and again when GitHub answers 401.
37pub type TokenSource = Box<dyn Fn() -> Result<Token, Error> + Send + Sync>;
38
39/// The `gh` CLI's token for github.com (`gh auth token`).
40pub fn gh_token() -> Result<Token, Error> {
41    let output = std::process::Command::new("gh")
42        .args(["auth", "token", "--hostname", "github.com"])
43        .output()
44        .map_err(|error| Error::Token(format!("cannot run gh: {error}")))?;
45    if !output.status.success() {
46        return Err(Error::Token(format!(
47            "gh auth token failed: {}",
48            String::from_utf8_lossy(&output.stderr).trim()
49        )));
50    }
51    let token = String::from_utf8(output.stdout)
52        .map_err(|_| Error::Token("gh auth token printed a token that is not UTF-8".to_owned()))?;
53    Ok(Token(token.trim().to_owned()))
54}
55
56/// Makes ring the process's TLS crypto provider, which every `reqwest::Client` needs: reqwest is built without
57/// one. Calling it again does nothing.
58pub fn install_crypto_provider() {
59    // Fails only when a provider is installed already.
60    let _ = rustls::crypto::ring::default_provider().install_default();
61}
62
63/// What a GitHub request can fail with. No variant holds the token.
64#[derive(Debug, thiserror::Error)]
65pub enum Error {
66    /// GitHub's rate limit is used up; no request is sent before `until`.
67    #[error("GitHub's rate limit is used up until {}s after the epoch", secs(*until))]
68    RateLimited {
69        /// When the limit resets.
70        until: SystemTime,
71    },
72    /// GitHub answered with an error status.
73    #[error("GitHub answered {status} to {method} {path}: {message}")]
74    Status {
75        /// The request's method.
76        method: Method,
77        /// The request's path, without the host.
78        path: String,
79        /// What GitHub answered.
80        status: StatusCode,
81        /// GitHub's error message, or the body as text.
82        message: String,
83    },
84    /// The request did not get an answer.
85    #[error("{method} {path}: {source}")]
86    Http {
87        /// The request's method.
88        method: Method,
89        /// The request's path, without the host.
90        path: String,
91        /// What failed, without the URL.
92        source: reqwest::Error,
93    },
94    /// The answer is not the JSON expected.
95    #[error("{path}: unexpected response: {source}")]
96    Decode {
97        /// The request's path, without the host.
98        path: String,
99        /// What did not parse.
100        source: serde_json::Error,
101    },
102    /// No token could be read.
103    #[error("no GitHub token: {0}")]
104    Token(String),
105}
106
107fn secs(time: SystemTime) -> u64 {
108    time.duration_since(SystemTime::UNIX_EPOCH)
109        .unwrap_or_default()
110        .as_secs()
111}
112
113/// A job waiting for a runner.
114#[derive(Debug, Clone, Default, PartialEq, Eq)]
115pub struct QueuedJob {
116    /// The job's ID; IDs grow with time.
117    pub id: u64,
118    /// The workflow run it belongs to.
119    pub run_id: u64,
120    /// The job's name in the workflow.
121    pub name: String,
122    /// The workflow's `runs-on` labels.
123    pub labels: Vec<String>,
124    /// When GitHub queued it, in milliseconds since the epoch; 0 if GitHub's answer did not say.
125    pub created_at: u64,
126    /// 1 for the run's first attempt.
127    pub run_attempt: u32,
128    /// The workflow's name.
129    pub workflow: Option<String>,
130    /// The branch the run is for.
131    pub branch: Option<String>,
132    /// The job's page on GitHub.
133    pub url: Option<String>,
134}
135
136/// A job as `GET /repos/{owner}/{repo}/actions/jobs/{id}` describes it: its progress and steps.
137#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
138pub struct JobDetails {
139    /// The job's ID.
140    pub id: u64,
141    /// The workflow run it belongs to.
142    pub run_id: u64,
143    /// 1 for the run's first attempt.
144    #[serde(default = "first_attempt")]
145    pub run_attempt: u32,
146    /// The job's name in the workflow.
147    pub name: String,
148    /// The workflow's name.
149    pub workflow_name: Option<String>,
150    /// The branch the run is for.
151    pub head_branch: Option<String>,
152    /// The job's page on GitHub.
153    pub html_url: Option<String>,
154    /// The workflow's `runs-on` labels.
155    #[serde(default)]
156    pub labels: Vec<String>,
157    /// When GitHub queued it.
158    pub created_at: Option<String>,
159    /// `queued`, `in_progress`, `completed`, `waiting`, ….
160    pub status: String,
161    /// Once completed: `success`, `failure`, `cancelled`, ….
162    pub conclusion: Option<String>,
163    /// When the runner started it.
164    pub started_at: Option<String>,
165    /// When it completed.
166    pub completed_at: Option<String>,
167    /// The runner that took it, if one did.
168    pub runner_name: Option<String>,
169    /// Its steps so far.
170    #[serde(default)]
171    pub steps: Vec<JobStep>,
172}
173
174fn first_attempt() -> u32 {
175    1
176}
177
178/// A step of `JobDetails`.
179#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
180pub struct JobStep {
181    /// GitHub's step number.
182    pub number: u32,
183    /// The name GitHub shows.
184    pub name: String,
185    /// `pending`, `queued`, `in_progress` or `completed`.
186    pub status: String,
187    /// Once completed: `success`, `failure`, `skipped`, ….
188    pub conclusion: Option<String>,
189    /// When it started.
190    pub started_at: Option<String>,
191    /// When it completed.
192    pub completed_at: Option<String>,
193}
194
195/// A runner registered on a repository.
196#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
197pub struct Runner {
198    /// The runner's ID.
199    pub id: u64,
200    /// Its name; runnerd's runners start with the host's prefix.
201    pub name: String,
202    /// `online` or `offline`.
203    pub status: String,
204    /// Whether it runs a job.
205    pub busy: bool,
206}
207
208/// What deleting a runner did.
209#[derive(Debug, Clone, Copy, PartialEq, Eq)]
210pub enum Deleted {
211    /// GitHub removed it.
212    Deleted,
213    /// It was gone already: an ephemeral runner removes itself after its job.
214    NotFound,
215    /// GitHub refused because the runner runs a job.
216    Busy,
217}
218
219/// The GitHub calls the scheduler makes. `Github` is the real client; the engine's tests use a fake.
220pub trait GithubApi: Send + Sync {
221    /// The repository's jobs waiting for a runner, in queued runs and in runs that started some of their jobs.
222    fn queued_jobs(
223        &self,
224        repository: &str,
225    ) -> impl Future<Output = Result<Vec<QueuedJob>, Error>> + Send;
226    /// The repository's self-hosted runners.
227    fn runners(&self, repository: &str) -> impl Future<Output = Result<Vec<Runner>, Error>> + Send;
228    /// Registers a runner for one job and returns it with the config `run.sh --jitconfig` takes.
229    fn jit_config(
230        &self,
231        repository: &str,
232        name: &str,
233        labels: &[String],
234    ) -> impl Future<Output = Result<(Runner, JitConfig), Error>> + Send;
235    /// Removes a runner from the repository.
236    fn delete_runner(
237        &self,
238        repository: &str,
239        id: u64,
240    ) -> impl Future<Output = Result<Deleted, Error>> + Send;
241    /// One job's progress and steps.
242    fn job(
243        &self,
244        repository: &str,
245        id: u64,
246    ) -> impl Future<Output = Result<JobDetails, Error>> + Send;
247    /// Cancels a workflow run, and so every job in it: GitHub cancels runs, not single jobs. A run that already
248    /// completed is left as it is.
249    fn cancel_run(
250        &self,
251        repository: &str,
252        run: u64,
253    ) -> impl Future<Output = Result<(), Error>> + Send;
254}
255
256/// A GitHub API client for one token. Requests on one URL are conditional on its last `ETag`, so an unchanged
257/// answer costs no rate limit.
258pub struct Github {
259    http: reqwest::Client,
260    base: String,
261    /// Run on Tokio's blocking pool: `gh` may wait on a keyring for seconds.
262    token_source: Arc<dyn Fn() -> Result<Token, Error> + Send + Sync>,
263    /// An async lock, held while the source runs: requests that need a token wait without blocking a worker, and
264    /// concurrent 401s read it once.
265    token: tokio::sync::Mutex<Option<Token>>,
266    // Locked with `sync::lock`: a panic while holding one of these locks leaves plain data behind, still usable.
267    etags: Mutex<HashMap<String, (HeaderValue, Vec<u8>)>>,
268    blocked_until: Mutex<Option<SystemTime>>,
269}
270
271impl fmt::Debug for Github {
272    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
273        f.debug_struct("Github")
274            .field("base", &self.base)
275            .finish_non_exhaustive()
276    }
277}
278
279#[derive(Deserialize)]
280struct Runs {
281    workflow_runs: Vec<Run>,
282}
283
284#[derive(Deserialize)]
285struct Run {
286    id: u64,
287}
288
289#[derive(Deserialize)]
290struct Jobs {
291    jobs: Vec<Job>,
292}
293
294#[derive(Deserialize)]
295struct Job {
296    id: u64,
297    run_id: u64,
298    name: String,
299    status: String,
300    labels: Vec<String>,
301    created_at: Option<String>,
302    run_attempt: Option<u32>,
303    workflow_name: Option<String>,
304    head_branch: Option<String>,
305    html_url: Option<String>,
306}
307
308#[derive(Deserialize)]
309struct Runners {
310    runners: Vec<Runner>,
311}
312
313#[derive(Deserialize)]
314struct JitResponse {
315    runner: Runner,
316    encoded_jit_config: String,
317}
318
319#[derive(Deserialize)]
320struct ErrorBody {
321    message: String,
322}
323
324impl Github {
325    /// A client for `base` (`API_URL`, or a test server) that takes its token from `token_source`.
326    pub fn new(base: &str, token_source: TokenSource) -> Result<Github, Error> {
327        install_crypto_provider();
328        let mut headers = HeaderMap::new();
329        headers.insert(
330            header::ACCEPT,
331            HeaderValue::from_static("application/vnd.github+json"),
332        );
333        headers.insert(
334            "x-github-api-version",
335            HeaderValue::from_static("2022-11-28"),
336        );
337        let http = reqwest::Client::builder()
338            .user_agent(concat!("runnerd/", env!("CARGO_PKG_VERSION")))
339            .default_headers(headers)
340            .timeout(REQUEST_TIMEOUT)
341            .build()
342            .map_err(|source| Error::Http {
343                method: Method::GET,
344                path: base.to_owned(),
345                source,
346            })?;
347        Ok(Github {
348            http,
349            base: base.trim_end_matches('/').to_owned(),
350            token_source: Arc::from(token_source),
351            token: tokio::sync::Mutex::new(None),
352            etags: Mutex::new(HashMap::new()),
353            blocked_until: Mutex::new(None),
354        })
355    }
356
357    /// When the next request may be sent: `None` now, otherwise when the rate limit resets.
358    pub fn blocked_until(&self) -> Option<SystemTime> {
359        let blocked = *lock(&self.blocked_until);
360        blocked.filter(|until| *until > SystemTime::now())
361    }
362
363    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, Error> {
364        let response = self.send(Method::GET, path, None).await?;
365        let bytes = if response.status == StatusCode::NOT_MODIFIED {
366            let cached = lock(&self.etags).get(path).map(|(_, body)| body.clone());
367            match cached {
368                Some(body) => body,
369                None => {
370                    return Err(Error::Status {
371                        method: Method::GET,
372                        path: path.to_owned(),
373                        status: response.status,
374                        message: "not modified, but nothing is cached".to_owned(),
375                    });
376                }
377            }
378        } else {
379            let etag = response.etag.clone();
380            let response = expect_success(Method::GET, path, response)?;
381            if let Some(etag) = etag {
382                lock(&self.etags).insert(path.to_owned(), (etag, response.bytes.clone()));
383            }
384            response.bytes
385        };
386        decode(path, &bytes)
387    }
388
389    /// Sends a request with the token, once more with a fresh token after a 401. Records the rate limit.
390    async fn send(
391        &self,
392        method: Method,
393        path: &str,
394        body: Option<&serde_json::Value>,
395    ) -> Result<Response, Error> {
396        if let Some(until) = self.blocked_until() {
397            return Err(Error::RateLimited { until });
398        }
399        // The token GitHub refused, once it did.
400        let mut refused: Option<Token> = None;
401        loop {
402            let token = self.token(refused.as_ref()).await?;
403            let mut request = self
404                .http
405                .request(method.clone(), format!("{}{path}", self.base))
406                .bearer_auth(&token.0);
407            if let Some(body) = body {
408                request = request
409                    .header(header::CONTENT_TYPE, "application/json")
410                    .body(body.to_string());
411            }
412            if method == Method::GET
413                && let Some((etag, _)) = lock(&self.etags).get(path)
414            {
415                request = request.header(header::IF_NONE_MATCH, etag.clone());
416            }
417            let response = request.send().await.map_err(|source| Error::Http {
418                method: method.clone(),
419                path: path.to_owned(),
420                source: source.without_url(),
421            })?;
422            let status = response.status();
423            if status == StatusCode::UNAUTHORIZED && refused.is_none() {
424                refused = Some(token);
425                continue;
426            }
427            if let Some(until) = rate_limited_until(status, response.headers(), SystemTime::now()) {
428                *lock(&self.blocked_until) = Some(until);
429                if matches!(
430                    status,
431                    StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
432                ) {
433                    return Err(Error::RateLimited { until });
434                }
435            }
436            let etag = response.headers().get(header::ETAG).cloned();
437            let bytes = response.bytes().await.map_err(|source| Error::Http {
438                method: method.clone(),
439                path: path.to_owned(),
440                source: source.without_url(),
441            })?;
442            return Ok(Response {
443                status,
444                etag,
445                bytes: bytes.to_vec(),
446            });
447        }
448    }
449
450    /// The token, read from the source if there is none yet or it is still the one GitHub `refused`: of several
451    /// requests refused at once, the first reads a new one and the others take it.
452    async fn token(&self, refused: Option<&Token>) -> Result<Token, Error> {
453        let mut token = self.token.lock().await;
454        let stale = match (&*token, refused) {
455            (None, _) => true,
456            (Some(current), Some(refused)) => current.0 == refused.0,
457            (Some(_), None) => false,
458        };
459        if stale {
460            let source = self.token_source.clone();
461            *token = Some(
462                tokio::task::spawn_blocking(move || source())
463                    .await
464                    .map_err(|error| {
465                        Error::Token(format!("the token source stopped: {error}"))
466                    })??,
467            );
468        }
469        token
470            .clone()
471            .ok_or_else(|| Error::Token("the token source returned nothing".to_owned()))
472    }
473}
474
475impl GithubApi for Github {
476    async fn queued_jobs(&self, repository: &str) -> Result<Vec<QueuedJob>, Error> {
477        let mut jobs = Vec::new();
478        let mut listed = Vec::new();
479        for status in ["queued", "in_progress"] {
480            let runs: Runs = self
481                .get(&format!(
482                    "/repos/{repository}/actions/runs?status={status}&per_page=100"
483                ))
484                .await?;
485            for run in runs.workflow_runs {
486                let path = jobs_path(repository, run.id);
487                let run_jobs: Jobs = self.get(&path).await?;
488                listed.push(path);
489                jobs.extend(
490                    run_jobs
491                        .jobs
492                        .into_iter()
493                        .filter(|job| job.status == "queued")
494                        .map(|job| QueuedJob {
495                            id: job.id,
496                            run_id: job.run_id,
497                            name: job.name,
498                            labels: job.labels,
499                            created_at: job
500                                .created_at
501                                .as_deref()
502                                .and_then(time::parse_ms)
503                                .unwrap_or(0),
504                            run_attempt: job.run_attempt.unwrap_or(1),
505                            workflow: job.workflow_name,
506                            branch: job.head_branch,
507                            url: job.html_url,
508                        }),
509                );
510            }
511        }
512        // The cached answers of runs that ended: nothing asks for them again.
513        let runs_prefix = format!("/repos/{repository}/actions/runs/");
514        lock(&self.etags).retain(|path, _| {
515            !(path.starts_with(&runs_prefix) && path.contains("/jobs?")) || listed.contains(path)
516        });
517        jobs.sort_by_key(|job| job.id);
518        jobs.dedup_by_key(|job| job.id);
519        Ok(jobs)
520    }
521
522    async fn runners(&self, repository: &str) -> Result<Vec<Runner>, Error> {
523        let runners: Runners = self
524            .get(&format!("/repos/{repository}/actions/runners?per_page=100"))
525            .await?;
526        Ok(runners.runners)
527    }
528
529    async fn jit_config(
530        &self,
531        repository: &str,
532        name: &str,
533        labels: &[String],
534    ) -> Result<(Runner, JitConfig), Error> {
535        let body = serde_json::json!({
536            "name": name,
537            // The repository's default runner group.
538            "runner_group_id": 1,
539            "labels": labels,
540            "work_folder": "_work",
541        });
542        let path = format!("/repos/{repository}/actions/runners/generate-jitconfig");
543        let response = self.send(Method::POST, &path, Some(&body)).await?;
544        let response = expect_success(Method::POST, &path, response)?;
545        let jit: JitResponse = decode(&path, &response.bytes)?;
546        Ok((jit.runner, JitConfig(jit.encoded_jit_config)))
547    }
548
549    async fn delete_runner(&self, repository: &str, id: u64) -> Result<Deleted, Error> {
550        let path = format!("/repos/{repository}/actions/runners/{id}");
551        let response = self.send(Method::DELETE, &path, None).await?;
552        match response.status {
553            StatusCode::NOT_FOUND => Ok(Deleted::NotFound),
554            StatusCode::UNPROCESSABLE_ENTITY => Ok(Deleted::Busy),
555            _ => expect_success(Method::DELETE, &path, response).map(|_| Deleted::Deleted),
556        }
557    }
558
559    async fn job(&self, repository: &str, id: u64) -> Result<JobDetails, Error> {
560        self.get(&format!("/repos/{repository}/actions/jobs/{id}"))
561            .await
562    }
563
564    async fn cancel_run(&self, repository: &str, run: u64) -> Result<(), Error> {
565        let path = format!("/repos/{repository}/actions/runs/{run}/cancel");
566        let response = self.send(Method::POST, &path, None).await?;
567        // 409: the run completed already.
568        if response.status == StatusCode::CONFLICT {
569            return Ok(());
570        }
571        expect_success(Method::POST, &path, response).map(|_| ())
572    }
573}
574
575fn jobs_path(repository: &str, run: u64) -> String {
576    format!("/repos/{repository}/actions/runs/{run}/jobs?filter=latest&per_page=100")
577}
578
579struct Response {
580    status: StatusCode,
581    etag: Option<HeaderValue>,
582    bytes: Vec<u8>,
583}
584
585fn expect_success(method: Method, path: &str, response: Response) -> Result<Response, Error> {
586    if response.status.is_success() {
587        return Ok(response);
588    }
589    let message = serde_json::from_slice::<ErrorBody>(&response.bytes).map_or_else(
590        |_| String::from_utf8_lossy(&response.bytes).into_owned(),
591        |body| body.message,
592    );
593    Err(Error::Status {
594        method,
595        path: path.to_owned(),
596        status: response.status,
597        message,
598    })
599}
600
601fn decode<T: DeserializeOwned>(path: &str, bytes: &[u8]) -> Result<T, Error> {
602    serde_json::from_slice(bytes).map_err(|source| Error::Decode {
603        path: path.to_owned(),
604        source,
605    })
606}
607
608/// When requests may resume, if this response says the rate limit is used up: the primary limit's reset time, or
609/// a secondary limit's `retry-after`.
610fn rate_limited_until(
611    status: StatusCode,
612    headers: &HeaderMap,
613    now: SystemTime,
614) -> Option<SystemTime> {
615    let number = |name: &str| {
616        headers
617            .get(name)
618            .and_then(|value| value.to_str().ok())
619            .and_then(|value| value.trim().parse::<u64>().ok())
620    };
621    if number("x-ratelimit-remaining") == Some(0)
622        && let Some(reset) = number("x-ratelimit-reset")
623    {
624        return Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset));
625    }
626    if matches!(
627        status,
628        StatusCode::FORBIDDEN | StatusCode::TOO_MANY_REQUESTS
629    ) && let Some(seconds) = number("retry-after")
630    {
631        return Some(now + Duration::from_secs(seconds));
632    }
633    None
634}
635
636#[cfg(test)]
637mod tests;
runnerd_core/github.rs

runnerd_core/
github.rs
