diff options
| -rw-r--r-- | src/api/peers.rs | 7 | ||||
| -rw-r--r-- | src/api/stats.rs | 4 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 35 | ||||
| -rw-r--r-- | src/main.rs | 21 | ||||
| -rw-r--r-- | src/state.rs | 2 | ||||
| -rw-r--r-- | src/swarm/members.rs | 89 |
6 files changed, 147 insertions, 11 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 0305647..7caaaf3 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -48,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, @@ -87,6 +89,8 @@ pub(crate) struct ThroughputInfo { #[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. @@ -125,6 +129,7 @@ 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 { @@ -155,6 +160,7 @@ pub(crate) async fn list_peers( }; peers.push(PeerStatus { + alias: aliases.remove(&public_key), public_key, validated_swarm: ps.standing.swarm_id().map(str::to_owned), status, @@ -178,6 +184,7 @@ pub(crate) async fn list_peers( 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, diff --git a/src/api/stats.rs b/src/api/stats.rs index b39e054..bc925ff 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -19,6 +19,8 @@ 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, @@ -147,6 +149,8 @@ async fn collect(state: &AppState) -> StatsResponse { } 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(), diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 16869f7..d29207c 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -68,6 +68,7 @@ fn flag(value: bool) -> f64 { /// What we show per peer. struct Peer { key: String, + alias: Option<String>, neighbour: bool, dormant: bool, last_seen_secs: Option<f64>, @@ -86,6 +87,7 @@ async fn peers(state: &AppState) -> Vec<Peer> { .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); @@ -111,6 +113,7 @@ async fn peers(state: &AppState) -> Vec<Peer> { incidents: peer_state.incidents, rtt_secs, throughput, + alias: aliases.remove(&key), key, }); } @@ -139,6 +142,22 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou 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", @@ -342,6 +361,22 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { 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( diff --git a/src/main.rs b/src/main.rs index 7261d6a..65134d8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -97,12 +97,32 @@ struct Args { #[arg(long, value_name = "COUNT", default_value_t = 12)] neighbours: usize, + /// A name for this node, shown next to its public key (/peers, metrics, logs). + /// Published in the swarm's member list; not unique, never an identity. + #[arg(long, value_name = "NAME", value_parser = parse_alias)] + alias: Option<String>, + /// Test hook: start on a non-genesis epoch of this generation (with a fixed dummy /// id), to check that nodes on different epochs refuse to sync. #[arg(long, hide = true, requires = "debug_api", value_name = "GENERATION")] debug_epoch: Option<u64>, } +/// Up to 64 characters, no control characters, not blank. +fn parse_alias(value: &str) -> Result<String, String> { + let alias = value.trim(); + if alias.is_empty() { + return Err("must not be blank".to_owned()); + } + if alias.chars().count() > 64 { + return Err("must be at most 64 characters".to_owned()); + } + if alias.chars().any(char::is_control) { + return Err("must not contain control characters".to_owned()); + } + Ok(alias.to_owned()) +} + #[tokio::main] async fn main() -> Result<()> { logging::init(); @@ -201,6 +221,7 @@ async fn main() -> Result<()> { swarm_key, run_id: uuid::Uuid::new_v4().into_bytes(), neighbours_k: args.neighbours.max(2), + alias: args.alias, neighbours: Default::default(), pulling: Default::default(), rtt: Default::default(), diff --git a/src/state.rs b/src/state.rs index d8f0bbf..7517b72 100644 --- a/src/state.rs +++ b/src/state.rs @@ -593,6 +593,8 @@ pub(crate) struct AppState { /// Target number of neighbours (`--neighbours`); up to this many + 1 members the /// swarm is a full mesh. pub(crate) neighbours_k: usize, + /// Our alias (`--alias`), published in the member list (`swarm::members`). + pub(crate) alias: Option<String>, /// Our own connectivity record and stability tier (`swarm::stability`). pub(crate) stability: Arc<std::sync::Mutex<crate::swarm::stability::Stability>>, /// Round-trip times to peers (`swarm::rtt`). diff --git a/src/swarm/members.rs b/src/swarm/members.rs index e64dd20..e02dba4 100644 --- a/src/swarm/members.rs +++ b/src/swarm/members.rs @@ -7,6 +7,12 @@ //! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry //! together with writing the forget record; a forgotten node that turns out to be alive //! cancels the record and registers again. +//! +//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name +//! for people reading /peers, metrics and logs. Not unique and never an identity: the +//! public key stays that. + +use std::collections::HashMap; use automerge::AutoCommit; use serde_json::{Value as JsonValue, json}; @@ -69,27 +75,88 @@ pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), store::write_path(doc, &tier_path(pk), &json!(tier.as_str())) } -/// Registers this node if it is ready and its entry is missing. +fn alias_path(pk: &str) -> Vec<String> { + let mut path = member_path(pk); + path.push("alias".to_owned()); + path +} + +/// `pk`'s alias, if it gave itself one. +pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> { + match store::read_path(doc, &alias_path(pk)) { + Ok(Some(JsonValue::String(alias))) => Some(alias), + _ => None, + } +} + +/// The aliases of all members that have one. +pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> { + members(doc) + .into_iter() + .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias))) + .collect() +} + +fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> { + match alias { + Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)), + None => match store::delete_path(doc, &alias_path(pk), false) { + Err(StoreError::NotFound) => Ok(()), + result => result, + }, + } +} + +/// Registers this node if it is ready and its entry is missing, and keeps its alias +/// as configured (set, changed or removed after a restart with another `--alias`). pub(crate) async fn ensure_registered(state: &AppState) { if !state.is_ready().await { return; } - let own = member_path(state.local_id()); - let registered = store::read_path(&state.doc.read().await.doc, &own) - .ok() - .flatten() - .is_some(); - if registered { + let local = state.local_id().to_owned(); + let own = member_path(&local); + let (registered, recorded_alias) = { + let replica = state.doc.read().await; + let registered = store::read_path(&replica.doc, &own) + .ok() + .flatten() + .is_some(); + (registered, alias(&replica.doc, &local)) + }; + let wanted = state.alias.clone(); + + if !registered { + let mut entry = json!({ "since": unix_millis() }); + if let Some(alias) = &wanted { + entry["alias"] = json!(alias); + } + match state + .mutate_doc(move |doc| store::write_path(doc, &own, &entry)) + .await + { + Ok(()) => info!( + alias = wanted.as_deref(), + "Registered in the swarm's member list" + ), + Err(error) => warn!(error = %error, "Failed to register in the member list"), + } return; } - let entry = json!({ "since": unix_millis() }); + if recorded_alias == wanted { + return; + } + let alias = wanted.clone(); match state - .mutate_doc(move |doc| store::write_path(doc, &own, &entry)) + .mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref())) .await { - Ok(()) => info!("Registered in the swarm's member list"), - Err(error) => warn!(error = %error, "Failed to register in the member list"), + Ok(()) => info!( + alias = wanted.as_deref(), + previous = recorded_alias.as_deref(), + "Alias updated in the member list" + ), + Err(error) => warn!(error = %error, "Failed to record our alias"), } } |