runnerd_core/
github.rs

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

runnerd_core/
github.rs
