738dd1c95b
no behaviour change: a pure move of items and tests into new modules, plus visibility edits (pub(crate)) and path updates in sibling modules that named crate-root items. same routes, same OpenAPI, 561 tests before and after. why: #5470, new features stop landing in one 6.4k-line file by default. files outside swarm-controller/src: docs/agent-lifecycle/agent-roster.md, swarm-ui JobsPage.tsx and AgentsPage.tsx, nix/host-modules/swarm-controller.nix and swarm-grafana.nix hold comments that named main.rs for items that moved.
391 lines
15 KiB
Rust
391 lines
15 KiB
Rust
//! `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::routes::{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::routes::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::startup::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);
|
|
}
|
|
}
|