From 4f681a97fb8a494b7e83b0fd0cb9aeb0370a819f Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 19:37:58 +0100 Subject: [PATCH 1/5] fix(logging): write dipper's logs without colour codes Dipper wrote terminal colour codes into its logs even when they went to a log store, which split fields like the event tag so a search for them found nothing. Logs are now plain text. --- bin/dipper-service/src/main.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index e6a243f1..788047e7 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -61,9 +61,10 @@ const STOP_STEP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5) reason = "predates this lint; fix when next touched" )] pub async fn main() -> anyhow::Result<()> { - // Set up logging + // Set up logging. Plain text, with no colour codes, so log stores can search it. tracing_subscriber::fmt() .with_env_filter(EnvFilter::from_default_env()) + .with_ansi(false) .init(); // Load the configuration From 1ba08b10e02ca7ac237db15e95139f2d0c284f7d Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 19:41:41 +0100 Subject: [PATCH 2/5] feat(alerts): post the log lines an operator must act on to Slack Dipper's tagged warnings and errors, such as a cancel that keeps failing, were only in its logs. With a Slack webhook in the config, each listed tag is now posted, at most once per tag every 15 minutes with a count of the rest, without ever holding up the code that logged it. --- bin/dipper-service/src/alerts.rs | 396 +++++++++++++++++++++++++++++++ bin/dipper-service/src/config.rs | 52 ++++ bin/dipper-service/src/main.rs | 26 +- k8s/configmap-example.yaml | 5 + 4 files changed, 470 insertions(+), 9 deletions(-) create mode 100644 bin/dipper-service/src/alerts.rs diff --git a/bin/dipper-service/src/alerts.rs b/bin/dipper-service/src/alerts.rs new file mode 100644 index 00000000..3e0be77c --- /dev/null +++ b/bin/dipper-service/src/alerts.rs @@ -0,0 +1,396 @@ +//! Sends the log lines an operator has to act on to Slack. A line is picked out by its `event` +//! tag, so the code that logs it needs no change. Posting happens in the background, at most once +//! per event in each throttle window, and never holds up or fails the code that logged it. + +use std::{ + collections::{HashMap, HashSet}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; + +use tokio::{sync::mpsc, time::Instant}; +use tracing::{ + Event, Level, Subscriber, + field::{Field, Visit}, +}; +use tracing_subscriber::{Layer, layer::Context}; +use url::Url; + +use crate::config::AlertsConfig; + +/// Alerts that can wait to be posted. More are dropped and counted, since a burst that size +/// is mostly held back by the throttle anyway. +const QUEUE: usize = 64; + +/// How long a single post to Slack may take. +const POST_TIMEOUT: Duration = Duration::from_secs(10); + +/// How often to check for events whose window has ended with alerts held back, to post a count. +const HELD_BACK_CHECK: Duration = Duration::from_secs(60); + +/// A tagged log line to alert on. +#[derive(Debug, Clone, PartialEq, Eq)] +struct Alert { + event: String, + level: Level, + message: String, + fields: String, +} + +/// Picks the tagged log lines out of dipper's logging and queues them for posting. +pub struct AlertLayer { + events: HashSet, + queue: mpsc::Sender, + dropped: Arc, +} + +/// The alert layer for dipper's logging, with its poster running in the background, or `None` +/// when no Slack webhook is configured. +pub fn layer(config: &AlertsConfig) -> Option { + let url = config.slack_webhook_url.as_ref()?.as_ref().clone(); + let (queue, alerts) = mpsc::channel(QUEUE); + let dropped = Arc::new(AtomicU64::new(0)); + tokio::spawn(post_alerts( + alerts, + url, + config.throttle, + Arc::clone(&dropped), + )); + Some(AlertLayer { + events: config.events.iter().cloned().collect(), + queue, + dropped, + }) +} + +impl Layer for AlertLayer { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + let mut line = LogLine::default(); + event.record(&mut line); + let Some(tag) = line.event.filter(|tag| self.events.contains(tag)) else { + return; + }; + let alert = Alert { + event: tag, + level: *event.metadata().level(), + message: line.message, + fields: line.fields, + }; + // Waiting here would hold up the code that logged it, so a full queue drops the alert; + // the poster reports how many. + if self.queue.try_send(alert).is_err() { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } +} + +/// A log line's `event` tag, message and other fields. +#[derive(Default)] +struct LogLine { + event: Option, + message: String, + fields: String, +} + +impl LogLine { + fn record(&mut self, field: &Field, value: String) { + match field.name() { + "event" => self.event = Some(value), + "message" => self.message = value, + name => { + if !self.fields.is_empty() { + self.fields.push(' '); + } + self.fields.push_str(&format!("{name}={value}")); + } + } + } +} + +impl Visit for LogLine { + fn record_str(&mut self, field: &Field, value: &str) { + self.record(field, value.to_owned()); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.record(field, format!("{value:?}")); + } +} + +/// Post each queued alert to Slack, subject to the throttle, until the queue closes. +async fn post_alerts( + mut alerts: mpsc::Receiver, + url: Url, + window: Duration, + dropped: Arc, +) { + let http = match reqwest::Client::builder().timeout(POST_TIMEOUT).build() { + Ok(http) => http, + Err(err) => { + tracing::error!(error = %err, "Failed to build the HTTP client for Slack alerts; none will be sent"); + return; + } + }; + let mut throttle = Throttle::new(window); + let mut check = tokio::time::interval(HELD_BACK_CHECK); + loop { + let mut texts = Vec::new(); + tokio::select! { + alert = alerts.recv() => { + let Some(alert) = alert else { + return; + }; + if let Some(held_back) = throttle.admit(&alert.event, Instant::now()) { + texts.push(alert_text(&alert, held_back)); + } + } + _ = check.tick() => { + for (event, held_back) in throttle.due(Instant::now()) { + texts.push(held_back_text(&event, held_back, window)); + } + } + } + let lost = dropped.swap(0, Ordering::Relaxed); + if lost > 0 { + tracing::warn!( + dropped = lost, + "Alerts arrived faster than they could be posted to Slack; some were dropped" + ); + } + for text in texts { + post(&http, &url, &text).await; + } + } +} + +/// At most 1 message per event in each window; alerts in between are counted and reported in +/// the next message for that event. +struct Throttle { + window: Duration, + events: HashMap, +} + +struct Window { + started: Instant, + held_back: u64, +} + +impl Throttle { + fn new(window: Duration) -> Self { + Self { + window, + events: HashMap::new(), + } + } + + /// Whether to post this alert now and, if so, how many were held back before it. + fn admit(&mut self, event: &str, now: Instant) -> Option { + if let Some(window) = self.events.get_mut(event) + && now.duration_since(window.started) < self.window + { + window.held_back += 1; + return None; + } + let held_back = self + .events + .insert( + event.to_owned(), + Window { + started: now, + held_back: 0, + }, + ) + .map_or(0, |window| window.held_back); + Some(held_back) + } + + /// Events whose window has ended with alerts held back, each with how many; the message + /// reporting them starts that event's next window. + fn due(&mut self, now: Instant) -> Vec<(String, u64)> { + let mut due = Vec::new(); + for (event, window) in &mut self.events { + if window.held_back > 0 && now.duration_since(window.started) >= self.window { + due.push((event.clone(), window.held_back)); + *window = Window { + started: now, + held_back: 0, + }; + } + } + due + } +} + +fn alert_text(alert: &Alert, held_back: u64) -> String { + let mut text = format!( + "dipper {} `{}`: {}", + alert.level, alert.event, alert.message + ); + if !alert.fields.is_empty() { + text.push_str(&format!("\n{}", alert.fields)); + } + if held_back > 0 { + text.push_str(&format!("\n{held_back} more since the last alert")); + } + text +} + +fn held_back_text(event: &str, held_back: u64, window: Duration) -> String { + format!( + "dipper `{event}`: {held_back} more in the last {} minutes", + window.as_secs() / 60 + ) +} + +/// Post a message to the Slack webhook, logging a failure without the webhook's URL, which is a +/// secret. +async fn post(http: &reqwest::Client, url: &Url, text: &str) { + let sent = http + .post(url.clone()) + .json(&serde_json::json!({ "text": text })) + .send() + .await + .and_then(reqwest::Response::error_for_status); + if let Err(err) = sent { + tracing::warn!(error = %err.without_url(), "Failed to post an alert to Slack"); + } +} + +#[cfg(test)] +mod tests { + use tracing_subscriber::layer::SubscriberExt; + use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{body_json, method}, + }; + + use super::*; + + fn test_layer(events: &[&str]) -> (AlertLayer, mpsc::Receiver) { + let (queue, alerts) = mpsc::channel(QUEUE); + let layer = AlertLayer { + events: events.iter().map(|event| (*event).to_owned()).collect(), + queue, + dropped: Arc::new(AtomicU64::new(0)), + }; + (layer, alerts) + } + + #[test] + fn picks_out_only_the_listed_events_whatever_their_level() { + let (layer, mut alerts) = test_layer(&["agreement_cancel_stuck", "nonce_gap_fill_failed"]); + let logging = tracing::Dispatch::new(tracing_subscriber::registry().with(layer)); + + tracing::dispatcher::with_default(&logging, || { + tracing::error!( + event = "agreement_cancel_stuck", + agreement_id = %"0xab", + attempts = 10u32, + "Cancelling an agreement keeps failing" + ); + tracing::warn!( + event = "nonce_gap_fill_failed", + nonce = 7u64, + "Gap fill failed" + ); + }); + tracing::dispatcher::with_default(&logging, || { + tracing::error!(event = "something_else", "Not listed"); + tracing::error!("Not tagged"); + }); + + assert_eq!( + alerts.try_recv().expect("first alert"), + Alert { + event: "agreement_cancel_stuck".to_owned(), + level: Level::ERROR, + message: "Cancelling an agreement keeps failing".to_owned(), + fields: "agreement_id=0xab attempts=10".to_owned(), + } + ); + assert_eq!( + alerts.try_recv().expect("second alert").event, + "nonce_gap_fill_failed" + ); + assert!(alerts.try_recv().is_err(), "nothing else"); + } + + #[test] + fn counts_alerts_dropped_when_the_queue_is_full() { + let (queue, _alerts) = mpsc::channel(1); + let layer = AlertLayer { + events: HashSet::from(["rpc_blocks_refused".to_owned()]), + queue, + dropped: Arc::new(AtomicU64::new(0)), + }; + let dropped = Arc::clone(&layer.dropped); + let subscriber = tracing_subscriber::registry().with(layer); + + tracing::subscriber::with_default(subscriber, || { + for _ in 0..3 { + tracing::error!(event = "rpc_blocks_refused", "Refused"); + } + }); + + assert_eq!(dropped.load(Ordering::Relaxed), 2); + } + + #[test] + fn sends_at_most_1_message_per_event_in_each_window() { + let window = Duration::from_secs(900); + let mut throttle = Throttle::new(window); + let start = Instant::now(); + + assert_eq!(throttle.admit("a", start), Some(0)); + assert_eq!(throttle.admit("a", start + Duration::from_secs(60)), None); + assert_eq!(throttle.admit("a", start + Duration::from_secs(120)), None); + assert_eq!(throttle.admit("b", start), Some(0), "each event on its own"); + + assert!(throttle.due(start + Duration::from_secs(600)).is_empty()); + let after = start + window; + assert_eq!(throttle.due(after), vec![("a".to_owned(), 2)]); + assert_eq!( + throttle.admit("a", after + Duration::from_secs(60)), + None, + "the count starts the next window" + ); + assert_eq!(throttle.admit("a", after + window), Some(1)); + } + + #[test] + fn says_what_happened_and_how_many_more() { + let alert = Alert { + event: "agreement_cancel_stuck".to_owned(), + level: Level::ERROR, + message: "Cancelling an agreement keeps failing".to_owned(), + fields: "agreement_id=0xab".to_owned(), + }; + + assert_eq!( + alert_text(&alert, 3), + "dipper ERROR `agreement_cancel_stuck`: Cancelling an agreement keeps failing\n\ + agreement_id=0xab\n3 more since the last alert" + ); + assert_eq!( + held_back_text("rpc_blocks_refused", 5, Duration::from_secs(900)), + "dipper `rpc_blocks_refused`: 5 more in the last 15 minutes" + ); + } + + #[tokio::test] + async fn posts_the_message_as_slack_text() { + let slack = MockServer::start().await; + Mock::given(method("POST")) + .and(body_json(serde_json::json!({ "text": "hello" }))) + .respond_with(ResponseTemplate::new(200)) + .expect(1) + .mount(&slack) + .await; + let url: Url = slack.uri().parse().expect("URL"); + + post(&reqwest::Client::new(), &url, "hello").await; + } +} diff --git a/bin/dipper-service/src/config.rs b/bin/dipper-service/src/config.rs index 4a85f18d..8bc4619c 100644 --- a/bin/dipper-service/src/config.rs +++ b/bin/dipper-service/src/config.rs @@ -97,6 +97,42 @@ pub struct Config { /// the server off. #[serde(default)] pub health: HealthConfig, + /// Where to send alerts for the log lines an operator has to act on. Without a webhook, + /// none are sent. + #[serde(default)] + pub alerts: AlertsConfig, +} + +/// Alerts for the log lines an operator has to act on, picked out by their `event` tag and +/// posted to Slack, at most once per event in each throttle window. +#[serde_as] +#[derive(Debug, serde::Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct AlertsConfig { + /// The Slack incoming-webhook URL to post alerts to. Without one, none are sent. + pub slack_webhook_url: Option>, + /// The `event` tags of the log lines to alert on. + pub events: Vec, + /// Least time, in seconds, between 2 messages for the same event; alerts in between are + /// counted into the next one. + #[serde_as(as = "serde_with::DurationSeconds")] + pub throttle: Duration, +} + +impl Default for AlertsConfig { + fn default() -> Self { + Self { + slack_webhook_url: None, + events: [ + "agreement_cancel_stuck", + "rpc_blocks_refused", + "nonce_gap_fill_failed", + ] + .map(str::to_owned) + .to_vec(), + throttle: Duration::from_secs(15 * 60), + } + } } /// Configuration for the HTTP health endpoint used by orchestrator liveness probes. Omitting the @@ -2082,6 +2118,22 @@ mod tests { ); } + #[test] + fn alerts_config_takes_a_webhook_and_keeps_the_default_events() { + let alerts = serde_json::from_str::( + r#"{"slack_webhook_url": "https://hooks.slack.com/services/T/B/x", "throttle": 60}"#, + ) + .expect("alerts config"); + + assert!(alerts.slack_webhook_url.is_some()); + assert_eq!(alerts.throttle, Duration::from_secs(60)); + assert_eq!(alerts.events, AlertsConfig::default().events); + assert!( + !format!("{alerts:?}").contains("hooks.slack.com"), + "the webhook is a secret, kept out of logs" + ); + } + /// A misspelled key would otherwise be dropped in silence, leaving an operator who meant to /// change the endpoint with the defaults and no sign that the setting never took effect. #[test] diff --git a/bin/dipper-service/src/main.rs b/bin/dipper-service/src/main.rs index 788047e7..4b52b7c2 100644 --- a/bin/dipper-service/src/main.rs +++ b/bin/dipper-service/src/main.rs @@ -8,7 +8,10 @@ use dipper_producer::events::{ use futures_lite::StreamExt; use thegraph_core::alloy::signers::local::PrivateKeySigner; use tokio::task::JoinSet; -use tracing_subscriber::EnvFilter; +use tracing_subscriber::{ + EnvFilter, Layer as _, filter::LevelFilter, layer::SubscriberExt as _, + util::SubscriberInitExt as _, +}; use self::{ config::DEFAULT_MAX_CANDIDATES, registry::RegistryProvider, signing::eip712::Eip712Signer, @@ -17,6 +20,7 @@ use self::{ use crate::config::EventStreamingConfig; mod admin_rpc_server; +mod alerts; mod cancel_dispatch; mod chain_client; mod config; @@ -61,20 +65,24 @@ const STOP_STEP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5) reason = "predates this lint; fix when next touched" )] pub async fn main() -> anyhow::Result<()> { - // Set up logging. Plain text, with no colour codes, so log stores can search it. - tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .with_ansi(false) - .init(); - - // Load the configuration - tracing::debug!("loading configuration"); + // Load the configuration first, since logging needs it to know where to send alerts let conf_path = env::args() .nth(1) .expect("Missing argument for config path") .parse::() .expect("Invalid path"); let conf = config::load_from_file(&conf_path).expect("Failed to load config"); + + // Set up logging. Plain text, with no colour codes, so log stores can search it. Alerts see + // warnings and errors whatever the log level is set to. + tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_ansi(false) + .with_filter(EnvFilter::from_default_env()), + ) + .with(alerts::layer(&conf.alerts).map(|layer| layer.with_filter(LevelFilter::WARN))) + .init(); tracing::debug!(conf=?conf, "configuration loaded"); // Reject a config the protocol-managed path can't run with before building diff --git a/k8s/configmap-example.yaml b/k8s/configmap-example.yaml index 2d369960..88222ea5 100644 --- a/k8s/configmap-example.yaml +++ b/k8s/configmap-example.yaml @@ -118,5 +118,10 @@ data: "enabled": true, "listen_addr": "0.0.0.0:8546", "threshold": 840 + }, + "alerts": { + "slack_webhook_url": "REPLACE_ME_OR_REMOVE_FOR_NO_ALERTS", + "events": ["agreement_cancel_stuck", "rpc_blocks_refused", "nonce_gap_fill_failed"], + "throttle": 900 } } From ee580bfbc113a0b34d88477a3671f273b40e39de Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 19:44:31 +0100 Subject: [PATCH 3/5] fix(alerts): escape the characters Slack reads as markup Slack treats &, < and > in a message as links, mentions or entities, so an error carrying an HTML page, such as a proxy's 502, came through mangled. They are now escaped. --- bin/dipper-service/src/alerts.rs | 26 ++++++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/bin/dipper-service/src/alerts.rs b/bin/dipper-service/src/alerts.rs index 3e0be77c..e9d640cf 100644 --- a/bin/dipper-service/src/alerts.rs +++ b/bin/dipper-service/src/alerts.rs @@ -227,10 +227,12 @@ impl Throttle { fn alert_text(alert: &Alert, held_back: u64) -> String { let mut text = format!( "dipper {} `{}`: {}", - alert.level, alert.event, alert.message + alert.level, + slack_escape(&alert.event), + slack_escape(&alert.message) ); if !alert.fields.is_empty() { - text.push_str(&format!("\n{}", alert.fields)); + text.push_str(&format!("\n{}", slack_escape(&alert.fields))); } if held_back > 0 { text.push_str(&format!("\n{held_back} more since the last alert")); @@ -238,6 +240,14 @@ fn alert_text(alert: &Alert, held_back: u64) -> String { text } +/// Escape the 3 characters Slack reads as markup, so text like an HTML error page shows as written +/// instead of turning into links or mentions. +fn slack_escape(text: &str) -> String { + text.replace('&', "&") + .replace('<', "<") + .replace('>', ">") +} + fn held_back_text(event: &str, held_back: u64, window: Duration) -> String { format!( "dipper `{event}`: {held_back} more in the last {} minutes", @@ -360,6 +370,18 @@ mod tests { assert_eq!(throttle.admit("a", after + window), Some(1)); } + #[test] + fn escapes_what_slack_reads_as_markup() { + let alert = Alert { + event: "nonce_gap_fill_failed".to_owned(), + level: Level::WARN, + message: "Gap fill failed".to_owned(), + fields: "error=502 & more".to_owned(), + }; + + assert!(alert_text(&alert, 0).ends_with("error=<html>502 & more</html>")); + } + #[test] fn says_what_happened_and_how_many_more() { let alert = Alert { From 235364fedece339db36ec2b371d50db06954f6e8 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 19:45:00 +0100 Subject: [PATCH 4/5] fix(alerts): describe a throttle window shorter than a minute correctly The count of held-back alerts named the window in whole minutes rounded down, so 30 seconds read as "the last 0 minutes", and a window under a minute was only checked once a minute. Both now follow the window as configured. --- bin/dipper-service/src/alerts.rs | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/bin/dipper-service/src/alerts.rs b/bin/dipper-service/src/alerts.rs index e9d640cf..877a9bd9 100644 --- a/bin/dipper-service/src/alerts.rs +++ b/bin/dipper-service/src/alerts.rs @@ -135,7 +135,8 @@ async fn post_alerts( } }; let mut throttle = Throttle::new(window); - let mut check = tokio::time::interval(HELD_BACK_CHECK); + // A window shorter than the usual check is checked as often as it ends. + let mut check = tokio::time::interval(HELD_BACK_CHECK.min(window).max(Duration::from_secs(1))); loop { let mut texts = Vec::new(); tokio::select! { @@ -250,11 +251,22 @@ fn slack_escape(text: &str) -> String { fn held_back_text(event: &str, held_back: u64, window: Duration) -> String { format!( - "dipper `{event}`: {held_back} more in the last {} minutes", - window.as_secs() / 60 + "dipper `{event}`: {held_back} more in the last {}", + describe(window) ) } +/// A throttle window in words: whole minutes where it divides into them, seconds otherwise. +fn describe(window: Duration) -> String { + let seconds = window.as_secs(); + match (seconds / 60, seconds % 60) { + (1, 0) => "minute".to_owned(), + (minutes, 0) if minutes > 0 => format!("{minutes} minutes"), + _ if seconds == 1 => "second".to_owned(), + _ => format!("{seconds} seconds"), + } +} + /// Post a message to the Slack webhook, logging a failure without the webhook's URL, which is a /// secret. async fn post(http: &reqwest::Client, url: &Url, text: &str) { @@ -400,6 +412,9 @@ mod tests { held_back_text("rpc_blocks_refused", 5, Duration::from_secs(900)), "dipper `rpc_blocks_refused`: 5 more in the last 15 minutes" ); + assert_eq!(describe(Duration::from_secs(60)), "minute"); + assert_eq!(describe(Duration::from_secs(90)), "90 seconds"); + assert_eq!(describe(Duration::from_secs(30)), "30 seconds"); } #[tokio::test] From d17e2bc1045c68ae09b3b1b0231adadf9a12cf36 Mon Sep 17 00:00:00 2001 From: MoonBoi9001 Date: Thu, 8 Oct 2026 19:45:43 +0100 Subject: [PATCH 5/5] fix(alerts): count alerts dropped in a burst into Slack's figure Alerts dropped because the queue was full only showed up as a warning in dipper's own logs, so the "N more" figure in Slack was too low during a burst. They are now counted by event into that figure. --- bin/dipper-service/src/alerts.rs | 75 +++++++++++++++++++++++--------- 1 file changed, 54 insertions(+), 21 deletions(-) diff --git a/bin/dipper-service/src/alerts.rs b/bin/dipper-service/src/alerts.rs index 877a9bd9..d2460b08 100644 --- a/bin/dipper-service/src/alerts.rs +++ b/bin/dipper-service/src/alerts.rs @@ -4,10 +4,7 @@ use std::{ collections::{HashMap, HashSet}, - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, + sync::{Arc, Mutex, PoisonError}, time::Duration, }; @@ -21,8 +18,8 @@ use url::Url; use crate::config::AlertsConfig; -/// Alerts that can wait to be posted. More are dropped and counted, since a burst that size -/// is mostly held back by the throttle anyway. +/// Alerts that can wait to be posted. More are dropped, as a burst that size is mostly held back +/// by the throttle anyway, and counted into the held-back figure for their event. const QUEUE: usize = 64; /// How long a single post to Slack may take. @@ -44,15 +41,18 @@ struct Alert { pub struct AlertLayer { events: HashSet, queue: mpsc::Sender, - dropped: Arc, + dropped: Dropped, } +/// Alerts dropped because the queue was full, counted by event. +type Dropped = Arc>>; + /// The alert layer for dipper's logging, with its poster running in the background, or `None` /// when no Slack webhook is configured. pub fn layer(config: &AlertsConfig) -> Option { let url = config.slack_webhook_url.as_ref()?.as_ref().clone(); let (queue, alerts) = mpsc::channel(QUEUE); - let dropped = Arc::new(AtomicU64::new(0)); + let dropped = Dropped::default(); tokio::spawn(post_alerts( alerts, url, @@ -80,9 +80,14 @@ impl Layer for AlertLayer { fields: line.fields, }; // Waiting here would hold up the code that logged it, so a full queue drops the alert; - // the poster reports how many. - if self.queue.try_send(alert).is_err() { - self.dropped.fetch_add(1, Ordering::Relaxed); + // the poster counts it into the next message for its event. + if let Err(full) = self.queue.try_send(alert) { + *self + .dropped + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(full.into_inner().event) + .or_default() += 1; } } } @@ -125,7 +130,7 @@ async fn post_alerts( mut alerts: mpsc::Receiver, url: Url, window: Duration, - dropped: Arc, + dropped: Dropped, ) { let http = match reqwest::Client::builder().timeout(POST_TIMEOUT).build() { Ok(http) => http, @@ -154,12 +159,9 @@ async fn post_alerts( } } } - let lost = dropped.swap(0, Ordering::Relaxed); - if lost > 0 { - tracing::warn!( - dropped = lost, - "Alerts arrived faster than they could be posted to Slack; some were dropped" - ); + let lost = std::mem::take(&mut *dropped.lock().unwrap_or_else(PoisonError::into_inner)); + for (event, count) in lost { + throttle.hold_back(&event, count, Instant::now()); } for text in texts { post(&http, &url, &text).await; @@ -208,6 +210,19 @@ impl Throttle { Some(held_back) } + /// Count alerts that never reached the throttle into its held-back figure for their event; + /// one with no window yet is reported at the next check. + fn hold_back(&mut self, event: &str, count: u64, now: Instant) { + let window = self.window; + self.events + .entry(event.to_owned()) + .or_insert_with(|| Window { + started: now.checked_sub(window).unwrap_or(now), + held_back: 0, + }) + .held_back += count; + } + /// Events whose window has ended with alerts held back, each with how many; the message /// reporting them starts that event's next window. fn due(&mut self, now: Instant) -> Vec<(String, u64)> { @@ -296,7 +311,7 @@ mod tests { let layer = AlertLayer { events: events.iter().map(|event| (*event).to_owned()).collect(), queue, - dropped: Arc::new(AtomicU64::new(0)), + dropped: Dropped::default(), }; (layer, alerts) } @@ -346,7 +361,7 @@ mod tests { let layer = AlertLayer { events: HashSet::from(["rpc_blocks_refused".to_owned()]), queue, - dropped: Arc::new(AtomicU64::new(0)), + dropped: Dropped::default(), }; let dropped = Arc::clone(&layer.dropped); let subscriber = tracing_subscriber::registry().with(layer); @@ -357,7 +372,10 @@ mod tests { } }); - assert_eq!(dropped.load(Ordering::Relaxed), 2); + assert_eq!( + *dropped.lock().unwrap(), + HashMap::from([("rpc_blocks_refused".to_owned(), 2)]) + ); } #[test] @@ -382,6 +400,21 @@ mod tests { assert_eq!(throttle.admit("a", after + window), Some(1)); } + #[test] + fn counts_dropped_alerts_into_the_held_back_figure() { + let window = Duration::from_secs(900); + let mut throttle = Throttle::new(window); + let start = Instant::now(); + assert_eq!(throttle.admit("a", start), Some(0)); + + throttle.hold_back("a", 5, start + Duration::from_secs(60)); + throttle.hold_back("b", 2, start + Duration::from_secs(60)); + + let mut due = throttle.due(start + window); + due.sort(); + assert_eq!(due, vec![("a".to_owned(), 5), ("b".to_owned(), 2)]); + } + #[test] fn escapes_what_slack_reads_as_markup() { let alert = Alert {