From d4e2c905f2ba34668b32543bb3643c8006b75e47 Mon Sep 17 00:00:00 2001 From: Jean Date: Mon, 13 Jul 2026 19:13:31 +0000 Subject: [PATCH] Shared stats registry + gRPC Stats API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single namespaced counter map on the engine that every service reports into: ServiceCtx::count(key) is folded in after each command, the engine bumps its own events (botserv.kick/ban/warn/greet/trigger), and a per-service command counter (nickserv.command, chanserv.command, …) is tallied in dispatch. stats_snapshot() merges those with live gauges from the store (accounts/channels/bots/opers total). Exposed over gRPC as a new Stats service (Get -> map), authorized by the same bearer token as Directory/Accounts — so any consumer (a dashboard, Django) reads every module's stats through one pipe. New service is additive to the proto; existing RPCs untouched. --- api/src/lib.rs | 10 +++++++ proto/fedserv.proto | 12 ++++++++ src/engine/mod.rs | 70 +++++++++++++++++++++++++++++++++++++++++---- src/grpc.rs | 27 +++++++++++++++-- 4 files changed, 111 insertions(+), 8 deletions(-) diff --git a/api/src/lib.rs b/api/src/lib.rs index 14f9e49..c871623 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -168,6 +168,10 @@ pub struct Sender<'a> { #[derive(Default)] pub struct ServiceCtx { pub actions: Vec, + // Namespaced stat counters to bump (e.g. "nickserv.register"). The engine + // folds these into the shared registry after the command runs — so every + // service reports its own stats through one shared pipe. + pub stats: Vec, } impl ServiceCtx { @@ -179,6 +183,12 @@ impl ServiceCtx { }); } + // Record a stat counter bump (namespaced, e.g. "chanserv.register"). Folded + // into the engine's shared registry, which the gRPC Stats API exposes. + pub fn count(&mut self, key: impl Into) { + self.stats.push(key.into()); + } + // A channel/target message sourced from one of our pseudo-clients (e.g. a // bot answering a fantasy command, or BotServ SAY). pub fn privmsg(&mut self, from: &str, to: &str, text: impl Into) { diff --git a/proto/fedserv.proto b/proto/fedserv.proto index febd970..c51b9b7 100644 --- a/proto/fedserv.proto +++ b/proto/fedserv.proto @@ -138,3 +138,15 @@ message ForceLogoutReply { Status status = 1; uint32 sessions_cleared = 2; } message GroupNickRequest { string nick = 1; string account = 2; } message UngroupNickRequest { string nick = 1; } + +// Stats API: a flat, namespaced counter map any service contributes to +// (e.g. "nickserv.command", "botserv.kick", "channels.total"). Read-only. +service Stats { + rpc Get(StatsRequest) returns (StatsResponse); +} + +message StatsRequest {} +message StatsResponse { + map counters = 1; + uint64 collected_at = 2; // unix seconds when this snapshot was taken +} diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 086b946..414d69d 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -115,6 +115,10 @@ pub struct Engine { trigger_cache: HashMap, // Kicker bans (times-to-ban) awaiting BANEXPIRE removal. Swept on each event. pending_unbans: Vec, + // Shared, namespaced stat counters any service contributes to (via + // ServiceCtx::count) plus engine-internal events; exposed over the gRPC + // Stats API. Ordered so a snapshot is stable. + stats: std::collections::BTreeMap, } struct CachedBadwords { @@ -172,9 +176,26 @@ impl Engine { badword_cache: HashMap::new(), trigger_cache: HashMap::new(), pending_unbans: Vec::new(), + stats: std::collections::BTreeMap::new(), } } + // Increment a shared stat counter. + pub fn bump(&mut self, key: &str) { + *self.stats.entry(key.to_string()).or_insert(0) += 1; + } + + // The shared counters plus live gauges derived from the store, for the gRPC + // Stats API. Any service's counters ride in here alongside these. + pub fn stats_snapshot(&self) -> std::collections::BTreeMap { + let mut m = self.stats.clone(); + m.insert("accounts.total".to_string(), self.db.accounts().count() as u64); + m.insert("channels.total".to_string(), self.db.channels().count() as u64); + m.insert("bots.total".to_string(), self.db.bots().count() as u64); + m.insert("opers.total".to_string(), self.opers.len() as u64); + m + } + // Wall-clock seconds, overridable in tests so the time-based FLOOD kicker is // deterministic. fn now_secs(&self) -> u64 { @@ -648,6 +669,7 @@ impl Engine { // access + a personal greet, the assigned bot displays it. if let Some(g) = greet { out.push(g); + self.bump("botserv.greet"); } out } @@ -974,14 +996,25 @@ impl Engine { } } } else { - let Self { services, network, db, .. } = self; - for svc in services.iter_mut() { - if to.eq_ignore_ascii_case(svc.uid()) || to.eq_ignore_ascii_case(svc.nick()) { - let args: Vec<&str> = text.split_whitespace().collect(); - svc.on_command(&sender, &args, &mut ctx, network, db); - break; + let mut matched: Option = None; + { + let Self { services, network, db, .. } = self; + for svc in services.iter_mut() { + if to.eq_ignore_ascii_case(svc.uid()) || to.eq_ignore_ascii_case(svc.nick()) { + let args: Vec<&str> = text.split_whitespace().collect(); + svc.on_command(&sender, &args, &mut ctx, network, db); + matched = Some(svc.nick().to_ascii_lowercase()); + break; + } } } + if let Some(nick) = matched { + self.bump(&format!("{nick}.command")); + } + } + // Fold any counters the command recorded into the shared registry. + for key in std::mem::take(&mut ctx.stats) { + self.bump(&key); } // A command may have changed the bot registry (BotServ BOT ADD/DEL). let mut out = ctx.actions; @@ -1091,6 +1124,7 @@ impl Engine { w }; if !already { + self.bump("botserv.warn"); return vec![NetAction::Notice { from: botuid, to: from.to_string(), text: format!("Please mind the channel rules — {reason} Next time you'll be kicked.") }]; } } @@ -1109,6 +1143,7 @@ impl Engine { if let Some(host) = self.network.host_of(from).map(str::to_string) { let mask = format!("*!*@{host}"); out.push(NetAction::ChannelMode { from: botuid.clone(), channel: channel.to_string(), modes: format!("+b {mask}") }); + self.bump("botserv.ban"); if ban_expire > 0 { let at = self.now_secs() + ban_expire as u64; self.pending_unbans.push(PendingUnban { at, from: botuid.clone(), channel: channel.to_string(), mask }); @@ -1117,6 +1152,7 @@ impl Engine { } } out.push(NetAction::Kick { from: botuid, channel: channel.to_string(), uid: from.to_string(), reason: reason.to_string() }); + self.bump("botserv.kick"); out } @@ -1142,6 +1178,7 @@ impl Engine { let response = cached.responses.get(idx)?.clone(); let nick = self.network.nick_of(from).unwrap_or(from); let response = response.replace("$nick", nick); + self.bump("botserv.trigger"); Some(NetAction::Privmsg { from: botuid, to: channel.to_string(), text: response }) } @@ -2476,6 +2513,27 @@ mod tests { assert!(!say(&mut e, "goodbye all").iter().any(|a| matches!(a, NetAction::Privmsg { .. })), "no response on miss"); } + // The shared stats registry gathers per-service command counts, BotServ + // events, and live gauges — the same pipe every module reports through. + #[test] + fn stats_registry_collects_across_services() { + let (mut e, _p) = kicker_fixture("bsstats"); + let bs = |e: &mut Engine, t: &str| e.handle(NetEvent::Privmsg { from: "000AAAAAB".into(), to: "42SAAAAAD".into(), text: t.into() }); + let say = |e: &mut Engine, t: &str| e.handle(NetEvent::Privmsg { from: "000AAAAAS".into(), to: "#c".into(), text: t.into() }); + // The fixture already ran a NickServ IDENTIFY and BotServ BOT ADD/ASSIGN. + bs(&mut e, "KICK #c CAPS ON"); + say(&mut e, "SHOUTING LOUDLY NOW"); // one kick + + let s = e.stats_snapshot(); + assert!(s.get("nickserv.command").copied().unwrap_or(0) >= 1, "nickserv counted: {s:?}"); + assert!(s.get("botserv.command").copied().unwrap_or(0) >= 2, "botserv commands counted"); + assert_eq!(s.get("botserv.kick").copied(), Some(1), "one kick counted"); + // Live gauges derived from the store. + assert_eq!(s.get("channels.total").copied(), Some(1)); + assert_eq!(s.get("bots.total").copied(), Some(1)); + assert!(s.contains_key("accounts.total")); + } + // DONTKICKVOICES exempts voiced users; removing the voice makes them kickable. #[test] fn botserv_dontkickvoices_exempts() { diff --git a/src/grpc.rs b/src/grpc.rs index 9e9657e..aa328cc 100644 --- a/src/grpc.rs +++ b/src/grpc.rs @@ -23,12 +23,14 @@ pub mod pb { use pb::accounts_server::{Accounts, AccountsServer}; use pb::directory_server::{Directory, DirectoryServer}; use pb::replication_event::Kind; +use pb::stats_server::{Stats, StatsServer}; use pb::{ AccountDropped, AccountEmailSet, AccountRecord, AccountRegistered, AccountReply, AccountVerified, AuthenticateReply, AuthenticateRequest, ChannelDescSet, ChannelDropped, ChannelFounderSet, ChannelRecord, ChannelRegistered, ConfirmRequest, DropRequest, ForceLogoutReply, ForceLogoutRequest, GroupNickRequest, NickGrouped, NickUngrouped, RegisterRequest, ReplicationEvent, SetEmailRequest, SetPasswordRequest, - SnapshotRequest, SnapshotResponse, Status as PbStatus, SubscribeRequest, UngroupNickRequest, + SnapshotRequest, SnapshotResponse, Status as PbStatus, StatsRequest, StatsResponse, SubscribeRequest, + UngroupNickRequest, }; type Shared = Arc>; @@ -298,6 +300,25 @@ fn describe(s: AuthorityStatus) -> &'static str { } } +// Read-only stats: the shared counter registry plus live gauges, for dashboards. +struct StatsService { + engine: Shared, + token: String, +} + +#[tonic::async_trait] +impl Stats for StatsService { + async fn get(&self, req: Request) -> Result, Status> { + authorize(&req, &self.token)?; + let counters = self.engine.lock().await.stats_snapshot().into_iter().collect(); + let collected_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + Ok(Response::new(StatsResponse { counters, collected_at })) + } +} + // Start the listener, if configured. Absent [grpc] in config.toml = no-op, same // pattern as gossip being optional. pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender) { @@ -307,6 +328,7 @@ pub async fn run(engine: Shared, cfg: GrpcCfg, outbound: broadcast::Sender return tracing::error!(%e, "grpc TLS setup failed"), }; } - tracing::info!(%addr, tls = has_tls, "grpc directory + accounts API listening"); + tracing::info!(%addr, tls = has_tls, "grpc directory + accounts + stats API listening"); if let Err(e) = server .add_service(DirectoryServer::new(directory_svc)) .add_service(AccountsServer::new(accounts_svc)) + .add_service(StatsServer::new(stats_svc)) .serve(addr) .await {