crates/hub/src/db/runner_updates.rs at f2d2776b2a14 · andymccall/oxygenhub

# f2d2776b2a14 oxygenhub / crates / hub / src / db / runner_updates.rs

7.9 KiB · Rust · last changed Runner updates in waves: a later wave takes a version once the waves before it proved it; an error stops it (refs andymccall/oxygenflow#84) 8 days ago source blame history raw

//! 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<bool>,
    pub channel: Option<String>,
    pub window_from: Option<i64>,
    pub window_to: Option<i64>,
    pub dry_run: Option<bool>,
}

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<i64>,
    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<i64>,
}

pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> {
    sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await
}

pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> {
    let p: Option<Policy> = 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<Vec<Group>> {
    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<Option<Group>> {
    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<i64>) -> sqlx::Result<Group> {
    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<i64>) -> 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<i64>) -> 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<Vec<(i64, i64)>> {
    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<Vec<Hold>> {
    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<Option<Hold>> {
    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<bool> {
    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<i64>,
}

/// Every runner's [`Soak`] on `version`.
pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result<Vec<Soak>> {
    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<Option<String>> {
    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}") })
    }))
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

1	//! Runner updates (oxygenhub#322): the policy at each level and the test
2	//! groups runners are put in.
3
4	use super::{now_ms, Pool};
5	use sqlx::FromRow;
6
7	/// One level's settings; `None` inherits from the next level out.
8	#[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)]
9	pub struct Policy {
10	pub scope: String,
11	pub key: String,
12	pub updates: Option<bool>,
13	pub channel: Option<String>,
14	pub window_from: Option<i64>,
15	pub window_to: Option<i64>,
16	pub dry_run: Option<bool>,
17	}
18
19	impl Policy {
20	/// Nothing set: the level inherits everything.
21	pub fn is_empty(&self) -> bool {
22	self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none()
23	}
24	}
25
26	#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
27	pub struct Group {
28	pub id: i64,
29	/// Empty: a group of the site's runners.
30	pub owner: String,
31	pub name: String,
32	/// A share of the runners in reach, by a stable hash of their ids;
33	/// `None`: only the chosen ones.
34	pub percent: Option<i64>,
35	pub created_ms: i64,
36	/// The wave its runners take a version in (oxygenflow#84): wave 1
37	/// first, each later one once those before it have proved the version;
38	/// `None`: after every wave.
39	pub wave: Option<i64>,
40	}
41
42	pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> {
43	sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await
44	}
45
46	pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> {
47	let p: Option<Policy> = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2")
48	.bind(scope)
49	.bind(key)
50	.fetch_optional(pool)
51	.await?;
52	Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() }))
53	}
54
55	/// Set a level's settings; one with nothing set is removed.
56	pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> {
57	sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?;
58	if p.is_empty() {
59	return Ok(());
60	}
61	sqlx::query(
62	"INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run)
63	VALUES ($1, $2, $3, $4, $5, $6, $7)",
64	)
65	.bind(&p.scope)
66	.bind(&p.key)
67	.bind(p.updates)
68	.bind(&p.channel)
69	.bind(p.window_from)
70	.bind(p.window_to)
71	.bind(p.dry_run)
72	.execute(pool)
73	.await?;
74	Ok(())
75	}
76
77	pub async fn groups(pool: &Pool) -> sqlx::Result<Vec<Group>> {
78	sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await
79	}
80
81	pub async fn group(pool: &Pool, id: i64) -> sqlx::Result<Option<Group>> {
82	sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await
83	}
84
85	pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> {
86	sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *")
87	.bind(owner)
88	.bind(name)
89	.bind(percent)
90	.bind(now_ms())
91	.fetch_one(pool)
92	.await
93	}
94
95	pub async fn set_percent(pool: &Pool, id: i64, percent: Option<i64>) -> sqlx::Result<()> {
96	sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?;
97	Ok(())
98	}
99
100	pub async fn set_wave(pool: &Pool, id: i64, wave: Option<i64>) -> sqlx::Result<()> {
101	sqlx::query("UPDATE runner_groups SET wave = $1 WHERE id = $2").bind(wave).bind(id).execute(pool).await?;
102	Ok(())
103	}
104
105	/// Remove a group, its members and its policy.
106	pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> {
107	sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?;
108	sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?;
109	sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?;
110	Ok(())
111	}
112
113	/// Every chosen member, as (group, runner).
114	pub async fn members(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
115	sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await
116	}
117
118	/// Make `runners` the group's chosen members.
119	pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> {
120	sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?;
121	for r in runners {
122	sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?;
123	}
124	Ok(())
125	}
126
127	/// What a runner said about its update.
128	pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> {
129	sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3")
130	.bind(note)
131	.bind(now_ms())
132	.bind(runner)
133	.execute(pool)
134	.await?;
135	Ok(())
136	}
137
138	/// A build held back from every runner (#326).
139	#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
140	pub struct Hold {
141	pub version: String,
142	/// The runner it rolled back on.
143	pub runner: String,
144	pub why: String,
145	pub at_ms: i64,
146	}
147
148	/// Hold `version` back, the first time it rolls back.
149	pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> {
150	sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING")
151	.bind(version)
152	.bind(runner)
153	.bind(why)
154	.bind(now_ms())
155	.execute(pool)
156	.await?;
157	Ok(())
158	}
159
160	pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> {
161	sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await
162	}
163
164	pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> {
165	sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await
166	}
167
168	/// Lift the hold on `version`; whether there was one.
169	pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> {
170	Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0)
171	}
172
173	/// How a runner has done on a version (oxygenflow#84): its finished jobs
174	/// on that build, how many ended in `error` (the runner's or the
175	/// infrastructure's fault, not a test's), and when the first started.
176	#[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)]
177	pub struct Soak {
178	pub runner_id: i64,
179	pub jobs: i64,
180	pub errors: i64,
181	pub first_ms: Option<i64>,
182	}
183
184	/// Every runner's [`Soak`] on `version`.
185	pub async fn soaks(pool: &Pool, version: &str) -> sqlx::Result<Vec<Soak>> {
186	sqlx::query_as(
187	"SELECT runner_id, COUNT(*) AS jobs, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, MIN(started_ms) AS first_ms
188	FROM flow_jobs WHERE runner_version = $1 AND runner_id IS NOT NULL AND status IN ('passed', 'failed', 'error')
189	GROUP BY runner_id",
190	)
191	.bind(version)
192	.fetch_all(pool)
193	.await
194	}
195
196	/// The first job on `version` that ended in error on runner `runner`, in
197	/// words: `job build of OWNER/NAME run #N: why`.
198	pub async fn first_error(pool: &Pool, version: &str, runner: i64) -> sqlx::Result<Option<String>> {
199	let row: Option<(String, String, String, i64, String)> = sqlx::query_as(
200	"SELECT j.name, p.owner, p.name, r.number, j.error FROM flow_jobs j
201	JOIN flow_runs r ON r.id = j.run_id JOIN repositories p ON p.id = r.repo_id
202	WHERE j.runner_version = $1 AND j.runner_id = $2 AND j.status = 'error'
203	ORDER BY j.finished_ms LIMIT 1",
204	)
205	.bind(version)
206	.bind(runner)
207	.fetch_optional(pool)
208	.await?;
209	Ok(row.map(\|(job, owner, name, n, error)\| {
210	let why: String = error.chars().take(120).collect();
211	format!("job {job} of {owner}/{name} run #{n}{}", if why.is_empty() { String::new() } else { format!(": {why}") })
212	}))
213	}

Keyboard