diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/auth.rs | 6 | ||||
| -rw-r--r-- | src/api/debug.rs | 2 | ||||
| -rw-r--r-- | src/api/document.rs | 77 | ||||
| -rw-r--r-- | src/api/log.rs | 32 | ||||
| -rw-r--r-- | src/api/mod.rs | 20 | ||||
| -rw-r--r-- | src/api/peers.rs | 75 | ||||
| -rw-r--r-- | src/api/stats.rs | 199 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 521 | ||||
| -rw-r--r-- | src/api/store.rs | 2 |
9 files changed, 918 insertions, 16 deletions
diff --git a/src/api/auth.rs b/src/api/auth.rs index 3926c70..70b1af5 100644 --- a/src/api/auth.rs +++ b/src/api/auth.rs @@ -1,14 +1,14 @@ -use argon2::{password_hash::PasswordVerifier, Argon2}; +use argon2::{Argon2, password_hash::PasswordVerifier}; use axum::extract::State; use axum::{ body::Body, http::{ - header::{AUTHORIZATION, WWW_AUTHENTICATE}, HeaderMap, Request, StatusCode, + header::{AUTHORIZATION, WWW_AUTHENTICATE}, }, response::{IntoResponse, Response}, }; -use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; use dashmap::DashMap; use password_hash::phc::PasswordHash; use std::{ diff --git a/src/api/debug.rs b/src/api/debug.rs index 929d665..6a56a9c 100644 --- a/src/api/debug.rs +++ b/src/api/debug.rs @@ -2,7 +2,7 @@ use std::collections::HashSet; -use axum::{body::Bytes, extract::State, http::StatusCode, Json}; +use axum::{Json, body::Bytes, extract::State, http::StatusCode}; use tracing::{info, warn}; use crate::state::AppState; diff --git a/src/api/document.rs b/src/api/document.rs new file mode 100644 index 0000000..303f5c9 --- /dev/null +++ b/src/api/document.rs @@ -0,0 +1,77 @@ +//! `GET /admin/document`: the local document as Automerge bytes, for backups and as +//! input to an `import` remake. +//! +//! The whole document is saved: history, and protocol data under `_continuum`. Where +//! it came from is in headers: `X-Continuum-Swarm`, `X-Continuum-Epoch` +//! (`<generation>:<id hex>`) and `X-Continuum-Heads` (hex, comma-separated). + +use axum::{ + extract::State, + http::{HeaderName, HeaderValue, StatusCode, header}, + response::{IntoResponse, Response}, +}; +use tokio::time::Instant; +use tracing::{info, warn}; + +use crate::state::AppState; + +pub(crate) async fn get_document(State(state): State<AppState>) -> Response { + if !state.is_ready().await { + return (StatusCode::SERVICE_UNAVAILABLE, "node is not ready\n").into_response(); + } + let swarm_id = state.swarm_id().await.unwrap_or_default(); + + let start = Instant::now(); + let doc = std::sync::Arc::clone(&state.doc); + // Both steps take time proportional to the document: off the async threads, and + // the lock only for the copy. + let saved = tokio::task::spawn_blocking(move || { + let (mut copy, epoch) = { + let replica = doc.blocking_read(); + (replica.doc.clone(), replica.epoch) + }; + let heads = copy.get_heads(); + (copy.save(), heads, epoch) + }) + .await; + let (bytes, heads, epoch) = match saved { + Ok(saved) => saved, + Err(error) => { + warn!(error = %error, "Saving the document for export failed"); + return (StatusCode::INTERNAL_SERVER_ERROR, "saving failed\n").into_response(); + } + }; + let took = start.elapsed(); + state.stats.saved_size_measured(bytes.len() as u64, took); + info!( + bytes = bytes.len(), + took_ms = took.as_millis() as u64, + "Exported the document" + ); + + let heads = heads + .iter() + .map(|head| head.to_string()) + .collect::<Vec<_>>() + .join(","); + let filename = format!( + "attachment; filename=\"continuum-{swarm_id}-{}.automerge\"", + epoch.generation + ); + let mut response = bytes.into_response(); + let headers = response.headers_mut(); + let mut set = |name: HeaderName, value: String| { + if let Ok(value) = HeaderValue::from_str(&value) { + headers.insert(name, value); + } + }; + set(header::CONTENT_TYPE, "application/octet-stream".to_owned()); + set(header::CONTENT_DISPOSITION, filename); + set(HeaderName::from_static("x-continuum-swarm"), swarm_id); + set( + HeaderName::from_static("x-continuum-epoch"), + format!("{}:{}", epoch.generation, epoch.id_hex()), + ); + set(HeaderName::from_static("x-continuum-heads"), heads); + response +} diff --git a/src/api/log.rs b/src/api/log.rs new file mode 100644 index 0000000..5d44926 --- /dev/null +++ b/src/api/log.rs @@ -0,0 +1,32 @@ +//! /admin/log: read and replace the log filter of the running node. + +use axum::{body::Bytes, http::StatusCode}; +use tracing::{info, warn}; + +use crate::logging; + +/// `GET /admin/log`: the filter in effect, as plain text in `RUST_LOG` syntax. +pub(crate) async fn get_log_filter() -> Result<String, StatusCode> { + logging::filter().map_err(|error| { + warn!(error = %format!("{error:#}"), "Cannot read the log filter"); + StatusCode::INTERNAL_SERVER_ERROR + }) +} + +/// `PUT /admin/log`: replace the filter; the body is plain text in `RUST_LOG` syntax, +/// e.g. `info,continuum::swarm::sync=trace`. Answers with the filter now in effect; +/// 400 (with the reason) if the body doesn't parse, and the old filter stays. +pub(crate) async fn put_log_filter(body: Bytes) -> Result<String, (StatusCode, String)> { + let directives = std::str::from_utf8(&body) + .map_err(|_| { + ( + StatusCode::BAD_REQUEST, + "The filter must be UTF-8\n".to_owned(), + ) + })? + .trim(); + let applied = logging::set_filter(directives) + .map_err(|error| (StatusCode::BAD_REQUEST, format!("{error:#}\n")))?; + info!(filter = %applied, "Log filter replaced"); + Ok(applied + "\n") +} diff --git a/src/api/mod.rs b/src/api/mod.rs index ea3ad41..3233b16 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -2,17 +2,20 @@ pub(crate) mod auth; mod debug; +mod document; +mod log; mod peers; +mod stats; mod store; use axum::{ + Router, http::{ - header::{CONTENT_TYPE, ETAG}, HeaderValue, StatusCode, + header::{CONTENT_TYPE, ETAG}, }, response::{IntoResponse, Response}, routing::{delete, get, put}, - Router, }; use serde_json::Value as JsonValue; use sha2::{Digest, Sha256}; @@ -20,7 +23,7 @@ use tracing::{error, warn}; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; -use crate::api::auth::{basic_auth, AuthState}; +use crate::api::auth::{AuthState, basic_auth}; use crate::document::store::StoreError; use crate::state::AppState; @@ -37,6 +40,12 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> .route("/stream/{*key}", get(store::stream_store_entry)) .route("/peers", get(peers::list_peers)) .route("/peers/{peer}", delete(peers::delete_peer)) + .route( + "/admin/log", + get(log::get_log_filter).put(log::put_log_filter), + ) + .route("/admin/stats", get(stats::get_stats)) + .route("/admin/document", get(document::get_document)) .merge(crate::remake::router::<AppState>()); if debug_api { @@ -148,6 +157,11 @@ fn response_for_value( peers::PeerListResponse, peers::PeerLivenessStatus, peers::EpochInfo, + crate::transport::traffic::PeerTraffic, + crate::transport::traffic::ByKind, + crate::transport::traffic::Counter, + crate::state::Incidents, + peers::ThroughputInfo, ) ), tags( diff --git a/src/api/peers.rs b/src/api/peers.rs index d60e7b3..7caaaf3 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -3,9 +3,9 @@ use std::sync::Arc; use axum::{ + Json, extract::{Path, Query, State}, http::StatusCode, - Json, }; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; @@ -16,8 +16,9 @@ use utoipa::ToSchema; use super::store_status; use crate::document::replica::Epoch; use crate::document::store; -use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState}; +use crate::state::{AppState, HelloState, Incidents, Liveness, PeerState, unix_millis}; use crate::swarm::peer::forgotten_path; +use crate::transport::traffic::PeerTraffic; #[derive(Debug, Serialize, ToSchema)] #[serde(rename_all = "lowercase")] @@ -47,6 +48,8 @@ impl From<Epoch> for EpochInfo { #[serde(rename_all = "camelCase")] pub(crate) struct PeerStatus { public_key: String, + /// The name the peer gave itself in the member list, if any (not unique). + alias: Option<String>, validated_swarm: Option<String>, status: PeerLivenessStatus, handshake_complete: bool, @@ -57,12 +60,41 @@ pub(crate) struct PeerStatus { epoch: Option<EpochInfo>, /// Whether that epoch equals ours, i.e. whether we sync with this peer. same_epoch: Option<bool>, + /// Number of heads in its last keepalive; absent before one, or when it had more + /// than 32 (then only their digest is sent). + heads: Option<usize>, + /// We keep it alive and sync with it on our own; other peers we only answer. + neighbour: bool, + /// Effective round-trip time in milliseconds (smoothed, raised by requests still + /// unanswered); absent without a recent sample. + rtt_ms: Option<f64>, + /// Messages and payload bytes exchanged with it, by kind, since we started (or + /// since it was last listed). + traffic: PeerTraffic, + incidents: Incidents, + /// Smoothed sync throughput (`swarm::throughput`); absent before the first large + /// exchange. + sync_throughput: Option<ThroughputInfo>, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ThroughputInfo { + bytes_per_sec: f64, + samples: u32, + /// Age of the last sample. + secs_ago: u64, } #[derive(Debug, Serialize, ToSchema)] #[serde(rename_all = "camelCase")] pub(crate) struct PeerListResponse { + /// Our own alias (`--alias`). + local_alias: Option<String>, local_epoch: EpochInfo, + /// Number of heads of our document. Several heads are concurrent changes no later + /// change has merged yet. + local_heads: usize, peers: Vec<PeerStatus>, } @@ -97,9 +129,26 @@ pub(crate) async fn list_peers( let now = Instant::now(); let local_epoch = state.local_epoch(); + let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc); let mut peers = Vec::with_capacity(snapshot.len()); for (public_key, peer_state_mutex) in snapshot { + let neighbour = crate::swarm::topology::is_neighbour(&state, &public_key); + let rtt_ms = state + .lock_rtt() + .effective(&public_key, now, state.timing.rtt_fresh_for()) + .map(|rtt| rtt.as_secs_f64() * 1000.0); + let traffic = state.traffic.peer(&public_key).unwrap_or_default(); + let sync_throughput = state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&public_key) + .map(|estimate| ThroughputInfo { + bytes_per_sec: estimate.bytes_per_sec.round(), + samples: estimate.samples, + secs_ago: now.saturating_duration_since(estimate.at).as_secs(), + }); let ps = peer_state_mutex.lock().await; let (status, dormant_for_secs) = match ps.liveness { @@ -111,26 +160,33 @@ pub(crate) async fn list_peers( }; peers.push(PeerStatus { + alias: aliases.remove(&public_key), public_key, - validated_swarm: match &ps.membership { - PeerMembership::Unknown => None, - PeerMembership::Member { swarm_id } => Some(swarm_id.clone()), - }, + validated_swarm: ps.standing.swarm_id().map(str::to_owned), status, handshake_complete: ps.hello_state == HelloState::Established, last_seen_secs_ago: ps .last_seen .map(|t| now.saturating_duration_since(t).as_secs()), dormant_for_secs, - has_pending_delivery: ps.pending_outbound_sync.is_some(), + has_pending_delivery: ps.sync.sending.is_some() || ps.sync.retry_at.is_some(), epoch: ps.epoch.map(EpochInfo::from), same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), + heads: ps.reported_heads, + neighbour, + rtt_ms, + traffic, + incidents: ps.incidents, + sync_throughput, }); } peers.sort_by(|a, b| a.public_key.cmp(&b.public_key)); + let local_heads = state.doc.write().await.doc.get_heads().len(); Ok(Json(PeerListResponse { + local_alias: state.alias.clone(), local_epoch: local_epoch.into(), + local_heads, peers, })) } @@ -178,7 +234,10 @@ pub(crate) async fn delete_peer( let peer_clone = peer.clone(); state - .mutate_doc(move |doc| store::write_path(doc, &forgotten_path(&peer_clone), &record)) + .mutate_doc(move |doc| { + store::write_path(doc, &forgotten_path(&peer_clone), &record)?; + crate::swarm::members::unregister(doc, &peer_clone) + }) .await .map_err(|e| store_status(e, "Failed to write forget record"))?; diff --git a/src/api/stats.rs b/src/api/stats.rs new file mode 100644 index 0000000..bc925ff --- /dev/null +++ b/src/api/stats.rs @@ -0,0 +1,199 @@ +//! /admin/stats: how this node is doing, in numbers. JSON, or the Prometheus text +//! format when the client asks for it (`prometheus`). + +mod prometheus; + +use axum::{ + Json, + extract::State, + http::{HeaderMap, header}, + response::{IntoResponse, Response}, +}; +use serde::Serialize; + +use super::peers::EpochInfo; +use crate::state::{AppState, Incidents}; +use crate::swarm::{members, stability, topology}; +use crate::transport::traffic::PeerTraffic; + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct StatsResponse { + public_key: String, + alias: Option<String>, + uptime_secs: u64, + ready: bool, + epoch: EpochInfo, + document: DocumentStats, + catch_up: CatchUpStats, + topology: TopologyStats, + stability: StabilityStats, + /// HELLO_REPLYs to an earlier run of ours, ignored. + stale_hello_replies: u64, + /// All peers together, including ones no longer listed in /peers. + traffic: PeerTraffic, + /// The peers listed in /peers together (a forgotten peer's are gone). + incidents: Incidents, + propagation: PropagationStats, +} + +/// How long changes written elsewhere took from their commit there until applied here, +/// once caught up. Includes the clock skew between the writer and us. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationStats { + count: u64, + sum_ms: u64, + /// Changes stamped later than our clock on arrival (writer's clock ahead). + ahead: u64, + /// Not cumulative; the last bucket has no bound. + buckets: Vec<PropagationBucket>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationBucket { + le_ms: Option<u64>, + count: u64, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct DocumentStats { + heads: usize, + /// Most heads after applying a sync message (concurrent changes not yet merged). + max_heads: usize, + /// Size as saved (compressed): about what a joining node downloads. Measured in the + /// background, at least a minute apart (longer for large documents); absent until + /// first measured. + saved_bytes: Option<u64>, + /// How long ago `saved_bytes` was measured. + saved_bytes_secs_ago: Option<u64>, + /// How long the last measurement took (copy and save), in milliseconds. + saved_bytes_took_ms: Option<u64>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct CatchUpStats { + /// Catch-up sources chosen (more than one: it switched). + sources: u32, + /// Seconds from start until the initial sync was done; absent until then (and on + /// the genesis node). + done_after_secs: Option<f64>, + /// Sync payload received until then. + sync_bytes: Option<u64>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct TopologyStats { + k: usize, + /// Absent until ready with a member list. + view: Option<TopologyView>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct TopologyView { + /// Registered members, us included. + members: usize, + /// Members other than us not judged unreachable. + reachable: usize, + stable: usize, + neighbours: usize, + stable_neighbours: usize, + full_mesh: bool, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct StabilityStats { + /// As published in the member registry. + tier: &'static str, + /// Fraction of the observed time (up to a day) with a live peer. + connected_fraction: f64, + observed_secs: u64, +} + +/// `GET /admin/stats`. +pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap) -> Response { + let stats = collect(&state).await; + let accept = headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()); + if prometheus::wanted(accept) { + let body = prometheus::render(&state, &stats).await; + ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response() + } else { + Json(stats).into_response() + } +} + +async fn collect(state: &AppState) -> StatsResponse { + let node = state.stats.snapshot(); + + let (heads, tier) = { + let mut replica = state.doc.write().await; + let heads = replica.doc.get_heads().len(); + (heads, members::tier(&replica.doc, state.local_id())) + }; + let saved = state.stats.saved_size(); + + let (connected_fraction, observed) = stability::measured(state); + + let mut incidents = Incidents::default(); + let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect(); + for peer_state in peers { + incidents.add(&peer_state.lock().await.incidents); + } + + StatsResponse { + public_key: state.local_id().to_owned(), + alias: state.alias.clone(), + uptime_secs: node.uptime.as_secs(), + ready: state.is_ready().await, + epoch: state.local_epoch().into(), + document: DocumentStats { + heads, + max_heads: node.max_heads.max(heads), + saved_bytes: saved.map(|s| s.bytes), + saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()), + saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64), + }, + catch_up: CatchUpStats { + sources: node.catch_up_sources, + done_after_secs: node.caught_up.map(|c| c.after.as_secs_f64()), + sync_bytes: node.caught_up.map(|c| c.sync_bytes), + }, + topology: TopologyStats { + k: state.neighbours_k, + view: topology::summary(state).map(|s| TopologyView { + members: s.members, + reachable: s.reachable, + stable: s.stable, + neighbours: s.neighbours, + stable_neighbours: s.stable_neighbours, + full_mesh: s.full_mesh, + }), + }, + stability: StabilityStats { + tier: tier.as_str(), + connected_fraction, + observed_secs: observed.as_secs(), + }, + stale_hello_replies: node.stale_hello_replies, + traffic: state.traffic.total(), + incidents, + propagation: PropagationStats { + count: node.propagation.count, + sum_ms: node.propagation.sum_ms, + ahead: node.propagation.ahead, + buckets: node + .propagation + .buckets() + .map(|(le_ms, count)| PropagationBucket { le_ms, count }) + .collect(), + }, + } +} diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs new file mode 100644 index 0000000..d29207c --- /dev/null +++ b/src/api/stats/prometheus.rs @@ -0,0 +1,521 @@ +//! /admin/stats in the Prometheus text exposition format (version 0.0.4), for a +//! scraper asking for it. Adds per-peer series (the figures /peers shows), labelled +//! with the peer's public key. + +use std::fmt::Write; + +use tokio::time::Instant; + +use super::StatsResponse; +use crate::state::{AppState, Incidents, Liveness}; +use crate::transport::traffic::{ByKind, Counter, PeerTraffic}; + +pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8"; + +/// Whether the client prefers this format to JSON: Prometheus asks for OpenMetrics or +/// plain text; browsers and curl don't name either. +pub(super) fn wanted(accept: Option<&str>) -> bool { + accept.is_some_and(|accept| { + accept.split(',').any(|item| { + let media = item.split(';').next().unwrap_or("").trim(); + media == "application/openmetrics-text" || media == "text/plain" + }) + }) +} + +#[derive(Default)] +struct Out(String); + +impl Out { + fn family(&mut self, name: &str, kind: &str, help: &str) { + let _ = writeln!(self.0, "# HELP continuum_{name} {help}"); + let _ = writeln!(self.0, "# TYPE continuum_{name} {kind}"); + } + + fn sample(&mut self, name: &str, labels: &[(&str, &str)], value: f64) { + let _ = write!(self.0, "continuum_{name}"); + if !labels.is_empty() { + self.0.push('{'); + for (i, (key, value)) in labels.iter().enumerate() { + if i > 0 { + self.0.push(','); + } + let _ = write!(self.0, "{key}=\"{}\"", escape(value)); + } + self.0.push('}'); + } + let _ = writeln!(self.0, " {value}"); + } + + /// A family with one unlabelled sample. + fn single(&mut self, name: &str, kind: &str, help: &str, value: f64) { + self.family(name, kind, help); + self.sample(name, &[], value); + } +} + +fn escape(value: &str) -> String { + value + .replace('\\', "\\\\") + .replace('"', "\\\"") + .replace('\n', "\\n") +} + +fn flag(value: bool) -> f64 { + if value { 1.0 } else { 0.0 } +} + +/// What we show per peer. +struct Peer { + key: String, + alias: Option<String>, + neighbour: bool, + dormant: bool, + last_seen_secs: Option<f64>, + traffic: PeerTraffic, + incidents: Incidents, + rtt_secs: Option<f64>, + throughput: Option<f64>, +} + +async fn peers(state: &AppState) -> Vec<Peer> { + let now = Instant::now(); + let snapshot: Vec<_> = state + .peer_states + .read() + .await + .iter() + .map(|(key, peer_state)| (key.clone(), peer_state.clone())) + .collect(); + let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc); + let mut peers = Vec::with_capacity(snapshot.len()); + for (key, peer_state) in snapshot { + let neighbour = crate::swarm::topology::is_neighbour(state, &key); + let rtt_secs = state + .lock_rtt() + .effective(&key, now, state.timing.rtt_fresh_for()) + .map(|rtt| rtt.as_secs_f64()); + let throughput = state + .throughput + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(&key) + .map(|estimate| estimate.bytes_per_sec); + let traffic = state.traffic.peer(&key).unwrap_or_default(); + let peer_state = peer_state.lock().await; + peers.push(Peer { + neighbour, + dormant: matches!(peer_state.liveness, Liveness::Dormant { .. }), + last_seen_secs: peer_state + .last_seen + .map(|t| now.saturating_duration_since(t).as_secs_f64()), + traffic, + incidents: peer_state.incidents, + rtt_secs, + throughput, + alias: aliases.remove(&key), + key, + }); + } + peers.sort_by(|a, b| a.key.cmp(&b.key)); + peers +} + +/// Every (direction, kind, counter) of `traffic`. +fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Counter)> { + let by_kind = |direction: &'static str, by: &ByKind| { + [ + (direction, "hello", by.hello), + (direction, "hello_reply", by.hello_reply), + (direction, "sync", by.sync), + (direction, "ping", by.ping), + (direction, "pong", by.pong), + ] + }; + let mut series = Vec::new(); + series.extend(by_kind("sent", &traffic.sent)); + series.extend(by_kind("received", &traffic.received)); + series.extend(by_kind("refused", &traffic.refused)); + series +} + +pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { + let mut out = Out::default(); + + // Names for joining (`* on(peer) group_left(alias) continuum_peer_info`), kept out + // of the other series so a rename doesn't start new ones. + out.family( + "info", + "gauge", + "This node's public key and alias (always 1).", + ); + out.sample( + "info", + &[ + ("public_key", &stats.public_key), + ("alias", stats.alias.as_deref().unwrap_or("")), + ], + 1.0, + ); + + out.single( + "uptime_seconds", + "gauge", + "Seconds since this node started.", + stats.uptime_secs as f64, + ); + out.single( + "ready", + "gauge", + "1 once the node holds the swarm's document.", + flag(stats.ready), + ); + out.single( + "epoch_generation", + "gauge", + "Generation of the local epoch (remakes so far).", + state.local_epoch().generation as f64, + ); + + let document = &stats.document; + out.single( + "document_heads", + "gauge", + "Heads of the local document.", + document.heads as f64, + ); + out.single( + "document_max_heads", + "gauge", + "Most heads after applying a sync message, since start.", + document.max_heads as f64, + ); + if let Some(bytes) = document.saved_bytes { + out.single( + "document_saved_bytes", + "gauge", + "Size of the document as saved (compressed); measured in the background.", + bytes as f64, + ); + } + + let catch_up = &stats.catch_up; + out.single( + "catch_up_sources_total", + "counter", + "Catch-up sources chosen (more than one: switched).", + catch_up.sources as f64, + ); + if let Some(secs) = catch_up.done_after_secs { + out.single( + "catch_up_duration_seconds", + "gauge", + "Seconds from start until the initial sync was done.", + secs, + ); + } + if let Some(bytes) = catch_up.sync_bytes { + out.single( + "catch_up_sync_bytes", + "gauge", + "Sync payload received until the initial sync was done.", + bytes as f64, + ); + } + + let topology = &stats.topology; + out.single( + "neighbours_target", + "gauge", + "Target number of neighbours (k).", + topology.k as f64, + ); + if let Some(view) = &topology.view { + out.single( + "members", + "gauge", + "Registered members, this node included.", + view.members as f64, + ); + out.single( + "reachable_members", + "gauge", + "Members other than this node not judged unreachable.", + view.reachable as f64, + ); + out.single( + "stable_members", + "gauge", + "Members with the stable tier.", + view.stable as f64, + ); + out.single( + "neighbours", + "gauge", + "Current neighbours.", + view.neighbours as f64, + ); + out.single( + "stable_neighbours", + "gauge", + "Current neighbours with the stable tier.", + view.stable_neighbours as f64, + ); + out.single( + "full_mesh", + "gauge", + "1 while every reachable member is a neighbour.", + flag(view.full_mesh), + ); + } + + let stability = &stats.stability; + out.single( + "stable", + "gauge", + "1 while this node's published tier is stable.", + flag(stability.tier == "stable"), + ); + out.single( + "connected_ratio", + "gauge", + "Fraction of the observed time (up to a day) with a live peer.", + stability.connected_fraction, + ); + out.single( + "connectivity_observed_seconds", + "gauge", + "Time the connected ratio is based on.", + stability.observed_secs as f64, + ); + + out.single( + "stale_hello_replies_total", + "counter", + "HELLO_REPLYs to an earlier run of this node, ignored.", + stats.stale_hello_replies as f64, + ); + + let propagation = &stats.propagation; + out.family( + "propagation_delay_seconds", + "histogram", + "Time from a change's commit elsewhere until applied here (includes clock skew).", + ); + let mut cumulative = 0; + for bucket in &propagation.buckets { + cumulative += bucket.count; + let le = match bucket.le_ms { + Some(ms) => (ms as f64 / 1000.0).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "propagation_delay_seconds_bucket", + &[("le", &le)], + cumulative as f64, + ); + } + out.sample( + "propagation_delay_seconds_sum", + &[], + propagation.sum_ms as f64 / 1000.0, + ); + out.sample( + "propagation_delay_seconds_count", + &[], + propagation.count as f64, + ); + out.single( + "propagation_clock_ahead_total", + "counter", + "Changes stamped later than this node's clock on arrival.", + propagation.ahead as f64, + ); + + // Node totals, including peers no longer listed. + let totals = traffic_series(&stats.traffic); + out.family( + "messages_total", + "counter", + "Messages by direction and kind, all peers.", + ); + for (direction, kind, counter) in &totals { + out.sample( + "messages_total", + &[("direction", direction), ("kind", kind)], + counter.messages as f64, + ); + } + out.family( + "payload_bytes_total", + "counter", + "Payload bytes by direction and kind, all peers.", + ); + for (direction, kind, counter) in &totals { + out.sample( + "payload_bytes_total", + &[("direction", direction), ("kind", kind)], + counter.bytes as f64, + ); + } + + let peers = peers(state).await; + + out.family( + "peer_info", + "gauge", + "A peer's alias from the member list (always 1).", + ); + for peer in &peers { + out.sample( + "peer_info", + &[ + ("peer", &peer.key), + ("alias", peer.alias.as_deref().unwrap_or("")), + ], + 1.0, + ); + } + + out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour."); + for peer in &peers { + out.sample( + "peer_neighbour", + &[("peer", &peer.key)], + flag(peer.neighbour), + ); + } + out.family("peer_dormant", "gauge", "1 while the peer is dormant."); + for peer in &peers { + out.sample("peer_dormant", &[("peer", &peer.key)], flag(peer.dormant)); + } + out.family( + "peer_last_seen_seconds", + "gauge", + "Seconds since the peer's last message.", + ); + for peer in &peers { + if let Some(secs) = peer.last_seen_secs { + out.sample("peer_last_seen_seconds", &[("peer", &peer.key)], secs); + } + } + out.family( + "peer_rtt_seconds", + "gauge", + "Effective round-trip time (smoothed, raised by unanswered requests).", + ); + for peer in &peers { + if let Some(rtt) = peer.rtt_secs { + out.sample("peer_rtt_seconds", &[("peer", &peer.key)], rtt); + } + } + out.family( + "peer_sync_throughput_bytes_per_second", + "gauge", + "Smoothed sync throughput in large exchanges.", + ); + for peer in &peers { + if let Some(throughput) = peer.throughput { + out.sample( + "peer_sync_throughput_bytes_per_second", + &[("peer", &peer.key)], + throughput, + ); + } + } + + out.family( + "peer_messages_total", + "counter", + "Messages by direction and kind.", + ); + for peer in &peers { + for (direction, kind, counter) in traffic_series(&peer.traffic) { + out.sample( + "peer_messages_total", + &[ + ("peer", &peer.key), + ("direction", direction), + ("kind", kind), + ], + counter.messages as f64, + ); + } + } + out.family( + "peer_payload_bytes_total", + "counter", + "Payload bytes by direction and kind.", + ); + for peer in &peers { + for (direction, kind, counter) in traffic_series(&peer.traffic) { + out.sample( + "peer_payload_bytes_total", + &[ + ("peer", &peer.key), + ("direction", direction), + ("kind", kind), + ], + counter.bytes as f64, + ); + } + } + + let incidents: [(&str, &str, fn(&Incidents) -> f64); 4] = [ + ( + "peer_reply_timeouts_total", + "Sync messages unanswered until the reply timeout.", + |i| i.reply_timeouts as f64, + ), + ( + "peer_heals_total", + "Re-handshakes because documents differed while sync was idle.", + |i| i.heals as f64, + ), + ( + "peer_dormancies_total", + "Times the peer went dormant.", + |i| i.dormancies as f64, + ), + ( + "peer_dormant_seconds_total", + "Time spent dormant, up to the last return.", + |i| i.dormant_ms as f64 / 1000.0, + ), + ]; + for (name, help, value) in incidents { + out.family(name, "counter", help); + for peer in &peers { + out.sample(name, &[("peer", &peer.key)], value(&peer.incidents)); + } + } + + out.0 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn negotiation() { + let prometheus = "application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1"; + assert!(wanted(Some(prometheus))); + assert!(wanted(Some("text/plain"))); + assert!(!wanted(Some("*/*"))); + assert!(!wanted(Some("application/json"))); + assert!(!wanted(Some("text/html,application/xhtml+xml,*/*;q=0.8"))); + assert!(!wanted(None)); + } + + #[test] + fn sample_lines() { + let mut out = Out::default(); + out.single("x_total", "counter", "An x.", 3.0); + out.sample("y", &[("peer", "a\"b"), ("kind", "sync")], 0.25); + assert_eq!( + out.0, + "# HELP continuum_x_total An x.\n# TYPE continuum_x_total counter\ncontinuum_x_total 3\ncontinuum_y{peer=\"a\\\"b\",kind=\"sync\"} 0.25\n" + ); + } +} diff --git a/src/api/store.rs b/src/api/store.rs index 409a530..d6ff306 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -7,8 +7,8 @@ use axum::{ body::{Body, Bytes}, extract::{Path, Query, State}, http::{ - header::{ACCEPT, CONTENT_TYPE}, HeaderMap, HeaderValue, StatusCode, + header::{ACCEPT, CONTENT_TYPE}, }, response::{IntoResponse, Response}, }; |