swarm UI: an alerts page, read by the controller with a Viewer Grafana token from bao
swarm-bao-grafana-controller-token, on the host that runs Grafana, logs in to Grafana's API over its socket as the built-in admin (the password swarm-bao-grafana-admin minted, #5138) and keeps a `swarm-controller` service account that is an enabled Viewer. It mints a token for that account and writes it to bao at swarm/controller/swarm-controller/grafana/token under Grafana's leaf. A ten-minute timer runs it again. The unit keeps a stored token Grafana accepts. If Grafana refuses the stored token with 401, or bao holds none, the unit deletes every token of the account and mints a new one. Grafana's grant gains create/read/update on that leaf, and the controller's gains read. swarm-controller serves GET /api/alerts from Grafana's /api/prometheus/grafana/api/v1/alerts. It reads the token from bao, keeps it in memory, and re-reads bao once on a 401. The HTTP client has 5s connect and 15s request timeouts. Any failure is a 503 that names its cause. swarm-ui gets an /alerts page listing them, built the same way as the hives page. This is option (c) on #5129, the operator's choice.
This commit is contained in:
atlas
mara
parent
8cb6273782
commit
7d09776742
23 changed files with 888 additions and 37 deletions
|
|
@ -508,6 +508,7 @@ pub(crate) mod tests {
|
|||
reviews: Arc::default(),
|
||||
try_outs: crate::review_classifier::TryOuts::default(),
|
||||
runs: crate::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
391
swarm-controller/src/alerts.rs
Normal file
391
swarm-controller/src/alerts.rs
Normal file
|
|
@ -0,0 +1,391 @@
|
|||
//! `GET /api/alerts`: the alerts the swarm's Grafana holds right now, read at
|
||||
//! request time for swarm-ui's alerts page.
|
||||
//!
|
||||
//! Grafana's URL comes from `swarm-controller.nix`
|
||||
//! (`SWARM_CONTROLLER_GRAFANA_URL`). The token is a Viewer service account's,
|
||||
//! read from the swarm's secret store at
|
||||
//! `swarm_secret_client::grafana::controller_token_path` under this daemon's
|
||||
//! own certificate. `swarm-bao-grafana-controller-token`, on Grafana's host,
|
||||
//! mints it and replaces it whenever Grafana refuses the stored one, so a 401
|
||||
//! here re-reads the store once before giving up.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use axum::Json;
|
||||
use axum::extract::State;
|
||||
use axum::http::StatusCode;
|
||||
use problem_details::ProblemDetails;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use utoipa::ToSchema;
|
||||
|
||||
use crate::{AppState, error_problem};
|
||||
|
||||
/// The env var `swarm-controller.nix` sets.
|
||||
const URL_ENV: &str = "SWARM_CONTROLLER_GRAFANA_URL";
|
||||
|
||||
/// Grafana's Prometheus-compatible list of alert instances, relative to its
|
||||
/// base URL. Instances in the `Normal` state are not listed.
|
||||
const ALERTS_PATH: &str = "api/prometheus/grafana/api/v1/alerts";
|
||||
|
||||
/// Bound on reaching Grafana.
|
||||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
/// Bound on one whole request to Grafana, body included.
|
||||
const TIMEOUT: Duration = Duration::from_secs(15);
|
||||
|
||||
type TokenFuture = Pin<Box<dyn Future<Output = Result<String>> + Send>>;
|
||||
|
||||
/// Where the token comes from: the store in the daemon, a stand-in in tests.
|
||||
pub(crate) type TokenSource = Arc<dyn Fn() -> TokenFuture + Send + Sync>;
|
||||
|
||||
/// One alert instance, as `GET /api/alerts` serves it.
|
||||
#[derive(Clone, Debug, PartialEq, Serialize, ToSchema)]
|
||||
pub(crate) struct Alert {
|
||||
/// The rule's title (Grafana's `alertname` label).
|
||||
pub name: String,
|
||||
/// Grafana's state for the instance: `Alerting`, `Pending`, `NoData` or
|
||||
/// `Error`.
|
||||
pub state: String,
|
||||
/// When the instance entered its state, RFC 3339, as Grafana reports it.
|
||||
pub active_at: Option<String>,
|
||||
/// The rule's `summary` annotation, templated by Grafana.
|
||||
pub summary: Option<String>,
|
||||
/// Every label of the instance, as Grafana reports them.
|
||||
pub labels: BTreeMap<String, String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PromResponse {
|
||||
data: PromData,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PromData {
|
||||
alerts: Vec<PromAlert>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PromAlert {
|
||||
#[serde(default)]
|
||||
labels: BTreeMap<String, String>,
|
||||
#[serde(default)]
|
||||
annotations: BTreeMap<String, String>,
|
||||
state: String,
|
||||
#[serde(rename = "activeAt")]
|
||||
active_at: Option<String>,
|
||||
}
|
||||
|
||||
impl From<PromAlert> for Alert {
|
||||
fn from(a: PromAlert) -> Self {
|
||||
Self {
|
||||
name: a.labels.get("alertname").cloned().unwrap_or_default(),
|
||||
state: a.state,
|
||||
active_at: a.active_at,
|
||||
summary: a.annotations.get("summary").cloned(),
|
||||
labels: a.labels,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Grafana's alerts API, with the token last read from the store.
|
||||
pub(crate) struct Grafana {
|
||||
http: reqwest::Client,
|
||||
alerts_url: url::Url,
|
||||
source: TokenSource,
|
||||
token: tokio::sync::Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
impl Grafana {
|
||||
/// # Errors
|
||||
/// A base URL the alerts path cannot be joined onto, or a client that
|
||||
/// cannot be built.
|
||||
pub(crate) fn new(base: &url::Url, source: TokenSource, timeout: Duration) -> Result<Self> {
|
||||
let http = reqwest::Client::builder()
|
||||
.connect_timeout(CONNECT_TIMEOUT)
|
||||
.timeout(timeout)
|
||||
.build()
|
||||
.context("building the Grafana HTTP client")?;
|
||||
Ok(Self {
|
||||
http,
|
||||
alerts_url: base
|
||||
.join(ALERTS_PATH)
|
||||
.with_context(|| format!("joining {ALERTS_PATH} onto {base}"))?,
|
||||
source,
|
||||
token: tokio::sync::Mutex::new(None),
|
||||
})
|
||||
}
|
||||
|
||||
/// The held token, or a fresh one from the source when `reread` or none is
|
||||
/// held yet.
|
||||
async fn token(&self, reread: bool) -> Result<String> {
|
||||
let mut held = self.token.lock().await;
|
||||
if let Some(t) = held.as_ref().filter(|_| !reread) {
|
||||
return Ok(t.clone());
|
||||
}
|
||||
let fresh = (self.source)().await?;
|
||||
*held = Some(fresh.clone());
|
||||
Ok(fresh)
|
||||
}
|
||||
|
||||
async fn get(&self, token: &str) -> Result<reqwest::Response> {
|
||||
self.http
|
||||
.get(self.alerts_url.clone())
|
||||
.bearer_auth(token)
|
||||
.send()
|
||||
.await
|
||||
.with_context(|| format!("GET {}", self.alerts_url))
|
||||
}
|
||||
|
||||
/// Every alert instance Grafana lists now.
|
||||
///
|
||||
/// # Errors
|
||||
/// No token in the store, Grafana unreachable or slower than the client's
|
||||
/// timeout, a 401 for both the held token and a re-read one, any other
|
||||
/// non-success status, or a body that is not the alerts shape.
|
||||
pub(crate) async fn alerts(&self) -> Result<Vec<Alert>> {
|
||||
let mut res = self.get(&self.token(false).await?).await?;
|
||||
if res.status() == reqwest::StatusCode::UNAUTHORIZED {
|
||||
res = self.get(&self.token(true).await?).await?;
|
||||
}
|
||||
let status = res.status();
|
||||
if !status.is_success() {
|
||||
anyhow::bail!("Grafana answered {status} to GET {}", self.alerts_url);
|
||||
}
|
||||
let body: PromResponse = res
|
||||
.json()
|
||||
.await
|
||||
.with_context(|| format!("reading Grafana's answer to GET {}", self.alerts_url))?;
|
||||
Ok(body.data.alerts.into_iter().map(Alert::from).collect())
|
||||
}
|
||||
}
|
||||
|
||||
/// One login and one read. An empty value is refused like an absent one.
|
||||
async fn token_from_store() -> Result<String> {
|
||||
let path = swarm_secret_client::grafana::controller_token_path()?;
|
||||
let store = crate::store::connect()
|
||||
.await
|
||||
.context("logging in to the swarm secret store")?;
|
||||
let stored: Option<swarm_secret_client::grafana::ControllerToken> = store
|
||||
.read_optional(&path)
|
||||
.await
|
||||
.with_context(|| format!("reading {path}"))?;
|
||||
match stored {
|
||||
Some(t) if !t.value.trim().is_empty() => Ok(t.value.trim().to_owned()),
|
||||
_ => anyhow::bail!(
|
||||
"nothing at {path}: swarm-bao-grafana-controller-token, on the host that runs \
|
||||
Grafana, has not stored this daemon's Grafana token"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// The daemon's Grafana client, or `None` when `SWARM_CONTROLLER_GRAFANA_URL`
|
||||
/// is unset or unusable, which is logged.
|
||||
pub(crate) fn from_env() -> Option<Arc<Grafana>> {
|
||||
let raw = std::env::var(URL_ENV).ok()?;
|
||||
let built = url::Url::parse(&raw)
|
||||
.with_context(|| format!("parse {URL_ENV} ({raw}) as a URL"))
|
||||
.and_then(|base| {
|
||||
let source: TokenSource = Arc::new(|| Box::pin(token_from_store()));
|
||||
Grafana::new(&base, source, TIMEOUT)
|
||||
});
|
||||
match built {
|
||||
Ok(g) => Some(Arc::new(g)),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "Grafana misconfigured; /api/alerts is off");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The alerts the swarm's Grafana holds now: firing, pending, and those whose
|
||||
/// rule has no data or failed to evaluate.
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "/api/alerts",
|
||||
responses(
|
||||
(status = 200, description = "every alert instance Grafana lists now", body = Vec<Alert>),
|
||||
(status = 503, description = "no Grafana is configured here, its token is not in the store \
|
||||
yet, or Grafana refused, failed or did not answer in time (problem+json)", body = String),
|
||||
),
|
||||
tag = "alerts"
|
||||
)]
|
||||
pub(crate) async fn get_alerts(
|
||||
State(state): State<AppState>,
|
||||
) -> Result<Json<Vec<Alert>>, ProblemDetails> {
|
||||
let Some(grafana) = state.alerts.as_ref() else {
|
||||
return Err(error_problem(
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"no Grafana is configured on this host",
|
||||
));
|
||||
};
|
||||
grafana.alerts().await.map(Json).map_err(|e| {
|
||||
let detail = format!("{e:#}");
|
||||
tracing::warn!(error = %detail, "reading Grafana's alerts failed");
|
||||
error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail)
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use axum::body::Body;
|
||||
use axum::http::{HeaderMap, Request};
|
||||
use std::sync::Mutex;
|
||||
use tower::ServiceExt as _;
|
||||
|
||||
/// What the stand-in Grafana answers to a request with the accepted token.
|
||||
const ALERTS_BODY: &str = r#"{"status":"success","data":{"alerts":[{
|
||||
"labels":{"alertname":"swarm queue connect failed","hive":"pr1ma","agent":"atlas"},
|
||||
"annotations":{"summary":"pr1ma/atlas could not connect"},
|
||||
"state":"Alerting","activeAt":"2026-10-15T09:00:00Z","value":"1e+00"}]}}"#;
|
||||
|
||||
/// A stand-in Grafana on a loopback port: the alerts path answers
|
||||
/// [`ALERTS_BODY`] to `Bearer <accepts>` and 401 to anything else, after
|
||||
/// `delay`.
|
||||
async fn fake_grafana(accepts: &'static str, delay: Duration) -> url::Url {
|
||||
let app = axum::Router::new().route(
|
||||
&format!("/{ALERTS_PATH}"),
|
||||
axum::routing::get(move |headers: HeaderMap| async move {
|
||||
tokio::time::sleep(delay).await;
|
||||
let ok = headers
|
||||
.get(axum::http::header::AUTHORIZATION)
|
||||
.is_some_and(|v| v == format!("Bearer {accepts}").as_str());
|
||||
if ok {
|
||||
(StatusCode::OK, ALERTS_BODY)
|
||||
} else {
|
||||
(StatusCode::UNAUTHORIZED, "")
|
||||
}
|
||||
}),
|
||||
);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move { axum::serve(listener, app).await });
|
||||
url::Url::parse(&format!("http://{addr}/")).unwrap()
|
||||
}
|
||||
|
||||
/// A token source handing out `tokens` in order, the last one forever,
|
||||
/// counting its calls.
|
||||
fn tokens(tokens: &'static [&'static str]) -> (TokenSource, Arc<Mutex<usize>>) {
|
||||
let calls = Arc::new(Mutex::new(0));
|
||||
let counted = Arc::clone(&calls);
|
||||
let source: TokenSource = Arc::new(move || {
|
||||
let mut n = counted.lock().unwrap();
|
||||
let t = tokens[(*n).min(tokens.len() - 1)];
|
||||
*n += 1;
|
||||
Box::pin(async move { Ok(t.to_owned()) })
|
||||
});
|
||||
(source, calls)
|
||||
}
|
||||
|
||||
async fn get_alerts_with(grafana: Option<Grafana>) -> (StatusCode, serde_json::Value) {
|
||||
let (mut state, _) = crate::tests::state_with_roster();
|
||||
state.alerts = grafana.map(Arc::new);
|
||||
let req = Request::get("/api/alerts")
|
||||
.body(Body::empty())
|
||||
.expect("a request");
|
||||
let res = crate::build_app(state)
|
||||
.oneshot(req)
|
||||
.await
|
||||
.expect("infallible");
|
||||
let status = res.status();
|
||||
let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
|
||||
.await
|
||||
.expect("a body");
|
||||
(status, serde_json::from_slice(&bytes).unwrap_or_default())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn serves_grafanas_alerts_read_with_the_stored_token() {
|
||||
let base = fake_grafana("viewer-token", Duration::ZERO).await;
|
||||
let (source, calls) = tokens(&["viewer-token"]);
|
||||
let grafana = Grafana::new(&base, source, TIMEOUT).unwrap();
|
||||
let (status, body) = get_alerts_with(Some(grafana)).await;
|
||||
assert_eq!(status, StatusCode::OK, "{body}");
|
||||
assert_eq!(
|
||||
body,
|
||||
serde_json::json!([{
|
||||
"name": "swarm queue connect failed",
|
||||
"state": "Alerting",
|
||||
"active_at": "2026-10-15T09:00:00Z",
|
||||
"summary": "pr1ma/atlas could not connect",
|
||||
"labels": {
|
||||
"alertname": "swarm queue connect failed",
|
||||
"hive": "pr1ma",
|
||||
"agent": "atlas",
|
||||
},
|
||||
}])
|
||||
);
|
||||
assert_eq!(*calls.lock().unwrap(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_held_token_is_reused_across_requests() {
|
||||
let base = fake_grafana("viewer-token", Duration::ZERO).await;
|
||||
let (source, calls) = tokens(&["viewer-token"]);
|
||||
let grafana = Grafana::new(&base, source, TIMEOUT).unwrap();
|
||||
grafana.alerts().await.unwrap();
|
||||
grafana.alerts().await.unwrap();
|
||||
assert_eq!(*calls.lock().unwrap(), 1);
|
||||
}
|
||||
|
||||
/// The minting unit replaces a token Grafana refuses, so a 401 re-reads
|
||||
/// the store and sends the request once more.
|
||||
#[tokio::test]
|
||||
async fn a_refused_token_is_read_again_from_the_store() {
|
||||
let base = fake_grafana("replaced", Duration::ZERO).await;
|
||||
let (source, calls) = tokens(&["stale", "replaced"]);
|
||||
let grafana = Grafana::new(&base, source, TIMEOUT).unwrap();
|
||||
let alerts = grafana.alerts().await.unwrap();
|
||||
assert_eq!(alerts.len(), 1);
|
||||
assert_eq!(*calls.lock().unwrap(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_token_refused_twice_is_a_503_naming_the_status() {
|
||||
let base = fake_grafana("replaced", Duration::ZERO).await;
|
||||
let (source, calls) = tokens(&["stale"]);
|
||||
let grafana = Grafana::new(&base, source, TIMEOUT).unwrap();
|
||||
let (status, body) = get_alerts_with(Some(grafana)).await;
|
||||
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
|
||||
assert!(body["detail"].as_str().unwrap().contains("401"), "{body}");
|
||||
assert_eq!(*calls.lock().unwrap(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_grafana_slower_than_the_timeout_is_a_503() {
|
||||
let base = fake_grafana("viewer-token", Duration::from_secs(5)).await;
|
||||
let (source, _) = tokens(&["viewer-token"]);
|
||||
let grafana = Grafana::new(&base, source, Duration::from_millis(200)).unwrap();
|
||||
let started = std::time::Instant::now();
|
||||
let (status, _) = get_alerts_with(Some(grafana)).await;
|
||||
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
|
||||
assert!(started.elapsed() < Duration::from_secs(4));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn no_token_in_the_store_is_a_503_saying_so() {
|
||||
let base = fake_grafana("viewer-token", Duration::ZERO).await;
|
||||
let source: TokenSource =
|
||||
Arc::new(|| Box::pin(async { anyhow::bail!("nothing at the token path") }));
|
||||
let grafana = Grafana::new(&base, source, TIMEOUT).unwrap();
|
||||
let (status, body) = get_alerts_with(Some(grafana)).await;
|
||||
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
|
||||
assert!(
|
||||
body["detail"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.contains("nothing at the token path"),
|
||||
"{body}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn no_grafana_configured_is_a_503() {
|
||||
let (status, _) = get_alerts_with(None).await;
|
||||
assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
|
||||
}
|
||||
}
|
||||
|
|
@ -168,6 +168,7 @@ mod tests {
|
|||
try_outs: super::super::review_classifier::TryOuts::default(),
|
||||
reviews: std::sync::Arc::default(),
|
||||
runs: crate::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -138,6 +138,7 @@ mod tests {
|
|||
try_outs: super::super::review_classifier::TryOuts::default(),
|
||||
reviews: std::sync::Arc::default(),
|
||||
runs: crate::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -973,6 +973,7 @@ pub(crate) mod tests {
|
|||
try_outs: super::super::review_classifier::TryOuts::default(),
|
||||
reviews: std::sync::Arc::default(),
|
||||
runs: crate::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -50,6 +50,7 @@ mod agent_login;
|
|||
mod agent_renewal;
|
||||
mod agent_state_stream;
|
||||
mod agent_status;
|
||||
mod alerts;
|
||||
mod auth;
|
||||
mod builder;
|
||||
mod claude_login;
|
||||
|
|
@ -839,6 +840,7 @@ fn socket_path() -> PathBuf {
|
|||
(name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"),
|
||||
(name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page), and the classifier's swarm-wide settings"),
|
||||
(name = "pulls", description = "open pull requests, for the try-out page's dropdowns"),
|
||||
(name = "alerts", description = "the swarm Grafana's current alerts (swarm-ui's alerts page)"),
|
||||
)
|
||||
)]
|
||||
struct ApiDoc;
|
||||
|
|
@ -972,6 +974,9 @@ struct AppState {
|
|||
reviews: Arc<review_classifier::review::Reviews>,
|
||||
/// The review classifier's finished runs.
|
||||
runs: review_classifier::runs::Runs,
|
||||
/// Grafana's alerts API, for `GET /api/alerts`. `None` when no Grafana URL
|
||||
/// is configured.
|
||||
alerts: Option<Arc<alerts::Grafana>>,
|
||||
}
|
||||
|
||||
/// Env var the controller's NixOS module sets from
|
||||
|
|
@ -3451,6 +3456,7 @@ async fn main() -> Result<()> {
|
|||
try_outs: review_classifier::TryOuts::default(),
|
||||
reviews: Arc::default(),
|
||||
runs,
|
||||
alerts: alerts::from_env(),
|
||||
};
|
||||
|
||||
let app = build_app(state);
|
||||
|
|
@ -3468,6 +3474,7 @@ async fn main() -> Result<()> {
|
|||
fn build_app(state: AppState) -> axum::Router {
|
||||
let (router, api) = OpenApiRouter::<AppState>::with_openapi(ApiDoc::openapi())
|
||||
.routes(routes!(health))
|
||||
.routes(routes!(alerts::get_alerts))
|
||||
.routes(routes!(get_hives))
|
||||
.routes(routes!(get_hives_status))
|
||||
.routes(routes!(get_links))
|
||||
|
|
@ -3660,6 +3667,7 @@ mod tests {
|
|||
try_outs: super::review_classifier::TryOuts::default(),
|
||||
reviews: std::sync::Arc::default(),
|
||||
runs: super::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
};
|
||||
(state, sched)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -585,6 +585,7 @@ mod tests {
|
|||
try_outs: super::super::review_classifier::TryOuts::default(),
|
||||
reviews: std::sync::Arc::default(),
|
||||
runs: crate::review_classifier::runs::Runs::Live(None),
|
||||
alerts: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue