phughk/ottyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Move the ntfy alert sink to Settings, shared by all alert rules

Browse Source 
Alert rules no longer carry their own ntfy topic/server; Settings now
holds one process-wide ntfy sink (with its own "send test notification"),
and the Alerts panel is left to just define rule conditions.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
Hugh Kaznowski
2026-09-05 18:13:51 +01:00
co-authored by Claude Sonnet 5
parent ae4e550c1f
commit 18c2d72d22
9 changed files with 148 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/alerts.rs
+11 -33
View File
@@ -25,7 +25,6 @@ pub struct AlertRule {
pub name: String,
pub enabled: bool,
pub condition: AlertCondition,
pub sink: AlertSink,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -54,17 +53,6 @@ pub enum AlertCondition {
},
}
/// Modeled as an enum (rather than bare fields on `AlertRule`) so more sink kinds can be added later without reshaping it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "lowercase")]
pub enum AlertSink {
Ntfy {
topic: String,
/// Defaults to `https://ntfy.sh` when absent.
server: Option<String>,
},
}
#[derive(Debug)]
pub enum AlertError {
NotFound,
@@ -107,9 +95,8 @@ impl Settings {
name: String,
enabled: bool,
condition: AlertCondition,
sink: AlertSink,
) -> Result<AlertRule, AlertError> {
let rule = AlertRule { id: generate_alert_rule_id(&name), name, enabled, condition, sink };
let rule = AlertRule { id: generate_alert_rule_id(&name), name, enabled, condition };
self.alert_rules.push(rule.clone());
self.save(settings_path)?;
Ok(rule)
@@ -122,7 +109,6 @@ impl Settings {
name: String,
enabled: bool,
condition: AlertCondition,
sink: AlertSink,
) -> Result<AlertRule, AlertError> {
let rule = self
.alert_rules
@@ -132,7 +118,6 @@ impl Settings {
rule.name = name;
rule.enabled = enabled;
rule.condition = condition;
rule.sink = sink;
let updated = rule.clone();
self.save(settings_path)?;
Ok(updated)
@@ -322,15 +307,16 @@ fn evaluate_condition(
}
}
/// Errors are only logged (no retry queue); spawned off the ingest hot path so a slow/unreachable sink never blocks ingestion.
pub async fn deliver_for_rule(rule: AlertRule, message: String) {
match rule.sink {
AlertSink::Ntfy { topic, server } => {
let server = server.unwrap_or_else(|| "https://ntfy.sh".to_string());
if let Err(err) = deliver_ntfy(&server, &topic, &message).await {
eprintln!("warning: failed to deliver alert '{}' via ntfy: {err}", rule.name);
}
}
/// Delivers to the single, process-wide ntfy sink configured in Settings. Errors are only logged
/// (no retry queue); spawned off the ingest hot path so a slow/unreachable sink never blocks ingestion.
/// A silently no-op'd delivery (no topic configured) is expected — a rule can be defined before the sink is.
pub async fn deliver_for_rule(rule_name: String, message: String, topic: Option<String>, server: Option<String>) {
let Some(topic) = topic else {
return;
};
let server = server.unwrap_or_else(|| "https://ntfy.sh".to_string());
if let Err(err) = deliver_ntfy(&server, &topic, &message).await {
eprintln!("warning: failed to deliver alert '{rule_name}' via ntfy: {err}");
}
}
@@ -364,7 +350,6 @@ mod tests {
name: "test rule".to_string(),
enabled: true,
condition: AlertCondition::Immediate { query: query.to_string() },
sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
}
}
@@ -380,7 +365,6 @@ mod tests {
"spike".to_string(),
true,
AlertCondition::Immediate { query: r#"regex("error")"#.to_string() },
AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
)
.unwrap();
assert_eq!(settings.alert_rules.len(), 1);
@@ -395,7 +379,6 @@ mod tests {
"renamed".to_string(),
false,
AlertCondition::Immediate { query: r#"regex("warn")"#.to_string() },
AlertSink::Ntfy { topic: "other".to_string(), server: Some("https://example.com".to_string()) },
)
.unwrap();
assert_eq!(updated.id, created.id);
@@ -420,7 +403,6 @@ mod tests {
"x".to_string(),
true,
AlertCondition::Immediate { query: r#"regex("x")"#.to_string() },
AlertSink::Ntfy { topic: "x".to_string(), server: None },
);
assert!(matches!(result, Err(AlertError::NotFound)));
@@ -463,7 +445,6 @@ mod tests {
threshold: 3.0,
window_ms: 10_000,
},
sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
};
let mut states = HashMap::new();
for (seq, ms) in [(0u64, 0i64), (1, 1000)] {
@@ -489,7 +470,6 @@ mod tests {
threshold: 2.0,
window_ms: 1_000,
},
sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
};
let mut states = HashMap::new();
assert!(evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "error").is_empty());
@@ -507,7 +487,6 @@ mod tests {
field_regex: r"agent=(?P<agent>\S+)".to_string(),
capture_group: "agent".to_string(),
},
sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
};
let mut states = HashMap::new();
let fired = evaluate_alerts(&[rule.clone()], &mut states, 0, Timestamp::from_millis(0), "agent=curl");
@@ -531,7 +510,6 @@ mod tests {
ratio: 0.5,
window_ms: 10_000,
},
sink: AlertSink::Ntfy { topic: "alerts".to_string(), server: None },
};
let mut states = HashMap::new();
// Only an "error" so far; no "request" yet, so denominator is 0 -> skip.
src/api.rs
+11 -8
View File
@@ -423,7 +423,7 @@ async fn create_alert_rule(
Json(body): Json<CreateAlertRuleRequest>,
) -> impl IntoResponse {
let mut settings = state.settings.lock().unwrap();
match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition, body.sink) {
match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition) {
Ok(rule) => (StatusCode::CREATED, Json(rule)).into_response(),
Err(err) => alert_error_response(err),
}
@@ -435,7 +435,7 @@ async fn update_alert_rule(
Json(body): Json<UpdateAlertRuleRequest>,
) -> impl IntoResponse {
let mut settings = state.settings.lock().unwrap();
match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition, body.sink) {
match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition) {
Ok(rule) => Json(rule).into_response(),
Err(err) => alert_error_response(err),
}
@@ -481,6 +481,12 @@ async fn update_settings(
if let Some(custom_theme_id) = body.custom_theme_id {
settings.custom_theme_id = custom_theme_id;
}
if let Some(ntfy_topic) = body.ntfy_topic {
settings.ntfy_topic = ntfy_topic;
}
if let Some(ntfy_server) = body.ntfy_server {
settings.ntfy_server = ntfy_server;
}
match settings.save(&state.settings_path) {
Ok(()) => (),
Err(err) => return (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
@@ -756,8 +762,7 @@ mod tests {
json!({
"name": "error spike",
"enabled": true,
"condition": { "kind": "immediate", "query": "regex(\"error\")" },
"sink": { "kind": "ntfy", "topic": "alerts", "server": null }
"condition": { "kind": "immediate", "query": "regex(\"error\")" }
}),
)
.await;
@@ -776,8 +781,7 @@ mod tests {
json!({
"name": "renamed",
"enabled": false,
"condition": { "kind": "immediate", "query": "regex(\"warn\")" },
"sink": { "kind": "ntfy", "topic": "alerts", "server": null }
"condition": { "kind": "immediate", "query": "regex(\"warn\")" }
}),
)
.await;
@@ -807,8 +811,7 @@ mod tests {
json!({
"name": "x",
"enabled": true,
"condition": { "kind": "immediate", "query": "regex(\"x\")" },
"sink": { "kind": "ntfy", "topic": "x", "server": null }
"condition": { "kind": "immediate", "query": "regex(\"x\")" }
})
.to_string(),
))
src/dto.rs
+12 -4
View File
@@ -1,4 +1,4 @@
use crate::alerts::{AlertCondition, AlertSink};
use crate::alerts::AlertCondition;
use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher};
use crate::metric_primitive::MetricPrimitive;
use crate::profile::{ChartType, Dashboard, Panel, Profile};
@@ -171,6 +171,10 @@ pub struct SettingsDto {
pub max_logs: Option<usize>,
/// Id of the selected custom theme entry, if any. `None` means `color_scheme` alone determines the active palette.
pub custom_theme_id: Option<String>,
/// Ntfy topic every alert rule delivers to; `None` until configured.
pub ntfy_topic: Option<String>,
/// Ntfy server override; `None` means `https://ntfy.sh`.
pub ntfy_server: Option<String>,
}
impl From<&Settings> for SettingsDto {
@@ -180,11 +184,13 @@ impl From<&Settings> for SettingsDto {
active_profile: settings.profiles.active_profile.clone(),
max_logs: settings.max_logs,
custom_theme_id: settings.custom_theme_id.clone(),
ntfy_topic: settings.ntfy_topic.clone(),
ntfy_server: settings.ntfy_server.clone(),
}
}
}
/// `max_logs`/`custom_theme_id` distinguish "omitted" (leave unchanged) from "present" (null clears, value sets), via `deserialize_double_option`.
/// `max_logs`/`custom_theme_id`/`ntfy_topic`/`ntfy_server` distinguish "omitted" (leave unchanged) from "present" (null clears, value sets), via `deserialize_double_option`.
#[derive(Debug, Deserialize)]
pub struct UpdateSettingsRequest {
pub color_scheme: Option<ColorScheme>,
@@ -192,6 +198,10 @@ pub struct UpdateSettingsRequest {
pub max_logs: Option<Option<usize>>,
#[serde(default, deserialize_with = "deserialize_double_option")]
pub custom_theme_id: Option<Option<String>>,
#[serde(default, deserialize_with = "deserialize_double_option")]
pub ntfy_topic: Option<Option<String>>,
#[serde(default, deserialize_with = "deserialize_double_option")]
pub ntfy_server: Option<Option<String>>,
}
/// Distinguishes a missing field (outer `None`) from present `null` (`Some(None)`) or present value (`Some(Some(value))`), which plain `Option<Option<T>>` can't do alone.
@@ -208,7 +218,6 @@ pub struct CreateAlertRuleRequest {
pub name: String,
pub enabled: bool,
pub condition: AlertCondition,
pub sink: AlertSink,
}
#[derive(Debug, Deserialize)]
@@ -216,7 +225,6 @@ pub struct UpdateAlertRuleRequest {
pub name: String,
pub enabled: bool,
pub condition: AlertCondition,
pub sink: AlertSink,
}
/// `server` defaults to `https://ntfy.sh` when absent.
src/engine.rs
+5 -2
View File
@@ -130,10 +130,13 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
evict_excess_logs(&mut index, &mut log_count, max_logs);
// Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path.
let rules = settings.lock().unwrap().alert_rules.clone();
let (rules, ntfy_topic, ntfy_server) = {
let settings = settings.lock().unwrap();
(settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone())
};
let fired = alerts::evaluate_alerts(&rules, &mut alert_states, sequence, timestamp, &text);
for (rule, message) in fired {
tokio::spawn(alerts::deliver_for_rule(rule, message));
tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone()));
}
}
EngineMessage::SetMaxLogs(new_max) => {
src/settings.rs
+9 -4
View File
@@ -78,6 +78,12 @@ pub struct Settings {
/// Alert rule definitions.
#[serde(default)]
pub alert_rules: Vec<AlertRule>,
/// Ntfy topic every alert rule delivers to; `None` until configured in Settings.
#[serde(default)]
pub ntfy_topic: Option<String>,
/// Ntfy server override; `None` means `https://ntfy.sh`.
#[serde(default)]
pub ntfy_server: Option<String>,
}
impl Default for Settings {
@@ -93,6 +99,8 @@ impl Default for Settings {
},
max_logs: None,
alert_rules: Vec::new(),
ntfy_topic: None,
ntfy_server: None,
}
}
}
@@ -333,11 +341,8 @@ mod tests {
condition: crate::alerts::AlertCondition::Immediate {
query: r#"regex("error")"#.to_string(),
},
sink: crate::alerts::AlertSink::Ntfy {
topic: "alerts".to_string(),
server: None,
},
});
settings.ntfy_topic = Some("alerts".to_string());
settings.save(&path).unwrap();