//! Runner updates (oxygenhub#322): the policy at each level and the test //! groups runners are put in. use super::{now_ms, Pool}; use sqlx::FromRow; /// One level's settings; `None` inherits from the next level out. #[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)] pub struct Policy { pub scope: String, pub key: String, pub updates: Option, pub channel: Option, pub window_from: Option, pub window_to: Option, pub dry_run: Option, } impl Policy { /// Nothing set: the level inherits everything. pub fn is_empty(&self) -> bool { self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none() } } #[derive(Clone, Debug, PartialEq, Eq, FromRow)] pub struct Group { pub id: i64, /// Empty: a group of the site's runners. pub owner: String, pub name: String, /// A share of the runners in reach, by a stable hash of their ids; /// `None`: only the chosen ones. pub percent: Option, pub created_ms: i64, /// The wave its runners take a version in (oxygenflow#84): wave 1 /// first, each later one once those before it have proved the version; /// `None`: after every wave. pub wave: Option, } pub async fn policies(pool: &Pool) -> sqlx::Result> { sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await } pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result { let p: Option = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2") .bind(scope) .bind(key) .fetch_optional(pool) .await?; Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() })) } /// Set a level's settings; one with nothing set is removed. pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> { sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?; if p.is_empty() { return Ok(()); } sqlx::query( "INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run) VALUES ($1, $2, $3, $4, $5, $6, $7)", ) .bind(&p.scope) .bind(&p.key) .bind(p.updates) .bind(&p.channel) .bind(p.window_from) .bind(p.window_to) .bind(p.dry_run) .execute(pool) .await?; Ok(()) } pub async fn groups(pool: &Pool) -> sqlx::Result> { sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await } pub async fn group(pool: &Pool, id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await } pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option) -> sqlx::Result { sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *") .bind(owner) .bind(name) .bind(percent) .bind(now_ms()) .fetch_one(pool) .await } pub async fn set_percent(pool: &Pool, id: i64, percent: Option) -> sqlx::Result<()> { sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?; Ok(()) } pub async fn set_wave(pool: &Pool, id: i64, wave: Option) -> sqlx::Result<()> { sqlx::query("UPDATE runner_groups SET wave = $1 WHERE id = $2").bind(wave).bind(id).execute(pool).await?; Ok(()) } /// Remove a group, its members and its policy. pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> { sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?; sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?; sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?; Ok(()) } /// Every chosen member, as (group, runner). pub async fn members(pool: &Pool) -> sqlx::Result> { sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await } /// Make `runners` the group's chosen members. pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> { sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?; for r in runners { sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?; } Ok(()) } /// What a runner said about its update. pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> { sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3") .bind(note) .bind(now_ms()) .bind(runner) .execute(pool) .await?; Ok(()) } /// A build held back from every runner (#326). #[derive(Clone, Debug, PartialEq, Eq, FromRow)] pub struct Hold { pub version: String, /// The runner it rolled back on. pub runner: String, pub why: String, pub at_ms: i64, } /// Hold `version` back, the first time it rolls back. pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> { sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING") .bind(version) .bind(runner) .bind(why) .bind(now_ms()) .execute(pool) .await?; Ok(()) } pub async fn holds(pool: &Pool) -> sqlx::Result> { sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await } pub async fn held(pool: &Pool, version: &str) -> sqlx::Result> { sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await } /// Lift the hold on `version`; whether there was one. pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result { Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0) } /// How a runner has done on a version (oxygenflow#84): its finished jobs /// on that build, how many ended in `error` (the runner's or the /// infrastructure's fault, not a test's), and when the first started. #[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)] pub struct Soak { pub runner_id: i64, pub jobs: i64, pub errors: i64, pub first_ms: Option, } /// Every runner's [`Soak`] on `version`. pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result> { sqlx::query_as( "SELECT runner_id, COUNT(*) AS jobs, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, MIN(started_ms) AS first_ms FROM flow_jobs WHERE runner_version = $1 AND runner_id IS NOT NULL AND status IN ('passed', 'failed', 'error') GROUP BY runner_id", ) .bind(version) .fetch_all(pool) .await } /// The first job on `version` that ended in error on runner `runner`, in /// words: `job build of OWNER/NAME run #N: why`. pub async fn first_error(pool: &Pool, version: &str, runner: i64) -> sqlx::Result> { let row: Option<(String, String, String, i64, String)> = sqlx::query_as( "SELECT j.name, p.owner, p.name, r.number, j.error FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id WHERE j.runner_version = $1 AND j.runner_id = $2 AND j.status = 'error' ORDER BY j.finished_ms LIMIT 1", ) .bind(version) .bind(runner) .fetch_optional(pool) .await?; Ok(row.map(|(job, owner, name, n, error)| { let why: String = error.chars().take(120).collect(); format!("job {job} of {owner}/{name} run #{n}{}", if why.is_empty() { String::new() } else { format!(": {why}") }) })) }