use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::state::AppState;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum AlertCondition {
DeadLetteredJob,
HealthIndicatorDown,
HighErrorRate,
ScheduledTaskFailure,
}
impl AlertCondition {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::DeadLetteredJob => "dead_lettered_job",
Self::HealthIndicatorDown => "health_indicator_down",
Self::HighErrorRate => "high_error_rate",
Self::ScheduledTaskFailure => "scheduled_task_failure",
}
}
#[must_use]
pub const fn where_to_look(self) -> &'static str {
match self {
Self::DeadLetteredJob => "/actuator/jobs",
Self::HealthIndicatorDown => "/actuator/health",
Self::HighErrorRate => "/actuator/metrics",
Self::ScheduledTaskFailure => "/actuator/tasks",
}
}
const fn actuator_suffix(self) -> &'static str {
match self {
Self::DeadLetteredJob => "/jobs",
Self::HealthIndicatorDown => "/health",
Self::HighErrorRate => "/metrics",
Self::ScheduledTaskFailure => "/tasks",
}
}
}
fn actuator_where_to_look(prefix: &str, condition: AlertCondition) -> String {
crate::actuator::actuator_route_path(prefix, condition.actuator_suffix())
}
fn sensitive_gated_where_to_look(
prefix: &str,
condition: AlertCondition,
sensitive: bool,
) -> String {
if sensitive {
actuator_where_to_look(prefix, condition)
} else {
let fallback = actuator_where_to_look(prefix, AlertCondition::HealthIndicatorDown);
let gated = actuator_where_to_look(prefix, condition);
format!("{fallback} ({gated} requires [actuator] sensitive = true)")
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[non_exhaustive]
pub enum AlertSeverity {
Critical,
Recovery,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum AlertRouting {
#[default]
All,
Critical,
}
impl AlertRouting {
#[must_use]
pub const fn accepts(self, severity: AlertSeverity) -> bool {
match self {
Self::All => true,
Self::Critical => matches!(severity, AlertSeverity::Critical),
}
}
}
impl std::str::FromStr for AlertRouting {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"all" => Ok(Self::All),
"critical" => Ok(Self::Critical),
_ => Err(()),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
#[non_exhaustive]
pub enum AlertEventKind {
Trigger,
Resolve,
}
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
pub struct Alert {
pub dedup_key: String,
pub condition: AlertCondition,
pub severity: AlertSeverity,
pub event: AlertEventKind,
pub title: String,
pub summary: String,
pub timestamp: DateTime<Utc>,
pub host: String,
pub where_to_look: String,
pub details: HashMap<String, String>,
}
impl Alert {
#[must_use]
pub fn trigger(condition: AlertCondition, key: impl Into<String>) -> AlertBuilder {
AlertBuilder::new(condition, key.into(), AlertEventKind::Trigger)
}
#[must_use]
pub fn recovery(condition: AlertCondition, key: impl Into<String>) -> AlertBuilder {
AlertBuilder::new(condition, key.into(), AlertEventKind::Resolve)
}
}
#[derive(Debug, Clone)]
pub struct AlertBuilder {
alert: Alert,
}
impl AlertBuilder {
fn new(condition: AlertCondition, dedup_key: String, event: AlertEventKind) -> Self {
let severity = match event {
AlertEventKind::Trigger => AlertSeverity::Critical,
AlertEventKind::Resolve => AlertSeverity::Recovery,
};
Self {
alert: Alert {
dedup_key,
condition,
severity,
event,
title: String::new(),
summary: String::new(),
timestamp: Utc::now(),
host: host_id(),
where_to_look: condition.where_to_look().to_owned(),
details: HashMap::new(),
},
}
}
#[must_use]
pub fn title(mut self, title: impl Into<String>) -> Self {
self.alert.title = title.into();
self
}
#[must_use]
pub fn summary(mut self, summary: impl Into<String>) -> Self {
self.alert.summary = summary.into();
self
}
#[must_use]
pub fn where_to_look(mut self, where_to_look: impl Into<String>) -> Self {
self.alert.where_to_look = where_to_look.into();
self
}
#[must_use]
pub fn detail(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.alert.details.insert(key.into(), value.into());
self
}
#[must_use]
pub fn build(self) -> Alert {
self.alert
}
}
#[must_use]
pub fn host_id() -> String {
std::env::var("AUTUMN_REPLICA_ID")
.ok()
.filter(|s| !s.trim().is_empty())
.or_else(|| {
std::env::var("HOSTNAME")
.ok()
.filter(|s| !s.trim().is_empty())
})
.unwrap_or_else(|| "unknown-host".to_owned())
}
fn error_rate_dedup_key(host: &str) -> String {
format!("high_error_rate:5xx:{host}")
}
fn health_indicator_dedup_key(indicator: &str, host: &str) -> String {
format!("health_indicator_down:{indicator}:{host}")
}
pub type AlertDeliveryFuture<'a> =
Pin<Box<dyn Future<Output = Result<(), AlertDeliveryError>> + Send + 'a>>;
#[derive(Debug, thiserror::Error)]
#[error("alert delivery via {channel} failed: {message}")]
pub struct AlertDeliveryError {
pub channel: &'static str,
pub message: String,
}
impl AlertDeliveryError {
#[must_use]
pub fn new(channel: &'static str, message: impl Into<String>) -> Self {
Self {
channel,
message: message.into(),
}
}
}
pub trait AlertChannel: Send + Sync + 'static {
fn name(&self) -> &'static str;
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a>;
fn accepts_severity(&self, _severity: AlertSeverity) -> bool {
true
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DedupDecision {
Send,
Suppress,
}
impl DedupDecision {
#[must_use]
pub const fn should_send(self) -> bool {
matches!(self, Self::Send)
}
}
#[derive(Debug, Clone)]
struct KeyState {
active: bool,
last_sent: DateTime<Utc>,
}
#[derive(Debug)]
pub struct AlertDeduplicator {
window: chrono::Duration,
keys: HashMap<String, KeyState>,
}
impl AlertDeduplicator {
#[must_use]
pub fn new(window: std::time::Duration) -> Self {
Self {
window: chrono::Duration::from_std(window)
.unwrap_or_else(|_| chrono::Duration::seconds(900)),
keys: HashMap::new(),
}
}
pub fn on_trigger(&mut self, key: &str, now: DateTime<Utc>) -> DedupDecision {
match self.keys.get_mut(key) {
Some(state) if state.active && (now - state.last_sent) < self.window => {
DedupDecision::Suppress
}
Some(state) => {
state.active = true;
state.last_sent = now;
DedupDecision::Send
}
None => {
self.keys.insert(
key.to_owned(),
KeyState {
active: true,
last_sent: now,
},
);
DedupDecision::Send
}
}
}
pub fn on_resolve(&mut self, key: &str) -> DedupDecision {
match self.keys.get_mut(key) {
Some(state) if state.active => {
state.active = false;
DedupDecision::Send
}
_ => DedupDecision::Suppress,
}
}
}
#[derive(Debug, Clone)]
pub struct AlerterSettings {
pub dedup_window: std::time::Duration,
pub health_grace: std::time::Duration,
pub error_rate_threshold: f64,
pub error_rate_min_requests: u64,
pub eval_interval: std::time::Duration,
pub actuator_prefix: String,
pub actuator_sensitive: bool,
}
fn sanitize_error_rate_threshold(threshold: f64) -> f64 {
if threshold.is_finite() && threshold > 0.0 && threshold <= 1.0 {
threshold
} else {
let default = default_error_rate_threshold();
tracing::warn!(
configured = threshold,
default,
"alerts: [alerts] error_rate_threshold ({threshold}) is not a valid 5xx rate in \
(0, 1]; falling back to the default ({default}) so 5xx alerting keeps working. Fix \
[alerts] error_rate_threshold (or the AUTUMN_ALERTS__ERROR_RATE_THRESHOLD env var)."
);
default
}
}
impl AlerterSettings {
fn from_config(
config: &AlertConfig,
actuator_prefix: String,
actuator_sensitive: bool,
) -> Self {
Self {
dedup_window: std::time::Duration::from_secs(config.dedup_window_secs.max(1)),
health_grace: std::time::Duration::from_secs(config.health_grace_secs),
error_rate_threshold: sanitize_error_rate_threshold(config.error_rate_threshold),
error_rate_min_requests: config.error_rate_min_requests.max(1),
eval_interval: std::time::Duration::from_secs(config.eval_interval_secs.max(1)),
actuator_prefix,
actuator_sensitive,
}
}
}
struct AlerterInner {
channels: Vec<Arc<dyn AlertChannel>>,
dedup: Mutex<AlertDeduplicator>,
settings: AlerterSettings,
}
#[derive(Clone)]
pub struct Alerter {
inner: Arc<AlerterInner>,
}
impl Alerter {
#[must_use]
pub fn new(channels: Vec<Arc<dyn AlertChannel>>, settings: AlerterSettings) -> Self {
let dedup = AlertDeduplicator::new(settings.dedup_window);
Self {
inner: Arc::new(AlerterInner {
channels,
dedup: Mutex::new(dedup),
settings,
}),
}
}
#[must_use]
pub fn has_channels(&self) -> bool {
!self.inner.channels.is_empty()
}
pub(crate) fn settings(&self) -> &AlerterSettings {
&self.inner.settings
}
#[must_use]
pub fn notify(&self, alert: Alert) -> bool {
let decision = self
.inner
.dedup
.lock()
.map_or(DedupDecision::Send, |mut d| {
d.on_trigger(&alert.dedup_key, alert.timestamp)
});
if decision.should_send() {
self.dispatch(alert);
true
} else {
false
}
}
#[must_use]
pub fn recover(&self, alert: Alert) -> bool {
let decision = self
.inner
.dedup
.lock()
.map_or(DedupDecision::Suppress, |mut d| {
d.on_resolve(&alert.dedup_key)
});
if decision.should_send() {
self.dispatch(alert);
true
} else {
false
}
}
fn dispatch(&self, alert: Alert) {
if self.inner.channels.is_empty() {
return;
}
let Ok(handle) = tokio::runtime::Handle::try_current() else {
tracing::warn!("no tokio runtime available for alert dispatch; skipping");
return;
};
let alert = Arc::new(alert);
for channel in &self.inner.channels {
if !channel.accepts_severity(alert.severity) {
continue;
}
let channel = Arc::clone(channel);
let alert = Arc::clone(&alert);
handle.spawn(async move {
let name = channel.name();
match channel.deliver(&alert).await {
Ok(()) => tracing::debug!(
channel = name,
dedup_key = %alert.dedup_key,
"operator alert delivered"
),
Err(error) => tracing::error!(
channel = name,
dedup_key = %alert.dedup_key,
error = %error,
"operator alert delivery failed; app continues serving"
),
}
});
}
}
}
#[must_use]
pub fn alerter(state: &AppState) -> Option<Arc<Alerter>> {
state.extension::<Alerter>()
}
pub fn notify_dead_lettered_job(state: &AppState, job_name: &str, job_id: &str, error: &str) {
let Some(alerter) = alerter(state) else {
return;
};
let alert = Alert::trigger(
AlertCondition::DeadLetteredJob,
format!("dead_lettered_job:{job_name}"),
)
.title(format!("Job '{job_name}' (id {job_id}) was dead-lettered"))
.summary(format!(
"Background job '{job_name}' (id {job_id}) exhausted its retries and was moved to the dead-letter queue. Last error: {error}"
))
.where_to_look(sensitive_gated_where_to_look(
&alerter.settings().actuator_prefix,
AlertCondition::DeadLetteredJob,
alerter.settings().actuator_sensitive,
))
.detail("job", job_name)
.detail("job_id", job_id)
.detail("error", error)
.build();
let _ = alerter.notify(alert);
}
pub fn notify_scheduled_task_failure(state: &AppState, task_name: &str, error: &str) {
let Some(alerter) = alerter(state) else {
return;
};
let alert = Alert::trigger(
AlertCondition::ScheduledTaskFailure,
format!("scheduled_task_failure:{task_name}"),
)
.title(format!("Scheduled task '{task_name}' failed"))
.summary(format!(
"The framework-scheduled task '{task_name}' returned an error on its last run: {error}"
))
.where_to_look(sensitive_gated_where_to_look(
&alerter.settings().actuator_prefix,
AlertCondition::ScheduledTaskFailure,
alerter.settings().actuator_sensitive,
))
.detail("task", task_name)
.detail("error", error)
.build();
let _ = alerter.notify(alert);
}
pub fn notify_scheduled_task_recovered(state: &AppState, task_name: &str) {
let Some(alerter) = alerter(state) else {
return;
};
let alert = Alert::recovery(
AlertCondition::ScheduledTaskFailure,
format!("scheduled_task_failure:{task_name}"),
)
.title(format!("Scheduled task '{task_name}' recovered"))
.summary(format!(
"The framework-scheduled task '{task_name}' completed successfully after a previous failure."
))
.where_to_look(sensitive_gated_where_to_look(
&alerter.settings().actuator_prefix,
AlertCondition::ScheduledTaskFailure,
alerter.settings().actuator_sensitive,
))
.detail("task", task_name)
.build();
let _ = alerter.recover(alert);
}
const fn default_alerts_enabled() -> bool {
true
}
const fn default_dedup_window_secs() -> u64 {
900
}
const fn default_health_grace_secs() -> u64 {
60
}
const fn default_error_rate_threshold() -> f64 {
0.05
}
const fn default_error_rate_min_requests() -> u64 {
20
}
const fn default_eval_interval_secs() -> u64 {
30
}
pub const PAGERDUTY_EVENTS_URL: &str = "https://events.pagerduty.com/v2/enqueue";
#[derive(Debug, Clone, Deserialize)]
#[serde(default)]
pub struct AlertConfig {
pub enabled: bool,
pub email: Option<String>,
pub webhook_url: Option<String>,
pub webhook_secret: Option<String>,
pub pagerduty_routing_key: Option<String>,
pub pagerduty_url: Option<String>,
#[serde(default)]
pub pagerduty_severities: AlertRouting,
pub slack_webhook_url: Option<String>,
#[serde(default)]
pub slack_severities: AlertRouting,
pub discord_webhook_url: Option<String>,
#[serde(default)]
pub discord_severities: AlertRouting,
#[serde(default)]
pub custom_channel: bool,
pub dedup_window_secs: u64,
pub health_grace_secs: u64,
pub error_rate_threshold: f64,
pub error_rate_min_requests: u64,
pub eval_interval_secs: u64,
}
impl Default for AlertConfig {
fn default() -> Self {
Self {
enabled: default_alerts_enabled(),
email: None,
webhook_url: None,
webhook_secret: None,
pagerduty_routing_key: None,
pagerduty_url: None,
pagerduty_severities: AlertRouting::All,
slack_webhook_url: None,
slack_severities: AlertRouting::All,
discord_webhook_url: None,
discord_severities: AlertRouting::All,
custom_channel: false,
dedup_window_secs: default_dedup_window_secs(),
health_grace_secs: default_health_grace_secs(),
error_rate_threshold: default_error_rate_threshold(),
error_rate_min_requests: default_error_rate_min_requests(),
eval_interval_secs: default_eval_interval_secs(),
}
}
}
impl AlertConfig {
#[must_use]
pub fn has_destination(&self) -> bool {
let set = |v: &Option<String>| v.as_ref().is_some_and(|s| !s.trim().is_empty());
set(&self.email)
|| set(&self.webhook_url)
|| set(&self.pagerduty_routing_key)
|| set(&self.slack_webhook_url)
|| set(&self.discord_webhook_url)
}
#[must_use]
pub fn is_active(&self) -> bool {
self.enabled && self.has_destination()
}
}
#[cfg(feature = "mail")]
pub struct MailAlertChannel {
mailer: Arc<crate::mail::Mailer>,
to: String,
}
#[cfg(feature = "mail")]
impl MailAlertChannel {
#[must_use]
pub fn new(mailer: Arc<crate::mail::Mailer>, to: impl Into<String>) -> Self {
Self {
mailer,
to: to.into(),
}
}
fn render_text(alert: &Alert) -> String {
use std::fmt::Write as _;
let mut body = format!(
"{}\n\nSeverity: {:?}\nEvent: {:?}\nCondition: {}\nHost/replica: {}\nWhen: {}\nWhere to look next: {}\n\n{}\n",
alert.title,
alert.severity,
alert.event,
alert.condition.as_str(),
alert.host,
alert.timestamp.to_rfc3339(),
alert.where_to_look,
alert.summary,
);
if !alert.details.is_empty() {
body.push_str("\nDetails:\n");
let mut keys: Vec<&String> = alert.details.keys().collect();
keys.sort();
for k in keys {
if let Some(v) = alert.details.get(k) {
let _ = writeln!(body, " {k}: {v}");
}
}
}
body
}
}
#[cfg(feature = "mail")]
impl AlertChannel for MailAlertChannel {
fn name(&self) -> &'static str {
"mail"
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a> {
Box::pin(async move {
let prefix = match alert.event {
AlertEventKind::Trigger => "[ALERT]",
AlertEventKind::Resolve => "[RECOVERED]",
};
let subject = format!("{prefix} {}", alert.title);
let mail = crate::mail::Mail::builder()
.to(self.to.clone())
.subject(subject)
.text(Self::render_text(alert))
.ignore_suppression()
.build()
.map_err(|e| AlertDeliveryError::new("mail", e.to_string()))?;
self.mailer
.send(mail)
.await
.map_err(|e| AlertDeliveryError::new("mail", e.to_string()))
})
}
}
#[cfg(feature = "http-client")]
pub struct WebhookAlertChannel {
client: crate::http_client::Client,
url: String,
secret: Option<String>,
}
#[cfg(feature = "http-client")]
impl WebhookAlertChannel {
#[must_use]
pub fn new(
client: crate::http_client::Client,
url: impl Into<String>,
secret: Option<String>,
) -> Self {
Self {
client,
url: url.into(),
secret,
}
}
}
#[cfg(feature = "http-client")]
impl AlertChannel for WebhookAlertChannel {
fn name(&self) -> &'static str {
"webhook"
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a> {
Box::pin(async move {
let body = serde_json::to_string(alert)
.map_err(|e| AlertDeliveryError::new("webhook", e.to_string()))?;
let mut req = self
.client
.named(&self.url)
.post(&self.url)
.header("Content-Type", "application/json");
if let Some(secret) = self.secret.as_ref() {
let timestamp = Utc::now().timestamp();
let signing_payload = format!("{timestamp}.{body}");
let signature = crate::security::config::hmac_sha256_hex(
secret.as_bytes(),
signing_payload.as_bytes(),
);
req = req.header("Autumn-Signature", format!("t={timestamp},v1={signature}"));
}
let response = req
.text_body(body)
.send()
.await
.map_err(|e| AlertDeliveryError::new("webhook", e.to_string()))?;
if response.is_success() {
Ok(())
} else {
Err(AlertDeliveryError::new(
"webhook",
format!("endpoint returned status {}", response.status()),
))
}
})
}
}
#[cfg(feature = "http-client")]
const fn pagerduty_severity(severity: AlertSeverity) -> &'static str {
match severity {
AlertSeverity::Critical => "critical",
AlertSeverity::Recovery => "info",
}
}
#[cfg(feature = "http-client")]
#[must_use]
pub fn pagerduty_event_payload(alert: &Alert, routing_key: &str) -> serde_json::Value {
const RESERVED: &[&str] = &["condition", "where_to_look", "detail"];
if alert.event == AlertEventKind::Resolve {
return serde_json::json!({
"routing_key": routing_key,
"event_action": "resolve",
"dedup_key": alert.dedup_key,
});
}
let mut custom = serde_json::Map::new();
let mut keys: Vec<&String> = alert.details.keys().collect();
keys.sort();
for k in keys {
if let Some(v) = alert.details.get(k) {
let key = if RESERVED.contains(&k.as_str()) {
format!("custom_{k}")
} else {
k.clone()
};
custom.insert(key, serde_json::Value::String(v.clone()));
}
}
custom.insert(
"condition".to_owned(),
serde_json::Value::String(alert.condition.as_str().to_owned()),
);
custom.insert(
"where_to_look".to_owned(),
serde_json::Value::String(alert.where_to_look.clone()),
);
if !alert.summary.is_empty() {
custom.insert(
"detail".to_owned(),
serde_json::Value::String(alert.summary.clone()),
);
}
serde_json::json!({
"routing_key": routing_key,
"event_action": "trigger",
"dedup_key": alert.dedup_key,
"payload": {
"summary": alert.title,
"source": alert.host,
"severity": pagerduty_severity(alert.severity),
"timestamp": alert.timestamp.to_rfc3339(),
"component": alert.condition.as_str(),
"custom_details": serde_json::Value::Object(custom),
},
})
}
#[cfg(feature = "http-client")]
pub struct PagerDutyAlertChannel {
client: crate::http_client::Client,
url: String,
routing_key: String,
routing: AlertRouting,
}
#[cfg(feature = "http-client")]
impl PagerDutyAlertChannel {
#[must_use]
pub fn new(
client: crate::http_client::Client,
url: impl Into<String>,
routing_key: impl Into<String>,
routing: AlertRouting,
) -> Self {
Self {
client,
url: url.into(),
routing_key: routing_key.into(),
routing,
}
}
}
#[cfg(feature = "http-client")]
impl AlertChannel for PagerDutyAlertChannel {
fn name(&self) -> &'static str {
"pagerduty"
}
fn accepts_severity(&self, severity: AlertSeverity) -> bool {
self.routing.accepts(severity)
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a> {
Box::pin(async move {
let payload = pagerduty_event_payload(alert, &self.routing_key);
let response = self
.client
.named(&self.url)
.post(&self.url)
.json(&payload)
.send()
.await
.map_err(|e| AlertDeliveryError::new("pagerduty", e.to_string()))?;
if response.is_success() {
Ok(())
} else {
Err(AlertDeliveryError::new(
"pagerduty",
format!("endpoint returned status {}", response.status()),
))
}
})
}
}
#[cfg(feature = "http-client")]
fn slack_message_text(alert: &Alert) -> String {
use std::fmt::Write as _;
let (icon, label) = match alert.event {
AlertEventKind::Trigger => ("\u{1f534}", "ALERT"),
AlertEventKind::Resolve => ("\u{2705}", "RECOVERED"),
};
let mut body = format!("{icon} *[{label}] {}*\n", alert.title);
let _ = writeln!(body, "*When:* {}", alert.timestamp.to_rfc3339());
let _ = writeln!(body, "*Host/replica:* {}", alert.host);
let _ = writeln!(body, "*Condition:* {}", alert.condition.as_str());
let _ = writeln!(body, "*Where to look:* {}", alert.where_to_look);
if !alert.summary.is_empty() {
let _ = write!(body, "\n{}", alert.summary);
}
if !alert.details.is_empty() {
body.push_str("\n\n*Details:*");
let mut keys: Vec<&String> = alert.details.keys().collect();
keys.sort();
for k in keys {
if let Some(v) = alert.details.get(k) {
let _ = write!(body, "\n• {k}: {v}");
}
}
}
body
}
#[cfg(feature = "http-client")]
#[must_use]
pub fn slack_message_payload(alert: &Alert) -> serde_json::Value {
serde_json::json!({ "text": slack_message_text(alert) })
}
#[cfg(feature = "http-client")]
pub struct SlackAlertChannel {
client: crate::http_client::Client,
url: String,
name: &'static str,
routing: AlertRouting,
}
#[cfg(feature = "http-client")]
impl SlackAlertChannel {
#[must_use]
pub fn slack(
client: crate::http_client::Client,
url: impl Into<String>,
routing: AlertRouting,
) -> Self {
Self {
client,
url: url.into(),
name: "slack",
routing,
}
}
#[must_use]
pub fn discord(
client: crate::http_client::Client,
url: impl Into<String>,
routing: AlertRouting,
) -> Self {
Self {
client,
url: url.into(),
name: "discord",
routing,
}
}
}
#[cfg(feature = "http-client")]
impl AlertChannel for SlackAlertChannel {
fn name(&self) -> &'static str {
self.name
}
fn accepts_severity(&self, severity: AlertSeverity) -> bool {
self.routing.accepts(severity)
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a> {
Box::pin(async move {
let payload = slack_message_payload(alert);
let response = self
.client
.named(&self.url)
.post(&self.url)
.json(&payload)
.send()
.await
.map_err(|e| AlertDeliveryError::new(self.name, e.to_string()))?;
if response.is_success() {
Ok(())
} else {
Err(AlertDeliveryError::new(
self.name,
format!("endpoint returned status {}", response.status()),
))
}
})
}
}
#[cfg(feature = "http-client")]
#[must_use]
pub fn native_transport_channels(
config: &AlertConfig,
client: &crate::http_client::Client,
) -> Vec<Arc<dyn AlertChannel>> {
let mut channels: Vec<Arc<dyn AlertChannel>> = Vec::new();
if let Some(routing_key) = config
.pagerduty_routing_key
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty())
{
let url = config
.pagerduty_url
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.unwrap_or(PAGERDUTY_EVENTS_URL);
if is_absolute_http_url(url) {
channels.push(Arc::new(PagerDutyAlertChannel::new(
client.clone(),
url.to_owned(),
routing_key.to_owned(),
config.pagerduty_severities,
)));
} else {
tracing::warn!(
pagerduty_url = url,
"alerts: the configured [alerts] pagerduty_url ({url}) is not an absolute \
http(s) URL; no PagerDuty alerts will be delivered. Fix [alerts] pagerduty_url \
(or the AUTUMN_ALERTS__PAGERDUTY_URL env var)."
);
}
}
push_chat_channel(
&mut channels,
client,
config.slack_webhook_url.as_deref(),
"slack",
config.slack_severities,
);
push_chat_channel(
&mut channels,
client,
config.discord_webhook_url.as_deref(),
"discord",
config.discord_severities,
);
channels
}
#[cfg(feature = "http-client")]
fn push_chat_channel(
channels: &mut Vec<Arc<dyn AlertChannel>>,
client: &crate::http_client::Client,
url: Option<&str>,
provider: &'static str,
routing: AlertRouting,
) {
let Some(url) = url.map(str::trim).filter(|s| !s.is_empty()) else {
return;
};
if !is_absolute_https_url(url) {
tracing::warn!(
provider,
webhook_url = url,
"alerts: the configured [alerts] {provider}_webhook_url ({url}) is not an absolute \
https URL; no {provider} alerts will be delivered ({provider} only exposes https \
webhook endpoints). Fix the URL (or the AUTUMN_ALERTS__{PROVIDER}_WEBHOOK_URL env \
var).",
PROVIDER = provider.to_uppercase(),
);
return;
}
let channel: Arc<dyn AlertChannel> = if provider == "discord" {
Arc::new(SlackAlertChannel::discord(
client.clone(),
url.to_owned(),
routing,
))
} else {
Arc::new(SlackAlertChannel::slack(
client.clone(),
url.to_owned(),
routing,
))
};
channels.push(channel);
}
#[cfg(feature = "http-client")]
fn is_absolute_http_url(url: &str) -> bool {
if !(url.starts_with("http://") || url.starts_with("https://")) {
return false;
}
::url::Url::parse(url)
.ok()
.and_then(|parsed| parsed.host_str().map(|h| !h.is_empty()))
.unwrap_or(false)
}
#[cfg(feature = "http-client")]
fn is_absolute_https_url(url: &str) -> bool {
url.starts_with("https://") && is_absolute_http_url(url)
}
#[cfg(feature = "mail")]
fn is_valid_alert_mailbox(email: &str) -> bool {
email.parse::<lettre::message::Mailbox>().is_ok()
}
#[cfg(feature = "mail")]
const fn mail_transport_requires_from(transport: crate::mail::Transport) -> bool {
matches!(transport, crate::mail::Transport::Smtp)
}
#[cfg(feature = "mail")]
fn build_mail_alert_channel(
state: &AppState,
mailer: Arc<crate::mail::Mailer>,
email: &str,
) -> Option<Arc<dyn AlertChannel>> {
let email_trimmed = email.trim();
if mailer.is_disabled() {
tracing::warn!(
"alerts: an operator email is configured but [mail] transport is disabled; \
no email alerts will be delivered. Set [mail] transport to a real backend \
or use a webhook destination."
);
return None;
}
if !is_valid_alert_mailbox(email_trimmed) {
tracing::warn!(
email = email_trimmed,
"alerts: the configured [alerts] email ({email_trimmed}) is not a valid email \
address; no email alerts will be delivered. Fix [alerts] email (or the \
AUTUMN_ALERTS__EMAIL env var)."
);
return None;
}
let mail_cfg = state.config().mail;
if mail_transport_requires_from(mail_cfg.transport)
&& mail_cfg
.from
.as_deref()
.is_none_or(|from| from.trim().is_empty())
{
tracing::warn!(
"alerts: an operator email is configured but [mail] from is not set, so SMTP alert \
delivery will fail; no email alerts will be delivered. Set [mail] from (or the \
AUTUMN_MAIL__FROM env var) to a sender address, or use a webhook destination."
);
return None;
}
Some(Arc::new(MailAlertChannel::new(
mailer,
email_trimmed.to_owned(),
)))
}
fn push_native_transport_channels(
state: &AppState,
config: &AlertConfig,
channels: &mut Vec<Arc<dyn AlertChannel>>,
) {
#[cfg(feature = "http-client")]
{
let client = crate::http_client::Client::from_state(state);
channels.extend(native_transport_channels(config, &client));
}
#[cfg(not(feature = "http-client"))]
{
let _ = (state, channels);
if config
.pagerduty_routing_key
.as_ref()
.is_some_and(|s| !s.trim().is_empty())
|| config
.slack_webhook_url
.as_ref()
.is_some_and(|s| !s.trim().is_empty())
|| config
.discord_webhook_url
.as_ref()
.is_some_and(|s| !s.trim().is_empty())
{
tracing::warn!(
"operator-alerts: a PagerDuty/Slack/Discord transport is configured but the \
`http-client` feature is not enabled; no such alerts will be delivered. Enable \
the `http-client` feature or use an email destination."
);
}
}
}
pub fn install_from_config(
state: &AppState,
config: &AlertConfig,
extra_channels: Vec<Arc<dyn AlertChannel>>,
) {
if !config.enabled {
return;
}
let mut channels: Vec<Arc<dyn AlertChannel>> = Vec::new();
#[cfg(feature = "mail")]
if let Some(email) = config.email.as_ref().filter(|s| !s.trim().is_empty()) {
if let Some(mailer) = state.extension::<crate::mail::Mailer>() {
if let Some(channel) = build_mail_alert_channel(state, mailer, email) {
channels.push(channel);
}
} else {
tracing::warn!(
"alerts: an operator email is configured but no mailer is installed; \
configure [mail] to enable email alerts"
);
}
}
#[cfg(not(feature = "mail"))]
if config.email.as_ref().is_some_and(|s| !s.trim().is_empty()) {
tracing::warn!(
"operator-alerts: an email destination is configured but the `mail` feature is not \
enabled; no email alerts will be delivered. Enable the `mail` feature or use a \
webhook destination."
);
}
#[cfg(feature = "http-client")]
if let Some(raw_url) = config.webhook_url.as_ref().filter(|s| !s.trim().is_empty()) {
let url = raw_url.trim();
let secret = config
.webhook_secret
.as_ref()
.map(|s| s.trim())
.filter(|s| !s.is_empty());
if !is_absolute_http_url(url) {
tracing::warn!(
webhook_url = url,
"alerts: the configured [alerts] webhook_url ({url}) is not an absolute http(s) \
URL (it must start with `http://` or `https://` and have a host); no webhook \
alerts will be delivered. Fix [alerts] webhook_url (or the \
AUTUMN_ALERTS__WEBHOOK_URL env var)."
);
} else if let Some(secret) = secret {
let client = crate::http_client::Client::from_state(state);
channels.push(Arc::new(WebhookAlertChannel::new(
client,
url.to_owned(),
Some(secret.to_owned()),
)));
} else {
tracing::warn!(
"alerts: a webhook destination is configured but no `webhook_secret` is set; \
alert webhooks are always signed, so no webhook alerts will be delivered. Set \
[alerts] webhook_secret (or the AUTUMN_ALERTS__WEBHOOK_SECRET env var)."
);
}
}
#[cfg(not(feature = "http-client"))]
if config
.webhook_url
.as_ref()
.is_some_and(|s| !s.trim().is_empty())
{
tracing::warn!(
"operator-alerts: a webhook destination is configured but the `http-client` feature \
is not enabled; no webhook alerts will be delivered. Enable the `http-client` \
feature or use an email destination."
);
}
push_native_transport_channels(state, config, &mut channels);
channels.extend(extra_channels);
if channels.is_empty() {
return;
}
let actuator_cfg = state.config().actuator;
let actuator_sensitive = actuator_cfg.sensitive;
let actuator_prefix = actuator_cfg.prefix;
let settings = AlerterSettings::from_config(config, actuator_prefix, actuator_sensitive);
let alerter = Alerter::new(channels, settings);
state.insert_extension(alerter.clone());
spawn_evaluation_loop(state.clone(), alerter);
}
fn spawn_evaluation_loop(state: AppState, alerter: Alerter) {
let settings = alerter.settings().clone();
tokio::spawn(async move {
let mut down_since: HashMap<String, DateTime<Utc>> = HashMap::new();
let (mut last_requests, mut last_5xx) = {
let snap = state.metrics().snapshot();
(snap.http.requests_total, snap.http.by_status.s5xx)
};
loop {
tokio::time::sleep(settings.eval_interval).await;
evaluate_error_rate(
&alerter,
&state,
&settings,
&mut last_requests,
&mut last_5xx,
);
evaluate_health(&alerter, &state, &settings, &mut down_since).await;
}
});
}
const fn sample_error_window(
total: u64,
s5xx: u64,
last_requests: &mut u64,
last_5xx: &mut u64,
min_requests: u64,
) -> Option<(u64, u64)> {
let req_delta = total.saturating_sub(*last_requests);
if req_delta < min_requests {
return None;
}
let err_delta = s5xx.saturating_sub(*last_5xx);
*last_requests = total;
*last_5xx = s5xx;
Some((req_delta, err_delta))
}
fn evaluate_error_rate(
alerter: &Alerter,
state: &AppState,
settings: &AlerterSettings,
last_requests: &mut u64,
last_5xx: &mut u64,
) {
let snap = state.metrics().snapshot();
let total = snap.http.requests_total;
let s5xx = snap.http.by_status.s5xx;
let Some((req_delta, err_delta)) = sample_error_window(
total,
s5xx,
last_requests,
last_5xx,
settings.error_rate_min_requests,
) else {
return;
};
#[allow(clippy::cast_precision_loss)]
let rate = err_delta as f64 / req_delta as f64;
let key = error_rate_dedup_key(&host_id());
if rate >= settings.error_rate_threshold {
let pct = rate * 100.0;
let threshold_pct = settings.error_rate_threshold * 100.0;
let alert = Alert::trigger(AlertCondition::HighErrorRate, key)
.title(format!("5xx error rate is {pct:.1}%"))
.summary(format!(
"{err_delta} of the last {req_delta} requests returned 5xx ({pct:.1}%), \
crossing the {threshold_pct:.1}% threshold."
))
.where_to_look(actuator_where_to_look(
&settings.actuator_prefix,
AlertCondition::HighErrorRate,
))
.detail("rate", format!("{rate:.4}"))
.detail("errors", err_delta.to_string())
.detail("requests", req_delta.to_string())
.build();
let _ = alerter.notify(alert);
} else {
let alert = Alert::recovery(AlertCondition::HighErrorRate, key)
.title("5xx error rate recovered")
.summary(format!(
"The 5xx error rate is back under the threshold ({pct:.1}% of {req_delta} requests).",
pct = rate * 100.0,
))
.where_to_look(actuator_where_to_look(
&settings.actuator_prefix,
AlertCondition::HighErrorRate,
))
.build();
let _ = alerter.recover(alert);
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum HealthTransition {
Trigger,
Recover,
}
const fn classify_health_transition(status: crate::actuator::HealthStatus) -> HealthTransition {
if status.is_healthy() {
HealthTransition::Recover
} else {
HealthTransition::Trigger
}
}
async fn evaluate_health(
alerter: &Alerter,
state: &AppState,
settings: &AlerterSettings,
down_since: &mut HashMap<String, DateTime<Utc>>,
) {
let results = state.health_indicator_registry().run_all().await;
let now = Utc::now();
let grace = chrono::Duration::from_std(settings.health_grace)
.unwrap_or_else(|_| chrono::Duration::seconds(60));
let host = host_id();
for result in &results {
let key = health_indicator_dedup_key(&result.name, &host);
match classify_health_transition(result.output.status) {
HealthTransition::Trigger => {
let status = result.output.status;
let first = *down_since.entry(result.name.clone()).or_insert(now);
if now - first >= grace {
let secs = (now - first).num_seconds().max(0);
let alert = Alert::trigger(AlertCondition::HealthIndicatorDown, key)
.title(format!("Health indicator '{}' is {status:?}", result.name))
.summary(format!(
"Health indicator '{}' has reported {status:?} for {secs}s, past the \
{grace_secs}s grace period.",
result.name,
grace_secs = settings.health_grace.as_secs(),
))
.where_to_look(actuator_where_to_look(
&settings.actuator_prefix,
AlertCondition::HealthIndicatorDown,
))
.detail("indicator", result.name.clone())
.detail("status", status.as_str())
.detail("down_seconds", secs.to_string())
.build();
let _ = alerter.notify(alert);
}
}
HealthTransition::Recover => {
if down_since.remove(&result.name).is_some() {
let alert = Alert::recovery(AlertCondition::HealthIndicatorDown, key)
.title(format!("Health indicator '{}' recovered", result.name))
.summary(format!(
"Health indicator '{}' is reporting healthy again.",
result.name
))
.where_to_look(actuator_where_to_look(
&settings.actuator_prefix,
AlertCondition::HealthIndicatorDown,
))
.detail("indicator", result.name.clone())
.build();
let _ = alerter.recover(alert);
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::Mutex as StdMutex;
fn ts(secs: i64) -> DateTime<Utc> {
DateTime::from_timestamp(1_700_000_000 + secs, 0).expect("valid timestamp")
}
#[test]
fn first_trigger_sends() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_trigger("k", ts(0)), DedupDecision::Send);
}
#[test]
fn repeated_triggers_within_window_are_suppressed() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_trigger("k", ts(0)), DedupDecision::Send);
for s in 1..100 {
assert_eq!(
d.on_trigger("k", ts(s)),
DedupDecision::Suppress,
"occurrence at {s}s inside window must be suppressed"
);
}
}
#[test]
fn trigger_renotifies_once_after_window_elapses() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_trigger("k", ts(0)), DedupDecision::Send);
assert_eq!(d.on_trigger("k", ts(500)), DedupDecision::Suppress);
assert_eq!(d.on_trigger("k", ts(901)), DedupDecision::Send);
assert_eq!(d.on_trigger("k", ts(902)), DedupDecision::Suppress);
}
#[test]
fn distinct_keys_are_independent() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_trigger("a", ts(0)), DedupDecision::Send);
assert_eq!(d.on_trigger("b", ts(0)), DedupDecision::Send);
}
#[test]
fn recovery_sends_only_for_active_key() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_resolve("k"), DedupDecision::Suppress);
assert_eq!(d.on_trigger("k", ts(0)), DedupDecision::Send);
assert_eq!(d.on_resolve("k"), DedupDecision::Send);
assert_eq!(d.on_resolve("k"), DedupDecision::Suppress);
}
#[test]
fn trigger_after_recovery_alerts_immediately() {
let mut d = AlertDeduplicator::new(std::time::Duration::from_secs(900));
assert_eq!(d.on_trigger("k", ts(0)), DedupDecision::Send);
assert_eq!(d.on_resolve("k"), DedupDecision::Send);
assert_eq!(d.on_trigger("k", ts(10)), DedupDecision::Send);
}
use crate::actuator::HealthStatus;
#[test]
fn health_down_status_drives_trigger_branch() {
assert_eq!(
classify_health_transition(HealthStatus::Down),
HealthTransition::Trigger
);
}
#[test]
fn healthy_statuses_recover() {
assert_eq!(
classify_health_transition(HealthStatus::Up),
HealthTransition::Recover
);
assert_eq!(
classify_health_transition(HealthStatus::Unknown),
HealthTransition::Recover
);
}
#[test]
fn out_of_service_triggers_like_down() {
assert!(!HealthStatus::OutOfService.is_healthy());
assert_eq!(
classify_health_transition(HealthStatus::OutOfService),
HealthTransition::Trigger
);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum HealthTick {
Triggered(DedupDecision),
Recovered(DedupDecision),
Idle,
}
fn health_tick_for_sequence(statuses: &[HealthStatus], grace: chrono::Duration) -> HealthTick {
let name = "db";
let key = format!("health_indicator_down:{name}");
let mut dedup = AlertDeduplicator::new(std::time::Duration::from_secs(900));
let mut down_since: HashMap<String, DateTime<Utc>> = HashMap::new();
let mut last = HealthTick::Idle;
for (i, status) in statuses.iter().enumerate() {
last = HealthTick::Idle;
let now = ts(i64::try_from(i).unwrap());
match classify_health_transition(*status) {
HealthTransition::Trigger => {
let first = *down_since.entry(name.to_owned()).or_insert(now);
if now - first >= grace {
last = HealthTick::Triggered(dedup.on_trigger(&key, now));
}
}
HealthTransition::Recover => {
if down_since.remove(name).is_some() {
last = HealthTick::Recovered(dedup.on_resolve(&key));
}
}
}
}
last
}
fn recovery_tick_for_sequence(statuses: &[HealthStatus]) -> HealthTick {
health_tick_for_sequence(statuses, chrono::Duration::seconds(0))
}
#[test]
fn out_of_service_without_prior_down_triggers_after_grace() {
let tick = recovery_tick_for_sequence(&[HealthStatus::OutOfService]);
assert_eq!(
tick,
HealthTick::Triggered(DedupDecision::Send),
"OutOfService with no prior Down must fire an alert once past the grace period"
);
}
#[test]
fn down_then_out_of_service_does_not_recover() {
let tick = recovery_tick_for_sequence(&[HealthStatus::Down, HealthStatus::OutOfService]);
assert_eq!(
tick,
HealthTick::Triggered(DedupDecision::Suppress),
"OutOfService after Down must stay active (no recovery, deduped trigger)"
);
}
#[test]
fn down_then_up_recovers() {
let tick = recovery_tick_for_sequence(&[HealthStatus::Down, HealthStatus::Up]);
assert_eq!(
tick,
HealthTick::Recovered(DedupDecision::Send),
"a genuinely healthy status after Down must emit a recovery"
);
}
#[test]
fn down_then_out_of_service_then_up_recovers_once() {
let tick = recovery_tick_for_sequence(&[
HealthStatus::Down,
HealthStatus::OutOfService,
HealthStatus::Up,
]);
assert_eq!(tick, HealthTick::Recovered(DedupDecision::Send));
}
#[test]
fn brief_unhealthy_blip_inside_grace_window_does_not_alert() {
let tick = health_tick_for_sequence(
&[HealthStatus::OutOfService, HealthStatus::Up],
chrono::Duration::seconds(60),
);
assert_eq!(
tick,
HealthTick::Recovered(DedupDecision::Suppress),
"an unhealthy blip inside the grace window must not send a trigger, and its \
resolve must be suppressed (no active alert to clear)"
);
}
#[test]
fn trigger_is_critical_resolve_is_recovery() {
let t = Alert::trigger(AlertCondition::DeadLetteredJob, "k").build();
assert_eq!(t.severity, AlertSeverity::Critical);
assert_eq!(t.event, AlertEventKind::Trigger);
let r = Alert::recovery(AlertCondition::DeadLetteredJob, "k").build();
assert_eq!(r.severity, AlertSeverity::Recovery);
assert_eq!(r.event, AlertEventKind::Resolve);
}
#[test]
fn where_to_look_defaults_per_condition() {
assert_eq!(
Alert::trigger(AlertCondition::DeadLetteredJob, "k")
.build()
.where_to_look,
"/actuator/jobs"
);
assert_eq!(
Alert::trigger(AlertCondition::HighErrorRate, "k")
.build()
.where_to_look,
"/actuator/metrics"
);
assert_eq!(
Alert::trigger(AlertCondition::HealthIndicatorDown, "k")
.build()
.where_to_look,
"/actuator/health"
);
assert_eq!(
Alert::trigger(AlertCondition::ScheduledTaskFailure, "k")
.build()
.where_to_look,
"/actuator/tasks"
);
}
#[test]
fn actuator_where_to_look_honors_prefix() {
assert_eq!(
actuator_where_to_look("/actuator", AlertCondition::DeadLetteredJob),
"/actuator/jobs"
);
assert_eq!(
actuator_where_to_look("/actuator", AlertCondition::HealthIndicatorDown),
"/actuator/health"
);
assert_eq!(
actuator_where_to_look("/_ops", AlertCondition::DeadLetteredJob),
"/_ops/jobs"
);
assert_eq!(
actuator_where_to_look("/_ops", AlertCondition::HighErrorRate),
"/_ops/metrics"
);
assert_eq!(
actuator_where_to_look("/_ops", AlertCondition::ScheduledTaskFailure),
"/_ops/tasks"
);
assert_eq!(
actuator_where_to_look("_ops/", AlertCondition::HealthIndicatorDown),
"/_ops/health"
);
assert_eq!(
actuator_where_to_look("/", AlertCondition::DeadLetteredJob),
"/jobs"
);
}
#[test]
fn sensitive_gated_where_to_look_points_at_mounted_endpoint() {
assert_eq!(
sensitive_gated_where_to_look("/actuator", AlertCondition::DeadLetteredJob, true),
"/actuator/jobs"
);
assert_eq!(
sensitive_gated_where_to_look("/actuator", AlertCondition::ScheduledTaskFailure, true),
"/actuator/tasks"
);
let dead =
sensitive_gated_where_to_look("/actuator", AlertCondition::DeadLetteredJob, false);
assert!(
!dead.starts_with("/actuator/jobs"),
"must not link the unmounted /jobs endpoint directly: {dead}"
);
assert!(
dead.contains("/actuator/health"),
"must point at the always-mounted /health endpoint: {dead}"
);
assert!(
dead.contains("/actuator/jobs") && dead.contains("[actuator] sensitive = true"),
"must hint that /jobs requires enabling [actuator] sensitive: {dead}"
);
let task =
sensitive_gated_where_to_look("/actuator", AlertCondition::ScheduledTaskFailure, false);
assert!(
!task.starts_with("/actuator/tasks"),
"must not link the unmounted /tasks endpoint directly: {task}"
);
assert!(
task.contains("/actuator/health"),
"must point at the always-mounted /health endpoint: {task}"
);
assert!(
task.contains("/actuator/tasks") && task.contains("[actuator] sensitive = true"),
"must hint that /tasks requires enabling [actuator] sensitive: {task}"
);
let custom = sensitive_gated_where_to_look("/_ops", AlertCondition::DeadLetteredJob, false);
assert!(custom.contains("/_ops/health"));
assert!(custom.contains("/_ops/jobs"));
}
#[test]
fn stable_dedup_key_survives_trigger_and_recovery() {
let key = "dead_lettered_job:emailer";
let t = Alert::trigger(AlertCondition::DeadLetteredJob, key).build();
let r = Alert::recovery(AlertCondition::DeadLetteredJob, key).build();
assert_eq!(t.dedup_key, r.dedup_key);
}
#[test]
fn error_rate_dedup_key_is_host_scoped() {
assert_ne!(
error_rate_dedup_key("replica-a"),
error_rate_dedup_key("replica-b"),
"two replicas must not share the 5xx dedup key"
);
let host = "replica-a";
let trigger_key = error_rate_dedup_key(host);
let resolve_key = error_rate_dedup_key(host);
assert_eq!(
trigger_key, resolve_key,
"same-host trigger and resolve must correlate"
);
assert_eq!(trigger_key, "high_error_rate:5xx:replica-a");
}
#[test]
fn health_indicator_dedup_key_is_host_scoped() {
assert_ne!(
health_indicator_dedup_key("db", "replica-a"),
health_indicator_dedup_key("db", "replica-b"),
"two replicas must not share a health dedup key for the same indicator"
);
assert_ne!(
health_indicator_dedup_key("db", "replica-a"),
health_indicator_dedup_key("cache", "replica-a"),
);
let host = "replica-a";
let trigger_key = health_indicator_dedup_key("db", host);
let resolve_key = health_indicator_dedup_key("db", host);
assert_eq!(
trigger_key, resolve_key,
"same-host trigger and resolve must correlate"
);
assert_eq!(trigger_key, "health_indicator_down:db:replica-a");
}
#[derive(Default)]
struct CapturingChannel {
received: Arc<StdMutex<Vec<Alert>>>,
name: &'static str,
}
impl AlertChannel for CapturingChannel {
fn name(&self) -> &'static str {
self.name
}
fn deliver<'a>(&'a self, alert: &'a Alert) -> AlertDeliveryFuture<'a> {
let received = Arc::clone(&self.received);
let cloned = alert.clone();
Box::pin(async move {
received.lock().expect("lock").push(cloned);
Ok(())
})
}
}
fn settings() -> AlerterSettings {
AlerterSettings {
dedup_window: std::time::Duration::from_secs(900),
health_grace: std::time::Duration::from_secs(60),
error_rate_threshold: 0.05,
error_rate_min_requests: 20,
eval_interval: std::time::Duration::from_secs(30),
actuator_prefix: "/actuator".to_owned(),
actuator_sensitive: false,
}
}
#[tokio::test]
async fn dispatch_fans_out_to_all_channels() {
let a = Arc::new(StdMutex::new(Vec::new()));
let b = Arc::new(StdMutex::new(Vec::new()));
let channels: Vec<Arc<dyn AlertChannel>> = vec![
Arc::new(CapturingChannel {
received: Arc::clone(&a),
name: "a",
}),
Arc::new(CapturingChannel {
received: Arc::clone(&b),
name: "b",
}),
];
let alerter = Alerter::new(channels, settings());
assert!(
alerter.notify(
Alert::trigger(AlertCondition::DeadLetteredJob, "dead_lettered_job:x")
.title("x failed")
.build()
)
);
tokio::task::yield_now().await;
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
assert_eq!(a.lock().expect("lock").len(), 1);
assert_eq!(b.lock().expect("lock").len(), 1);
assert_eq!(a.lock().expect("lock")[0].title, "x failed");
}
#[tokio::test]
async fn sustained_condition_dispatches_once_per_window() {
let seen = Arc::new(StdMutex::new(Vec::new()));
let channels: Vec<Arc<dyn AlertChannel>> = vec![Arc::new(CapturingChannel {
received: Arc::clone(&seen),
name: "cap",
})];
let alerter = Alerter::new(channels, settings());
let mut sent = 0;
for _ in 0..50 {
if alerter.notify(
Alert::trigger(AlertCondition::HighErrorRate, "high_error_rate:5xx").build(),
) {
sent += 1;
}
}
assert_eq!(sent, 1, "sustained condition must alert once per window");
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
assert_eq!(seen.lock().expect("lock").len(), 1);
}
#[test]
fn config_active_requires_enabled_and_destination() {
let mut c = AlertConfig::default();
assert!(!c.is_active(), "no destination -> inactive");
c.email = Some("ops@example.com".to_owned());
assert!(c.is_active());
c.enabled = false;
assert!(!c.is_active(), "disabled -> inactive even with destination");
}
fn from_config_with_threshold(threshold: f64) -> AlerterSettings {
let config = AlertConfig {
error_rate_threshold: threshold,
..AlertConfig::default()
};
AlerterSettings::from_config(&config, "/actuator".to_owned(), false)
}
fn assert_threshold_eq(actual: f64, expected: f64, msg: &str) {
assert_eq!(actual.to_bits(), expected.to_bits(), "{msg}");
}
#[test]
fn valid_error_rate_threshold_is_kept() {
assert_threshold_eq(
from_config_with_threshold(0.2).error_rate_threshold,
0.2,
"a valid (0, 1] threshold must be kept",
);
assert_threshold_eq(
from_config_with_threshold(1.0).error_rate_threshold,
1.0,
"the boundary 1.0 must be kept",
);
}
#[test]
fn nan_error_rate_threshold_falls_back_to_default() {
assert_threshold_eq(
from_config_with_threshold(f64::NAN).error_rate_threshold,
default_error_rate_threshold(),
"NaN threshold must fall back to the default (can't be clamped)",
);
assert_threshold_eq(
from_config_with_threshold(f64::INFINITY).error_rate_threshold,
default_error_rate_threshold(),
"infinity threshold must fall back to the default",
);
}
#[test]
fn out_of_range_error_rate_threshold_falls_back_to_default() {
assert_threshold_eq(
from_config_with_threshold(1.5).error_rate_threshold,
default_error_rate_threshold(),
"a threshold > 1 must fall back to the default",
);
assert_threshold_eq(
from_config_with_threshold(0.0).error_rate_threshold,
default_error_rate_threshold(),
"a zero threshold must fall back to the default",
);
assert_threshold_eq(
from_config_with_threshold(-0.1).error_rate_threshold,
default_error_rate_threshold(),
"a negative threshold must fall back to the default",
);
}
#[test]
fn sanitized_threshold_behaves_as_default_in_comparison() {
let valid = from_config_with_threshold(0.10).error_rate_threshold;
assert!(0.10 >= valid, "a rate at the threshold fires");
assert!(0.05 < valid, "a rate below the threshold does not fire");
let sanitized = from_config_with_threshold(f64::NAN).error_rate_threshold;
let default = default_error_rate_threshold();
assert!(default >= sanitized, "a rate at the default fires");
assert!(
default - 0.001 < sanitized,
"a rate below the default does not"
);
}
#[test]
fn sub_threshold_ticks_accumulate_until_window_crosses_min_requests() {
let min_requests = 10;
let mut last_requests = 0;
let mut last_5xx = 0;
assert_eq!(
sample_error_window(4, 1, &mut last_requests, &mut last_5xx, min_requests),
None,
"first sub-threshold tick must not evaluate"
);
assert_eq!(
last_requests, 0,
"sub-threshold tick must not reset baseline"
);
assert_eq!(
last_5xx, 0,
"sub-threshold tick must not reset 5xx baseline"
);
assert_eq!(
sample_error_window(8, 2, &mut last_requests, &mut last_5xx, min_requests),
None,
"second sub-threshold tick must not evaluate"
);
assert_eq!(last_requests, 0);
assert_eq!(last_5xx, 0);
assert_eq!(
sample_error_window(12, 3, &mut last_requests, &mut last_5xx, min_requests),
Some((12, 3)),
"window must evaluate once summed requests cross the threshold"
);
assert_eq!(last_requests, 12, "baseline advances only after evaluation");
assert_eq!(last_5xx, 3);
assert_eq!(
sample_error_window(15, 3, &mut last_requests, &mut last_5xx, min_requests),
None
);
assert_eq!(
last_requests, 12,
"baseline held across the next sub-threshold tick"
);
assert_eq!(last_5xx, 3);
}
#[cfg(feature = "http-client")]
#[test]
fn is_absolute_http_url_matches_runtime_absoluteness() {
assert!(is_absolute_http_url(
"https://alerts.example.com/hooks/autumn"
));
assert!(is_absolute_http_url(
"http://alerts.example.com/hooks/autumn"
));
assert!(is_absolute_http_url("http://h.example.com:8080/x?a=1#f"));
assert!(!is_absolute_http_url("hooks.example/x"));
assert!(!is_absolute_http_url("/hooks/autumn"));
assert!(!is_absolute_http_url("ftp://h.example.com/x"));
assert!(!is_absolute_http_url("https://"));
assert!(!is_absolute_http_url(""));
}
#[test]
fn alert_serializes_to_json_for_webhook() {
let alert = Alert::trigger(AlertCondition::DeadLetteredJob, "dead_lettered_job:x")
.title("x failed")
.detail("job", "x")
.build();
let v: serde_json::Value = serde_json::to_value(&alert).expect("serialize");
assert_eq!(v["dedup_key"], "dead_lettered_job:x");
assert_eq!(v["condition"], "dead_lettered_job");
assert_eq!(v["severity"], "critical");
assert_eq!(v["event"], "trigger");
assert_eq!(v["where_to_look"], "/actuator/jobs");
}
#[test]
fn alert_routing_accepts_matches_severity() {
assert!(AlertRouting::All.accepts(AlertSeverity::Critical));
assert!(AlertRouting::All.accepts(AlertSeverity::Recovery));
assert!(AlertRouting::Critical.accepts(AlertSeverity::Critical));
assert!(!AlertRouting::Critical.accepts(AlertSeverity::Recovery));
}
#[test]
fn alert_routing_defaults_to_all() {
assert_eq!(AlertRouting::default(), AlertRouting::All);
}
#[test]
fn routing_deserializes_from_toml() {
#[derive(Deserialize)]
struct Holder {
r: AlertRouting,
}
let all: Holder = toml::from_str(r#"r = "all""#).expect("parse all");
assert_eq!(all.r, AlertRouting::All);
let crit: Holder = toml::from_str(r#"r = "critical""#).expect("parse critical");
assert_eq!(crit.r, AlertRouting::Critical);
}
#[test]
fn routing_from_str_matches_toml_spellings() {
assert_eq!("all".parse::<AlertRouting>(), Ok(AlertRouting::All));
assert_eq!(
"critical".parse::<AlertRouting>(),
Ok(AlertRouting::Critical)
);
assert!("warning".parse::<AlertRouting>().is_err());
assert!("All".parse::<AlertRouting>().is_err());
assert!("".parse::<AlertRouting>().is_err());
}
#[cfg(feature = "http-client")]
#[test]
fn pagerduty_trigger_payload_is_events_v2_shaped() {
let alert = Alert::trigger(AlertCondition::DeadLetteredJob, "dead_lettered_job:emailer")
.title("Job 'emailer' was dead-lettered")
.summary("exhausted retries")
.detail("job", "emailer")
.build();
let v = pagerduty_event_payload(&alert, "R0UT1NGKEY");
assert_eq!(v["routing_key"], "R0UT1NGKEY");
assert_eq!(v["event_action"], "trigger");
assert_eq!(v["dedup_key"], "dead_lettered_job:emailer");
assert_eq!(v["payload"]["severity"], "critical");
assert_eq!(v["payload"]["source"], alert.host);
assert_eq!(v["payload"]["summary"], "Job 'emailer' was dead-lettered");
assert_eq!(v["payload"]["component"], "dead_lettered_job");
assert_eq!(
v["payload"]["custom_details"]["condition"],
"dead_lettered_job"
);
assert_eq!(
v["payload"]["custom_details"]["where_to_look"],
"/actuator/jobs"
);
assert_eq!(v["payload"]["custom_details"]["job"], "emailer");
}
#[cfg(feature = "http-client")]
#[test]
fn pagerduty_custom_details_preserves_user_key_colliding_with_reserved_field() {
let alert = Alert::trigger(AlertCondition::DeadLetteredJob, "dead_lettered_job:emailer")
.title("Job 'emailer' was dead-lettered")
.summary("exhausted retries")
.detail("condition", "user-supplied-condition")
.detail("where_to_look", "user-supplied-pointer")
.detail("detail", "user-supplied-detail")
.build();
let v = pagerduty_event_payload(&alert, "R0UT1NGKEY");
let cd = &v["payload"]["custom_details"];
assert_eq!(cd["condition"], "dead_lettered_job");
assert_eq!(cd["where_to_look"], "/actuator/jobs");
assert_eq!(cd["detail"], "exhausted retries");
assert_eq!(cd["custom_condition"], "user-supplied-condition");
assert_eq!(cd["custom_where_to_look"], "user-supplied-pointer");
assert_eq!(cd["custom_detail"], "user-supplied-detail");
}
#[cfg(feature = "http-client")]
#[test]
fn pagerduty_resolve_payload_is_minimal_and_correlates() {
let alert = Alert::recovery(
AlertCondition::ScheduledTaskFailure,
"scheduled_task_failure:backup",
)
.title("Scheduled task 'backup' recovered")
.build();
let v = pagerduty_event_payload(&alert, "R0UT1NGKEY");
assert_eq!(v["event_action"], "resolve");
assert_eq!(v["dedup_key"], "scheduled_task_failure:backup");
assert!(v.get("payload").is_none(), "resolve must omit payload: {v}");
}
#[cfg(feature = "http-client")]
#[test]
fn slack_payload_carries_required_operator_fields() {
let alert = Alert::trigger(AlertCondition::HighErrorRate, "high_error_rate:5xx:h1")
.title("5xx error rate is 12.0%")
.summary("60 of the last 500 requests returned 5xx")
.where_to_look("/actuator/metrics")
.detail("rate", "0.1200")
.build();
let v = slack_message_payload(&alert);
let text = v["text"].as_str().expect("text field");
assert!(text.contains("5xx error rate is 12.0%"), "title: {text}");
assert!(text.contains(&alert.host), "host: {text}");
assert!(text.contains("/actuator/metrics"), "where_to_look: {text}");
assert!(
text.contains(&alert.timestamp.to_rfc3339()),
"timestamp: {text}"
);
assert!(text.contains("rate: 0.1200"), "details: {text}");
}
#[cfg(feature = "http-client")]
#[test]
fn pagerduty_channel_accepts_severity_reflects_routing() {
let crit = PagerDutyAlertChannel::new(
crate::http_client::Client::new(),
PAGERDUTY_EVENTS_URL,
"k",
AlertRouting::Critical,
);
assert_eq!(crit.name(), "pagerduty");
assert!(crit.accepts_severity(AlertSeverity::Critical));
assert!(!crit.accepts_severity(AlertSeverity::Recovery));
}
#[cfg(feature = "http-client")]
#[test]
fn slack_and_discord_channels_name_and_route() {
let slack = SlackAlertChannel::slack(
crate::http_client::Client::new(),
"https://hooks.slack.com/services/x",
AlertRouting::Critical,
);
assert_eq!(slack.name(), "slack");
assert!(!slack.accepts_severity(AlertSeverity::Recovery));
let discord = SlackAlertChannel::discord(
crate::http_client::Client::new(),
"https://discord.com/api/webhooks/x/y/slack",
AlertRouting::All,
);
assert_eq!(discord.name(), "discord");
assert!(discord.accepts_severity(AlertSeverity::Recovery));
}
#[cfg(feature = "http-client")]
fn channel_names(config: &AlertConfig) -> Vec<&'static str> {
let client = crate::http_client::Client::new();
native_transport_channels(config, &client)
.iter()
.map(|c| c.name())
.collect()
}
#[cfg(feature = "http-client")]
#[test]
fn native_transports_build_all_three_when_configured() {
let config = AlertConfig {
pagerduty_routing_key: Some("R0UT1NGKEY".to_owned()),
slack_webhook_url: Some("https://hooks.slack.com/services/x".to_owned()),
discord_webhook_url: Some("https://discord.com/api/webhooks/x/y/slack".to_owned()),
..AlertConfig::default()
};
let names = channel_names(&config);
assert!(names.contains(&"pagerduty"), "{names:?}");
assert!(names.contains(&"slack"), "{names:?}");
assert!(names.contains(&"discord"), "{names:?}");
}
#[cfg(feature = "http-client")]
#[test]
fn native_transports_skip_blank_routing_key_and_relative_urls() {
let config = AlertConfig {
pagerduty_routing_key: Some(" ".to_owned()),
slack_webhook_url: Some("hooks.slack/x".to_owned()),
discord_webhook_url: Some("/relative".to_owned()),
..AlertConfig::default()
};
assert!(
channel_names(&config).is_empty(),
"a blank routing key and relative URLs must register no channel"
);
}
#[cfg(feature = "http-client")]
#[test]
fn native_transports_skip_non_absolute_pagerduty_url() {
let config = AlertConfig {
pagerduty_routing_key: Some("R0UT1NGKEY".to_owned()),
pagerduty_url: Some("events.pagerduty.com/v2/enqueue".to_owned()),
..AlertConfig::default()
};
assert!(
channel_names(&config).is_empty(),
"a non-absolute pagerduty_url must skip the PagerDuty channel"
);
}
#[cfg(feature = "http-client")]
#[test]
fn native_transports_reject_non_https_slack_and_discord() {
let slack = AlertConfig {
slack_webhook_url: Some("http://hooks.slack.com/services/x".to_owned()),
..AlertConfig::default()
};
assert!(
channel_names(&slack).is_empty(),
"a non-https slack_webhook_url must not register a Slack channel"
);
let discord = AlertConfig {
discord_webhook_url: Some("http://discord.com/api/webhooks/x/slack".to_owned()),
..AlertConfig::default()
};
assert!(
channel_names(&discord).is_empty(),
"a non-https discord_webhook_url must not register a Discord channel"
);
let ok = AlertConfig {
slack_webhook_url: Some("https://hooks.slack.com/services/x".to_owned()),
..AlertConfig::default()
};
assert_eq!(channel_names(&ok), vec!["slack"]);
}
#[cfg(feature = "http-client")]
#[test]
fn has_destination_counts_native_transports() {
let pd = AlertConfig {
pagerduty_routing_key: Some("k".to_owned()),
..AlertConfig::default()
};
assert!(pd.has_destination() && pd.is_active());
let slack = AlertConfig {
slack_webhook_url: Some("https://hooks.slack.com/x".to_owned()),
..AlertConfig::default()
};
assert!(slack.has_destination());
assert!(!AlertConfig::default().has_destination());
}
#[cfg(feature = "http-client")]
#[tokio::test]
async fn pagerduty_channel_delivers_trigger_to_events_endpoint() {
use crate::http_client::{Client, MockEntry, MockRegistry};
use std::sync::atomic::AtomicUsize;
let registry = Arc::new(MockRegistry::new());
let calls = Arc::new(AtomicUsize::new(0));
registry.register(MockEntry {
method: Some(reqwest::Method::POST),
path: "/v2/enqueue".to_owned(),
alias: None,
status: 202,
body: Some(serde_json::json!({"status": "success"})),
call_count: calls.clone(),
});
let client = Client::new().with_mock(registry);
let channel = PagerDutyAlertChannel::new(
client,
PAGERDUTY_EVENTS_URL,
"R0UT1NGKEY",
AlertRouting::All,
);
let alert = Alert::trigger(AlertCondition::DeadLetteredJob, "dead_lettered_job:x")
.title("x failed")
.build();
channel.deliver(&alert).await.expect("delivered");
assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
}
#[cfg(feature = "http-client")]
#[tokio::test]
async fn slack_channel_reports_non_2xx_as_delivery_error() {
use crate::http_client::{Client, MockEntry, MockRegistry};
use std::sync::atomic::AtomicUsize;
let registry = Arc::new(MockRegistry::new());
registry.register(MockEntry {
method: Some(reqwest::Method::POST),
path: "/services/x".to_owned(),
alias: None,
status: 500,
body: None,
call_count: Arc::new(AtomicUsize::new(0)),
});
let client = Client::new().with_mock(registry);
let channel = SlackAlertChannel::slack(
client,
"https://hooks.slack.com/services/x",
AlertRouting::All,
);
let alert = Alert::trigger(AlertCondition::HighErrorRate, "high_error_rate:5xx").build();
let err = channel
.deliver(&alert)
.await
.expect_err("must error on 500");
assert_eq!(err.channel, "slack");
}
}