diff --git a/src/core/src/alerts/alert.rs b/src/core/src/alerts/alert.rs index fba317ecd6..10d5d75e48 100644 --- a/src/core/src/alerts/alert.rs +++ b/src/core/src/alerts/alert.rs @@ -80,6 +80,7 @@ use crate::{ custom::{VarValue, process_variable_replace}, derive_channel_format, format::ChannelFormat, + modifiers::{PreparedTemplate, prepare_modifiers}, platform::{self, platform_of}, render, render::slack as slack_render, @@ -3563,7 +3564,49 @@ async fn send_sns_notification( fn process_row_template( org_name: &str, - tpl: &String, + tpl: &str, + alert: &Alert, + row_type: RowTemplateType, + rows: &[Map], +) -> Vec { + let mut fields = std::collections::HashMap::>::new(); + let templates = (0..rows.len()) + .map(|index| { + prepare_modifiers( + tpl, + |field| { + let placeholder = format!("{{{field}}}"); + let values = fields.entry(field.to_string()).or_insert_with(|| { + process_row_templates_plain( + org_name, + &(0..rows.len()) + .map(|_| PreparedTemplate::plain(placeholder.clone())) + .collect::>(), + alert, + RowTemplateType::String, + rows, + ) + }); + let value = values[index].as_str()?; + (value != placeholder).then(|| value.to_string()) + }, + |field| { + rows[index].contains_key(field) + || alert + .context_attributes + .as_ref() + .is_some_and(|attrs| attrs.contains_key(field)) + }, + false, + ) + }) + .collect::>(); + process_row_templates_plain(org_name, &templates, alert, row_type, rows) +} + +fn process_row_templates_plain( + org_name: &str, + templates: &[PreparedTemplate], alert: &Alert, row_type: RowTemplateType, rows: &[Map], @@ -3579,8 +3622,8 @@ fn process_row_template( // For JSON row template type, try to parse the template as JSON let is_json_template = row_type == RowTemplateType::Json; - for row in rows.iter() { - let mut resp = tpl.to_string(); + for (row, tpl) in rows.iter().zip(templates) { + let mut resp = tpl.template.clone(); let mut alert_start_time = 0; let mut alert_end_time = 0; for (key, value) in row.iter() { @@ -3677,6 +3720,8 @@ fn process_row_template( } } + resp = tpl.finish(resp); + // If this is a JSON row template, try to parse it as JSON if is_json_template { match serde_json::from_str::(&resp) { @@ -7315,7 +7360,7 @@ mod tests { let out = process_row_template( "default", - &STREAM_TPL.to_string(), + STREAM_TPL, &alert, RowTemplateType::String, &[row], @@ -8297,3 +8342,208 @@ fn collapse_slo_evals(evals: &[crate::slo::evaluate::SloEvalResult]) -> SloColla } SloCollapse::Healthy } + +#[cfg(test)] +mod modifier_tests { + use serde_json::json; + + use super::*; + + #[test] + fn row_numeric_modifiers_numbers_strings_and_precedence() { + let rows = vec![ + serde_json::from_value(json!({ + "ratio": 0.9123, "bytes": "1048576", "rate": 1234567, + "latency": "3661.9", "alert_count": 1200, "pipe|humanize": "literal", + "message": "abc\"def", "payload": "{ratio|humanizePercentage}" + })) + .unwrap(), + serde_json::from_value(json!({ + "ratio": "0.5", "bytes": 1024, "rate": "2000", "latency": 0.25, + "message": "second" + })) + .unwrap(), + ]; + let mut alert = Alert::default(); + alert.context_attributes = Some([(String::from("rate"), String::from("10"))].into()); + let tpl = r#"{"ratio":"{ratio|humanizePercentage}","bytes":"{bytes|humanize1024}","rate":"{rate|humanize}","latency":"{latency|humanizeDuration}","count":"{alert_count|humanize}","pipe":"{pipe|humanize}","message":"{message:4}","payload":"{payload}"}"#.to_string(); + let rendered = process_row_template("default", &tpl, &alert, RowTemplateType::Json, &rows); + assert_eq!( + rendered[0], + json!({"ratio":"91.23%", "bytes":"1Mi", "rate":"1.235M", "latency":"1h 1m 1s", "count":"1.2k", "pipe":"literal", "message":"abc\"", "payload":"{ratio|humanizePercentage}"}) + ); + assert_eq!(rendered[1]["count"], "2"); + assert_eq!(rendered[1]["ratio"], "50%"); + assert_eq!(rendered[1]["bytes"], "1ki"); + assert_eq!(rendered[1]["rate"], "2k"); + assert_eq!(rendered[1]["latency"], "250ms"); + } + + #[test] + fn row_modifiers_fallback_special_values_and_injection() { + let rows = vec![serde_json::from_value(json!({"bad": "oops", "nan": "NaN", "inf": "+Inf", "negative_inf": "-Inf", "payload": "{alert_count|humanizePercentage}"})).unwrap()]; + let tpl = "{bad|humanize} {missing|humanize} {nan|humanize} {inf|humanizeDuration} {negative_inf|humanizePercentage} {alert_count|unknown} {alert_count|humanize:2} {payload}".to_string(); + let rendered = process_row_template( + "default", + &tpl, + &Alert::default(), + RowTemplateType::String, + &rows, + ); + assert_eq!(rendered, vec![Value::String("{bad|humanize} {missing|humanize} NaN +Inf -Inf% {alert_count|unknown} {alert_count|humanize:2} {alert_count|humanizePercentage}".into())]); + let ctx = NotificationContext { + alert_count: "1".into(), + rows_tpl_val: rendered, + ..Default::default() + }; + let embedded = apply_custom_template("{rows} {alert_count|humanizePercentage}", &ctx, true); + assert!(embedded.ends_with("{alert_count|humanizePercentage} 100%")); + } + + #[test] + fn row_timestamp_and_size_modifiers() { + let rows = vec![serde_json::from_value(json!({"t":1700000000123456_i64,"seconds":"-0.1","bytes":1536,"payload":"{t|formatTimestampMicros}","host":"expanded"})).unwrap()]; + let tpl = r#"{"timestamp":"{t|formatTimestampMicros}","seconds":"{seconds|formatTimestamp}","size":"{bytes|humanSize}","custom":"{t|formatTimestampMicros("{host} %Y%m%d%H%M%S%z \"quoted\"", "Asia/Shanghai")}","payload":"{payload}"}"#.to_string(); + let output = process_row_template( + "default", + &tpl, + &Alert::default(), + RowTemplateType::Json, + &rows, + ); + assert_eq!( + output[0], + json!({ + "timestamp":"2023-11-14T22:13:20.123456Z", "seconds":"1969-12-31T23:59:59.900Z", + "size":"1.5 KiB", "custom":"{host} 20231115061320+0800 \"quoted\"", "payload":"{t|formatTimestampMicros}" + }) + ); + let bad = r#"{seconds|formatTimestamp("%Q")} {t|formatTimestampMicros("%Y", "+25:00")} {bytes|humanSize()}"#.to_string(); + assert_eq!( + process_row_template( + "default", + &bad, + &Alert::default(), + RowTemplateType::String, + &rows + ), + vec![Value::String(bad)] + ); + } + + #[test] + fn row_rejects_lengths_and_preserves_literal_field_names() { + let rows = vec![ + serde_json::from_value( + json!({"bytes":1536,"metric:2":"1536","literal:2|humanSize":"literal value"}), + ) + .unwrap(), + ]; + let tpl = "{bytes:2|humanSize} {metric:2|humanSize} {literal:2|humanSize} {literal:2|humanSize:4} {bytes|humanSize} {alert_count:1|humanize}"; + assert_eq!( + process_row_template( + "default", + tpl, + &Alert::default(), + RowTemplateType::String, + &rows + ), + vec![Value::String( + "{bytes:2|humanSize} 1.5 KiB literal value lite 1.5 KiB {alert_count:1|humanize}" + .into() + )] + ); + } + + #[test] + fn row_rejected_modifiers_preserve_exact_source() { + let rows = + vec![serde_json::from_value(json!({"t":0,"host":"expanded","bad":"NaN"})).unwrap()]; + for tpl in [ + r#"{t|formatTimestamp("{host} %Q")}"#, + r#"{t|formatTimestamp("{host} %Y", "invalid-zone")}"#, + r#"{missing|formatTimestamp("{host} %Y")}"#, + r#"{bad|formatTimestamp("{host} %Y")}"#, + r#"{t|formatTimestamp("{host} %Y",)}"#, + r#"{t|formatTimestamp({host})}"#, + r#"{t|unknown("{host}")}"#, + ] { + for row_type in [RowTemplateType::String, RowTemplateType::Json] { + assert_eq!( + process_row_template("default", tpl, &Alert::default(), row_type, &rows), + vec![Value::String(tpl.into())] + ); + } + } + for (tpl, expected) in [ + ( + r#"{t|formatTimestamp("{host} %Q")} {host}"#, + r#"{t|formatTimestamp("{host} %Q")} expanded"#, + ), + ( + r#"{t|formatTimestamp("{host} %Y)} {host} {t|formatTimestamp("%Y")}"#, + r#"{t|formatTimestamp("expanded %Y)} expanded 1970"#, + ), + ( + "{t|formatTimestamp fired on {host} at {t}", + "{t|formatTimestamp fired on expanded at 0", + ), + ] { + assert_eq!( + process_row_template( + "default", + tpl, + &Alert::default(), + RowTemplateType::String, + &rows + ), + vec![Value::String(expected.into())] + ); + } + } + + #[test] + fn row_json_pipes_and_links_preserve_placeholders() { + let rows = vec![serde_json::from_value(json!({"bytes":1536})).unwrap()]; + for prefix in ["a|b", ""] { + let tpl = format!( + r#"{{"text":"{prefix}","count":"{{alert_count}}","size":"{{bytes|humanSize}}","invalid":"{{bytes:2|humanSize}}","unknown":"{{bytes|unknown}}"}}"# + ); + assert_eq!( + process_row_template( + "default", + &tpl, + &Alert::default(), + RowTemplateType::Json, + &rows + ), + vec![ + json!({"text":prefix,"count":"1","size":"1.5 KiB","invalid":"{bytes:2|humanSize}","unknown":"{bytes|unknown}"}) + ] + ); + } + } + + #[test] + fn row_nested_json_pipes_quotes_and_newlines() { + let rows = vec![serde_json::from_value(json!({"ratio":0.9123,"host":"web\"1"})).unwrap()]; + let tpl = r#"{ + "text": "a|b \"quoted\"", + "nested": { "link": "", + "host": "{host}", "ratio": "{ratio|humanizePercentage}", + "fallback": "{missing|humanSize}" } + }"#; + assert_eq!( + process_row_template( + "default", + tpl, + &Alert::default(), + RowTemplateType::Json, + &rows + ), + vec![json!({ + "text":"a|b \"quoted\"", "nested":{"link":"","host":"web\"1","ratio":"91.23%","fallback":"{missing|humanSize}"} + })] + ); + } +} diff --git a/src/core/src/alerts/notifications/custom.rs b/src/core/src/alerts/notifications/custom.rs index 161bc212a2..73e74dd97f 100644 --- a/src/core/src/alerts/notifications/custom.rs +++ b/src/core/src/alerts/notifications/custom.rs @@ -28,6 +28,37 @@ use super::context::NotificationContext; /// Render `tpl` against `ctx`. Pure: no I/O, no clock, no DB. pub fn apply_custom_template(tpl: &str, ctx: &NotificationContext, is_email: bool) -> String { + let prepared = super::modifiers::prepare_modifiers( + tpl, + |field| { + let placeholder = format!("{{{field}}}"); + let value = apply_custom_template_plain(&placeholder, ctx, is_email); + (value != placeholder).then_some(value) + }, + |field| { + ctx.row_columns.iter().any(|(key, _)| key == field) + || ctx.context_attributes.iter().any(|(key, _)| key == field) + || ctx.metadata.iter().any(|(key, _)| key == field) + || field.strip_prefix("group.").is_some_and(|name| { + ctx.group_labels + .as_ref() + .is_some_and(|labels| labels.contains_key(name)) + }) + }, + is_email, + ); + prepared.finish(apply_custom_template_plain( + &prepared.template, + ctx, + is_email, + )) +} + +pub(crate) fn apply_custom_template_plain( + tpl: &str, + ctx: &NotificationContext, + is_email: bool, +) -> String { let evaluation_timestamp = ctx.alert_trigger_time; let evaluation_timestamp_millis = evaluation_timestamp / 1000; let evaluation_timestamp_seconds = evaluation_timestamp_millis / 1000; @@ -677,4 +708,190 @@ mod golden { let out = apply_custom_template(r#"{"hosts":"{host}"}"#, &fixture_ctx(), false); assert_eq!(out, GOLDEN_HOSTS); } + + #[test] + fn custom_numeric_modifiers_and_precedence() { + let mut ctx = fixture_ctx(); + ctx.alert_agg_value = "0.9123".into(); + ctx.row_columns = vec![ + ("bytes".into(), vec!["1048576".into()]), + ("rate".into(), vec!["1234567".into()]), + ("latency".into(), vec!["3661.9".into()]), + ("alert_agg_value".into(), vec!["0.5".into()]), + ("multi".into(), vec!["1".into(), "2".into()]), + ("literal|humanize".into(), vec!["raw".into()]), + ]; + ctx.context_attributes.push(("rate".into(), "10".into())); + ctx.metadata.push(("rate".into(), "20".into())); + assert_eq!( + apply_custom_template( + "{alert_agg_value|humanizePercentage} {bytes|humanize1024} {rate|humanize} {latency|humanizeDuration} {multi|humanize} {literal|humanize} {rate:2}", + &ctx, + false + ), + "91.23% 1Mi 1.235M 1h 1m 1s {multi|humanize} raw 12" + ); + } + + #[test] + fn custom_modifiers_leave_fallbacks_and_data_untouched() { + let mut ctx = fixture_ctx(); + ctx.context_attributes.extend([ + ("bad".into(), "abc".into()), + ("nan".into(), "NaN".into()), + ("inf".into(), "+Inf".into()), + ("negative_inf".into(), "-Inf".into()), + ("payload".into(), "{alert_count|humanizePercentage}".into()), + ]); + ctx.rows_tpl_val = vec![Value::String("{alert_count|humanizePercentage}".into())]; + let tpl = "{bad|humanize} {missing|humanize} {nan|humanizeDuration} {inf|humanize1024} {negative_inf|humanizePercentage} {alert_count|unknown} {alert_count|} {payload} {rows}"; + assert_eq!( + apply_custom_template(tpl, &ctx, true), + "{bad|humanize} {missing|humanize} NaN +Inf -Inf% {alert_count|unknown} {alert_count|} {alert_count|humanizePercentage} {alert_count|humanizePercentage}" + ); + } + + #[test] + fn custom_timestamp_and_size_modifiers() { + let mut ctx = fixture_ctx(); + ctx.row_columns = vec![ + ("bytes".into(), vec!["1536".into()]), + ("t".into(), vec!["1700000000123456".into()]), + ]; + ctx.context_attributes = vec![("zero".into(), "0".into())]; + assert_eq!( + apply_custom_template( + r#"{bytes|humanSize} {t|formatTimestampMicros} {zero|formatTimestamp("%Y-%m-%d %H:%M %:z", "Asia/Shanghai")}"#, + &ctx, + true + ), + "1.5 KiB 2023-11-14T22:13:20.123456Z 1970-01-01 08:00 +08:00" + ); + let tpl = r#"{"date":"{zero|formatTimestamp("{host} %Y \"quoted\"", "UTC")}"}"#; + assert_eq!( + apply_custom_template(tpl, &ctx, false), + r#"{"date":"{host} 1970 \"quoted\""}"# + ); + let bad = r#"{zero|formatTimestamp("%Q")} {zero|formatTimestamp("%Y", "bad")} {bytes|humanSize()}"#; + assert_eq!(apply_custom_template(bad, &ctx, true), bad); + } + + #[test] + fn custom_modifiers_use_existing_json_column_aggregation() { + let rows = vec![serde_json::from_value(serde_json::json!({"ratio":0.9123,"bytes":"1048576","rate":1234567,"latency":"61.9"})).unwrap(), serde_json::from_value(serde_json::json!({"ratio":"0.9123","bytes":1048576,"rate":"1234567","latency":61.9})).unwrap()]; + let ctx = NotificationContext { + row_columns: super::super::context::build_row_columns(&rows), + ..Default::default() + }; + assert_eq!( + apply_custom_template( + "{ratio|humanizePercentage} {bytes|humanize1024} {rate|humanize} {latency|humanizeDuration}", + &ctx, + false + ), + "91.23% 1Mi 1.235M 1m 1s" + ); + } + + #[test] + fn custom_rejects_lengths_and_preserves_literal_field_names() { + let ctx = NotificationContext { + alert_count: "1536".into(), + row_columns: vec![ + ("bytes".into(), vec!["1536".into()]), + ("metric:2".into(), vec!["1536".into()]), + ("literal:2|humanSize".into(), vec!["literal value".into()]), + ], + ..Default::default() + }; + let tpl = "{bytes:2|humanSize} {metric:2|humanSize} {literal:2|humanSize} {literal:2|humanSize:4} {bytes|humanSize} {alert_count:1|humanize}"; + assert_eq!( + apply_custom_template(tpl, &ctx, false), + "{bytes:2|humanSize} 1.5 KiB literal value lite 1.5 KiB {alert_count:1|humanize}" + ); + } + + #[test] + fn custom_rejected_modifiers_preserve_exact_source() { + let ctx = NotificationContext { + row_columns: vec![ + ("t".into(), vec!["0".into()]), + ("host".into(), vec!["expanded".into()]), + ("bad".into(), vec!["NaN".into()]), + ], + ..Default::default() + }; + for tpl in [ + r#"{t|formatTimestamp("{host} %Q")}"#, + r#"{t|formatTimestamp("{host} %Y", "invalid-zone")}"#, + r#"{missing|formatTimestamp("{host} %Y")}"#, + r#"{bad|formatTimestamp("{host} %Y")}"#, + r#"{t|formatTimestamp("{host} %Y",)}"#, + r#"{t|formatTimestamp({host})}"#, + r#"{t|unknown("{host}")}"#, + ] { + for is_email in [false, true] { + assert_eq!(apply_custom_template(tpl, &ctx, is_email), tpl); + } + } + for (tpl, expected) in [ + ( + r#"{t|formatTimestamp("{host} %Q")} {host}"#, + r#"{t|formatTimestamp("{host} %Q")} expanded"#, + ), + ( + r#"{t|formatTimestamp("{host} %Y)} {host} {t|formatTimestamp("%Y")}"#, + r#"{t|formatTimestamp("expanded %Y)} expanded 1970"#, + ), + ( + "{t|formatTimestamp fired on {host} at {t}", + "{t|formatTimestamp fired on expanded at 0", + ), + ] { + assert_eq!(apply_custom_template(tpl, &ctx, true), expected); + } + } + + #[test] + fn custom_json_pipes_and_links_preserve_placeholders() { + let ctx = NotificationContext { + alert_count: "3".into(), + row_columns: vec![("bytes".into(), vec!["1536".into()])], + ..Default::default() + }; + for prefix in ["a|b", ""] { + let tpl = format!( + r#"{{"text":"{prefix}","count":"{{alert_count}}","size":"{{bytes|humanSize}}","invalid":"{{bytes:2|humanSize}}","unknown":"{{bytes|unknown}}"}}"# + ); + let expected = serde_json::json!({"text":prefix,"count":"3","size":"1.5 KiB","invalid":"{bytes:2|humanSize}","unknown":"{bytes|unknown}"}); + for is_email in [false, true] { + let rendered = apply_custom_template(&tpl, &ctx, is_email); + assert_eq!(serde_json::from_str::(&rendered).unwrap(), expected); + } + } + } + + #[test] + fn custom_nested_json_pipes_quotes_and_newlines() { + let ctx = NotificationContext { + row_columns: vec![ + ("ratio".into(), vec!["0.9123".into()]), + ("host".into(), vec!["web\"1".into()]), + ], + ..Default::default() + }; + let tpl = r#"{ + "text": "a|b \"quoted\"", + "nested": { "link": "", + "host": "{host}", "ratio": "{ratio|humanizePercentage}", + "fallback": "{missing|humanSize}" } + }"#; + let rendered = apply_custom_template(tpl, &ctx, false); + assert_eq!( + serde_json::from_str::(&rendered).unwrap(), + serde_json::json!({ + "text":"a|b \"quoted\"", "nested":{"link":"","host":"web\"1","ratio":"91.23%","fallback":"{missing|humanSize}"} + }) + ); + } } diff --git a/src/core/src/alerts/notifications/mod.rs b/src/core/src/alerts/notifications/mod.rs index 5475825c55..a70cdc303c 100644 --- a/src/core/src/alerts/notifications/mod.rs +++ b/src/core/src/alerts/notifications/mod.rs @@ -24,6 +24,7 @@ pub mod context; pub mod custom; pub mod default_template; pub mod format; +pub(crate) mod modifiers; pub mod org_default; pub mod platform; pub mod preview; diff --git a/src/core/src/alerts/notifications/modifiers/duration.rs b/src/core/src/alerts/notifications/modifiers/duration.rs new file mode 100644 index 0000000000..f0c4515582 --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/duration.rs @@ -0,0 +1,89 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use super::numeric::{DECIMAL_PREFIXES, humanize, significant}; + +pub(super) fn format(input: &str) -> Option { + Some(humanize_duration(input.parse::().ok()?)) +} + +fn humanize_duration(value: f64) -> String { + if !value.is_finite() { + return significant(value); + } + if value.abs() < 1.0 { + return format!("{}s", humanize(value, 1000.0, &DECIMAL_PREFIXES, true)); + } + // Avoid an overflowing integer conversion for durations outside Prometheus's int64 range. + if value.abs() >= i64::MAX as f64 { + return format!("{}s", significant(value)); + } + let seconds = value.abs() as i64; + let sign = if value < 0.0 { "-" } else { "" }; + let (days, hours, minutes, seconds) = ( + seconds / 86400, + seconds / 3600 % 24, + seconds / 60 % 60, + seconds % 60, + ); + if days != 0 { + format!("{sign}{days}d {hours}h {minutes}m {seconds}s") + } else if hours != 0 { + format!("{sign}{hours}h {minutes}m {seconds}s") + } else if minutes != 0 { + format!("{sign}{minutes}m {seconds}s") + } else { + format!("{}s", significant(value)) + } +} + +#[cfg(test)] +mod tests { + use super::super::format_modifier; + + #[test] + fn prometheus_numeric_formats() { + for (input, function, expected) in [ + ("0.000000001", "humanizeDuration", "1ns"), + ("0.12345", "humanizeDuration", "123.5ms"), + ("0", "humanizeDuration", "0s"), + ("-0", "humanizeDuration", "-0s"), + ("12.3456", "humanizeDuration", "12.35s"), + ("61.9", "humanizeDuration", "1m 1s"), + ("-3661.9", "humanizeDuration", "-1h 1m 1s"), + ("90061", "humanizeDuration", "1d 1h 1m 1s"), + ("1e100", "humanizeDuration", "1e+100s"), + ] { + assert_eq!( + format_modifier(input, function).as_deref(), + Some(expected), + "{function}({input})" + ); + } + } + + #[test] + fn special_and_invalid_numbers() { + for value in ["NaN", "+Inf", "-Inf"] { + assert_eq!( + format_modifier(value, "humanizeDuration"), + Some(value.to_string()) + ); + } + for invalid in ["", "abc", "1, 2", "null", "true", " 12 "] { + assert_eq!(format_modifier(invalid, "humanizeDuration"), None); + } + } +} diff --git a/src/core/src/alerts/notifications/modifiers/humanize.rs b/src/core/src/alerts/notifications/modifiers/humanize.rs new file mode 100644 index 0000000000..5e2bf4a55b --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/humanize.rs @@ -0,0 +1,63 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use super::numeric::{BINARY_PREFIXES, DECIMAL_PREFIXES, humanize}; + +pub(super) fn format(input: &str, binary: bool) -> Option { + let value = input.parse::().ok()?; + Some(if binary { + humanize(value, 1024.0, &BINARY_PREFIXES, false) + } else { + humanize(value, 1000.0, &DECIMAL_PREFIXES, true) + }) +} + +#[cfg(test)] +mod tests { + use super::super::format_modifier; + + #[test] + fn prometheus_numeric_formats() { + for (input, function, expected) in [ + ("1234567", "humanize", "1.235M"), + ("-1234567", "humanize", "-1.235M"), + ("0.00000123", "humanize", "1.23u"), + ("0", "humanize", "0"), + ("1048576", "humanize1024", "1Mi"), + ("-1536", "humanize1024", "-1.5ki"), + ("0.000012345", "humanize1024", "1.234e-05"), + ("1e-30", "humanize", "1e-06y"), + ("1e30", "humanize", "1e+06Y"), + ] { + assert_eq!( + format_modifier(input, function).as_deref(), + Some(expected), + "{function}({input})" + ); + } + } + + #[test] + fn special_and_invalid_numbers() { + for function in ["humanize", "humanize1024"] { + for value in ["NaN", "+Inf", "-Inf"] { + assert_eq!(format_modifier(value, function), Some(value.to_string())); + } + for invalid in ["", "abc", "1, 2", "null", "true", " 12 "] { + assert_eq!(format_modifier(invalid, function), None); + } + } + } +} diff --git a/src/core/src/alerts/notifications/modifiers/mod.rs b/src/core/src/alerts/notifications/modifiers/mod.rs new file mode 100644 index 0000000000..1d7b91b431 --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/mod.rs @@ -0,0 +1,231 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +mod duration; +mod humanize; +mod numeric; +mod percentage; +mod size; +mod timestamp; + +pub(crate) struct PreparedTemplate { + pub(crate) template: String, + replacements: Vec<(String, String)>, +} + +impl PreparedTemplate { + pub(crate) fn plain(template: String) -> Self { + Self { + template, + replacements: Vec::new(), + } + } + + pub(crate) fn finish(&self, mut rendered: String) -> String { + for (marker, output) in &self.replacements { + rendered = rendered.replace(marker, output); + } + rendered + } + + fn protect(&mut self, output: String) { + // Opaque markers keep formatted and rejected expressions inert until substitution ends. + let marker = format!("\u{e000}{:032x}\u{e001}", rand::random::()); + self.template.push_str(&marker); + self.replacements.push((marker, output)); + } +} + +pub(crate) fn prepare_modifiers( + tpl: &str, + mut lookup: impl FnMut(&str) -> Option, + mut literal_field_exists: impl FnMut(&str) -> bool, + is_email: bool, +) -> PreparedTemplate { + let mut prepared = PreparedTemplate::plain(String::with_capacity(tpl.len())); + let mut cursor = 0; + while let Some((start, end)) = next_placeholder(tpl, cursor) { + prepared.template.push_str(&tpl[cursor..start]); + let expression = &tpl[start + 1..end - 1]; + let literal = is_literal_placeholder(expression, &mut literal_field_exists); + if let Some((field, function)) = expression.split_once('|').filter(|_| !literal) { + let replacement = if !field.is_empty() + && !tpl[..start].ends_with('{') + && (!field.contains(':') || literal_field_exists(field)) + { + lookup(field).and_then(|value| format_modifier(&value, function)) + } else { + None + }; + let output = match replacement { + Some(output) if !is_email => super::custom::format_variable_value(output), + Some(output) => output, + None => tpl[start..end].to_string(), + }; + prepared.protect(output); + } else { + prepared.template.push_str(&tpl[start..end]); + } + cursor = end; + } + prepared.template.push_str(&tpl[cursor..]); + prepared +} + +fn next_placeholder(tpl: &str, mut cursor: usize) -> Option<(usize, usize)> { + loop { + let mut start = None; + let mut modifier = false; + let mut quoted = false; + let mut escaped = false; + let mut nested = 0usize; + for (index, ch) in tpl[cursor..].char_indices() { + let index = cursor + index; + if quoted { + if escaped { + escaped = false; + } else if ch == '\\' { + escaped = true; + } else if ch == '"' { + quoted = false; + } + continue; + } + match ch { + '{' if modifier => nested += 1, + '{' => { + start = (!tpl[index + 1..].trim_start().starts_with('"')).then_some(index); + modifier = false; + } + '|' if start.is_some() => modifier = true, + '"' if modifier => quoted = true, + '}' if nested > 0 => nested -= 1, + '}' => { + if let Some(start) = start { + return Some((start, index + 1)); + } + } + _ => {} + } + } + // An unclosed modifier is literal text, so later placeholders still substitute. + cursor = start.filter(|_| modifier)? + 1; + } +} + +fn is_literal_placeholder( + expression: &str, + literal_field_exists: &mut impl FnMut(&str) -> bool, +) -> bool { + literal_field_exists(expression) + || expression.rsplit_once(':').is_some_and(|(field, length)| { + length.parse::().is_ok_and(|length| length > 0) && literal_field_exists(field) + }) +} + +fn format_modifier(input: &str, function: &str) -> Option { + if let Some(args) = function.strip_prefix("formatTimestampMicros") { + return timestamp::format(input, args, true); + } + if let Some(args) = function.strip_prefix("formatTimestamp") { + return timestamp::format(input, args, false); + } + match function { + "humanize" => humanize::format(input, false), + "humanize1024" => humanize::format(input, true), + "humanizeDuration" => duration::format(input), + "humanizePercentage" => percentage::format(input), + "humanSize" => size::format(input), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn render_modifiers(tpl: &str, lookup: impl FnMut(&str) -> Option) -> String { + let lookup = std::cell::RefCell::new(lookup); + let prepared = prepare_modifiers( + tpl, + |field| lookup.borrow_mut()(field), + |field| lookup.borrow_mut()(field).is_some(), + true, + ); + prepared.finish(prepared.template.clone()) + } + + #[test] + fn parser_preserves_unsupported_and_literal_fields() { + let tpl = "{v} {v|unknown} {missing|humanize} {v|} {|humanize} {v|humanize|humanize} {v|humanize:2} {{v|humanize}} {v|humanize"; + assert_eq!( + render_modifiers(tpl, |field| (field == "v").then(|| "1200".into())), + tpl + ); + assert_eq!( + render_modifiers("{v|humanize}", |field| match field { + "v" => Some("1200".into()), + "v|humanize" => Some("literal".into()), + _ => None, + }), + "{v|humanize}" + ); + assert_eq!( + render_modifiers(r#"{"value":"{v|humanize}"}"#, |field| (field == "v") + .then(|| "1200".into())), + r#"{"value":"1.2k"}"# + ); + } + + #[test] + fn unclosed_modifier_does_not_swallow_later_placeholders() { + let lookup = |field: &str| (field == "v").then(|| "1200".to_string()); + for (tpl, expected) in [ + ( + "{v|humanize fired {v|humanize} {v}", + "{v|humanize fired 1.2k {v}", + ), + ( + r#"{v|formatTimestamp("x} {v|humanize}"#, + r#"{v|formatTimestamp("x} 1.2k"#, + ), + ( + "{v|humanize {v|humanize {v|humanize}", + "{v|humanize {v|humanize 1.2k", + ), + ("{v|humanize", "{v|humanize"), + ] { + let prepared = prepare_modifiers(tpl, lookup, |_| false, true); + assert_eq!( + prepared.finish(prepared.template.clone()), + expected, + "{tpl}" + ); + } + } + + #[test] + fn quoted_format_braces_stay_inert() { + let tpl = r#"{t|formatTimestamp("{alert_count} %Y \"quoted\"", "UTC")}"#; + let prepared = prepare_modifiers( + tpl, + |field| (field == "t").then(|| "0".into()), + |_| false, + false, + ); + let plain = prepared.template.replace("{alert_count}", "99"); + assert_eq!(prepared.finish(plain), r#"{alert_count} 1970 \"quoted\""#); + } +} diff --git a/src/core/src/alerts/notifications/modifiers/numeric.rs b/src/core/src/alerts/notifications/modifiers/numeric.rs new file mode 100644 index 0000000000..58206c19d2 --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/numeric.rs @@ -0,0 +1,73 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +pub(super) const DECIMAL_PREFIXES: [&str; 8] = ["k", "M", "G", "T", "P", "E", "Z", "Y"]; +pub(super) const FRACTIONAL_PREFIXES: [&str; 8] = ["m", "u", "n", "p", "f", "a", "z", "y"]; +pub(super) const BINARY_PREFIXES: [&str; 8] = ["ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi", "Yi"]; + +pub(super) fn humanize(mut value: f64, base: f64, prefixes: &[&str], fractional: bool) -> String { + if value == 0.0 || !value.is_finite() { + return significant(value); + } + let mut prefix = ""; + if fractional && value.abs() < 1.0 { + for next in FRACTIONAL_PREFIXES { + if value.abs() >= 1.0 { + break; + } + prefix = next; + value *= 1000.0; + } + } else { + for next in prefixes { + if value.abs() < base { + break; + } + prefix = next; + value /= base; + } + } + format!("{}{prefix}", significant(value)) +} + +pub(super) fn significant(value: f64) -> String { + if value.is_nan() { + return "NaN".into(); + } + if value.is_infinite() { + return if value.is_sign_negative() { + "-Inf" + } else { + "+Inf" + } + .into(); + } + let scientific = format!("{value:.3e}"); + let (mantissa, exponent) = scientific.split_once('e').unwrap(); + let exponent = exponent.parse::().unwrap(); + if !(-4..4).contains(&exponent) { + format!("{}e{exponent:+03}", trim_fraction(mantissa)) + } else { + trim_fraction(&format!("{:.*}", (3 - exponent) as usize, value)).to_string() + } +} + +fn trim_fraction(value: &str) -> &str { + if value.contains('.') { + value.trim_end_matches('0').trim_end_matches('.') + } else { + value + } +} diff --git a/src/core/src/alerts/notifications/modifiers/percentage.rs b/src/core/src/alerts/notifications/modifiers/percentage.rs new file mode 100644 index 0000000000..2affc46ab9 --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/percentage.rs @@ -0,0 +1,58 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use super::numeric::significant; + +pub(super) fn format(input: &str) -> Option { + Some(format!( + "{}%", + significant(input.parse::().ok()? * 100.0) + )) +} + +#[cfg(test)] +mod tests { + use super::super::format_modifier; + + #[test] + fn prometheus_numeric_formats() { + for (input, function, expected) in [ + ("0.9123", "humanizePercentage", "91.23%"), + ("-0.25", "humanizePercentage", "-25%"), + ("0", "humanizePercentage", "0%"), + ("123.4567", "humanizePercentage", "1.235e+04%"), + ("1e308", "humanizePercentage", "+Inf%"), + ] { + assert_eq!( + format_modifier(input, function).as_deref(), + Some(expected), + "{function}({input})" + ); + } + } + + #[test] + fn special_and_invalid_numbers() { + for value in ["NaN", "+Inf", "-Inf"] { + assert_eq!( + format_modifier(value, "humanizePercentage"), + Some(format!("{value}%")) + ); + } + for invalid in ["", "abc", "1, 2", "null", "true", " 12 "] { + assert_eq!(format_modifier(invalid, "humanizePercentage"), None); + } + } +} diff --git a/src/core/src/alerts/notifications/modifiers/size.rs b/src/core/src/alerts/notifications/modifiers/size.rs new file mode 100644 index 0000000000..d28eebd817 --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/size.rs @@ -0,0 +1,58 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use super::numeric::significant; + +pub(super) fn format(input: &str) -> Option { + let mut value = input.parse::().ok()?; + if !value.is_finite() { + return Some(significant(value)); + } + let mut unit = "B"; + for next in ["KiB", "MiB", "GiB", "TiB", "PiB", "EiB", "ZiB", "YiB"] { + if value.abs() < 1024.0 { + break; + } + value /= 1024.0; + unit = next; + } + Some(format!("{} {unit}", significant(value))) +} + +#[cfg(test)] +mod tests { + use super::super::format_modifier; + + #[test] + fn human_size_bytes() { + for (input, expected) in [ + ("0", "0 B"), + ("1", "1 B"), + ("1536", "1.5 KiB"), + ("1048576", "1 MiB"), + ("-1536", "-1.5 KiB"), + ("NaN", "NaN"), + ("+Inf", "+Inf"), + ("-Inf", "-Inf"), + ] { + assert_eq!( + format_modifier(input, "humanSize").as_deref(), + Some(expected) + ); + } + assert_eq!(format_modifier("bad", "humanSize"), None); + assert_eq!(format_modifier("1536", "humanSize()"), None); + } +} diff --git a/src/core/src/alerts/notifications/modifiers/timestamp.rs b/src/core/src/alerts/notifications/modifiers/timestamp.rs new file mode 100644 index 0000000000..ea0c3c14cf --- /dev/null +++ b/src/core/src/alerts/notifications/modifiers/timestamp.rs @@ -0,0 +1,231 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::fmt::Write; + +pub(super) fn format(input: &str, args: &str, micros: bool) -> Option { + let args = if args.is_empty() { + Vec::new() + } else { + let args = args.strip_prefix('(')?.strip_suffix(')')?; + let args: Vec = serde_json::from_str(&format!("[{args}]")).ok()?; + if args.is_empty() || args.len() > 2 { + return None; + } + args + }; + let timestamp = timestamp_value(input, micros)?; + let reference_micros = timestamp.timestamp_micros(); + let timezone = args.get(1).map(String::as_str).unwrap_or("UTC"); + if !valid_timezone(timezone) { + return None; + } + let offset = config::utils::time::parse_timezone_to_offset_at(timezone, reference_micros)?; + timestamp + .naive_utc() + .checked_add_signed(chrono::Duration::seconds(offset))?; + let offset = chrono::FixedOffset::east_opt(i32::try_from(offset).ok()?)?; + let timestamp = timestamp.with_timezone(&offset); + let Some(format) = args.first() else { + return Some(timestamp.to_rfc3339_opts(chrono::SecondsFormat::AutoSi, true)); + }; + let items = chrono::format::StrftimeItems::new(format).collect::>(); + if items + .iter() + .any(|item| matches!(item, chrono::format::Item::Error)) + { + return None; + } + let mut output = String::new(); + write!(&mut output, "{}", timestamp.format_with_items(items.iter())).ok()?; + Some(output) +} + +fn timestamp_value(input: &str, micros: bool) -> Option> { + if let Ok(integer) = input.parse::() { + return if micros { + chrono::DateTime::from_timestamp_micros(integer) + } else { + chrono::DateTime::from_timestamp(integer, 0) + }; + } + if input.contains('.') && !input.contains(['e', 'E']) { + return decimal_timestamp(input, micros); + } + let value = input.parse::().ok()?; + let value = if micros { value / 1_000_000.0 } else { value }; + if !value.is_finite() || value.abs() >= i64::MAX as f64 { + return None; + } + let seconds = value.floor(); + let nanos = ((value - seconds) * 1_000_000_000.0).round() as u32; + let (seconds, nanos) = if nanos == 1_000_000_000 { + (seconds as i64 + 1, 0) + } else { + (seconds as i64, nanos) + }; + chrono::DateTime::from_timestamp(seconds, nanos) +} + +fn decimal_timestamp(input: &str, micros: bool) -> Option> { + let negative = input.starts_with('-'); + let unsigned = input.strip_prefix(['-', '+']).unwrap_or(input); + let (whole, fraction) = unsigned.split_once('.')?; + if (whole.is_empty() && fraction.is_empty()) + || !whole.bytes().all(|c| c.is_ascii_digit()) + || !fraction.bytes().all(|c| c.is_ascii_digit()) + { + return None; + } + let whole = if whole.is_empty() { + 0 + } else { + whole.parse::().ok()? + }; + let precision = if micros { 3 } else { 9 }; + let fraction = &fraction[..fraction.len().min(precision)]; + let fractional = if fraction.is_empty() { + 0 + } else { + fraction.parse::().ok()? + }; + let scale = 10_i128.pow(precision as u32); + let nanos = whole + .checked_mul(scale)? + .checked_add(fractional * 10_i128.pow((precision - fraction.len()) as u32))?; + let nanos = if negative { -nanos } else { nanos }; + chrono::DateTime::from_timestamp( + i64::try_from(nanos.div_euclid(1_000_000_000)).ok()?, + nanos.rem_euclid(1_000_000_000) as u32, + ) +} + +fn valid_timezone(timezone: &str) -> bool { + // The shared parser aliases CST to China Standard Time, which is ambiguous for users. + if timezone.is_empty() || timezone.eq_ignore_ascii_case("cst") { + return false; + } + if !timezone.starts_with(['+', '-']) { + return true; + } + let bytes = timezone.as_bytes(); + bytes.len() == 6 + && bytes[3] == b':' + && [bytes[1], bytes[2], bytes[4], bytes[5]] + .iter() + .all(u8::is_ascii_digit) + && &timezone[1..3] < "24" + && &timezone[4..6] < "60" +} + +#[cfg(test)] +mod tests { + use super::super::format_modifier; + + #[test] + fn timestamp_units_precision_timezones_and_formats() { + for (input, function, expected) in [ + ("0", "formatTimestamp", "1970-01-01T00:00:00Z"), + ("-0.1", "formatTimestamp", "1969-12-31T23:59:59.900Z"), + ( + "1700000000.123456", + "formatTimestamp", + "2023-11-14T22:13:20.123456Z", + ), + ( + "1700000000123456", + "formatTimestampMicros", + "2023-11-14T22:13:20.123456Z", + ), + ("-1", "formatTimestampMicros", "1969-12-31T23:59:59.999999Z"), + ( + "1.5", + "formatTimestampMicros", + "1970-01-01T00:00:00.000001500Z", + ), + ( + "0", + r#"formatTimestamp("%Y-%m-%d %H:%M:%S %:z", "Asia/Shanghai")"#, + "1970-01-01 08:00:00 +08:00", + ), + ( + "0", + r#"formatTimestamp("%Y%m%d%H%M%S%z", "-07:30")"#, + "19691231163000-0730", + ), + ( + "1704067200", + r#"formatTimestamp("%H:%M %:z", "America/New_York")"#, + "19:00 -05:00", + ), + ( + "1719792000", + r#"formatTimestamp("%H:%M %:z", "America/New_York")"#, + "20:00 -04:00", + ), + ("0", r#"formatTimestamp("%Y")"#, "1970"), + ] { + assert_eq!( + format_modifier(input, function).as_deref(), + Some(expected), + "{function}({input})" + ); + } + } + + #[test] + fn timestamp_invalid_inputs_and_arguments() { + for input in [ + "NaN", + "+Inf", + "-Inf", + "oops", + "1e100", + "9223372036854775807", + " 0 ", + ".", + "+.", + ] { + assert_eq!(format_modifier(input, "formatTimestamp"), None); + assert_eq!(format_modifier(input, "formatTimestampMicros"), None); + } + let max_timestamp = chrono::NaiveDate::MAX + .and_hms_opt(23, 59, 59) + .unwrap() + .and_utc() + .timestamp() + .to_string(); + assert_eq!( + format_modifier(&max_timestamp, r#"formatTimestamp("%Y", "+23:59")"#), + None + ); + for function in [ + "formatTimestamp()", + "formatTimestampExtra", + "formatTimestamp(1)", + r#"formatTimestamp("%Q")"#, + r#"formatTimestamp("%Y", "Unknown/Zone")"#, + r#"formatTimestamp("%Y", "CST")"#, + r#"formatTimestamp("%Y", "cst")"#, + r#"formatTimestamp("%Y", "+08:99")"#, + r#"formatTimestamp("%Y", "+8")"#, + r#"formatTimestamp("%Y", "")"#, + r#"formatTimestamp("%Y", "UTC", "extra")"#, + r#"formatTimestamp("%Y", "UTC")suffix"#, + ] { + assert_eq!(format_modifier("0", function), None, "{function}"); + } + } +}