statserv: persist per-channel activity across restart too
All checks were successful
CI / check (push) Successful in 3m49s
All checks were successful
CI / check (push) Successful in 3m49s
The earlier StatsSet fix persisted only the shared namespaced counters. The per-channel BOTSTATS view (lines seen + top talkers, from /statserv #chan) lived on the LiveChannel struct, which is rebuilt from each burst — so it reset to zero on restart. Move it to a name-keyed Network.chan_activity map (independent of live membership, seedable before the burst) and carry it in the StatsSet snapshot (new serde-default `channels` field) alongside the counters. Reworded the notice from 'seen this session' now that it persists.
This commit is contained in:
parent
8a9b6a37ca
commit
1f6a11dba2
8 changed files with 91 additions and 26 deletions
|
|
@ -10,7 +10,7 @@ pub fn handle(me: &str, from: &Sender, chan: &str, ctx: &mut ServiceCtx, net: &d
|
||||||
ctx.notice(me, from.uid, format!("No activity recorded for \x02{chan}\x02 yet."));
|
ctx.notice(me, from.uid, format!("No activity recorded for \x02{chan}\x02 yet."));
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
ctx.notice(me, from.uid, format!("\x02{chan}\x02 — \x02{lines}\x02 line(s) seen this session."));
|
ctx.notice(me, from.uid, format!("\x02{chan}\x02 — \x02{lines}\x02 line(s) seen."));
|
||||||
if top.is_empty() {
|
if top.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -136,9 +136,15 @@ pub enum Event {
|
||||||
// by minting a fake one on `sid`), so Local scope but still persisted.
|
// by minting a fake one on `sid`), so Local scope but still persisted.
|
||||||
JupeAdded { name: String, sid: String, reason: String },
|
JupeAdded { name: String, sid: String, reason: String },
|
||||||
JupeRemoved { name: String },
|
JupeRemoved { name: String },
|
||||||
// A full snapshot of the shared stat counters (StatServ / gRPC Stats),
|
// A full snapshot of the stats (StatServ / gRPC Stats), written periodically
|
||||||
// written periodically and on shutdown so they survive a restart.
|
// and on shutdown so they survive a restart: shared namespaced counters, plus
|
||||||
StatsSet { counters: Vec<(String, u64)> },
|
// per-channel activity (channel, line count, top talkers). `channels` defaults
|
||||||
|
// for log entries written before it existed.
|
||||||
|
StatsSet {
|
||||||
|
counters: Vec<(String, u64)>,
|
||||||
|
#[serde(default)]
|
||||||
|
channels: super::ChanStats,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
// Whether an event replicates across the federation. Account identity is Global
|
// Whether an event replicates across the federation. Account identity is Global
|
||||||
|
|
@ -627,9 +633,10 @@ pub(crate) fn apply(accounts: &mut HashMap<String, Account>, channels: &mut Hash
|
||||||
Event::JupeRemoved { name } => {
|
Event::JupeRemoved { name } => {
|
||||||
net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name));
|
net.jupes.retain(|j| !j.name.eq_ignore_ascii_case(&name));
|
||||||
}
|
}
|
||||||
Event::StatsSet { counters } => {
|
Event::StatsSet { counters, channels } => {
|
||||||
// A full snapshot: replace, so replaying the newest wins.
|
// A full snapshot: replace, so replaying the newest wins.
|
||||||
net.stats = counters.into_iter().collect();
|
net.stats = counters.into_iter().collect();
|
||||||
|
net.chan_stats = channels;
|
||||||
}
|
}
|
||||||
Event::AccountExpiryWarned { account } => {
|
Event::AccountExpiryWarned { account } => {
|
||||||
if let Some(a) = accounts.get_mut(&key(&account)) {
|
if let Some(a) = accounts.get_mut(&key(&account)) {
|
||||||
|
|
|
||||||
|
|
@ -282,6 +282,9 @@ pub struct Report {
|
||||||
// The network-wide lists that aren't accounts/channels/grouped/bots: the network
|
// The network-wide lists that aren't accounts/channels/grouped/bots: the network
|
||||||
// bans and the news items. Bundled so `apply` threads one parameter for them
|
// bans and the news items. Bundled so `apply` threads one parameter for them
|
||||||
// (and stays within its argument budget as more are added).
|
// (and stays within its argument budget as more are added).
|
||||||
|
// Persisted per-channel BOTSTATS activity: (channel, line count, top talkers).
|
||||||
|
pub type ChanStats = Vec<(String, u64, Vec<(String, u64)>)>;
|
||||||
|
|
||||||
#[derive(Debug, Clone, Default)]
|
#[derive(Debug, Clone, Default)]
|
||||||
pub struct NetData {
|
pub struct NetData {
|
||||||
pub akills: Vec<Akill>,
|
pub akills: Vec<Akill>,
|
||||||
|
|
@ -310,6 +313,8 @@ pub struct NetData {
|
||||||
// Persisted shared stat counters (StatServ / gRPC Stats), snapshotted so they
|
// Persisted shared stat counters (StatServ / gRPC Stats), snapshotted so they
|
||||||
// survive a restart (live gauges are re-derived, not stored here).
|
// survive a restart (live gauges are re-derived, not stored here).
|
||||||
pub stats: std::collections::BTreeMap<String, u64>,
|
pub stats: std::collections::BTreeMap<String, u64>,
|
||||||
|
// Persisted per-channel BOTSTATS activity: (channel, lines, top talkers).
|
||||||
|
pub chan_stats: ChanStats,
|
||||||
}
|
}
|
||||||
|
|
||||||
// A session-limit exception: an IP-mask glob and the session allowance for IPs
|
// A session-limit exception: an IP-mask glob and the session allowance for IPs
|
||||||
|
|
@ -1168,8 +1173,11 @@ impl Db {
|
||||||
if self.net.default_bot.is_some() {
|
if self.net.default_bot.is_some() {
|
||||||
snapshot.push(Event::DefaultBotSet { bot: self.net.default_bot.clone() });
|
snapshot.push(Event::DefaultBotSet { bot: self.net.default_bot.clone() });
|
||||||
}
|
}
|
||||||
if !self.net.stats.is_empty() {
|
if !self.net.stats.is_empty() || !self.net.chan_stats.is_empty() {
|
||||||
snapshot.push(Event::StatsSet { counters: self.net.stats.iter().map(|(k, v)| (k.clone(), *v)).collect() });
|
snapshot.push(Event::StatsSet {
|
||||||
|
counters: self.net.stats.iter().map(|(k, v)| (k.clone(), *v)).collect(),
|
||||||
|
channels: self.net.chan_stats.clone(),
|
||||||
|
});
|
||||||
}
|
}
|
||||||
for host in &self.host_cfg.offers {
|
for host in &self.host_cfg.offers {
|
||||||
snapshot.push(Event::VhostOfferAdded { host: host.clone() });
|
snapshot.push(Event::VhostOfferAdded { host: host.clone() });
|
||||||
|
|
|
||||||
|
|
@ -250,10 +250,24 @@ impl Db {
|
||||||
self.net.stats.clone()
|
self.net.stats.clone()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Snapshot the live stat counters to the log so they survive a restart.
|
/// The persisted per-channel activity, to seed BOTSTATS on startup.
|
||||||
pub fn persist_stats(&mut self, counters: &std::collections::BTreeMap<String, u64>) -> std::io::Result<()> {
|
pub fn persisted_chan_stats(&self) -> ChanStats {
|
||||||
self.log.append(Event::StatsSet { counters: counters.iter().map(|(k, v)| (k.clone(), *v)).collect() })?;
|
self.net.chan_stats.clone()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Snapshot the live stats (shared counters + per-channel activity) to the log
|
||||||
|
/// so they survive a restart.
|
||||||
|
pub fn persist_stats(
|
||||||
|
&mut self,
|
||||||
|
counters: &std::collections::BTreeMap<String, u64>,
|
||||||
|
chan_stats: ChanStats,
|
||||||
|
) -> std::io::Result<()> {
|
||||||
|
self.log.append(Event::StatsSet {
|
||||||
|
counters: counters.iter().map(|(k, v)| (k.clone(), *v)).collect(),
|
||||||
|
channels: chan_stats.clone(),
|
||||||
|
})?;
|
||||||
self.net.stats = counters.clone();
|
self.net.stats = counters.clone();
|
||||||
|
self.net.chan_stats = chan_stats;
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -532,13 +532,15 @@
|
||||||
let p = tmp("statspersist");
|
let p = tmp("statspersist");
|
||||||
let counters: std::collections::BTreeMap<String, u64> =
|
let counters: std::collections::BTreeMap<String, u64> =
|
||||||
[("chanserv.register".to_string(), 7u64), ("nickserv.identify".to_string(), 42u64)].into_iter().collect();
|
[("chanserv.register".to_string(), 7u64), ("nickserv.identify".to_string(), 42u64)].into_iter().collect();
|
||||||
|
let chans = vec![("#devs".to_string(), 6u64, vec![("reverse".to_string(), 5u64), ("Keiko".to_string(), 1u64)])];
|
||||||
{
|
{
|
||||||
let mut db = Db::open(&p, "N1");
|
let mut db = Db::open(&p, "N1");
|
||||||
db.persist_stats(&counters).unwrap();
|
db.persist_stats(&counters, chans.clone()).unwrap();
|
||||||
assert_eq!(db.persisted_stats(), counters, "stored while live");
|
assert_eq!(db.persisted_stats(), counters, "stored while live");
|
||||||
}
|
}
|
||||||
let db = Db::open(&p, "N1");
|
let db = Db::open(&p, "N1");
|
||||||
assert_eq!(db.persisted_stats(), counters, "counters replay from the log after a restart");
|
assert_eq!(db.persisted_stats(), counters, "counters replay from the log after a restart");
|
||||||
|
assert_eq!(db.persisted_chan_stats(), chans, "per-channel activity replays too");
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fold parity: the live state after a broad sequence of writes must equal the
|
// Fold parity: the live state after a broad sequence of writes must equal the
|
||||||
|
|
|
||||||
|
|
@ -213,8 +213,9 @@ impl Engine {
|
||||||
(!topics.is_empty()).then(|| (s.nick().to_string(), blurb, topics))
|
(!topics.is_empty()).then(|| (s.nick().to_string(), blurb, topics))
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
// Restore the persisted stat counters so they survive a restart.
|
// Restore persisted stats so they survive a restart.
|
||||||
network.seed_stats(db.persisted_stats());
|
network.seed_stats(db.persisted_stats());
|
||||||
|
network.seed_chan_activity(db.persisted_chan_stats());
|
||||||
Self {
|
Self {
|
||||||
services,
|
services,
|
||||||
network,
|
network,
|
||||||
|
|
@ -275,10 +276,11 @@ impl Engine {
|
||||||
// Only the real counters are persisted; the live gauges above are re-derived.
|
// Only the real counters are persisted; the live gauges above are re-derived.
|
||||||
pub fn persist_stats(&mut self) {
|
pub fn persist_stats(&mut self) {
|
||||||
let counters = self.network.stat_counters().clone();
|
let counters = self.network.stat_counters().clone();
|
||||||
if counters.is_empty() {
|
let chan_stats = self.network.chan_activity_snapshot();
|
||||||
|
if counters.is_empty() && chan_stats.is_empty() {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if let Err(err) = self.db.persist_stats(&counters) {
|
if let Err(err) = self.db.persist_stats(&counters, chan_stats) {
|
||||||
tracing::warn!(%err, "failed to persist stat counters");
|
tracing::warn!(%err, "failed to persist stat counters");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -23,9 +23,12 @@ pub struct Network {
|
||||||
// Downstream server tree: SID -> the SID that introduced it. Lets a hub's
|
// Downstream server tree: SID -> the SID that introduced it. Lets a hub's
|
||||||
// SQUIT cascade to every server (and user) behind it. Rebuilt each link.
|
// SQUIT cascade to every server (and user) behind it. Rebuilt each link.
|
||||||
servers: HashMap<String, String>,
|
servers: HashMap<String, String>,
|
||||||
// Shared, namespaced stat counters any service contributes to (ephemeral,
|
// Shared, namespaced stat counters any service contributes to (persisted via
|
||||||
// ordered for a stable snapshot). Read by StatServ and the gRPC Stats API.
|
// StatsSet). Read by StatServ and the gRPC Stats API.
|
||||||
stats: BTreeMap<String, u64>,
|
stats: BTreeMap<String, u64>,
|
||||||
|
// Per-channel BOTSTATS activity (lines + top talkers), name-keyed and
|
||||||
|
// persisted alongside `stats` so it survives a restart.
|
||||||
|
chan_activity: HashMap<String, ChanActivity>,
|
||||||
// Live session count per connecting IP, for OperServ session limiting.
|
// Live session count per connecting IP, for OperServ session limiting.
|
||||||
sessions: HashMap<String, u32>,
|
sessions: HashMap<String, u32>,
|
||||||
// Recent moderation/action incidents (bounded ring), for OperServ LOGSEARCH.
|
// Recent moderation/action incidents (bounded ring), for OperServ LOGSEARCH.
|
||||||
|
|
@ -85,7 +88,13 @@ pub struct Channel {
|
||||||
pub ops: HashSet<String>, // uids holding channel-operator status
|
pub ops: HashSet<String>, // uids holding channel-operator status
|
||||||
pub voices: HashSet<String>, // uids holding +v
|
pub voices: HashSet<String>, // uids holding +v
|
||||||
pub key: Option<String>,
|
pub key: Option<String>,
|
||||||
// BOTSTATS: lines seen this session, and per-nick counts (top-talkers). Bounded.
|
}
|
||||||
|
|
||||||
|
// Per-channel BOTSTATS activity: total lines and per-nick counts (top talkers).
|
||||||
|
// Held name-keyed (not on the live Channel, which is rebuilt from each burst) and
|
||||||
|
// persisted, so a channel's history survives a services restart.
|
||||||
|
#[derive(Default, Clone)]
|
||||||
|
pub struct ChanActivity {
|
||||||
pub lines: u64,
|
pub lines: u64,
|
||||||
pub talkers: HashMap<String, u64>,
|
pub talkers: HashMap<String, u64>,
|
||||||
}
|
}
|
||||||
|
|
@ -332,10 +341,10 @@ impl Network {
|
||||||
// is capped so a busy channel can't grow it without bound.
|
// is capped so a busy channel can't grow it without bound.
|
||||||
pub fn record_line(&mut self, channel: &str, nick: &str) {
|
pub fn record_line(&mut self, channel: &str, nick: &str) {
|
||||||
const TALKER_CAP: usize = 512;
|
const TALKER_CAP: usize = 512;
|
||||||
let c = self.channels.entry(lc(channel)).or_default();
|
let a = self.chan_activity.entry(lc(channel)).or_default();
|
||||||
c.lines = c.lines.saturating_add(1);
|
a.lines = a.lines.saturating_add(1);
|
||||||
if c.talkers.len() < TALKER_CAP || c.talkers.contains_key(nick) {
|
if a.talkers.len() < TALKER_CAP || a.talkers.contains_key(nick) {
|
||||||
*c.talkers.entry(nick.to_string()).or_insert(0) += 1;
|
*a.talkers.entry(nick.to_string()).or_insert(0) += 1;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -356,11 +365,34 @@ impl Network {
|
||||||
|
|
||||||
// BOTSTATS view: (total lines, top talkers by count, descending).
|
// BOTSTATS view: (total lines, top talkers by count, descending).
|
||||||
pub fn channel_activity(&self, channel: &str) -> Option<(u64, Vec<(String, u64)>)> {
|
pub fn channel_activity(&self, channel: &str) -> Option<(u64, Vec<(String, u64)>)> {
|
||||||
let c = self.channels.get(&lc(channel))?;
|
let a = self.chan_activity.get(&lc(channel))?;
|
||||||
let mut top: Vec<(String, u64)> = c.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect();
|
let mut top: Vec<(String, u64)> = a.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect();
|
||||||
top.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
|
top.sort_by(|x, y| y.1.cmp(&x.1).then_with(|| x.0.cmp(&y.0)));
|
||||||
top.truncate(10);
|
top.truncate(10);
|
||||||
Some((c.lines, top))
|
Some((a.lines, top))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Seed per-channel activity from persisted state at startup.
|
||||||
|
pub fn seed_chan_activity(&mut self, data: crate::engine::db::ChanStats) {
|
||||||
|
self.chan_activity = data
|
||||||
|
.into_iter()
|
||||||
|
.map(|(c, lines, talkers)| (c, ChanActivity { lines, talkers: talkers.into_iter().collect() }))
|
||||||
|
.collect();
|
||||||
|
}
|
||||||
|
|
||||||
|
// Snapshot per-channel activity for persistence: each active channel's line
|
||||||
|
// count and its top talkers (capped, so a busy channel bounds the event).
|
||||||
|
pub fn chan_activity_snapshot(&self) -> crate::engine::db::ChanStats {
|
||||||
|
self.chan_activity
|
||||||
|
.iter()
|
||||||
|
.filter(|(_, a)| a.lines > 0)
|
||||||
|
.map(|(c, a)| {
|
||||||
|
let mut talkers: Vec<(String, u64)> = a.talkers.iter().map(|(n, v)| (n.clone(), *v)).collect();
|
||||||
|
talkers.sort_by(|x, y| y.1.cmp(&x.1).then_with(|| x.0.cmp(&y.0)));
|
||||||
|
talkers.truncate(50);
|
||||||
|
(c.clone(), a.lines, talkers)
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Uids currently in `channel`.
|
// Uids currently in `channel`.
|
||||||
|
|
|
||||||
|
|
@ -2723,7 +2723,7 @@
|
||||||
|
|
||||||
// Per-channel view.
|
// Per-channel view.
|
||||||
let out = ss(&mut e, "#c");
|
let out = ss(&mut e, "#c");
|
||||||
assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("line(s) seen this session"))), "channel lines: {out:?}");
|
assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("line(s) seen"))), "channel lines: {out:?}");
|
||||||
assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("spammer") && text.contains("3"))), "top talker: {out:?}");
|
assert!(out.iter().any(|a| matches!(a, NetAction::Notice { text, .. } if text.contains("spammer") && text.contains("3"))), "top talker: {out:?}");
|
||||||
// Global view (oper) shows the shared counters.
|
// Global view (oper) shows the shared counters.
|
||||||
let out = ss(&mut e, "SERVER");
|
let out = ss(&mut e, "SERVER");
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue