hyperhive/swarm-ci-metrics/src/main.rs

//! The swarm's CI run poller: one pass per start, run by a timer.
//!
//! Reads every repo with Actions that the `ci-metrics` forge account can see,
//! lists each repo's runs newest first down to its cursor, and pushes each
//! newly finished run as two OTLP gauges (`sample`) to the swarm collector,
//! which forwards them to the metrics store. With no cursor the walk reaches
//! run 1, so the first pass is the backfill; a page is read at most every
//! [`PAGE_DELAY`].
//!
//! The store identity comes from the `BAO_*` environment. Under it this reads
//! the forge token swarm-controller mints for the account, and the OIDC
//! client secret the collector's route for this service checks, once per pass.
//!
//! Each pass also pushes the two `bao.client_cert.*` gauges for that store
//! certificate ([`push_cert`]).

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::{Context as _, Result, bail};
use forgejo_api::structs::{ActionRun, ListActionRunsQuery, RepoSearchQuery};
use forgejo_api::{Auth, Forgejo};
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::ci_metrics::{ClientSecret as StoredClientSecret, client_secret_path};
use swarm_secret_client::client::ENV_CLIENT_CERT;
use swarm_secret_client::forge::Credential;
use swarm_secret_client::stored_secret::StoredSecret;
use swarm_secret_client::{SecretStore, cert};

mod cursor;
mod otlp;
mod pass;
mod sample;

/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
/// writes it, matching the leaf's subject.
const CERT_ROLE: &str = "swarm-ci-metrics";

/// The environment prefix `swarm-ci-metrics.nix` sets.
const PREFIX: &str = "SWARM_CI_METRICS";

/// Runs and repos per forge page: the forge's own cap.
const PAGE: u32 = 50;

/// The pause before each further page of one listing.
const PAGE_DELAY: Duration = Duration::from_secs(1);

struct Config {
    forge_url: url::Url,
    /// `…/v1/metrics` on this service's collector route.
    metrics_url: String,
    queue: QueueConfig,
    state_file: PathBuf,
}

#[tokio::main]
async fn main() -> Result<()> {
    hive_log::init();
    swarm_secret_client::install_crypto_provider();
    let var = |name: &str| {
        let full = format!("{PREFIX}_{name}");
        std::env::var(&full).with_context(|| format!("{full} is unset"))
    };
    let client_id = var("OIDC_CLIENT_ID")?;
    let secret = client_secret(&client_id).await?;
    let reread_id = client_id.clone();
    let secret = StoredSecret::new(client_id.clone(), secret, move || {
        let client_id = reread_id.clone();
        async move { client_secret(&client_id).await.map_err(Into::into) }
    });
    let state_dir = std::env::var("STATE_DIRECTORY").context("STATE_DIRECTORY is unset")?;
    let cfg = Config {
        forge_url: var("FORGE_URL")?.parse().context("parsing the forge URL")?,
        metrics_url: format!("{}/v1/metrics", var("OTLP_ENDPOINT")?.trim_end_matches('/')),
        queue: QueueConfig {
            // `mint_token_for` reads only the token endpoint and the client.
            url: String::new(),
            token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
            client_id,
            client_secret: ClientSecret::Stored(secret),
            ca_file: None,
        },
        state_file: PathBuf::from(state_dir).join("cursor.json"),
    };

    let forge = Forgejo::new(Auth::Token(&forge_token().await?), cfg.forge_url.clone())
        .context("building the forge client")?;
    let http = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .build()
        .context("building the OTLP client")?;
    push_cert(&cfg, &http).await;
    let mut state = cursor::State::load(&cfg.state_file)?;
    let repos = repos(&forge).await?;
    tracing::info!(repos = repos.len(), "ci metrics: pass starting");

    let mut failed = 0;
    for repo in &repos {
        if let Err(e) = repo_pass(&cfg, &forge, &http, &mut state, repo).await {
            tracing::warn!(repo, error = %format!("{e:#}"), "ci metrics: repo pass failed");
            failed += 1;
        }
    }
    if failed > 0 {
        bail!(
            "{failed} of {} repos failed; their cursors stay where they were",
            repos.len()
        );
    }
    Ok(())
}

/// One repo: walk, plan, push, move the cursor.
async fn repo_pass(
    cfg: &Config,
    forge: &Forgejo,
    http: &reqwest::Client,
    state: &mut cursor::State,
    repo: &str,
) -> Result<()> {
    let runs = walk(forge, repo, state.repo(repo)).await?;
    let plan = pass::plan(state, repo, &runs, time::OffsetDateTime::now_utc());
    let mut skipped: BTreeMap<sample::Skip, usize> = BTreeMap::new();
    for (_, why) in &plan.skip {
        *skipped.entry(*why).or_default() += 1;
    }
    let pushing = plan.push.len();
    let pending = plan.lowest_pending;
    let bearer = if pushing > 0 {
        swarm_queue_client::mint_token_for(&cfg.queue, Some(&cfg.queue.client_id), None)
            .await
            .context("minting a token for the collector")?
    } else {
        String::new()
    };
    pass::apply(
        state,
        repo,
        plan,
        async |samples| push(http, &cfg.metrics_url, &bearer, samples).await,
        |s| s.save(&cfg.state_file),
    )
    .await?;
    tracing::info!(
        repo,
        read = runs.len(),
        pushed = pushing,
        ?skipped,
        ?pending,
        "ci metrics: repo done"
    );
    Ok(())
}

