pager: event deduplication, more data, and cleanup (#1259)

This commit is contained in:
Brandon Thomas
2026-03-30 21:30:57 -04:00
committed by GitHub
parent dbe42447af
commit 67df86e54e
12 changed files with 480 additions and 41 deletions
Generated
+1
View File
@@ -8038,6 +8038,7 @@ dependencies = [
"errors",
"log",
"rootly_client",
"sha2 0.11.0-rc.5",
"tokio",
"workspace-build-acceleration",
]
+1
View File
@@ -21,6 +21,7 @@ rootly_client.workspace = true
# External
chrono.workspace = true
sha2.workspace = true
log.workspace = true
tokio.workspace = true
+37 -5
View File
@@ -112,12 +112,44 @@ impl PagerClient {
labels.push(("hostname".to_string(), h.clone()));
}
if let Some(method) = &notification.http_method {
labels.push(("http_method".to_string(), method.clone()));
}
if let Some(path) = &notification.http_path {
labels.push(("http_path".to_string(), path.clone()));
}
if let Some(status_code) = notification.http_status_code {
labels.push(("http_status_code".to_string(), status_code.to_string()));
}
let labels = if labels.is_empty() { None } else { Some(labels) };
// Enrich the description with hostname if both are present.
let description = match (&notification.description, &self.hostname) {
(Some(desc), Some(h)) => Some(format!("{}\n\nHostname: {}", desc, h)),
(desc, _) => desc.clone(),
// Enrich the description with HTTP context and hostname if present.
let description = match &notification.description {
Some(desc) => {
let mut parts = vec![desc.clone()];
if let Some(method) = &notification.http_method {
parts.push(format!("HTTP Method: {}", method));
}
if let Some(path) = &notification.http_path {
parts.push(format!("HTTP Path: {}", path));
}
if let Some(status_code) = notification.http_status_code {
parts.push(format!("HTTP Status Code: {}", status_code));
}
if let Some(h) = &self.hostname {
parts.push(format!("Hostname: {}", h));
}
Some(parts.join("\n\n"))
}
None => None,
};
let result = create_alert(CreateAlertArgs {
@@ -135,7 +167,7 @@ impl PagerClient {
notification_target_type: notification_target_type.clone(),
notification_target_id: notification_target_id.clone(),
labels,
deduplication_key: None,
deduplication_key: Some(notification.to_deduplication_key()),
}).await;
match result {
@@ -0,0 +1,284 @@
use chrono::{DateTime, Utc};
use sha2::{Sha256, Digest};
use crate::notification::notification_details::NotificationDetails;
/// Returns the number of whole hours since the Unix epoch for the given time.
fn hours_since_epoch(time: DateTime<Utc>) -> i64 {
time.timestamp() / 3600
}
/// Generate a deduplication key for a notification.
///
/// Notifications with the same key are considered duplicates and can be suppressed.
/// The key is a hex-encoded SHA-256 hash of the notification's identifying fields.
///
/// When both `http_method` and `endpoint_path` are present, they are included in the
/// hash so that different endpoints with the same summary produce distinct keys.
/// The `hours_since_epoch` of the event time is included so that the same notification
/// firing in different hours produces a different key (allowing re-alerting).
pub(crate) fn generate_deduplication_key(details: &NotificationDetails) -> String {
let mut hasher = Sha256::new();
hasher.update(details.summary.as_bytes());
hasher.update(if details.is_from_error { b"1" } else { b"0" });
if let (Some(method), Some(path)) = (&details.http_method, &details.http_path) {
hasher.update(method.as_bytes());
hasher.update(path.as_bytes());
}
if let Some(status_code) = details.http_status_code {
hasher.update(status_code.to_le_bytes());
}
hasher.update(hours_since_epoch(details.event_time).to_le_bytes());
let result = hasher.finalize();
result.iter().map(|b| format!("{:02x}", b)).collect()
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::TimeZone;
fn make_details(
summary: &str,
is_from_error: bool,
http_method: Option<&str>,
http_path: Option<&str>,
event_time: DateTime<Utc>,
) -> NotificationDetails {
NotificationDetails {
summary: summary.to_string(),
description: None,
event_time,
http_method: http_method.map(|s| s.to_string()),
http_path: http_path.map(|s| s.to_string()),
http_status_code: None,
is_from_error,
}
}
// -- hours_since_epoch tests --
mod hours_since_epoch_tests {
use super::*;
#[test]
fn hours_since_epoch_at_unix_epoch() {
let epoch = Utc.with_ymd_and_hms(1970, 1, 1, 0, 0, 0).unwrap();
assert_eq!(hours_since_epoch(epoch), 0);
}
#[test]
fn hours_since_epoch_at_one_hour() {
let time = Utc.with_ymd_and_hms(1970, 1, 1, 1, 0, 0).unwrap();
assert_eq!(hours_since_epoch(time), 1);
}
#[test]
fn hours_since_epoch_mid_hour_rounds_down() {
let time = Utc.with_ymd_and_hms(1970, 1, 1, 1, 30, 0).unwrap();
assert_eq!(hours_since_epoch(time), 1);
}
#[test]
fn hours_since_epoch_just_before_next_hour() {
let time = Utc.with_ymd_and_hms(1970, 1, 1, 1, 59, 59).unwrap();
assert_eq!(hours_since_epoch(time), 1);
}
#[test]
fn hours_since_epoch_known_date() {
// 2026-03-30 14:00:00 UTC
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
assert_eq!(hours_since_epoch(time), 493022);
}
}
// -- Consistent hashing tests --
mod consistent_hashing_tests {
use super::*;
#[test]
fn consistent_hash_same_inputs() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("db connection failed", true, Some("POST"), Some("/v1/jobs"), time);
let b = make_details("db connection failed", true, Some("POST"), Some("/v1/jobs"), time);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn consistent_hash_without_http_fields() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("something broke", true, None, None, time);
let b = make_details("something broke", true, None, None, time);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
// -- Same hour deduplication tests --
#[test]
fn same_key_within_same_hour() {
let t1 = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let t2 = Utc.with_ymd_and_hms(2026, 3, 30, 14, 45, 30).unwrap();
let a = make_details("timeout", true, Some("GET"), Some("/v1/users"), t1);
let b = make_details("timeout", true, Some("GET"), Some("/v1/users"), t2);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
}
// -- Differentiation tests --
mod differentiation_tests {
use super::*;
#[test]
fn different_key_across_hour_boundary() {
let t1 = Utc.with_ymd_and_hms(2026, 3, 30, 14, 59, 59).unwrap();
let t2 = Utc.with_ymd_and_hms(2026, 3, 30, 15, 0, 0).unwrap();
let a = make_details("timeout", true, Some("GET"), Some("/v1/users"), t1);
let b = make_details("timeout", true, Some("GET"), Some("/v1/users"), t2);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn different_summary_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("error A", true, None, None, time);
let b = make_details("error B", true, None, None, time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn different_is_from_error_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("something happened", true, None, None, time);
let b = make_details("something happened", false, None, None, time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn different_endpoint_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("db error", true, Some("POST"), Some("/v1/jobs"), time);
let b = make_details("db error", true, Some("POST"), Some("/v1/users"), time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn different_method_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("db error", true, Some("GET"), Some("/v1/jobs"), time);
let b = make_details("db error", true, Some("POST"), Some("/v1/jobs"), time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
}
mod http_status_code_tests {
use super::*;
fn make_details_with_status(
summary: &str,
is_from_error: bool,
http_method: Option<&str>,
http_path: Option<&str>,
http_status_code: Option<u16>,
event_time: DateTime<Utc>,
) -> NotificationDetails {
NotificationDetails {
summary: summary.to_string(),
description: None,
event_time,
http_method: http_method.map(|s| s.to_string()),
http_path: http_path.map(|s| s.to_string()),
http_status_code,
is_from_error,
}
}
#[test]
fn different_status_code_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details_with_status("db error", true, Some("POST"), Some("/v1/jobs"), Some(500), time);
let b = make_details_with_status("db error", true, Some("POST"), Some("/v1/jobs"), Some(503), time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn status_code_vs_none_different_key() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details_with_status("db error", true, Some("POST"), Some("/v1/jobs"), Some(500), time);
let b = make_details_with_status("db error", true, Some("POST"), Some("/v1/jobs"), None, time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn same_status_code_same_key() {
let t1 = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let t2 = Utc.with_ymd_and_hms(2026, 3, 30, 14, 30, 0).unwrap();
let a = make_details_with_status("timeout", true, Some("GET"), Some("/v1/users"), Some(504), t1);
let b = make_details_with_status("timeout", true, Some("GET"), Some("/v1/users"), Some(504), t2);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn status_code_without_http_fields_still_differentiates() {
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details_with_status("error", true, None, None, Some(500), time);
let b = make_details_with_status("error", true, None, None, Some(502), time);
assert_ne!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
}
mod other_behaviors {
use super::*;
#[test]
fn only_method_set_ignores_it() {
// When only http_method is Some but endpoint_path is None, HTTP fields are excluded.
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("db error", true, Some("POST"), None, time);
let b = make_details("db error", true, None, None, time);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
#[test]
fn only_path_set_ignores_it() {
// When only endpoint_path is Some but http_method is None, HTTP fields are excluded.
let time = Utc.with_ymd_and_hms(2026, 3, 30, 14, 0, 0).unwrap();
let a = make_details("db error", true, None, Some("/v1/jobs"), time);
let b = make_details("db error", true, None, None, time);
assert_eq!(generate_deduplication_key(&a), generate_deduplication_key(&b));
}
}
}
+2
View File
@@ -1 +1,3 @@
mod generate_deduplication_key;
pub mod notification_details;
pub mod notification_details_builder;
@@ -1,6 +1,8 @@
use chrono::{DateTime, Utc};
use std::fmt::{Debug, Display};
use crate::notification::generate_deduplication_key::generate_deduplication_key;
/// Details for a pager notification.
#[derive(Clone, Debug)]
pub struct NotificationDetails {
@@ -12,15 +14,35 @@ pub struct NotificationDetails {
/// When the event occurred.
pub event_time: DateTime<Utc>,
/// HTTP method associated with the event, if any.
pub http_method: Option<String>,
/// HTTP endpoint path associated with the event, if any.
pub http_path: Option<String>,
/// HTTP status code associated with the event, if any.
pub http_status_code: Option<u16>,
/// Whether this notification originated from an error.
pub is_from_error: bool,
}
impl NotificationDetails {
pub fn to_deduplication_key(&self) -> String {
generate_deduplication_key(self)
}
/// Create a notification with a summary and description.
pub fn with_summary_and_description(summary: String, description: String) -> Self {
Self {
summary,
description: Some(description),
event_time: Utc::now(),
http_method: None,
http_path: None,
http_status_code: None,
is_from_error: false,
}
}
@@ -30,6 +52,10 @@ impl NotificationDetails {
summary,
description: None,
event_time: Utc::now(),
http_method: None,
http_path: None,
http_status_code: None,
is_from_error: false,
}
}
@@ -86,6 +112,10 @@ impl NotificationDetails {
summary,
description: Some(description),
event_time,
http_method: None,
http_path: None,
http_status_code: None,
is_from_error: true,
}
}
@@ -0,0 +1,79 @@
use chrono::{DateTime, Utc};
use std::fmt::{Debug, Display};
use crate::notification::notification_details::NotificationDetails;
pub struct NotificationDetailsBuilder {
summary: String,
description: Option<String>,
event_time: DateTime<Utc>,
http_method: Option<String>,
http_path: Option<String>,
http_status_code: Option<u16>,
is_from_error: bool,
}
impl NotificationDetailsBuilder {
/// Create a builder from a summary string.
pub fn from_summary(summary: String) -> Self {
Self {
summary,
description: None,
event_time: Utc::now(),
http_method: None,
http_path: None,
http_status_code: None,
is_from_error: false,
}
}
/// Create a builder from an error.
///
/// Sets `is_from_error` to true and derives the summary and description
/// from the error, matching the behavior of `NotificationDetails::from_error`.
pub fn from_error<E: Debug + Display>(error: &E) -> Self {
// TODO(bt,2026-03-30): Clean this up
let details = NotificationDetails::from_error(error);
Self {
summary: details.summary,
description: details.description,
event_time: details.event_time,
http_method: None,
http_path: None,
http_status_code: None,
is_from_error: true,
}
}
pub fn set_description(mut self, description: Option<String>) -> Self {
self.description = description;
self
}
pub fn set_http_method(mut self, http_method: Option<String>) -> Self {
self.http_method = http_method;
self
}
pub fn set_http_path(mut self, http_path: Option<String>) -> Self {
self.http_path = http_path;
self
}
pub fn set_http_status_code(mut self, http_status_code: Option<u16>) -> Self {
self.http_status_code = http_status_code;
self
}
pub fn build(self) -> NotificationDetails {
NotificationDetails {
summary: self.summary,
description: self.description,
event_time: self.event_time,
http_method: self.http_method,
http_path: self.http_path,
http_status_code: self.http_status_code,
is_from_error: self.is_from_error,
}
}
}
@@ -3,6 +3,7 @@ use std::sync::Arc;
use crate::http_server::endpoints::webhooks::handle_image_payload::handle_image_payload;
use crate::http_server::endpoints::webhooks::handle_images_payload::handle_images_payload;
use crate::http_server::endpoints::webhooks::handle_model_glb_payload::handle_model_glb_payload;
use crate::http_server::endpoints::webhooks::handle_model_mesh_payload::handle_model_mesh_payload;
use crate::http_server::endpoints::webhooks::handle_video_payload::handle_video_payload;
use crate::state::server_state::ServerState;
@@ -18,7 +19,7 @@ use mysql_queries::queries::generic_inference::fal::get_inference_job_by_fal_id:
use mysql_queries::queries::generic_inference::fal::mark_fal_generic_inference_job_successfully_done::{mark_fal_generic_inference_job_successfully_done, MarkJobArgs};
use serde_json::Value;
use utoipa::ToSchema;
use crate::http_server::endpoints::webhooks::handle_model_glb_payload::handle_model_glb_payload;
// 1. tauri --> hit endpoint to enqueue
//
// 2. webhook
@@ -1,7 +1,9 @@
use log::{debug, warn};
use pager::client::pager::Pager;
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
use crate::http_server::common_responses::common_web_error::CommonWebError;
use super::enqueue_alert::enqueue_alert;
/// Check `CommonWebError` and alert on server errors.
/// Returns `true` if the error was handled (alerted or intentionally skipped).
@@ -13,18 +15,29 @@ pub(super) fn check_common_web_error(
) -> bool {
match error {
CommonWebError::ServerError => {
enqueue_alert(
pager,
format!("CommonWebError::ServerError on {} {}", method, path),
format!(
"A CommonWebError::ServerError was returned.\n\n\
let summary = format!("CommonWebError::ServerError on {} {}", method, path);
let description = format!(
"A CommonWebError::ServerError was returned.\n\n\
Endpoint: {} {}\n\
Error: {:?}\n\
Time: {}",
method, path, error,
chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC"),
),
method, path, error,
chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC"),
);
let notification = NotificationDetailsBuilder::from_summary(summary)
.set_description(Some(description))
.set_http_method(Some(method.to_string()))
.set_http_path(Some(path.to_string()))
.set_http_status_code(Some(500))
.build();
if let Err(err) = pager.enqueue_page(notification) {
warn!("Error alerting middleware: failed to enqueue page: {:?}", err);
} else {
debug!("Error alerting middleware: enqueued alert for CommonWebError::ServerError");
}
true
}
// Don't alert on client errors (400, 401, 404, 402).
@@ -1,6 +1,7 @@
use pager::client::pager::Pager;
use log::{debug, warn};
use super::enqueue_alert::enqueue_alert;
use pager::client::pager::Pager;
use pager::notification::notification_details_builder::NotificationDetailsBuilder;
/// Fallback alerting based on HTTP status code when no typed error matched.
pub(super) fn check_status_code_fallback(
@@ -11,20 +12,30 @@ pub(super) fn check_status_code_fallback(
) {
match status_code {
500 => {
enqueue_alert(
pager,
format!("HTTP 500: {} {}", method, path),
format!(
"An untyped HTTP 500 response was returned (no typed error matched).\n\n\
let summary = format!("HTTP 500: {} {}", method, path);
let description = format!(
"An untyped HTTP 500 response was returned (no typed error matched).\n\n\
Endpoint: {} {}\n\
Time: {}",
method, path,
chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC"),
),
method, path,
chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC"),
);
let notification = NotificationDetailsBuilder::from_summary(summary)
.set_description(Some(description))
.set_http_method(Some(method.to_string()))
.set_http_path(Some(path.to_string()))
.set_http_status_code(Some(status_code))
.build();
if let Err(err) = pager.enqueue_page(notification) {
warn!("Error alerting middleware: failed to enqueue page: {:?}", err);
} else {
debug!("Error alerting middleware: enqueued alert for HTTP {}", status_code);
}
}
// Add more status-based rules here:
// 503 => { enqueue_alert(...); }
// 503 => { ... }
_ => {}
}
}
@@ -1,14 +0,0 @@
use log::{debug, warn};
use pager::client::pager::Pager;
use pager::notification::notification_details::NotificationDetails;
pub(super) fn enqueue_alert(pager: &Pager, summary: String, description: String) {
let notification = NotificationDetails::with_summary_and_description(summary, description);
if let Err(err) = pager.enqueue_page(notification) {
warn!("Error alerting middleware: failed to enqueue page: {:?}", err);
} else {
debug!("Error alerting middleware: enqueued alert");
}
}
@@ -1,5 +1,4 @@
mod check_common_web_error;
mod check_status_code_fallback;
mod enqueue_alert;
pub mod error_alerting_middleware;