/// The `cert` label this service's certificate is reported under.
const CERT_LABEL: &str = "ci-metrics";

/// Pushes the two `bao.client_cert.*` gauges for this service's own store
/// certificate ([`ENV_CLIENT_CERT`]), once per pass, stamped now.
///
/// An unreadable file or a refused push is logged and the pass carries on: the
/// runs are the pass's work, and this is a report about the process doing it.
async fn push_cert(cfg: &Config, http: &reqwest::Client) {
    let Ok(path) = std::env::var(ENV_CLIENT_CERT) else {
        tracing::debug!("ci metrics: {ENV_CLIENT_CERT} is unset, no client-cert gauges");
        return;
    };
    let (not_before, not_after) = match cert::file_validity(Path::new(&path)) {
        Ok(window) => window,
        Err(e) => {
            tracing::debug!(error = %e, "ci metrics: no client-cert gauges");
            return;
        }
    };
    let now = time::OffsetDateTime::now_utc().unix_timestamp_nanos();
    let body = otlp::cert_body(CERT_LABEL, not_before, not_after, now);
    let pushed = async {
        let bearer =
            swarm_queue_client::mint_token_for(&cfg.queue, Some(&cfg.queue.client_id), None)
                .await
                .context("minting a token for the collector")?;
        post(http, &cfg.metrics_url, &bearer, &body).await
    };
    if let Err(e) = pushed.await {
        tracing::warn!(error = %format!("{e:#}"), "ci metrics: pushing the client-cert gauges failed");
    }
}

/// Every repo the account sees that has Actions and is neither archived nor a
/// mirror, as `owner/name`.
async fn repos(forge: &Forgejo) -> Result<Vec<String>> {
    let mut out = Vec::new();
    for page in 1.. {
        let found = forge
            .repo_search(RepoSearchQuery::default())
            .page(page)
            .page_size(PAGE)
            .send()
            .await
            .context("searching the forge's repos")?
            .data
            .unwrap_or_default();
        let last = found.len() < PAGE as usize;
        out.extend(found.into_iter().filter_map(|r| {
            let wanted =
                r.has_actions == Some(true) && r.archived != Some(true) && r.mirror != Some(true);
            r.full_name.filter(|_| wanted)
        }));
        if last {
            break;
        }
        tokio::time::sleep(PAGE_DELAY).await;
    }
    Ok(out)
}

/// The runs of `repo`, newest first, down to the first page that reaches below
/// the cursor.
async fn walk(forge: &Forgejo, repo: &str, cursor: &cursor::RepoCursor) -> Result<Vec<ActionRun>> {
    let (owner, name) = repo
        .split_once('/')
        .with_context(|| format!("{repo:?} is not owner/name"))?;
    let mut runs = Vec::new();
    for page in 1.. {
        let found = forge
            .list_action_runs(owner, name, ListActionRunsQuery::default())
            .page(page)
            .page_size(PAGE)
            .send()
            .await
            .with_context(|| format!("listing the runs of {repo}, page {page}"))?
            .workflow_runs
            .unwrap_or_default();
        let done = found.len() < PAGE as usize
            || found.iter().any(|r| {
                r.index_in_repo
                    .and_then(|n| u64::try_from(n).ok())
                    .is_some_and(|n| cursor.is_below(n))
            });
        runs.extend(found);
        if done {
            break;
        }
        tokio::time::sleep(PAGE_DELAY).await;
    }
    Ok(runs)
}

/// POST `samples` to the collector; an error unless it answers 2xx.
async fn push(
    http: &reqwest::Client,
    url: &str,
    bearer: &str,
    samples: &[sample::RunSamples],
) -> Result<()> {
    post(http, url, bearer, &otlp::body(samples)).await
}

/// POST one OTLP JSON `body` to the collector; an error unless it answers 2xx.
async fn post(
    http: &reqwest::Client,
    url: &str,
    bearer: &str,
    body: &serde_json::Value,
) -> Result<()> {
    let resp = http
        .post(url)
        .bearer_auth(bearer)
        .json(body)
        .send()
        .await
        .with_context(|| format!("pushing to {url}"))?;
    let status = resp.status();
    if !status.is_success() {
        let body = resp.text().await.unwrap_or_default();
        bail!(
            "{url} answered {status}: {}",
            body.chars().take(300).collect::<String>()
        );
    }
    Ok(())
}

/// The `ci-metrics` account's forge token, which swarm-controller mints.
async fn forge_token() -> Result<String> {
    let path = swarm_secret_client::ci_metrics::forge_token_path()?;
    let stored: Option<Credential> = login()
        .await?
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
        _ => bail!("nothing at {path}: swarm-controller has not minted the ci-metrics forge token"),
    }
}

/// This service's OIDC client secret, which authelia's mint writes.
async fn client_secret(client_id: &str) -> Result<String> {
    let path = client_secret_path(client_id)?;
    let stored: Option<StoredClientSecret> = login()
        .await?
        .read_optional(&path)
        .await
        .with_context(|| format!("reading {path}"))?;
    match stored {
        Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
        _ => bail!(
            "nothing at {path}: authelia's mint, in its container, has not \
             stored this service's client secret"
        ),
    }
}

async fn login() -> Result<SecretStore> {
    SecretStore::from_env(CERT_ROLE)
        .await
        .context("logging in to the secret store")
}