fix(alerts): validate and quote group-by fields of custom aggregation alerts (#15649)

This commit is contained in:
Hengfei Yang
2026-10-22 12:43:10 +00:00
committed by GitHub
parent 110251be45
commit 0863a0d69c
3 changed files with 307 additions and 95 deletions
+94
View File
@@ -358,6 +358,8 @@ pub enum AlertError {
RecoveryNeedsRoutingKey { destination: String },
#[error("Error in multi alert grouping: {0}")]
MultiAlertGroupingError(String),
#[error("Group by column {column} not found on stream {stream_name}")]
GroupByColumnNotFound { column: String, stream_name: String },
}
/// What a manual trigger did: sent, with its messages, or held by an active downtime.
@@ -477,6 +479,30 @@ fn validate_multi_alert_config(alert: &Alert) -> Result<(), AlertError> {
Ok(())
}
/// Refuses, at save time, a custom-aggregation `group_by` entry that is not a column of the stream.
#[allow(clippy::result_large_err)]
fn validate_custom_group_by(
stream_name: &str,
schema: &arrow_schema::Schema,
alert: &Alert,
) -> Result<(), AlertError> {
if alert.query_condition.query_type != QueryType::Custom {
return Ok(());
}
let group_by = alert
.query_condition
.aggregation
.as_ref()
.and_then(|agg| agg.group_by.as_deref())
.unwrap_or(&[]);
super::quoted_group_by(schema, group_by)
.map(drop)
.map_err(|column| AlertError::GroupByColumnNotFound {
column,
stream_name: stream_name.to_owned(),
})
}
/// Structural half only: `{var}` tokens are unresolved until send time, so a body shaped right but
/// malformed still fails at send. A CONTENT template always renders an object.
fn platform_body_is_addressable(kind: TemplateKind, body: &str) -> bool {
@@ -842,6 +868,7 @@ async fn prepare_alert(
stream_name: stream_name.to_owned(),
});
}
validate_custom_group_by(stream_name, &schema, alert)?;
// Alerts must follow the max_query_range of the stream as set in the schema
if let Some(settings) = unwrap_stream_settings(&schema) {
@@ -4851,6 +4878,73 @@ mod threshold_validation_tests {
}
}
#[cfg(test)]
mod group_by_validation_tests {
use arrow_schema::{DataType, Field, Schema};
use config::{
meta::alerts::{AggFunction, Aggregation, Condition, Operator, QueryType, alert::Alert},
utils::json::json,
};
use super::{AlertError, validate_custom_group_by};
fn schema() -> Schema {
Schema::new(vec![
Field::new("host", DataType::Utf8, false),
Field::new("latency", DataType::Int64, false),
])
}
fn custom_alert(group_by: &[&str]) -> Alert {
let mut alert = Alert::default();
alert.query_condition.query_type = QueryType::Custom;
alert.query_condition.aggregation = Some(Aggregation {
group_by: Some(group_by.iter().map(|s| s.to_string()).collect()),
function: AggFunction::Avg,
having: Condition {
column: "latency".into(),
operator: Operator::GreaterThan,
value: json!(100),
ignore_case: false,
},
warning_value: None,
multi_alert: false,
});
alert
}
#[test]
fn test_group_by_columns_in_the_schema_pass() {
assert!(validate_custom_group_by("svc", &schema(), &custom_alert(&["host"])).is_ok());
assert!(validate_custom_group_by("svc", &schema(), &custom_alert(&[])).is_ok());
assert!(validate_custom_group_by("svc", &schema(), &Alert::default()).is_ok());
}
#[test]
fn test_group_by_column_outside_the_schema_is_rejected_and_named() {
let alert = custom_alert(&["host", "host, (SELECT 1)"]);
let err = validate_custom_group_by("svc", &schema(), &alert).unwrap_err();
assert!(matches!(
&err,
AlertError::GroupByColumnNotFound { column, stream_name }
if column == "host, (SELECT 1)" && stream_name == "svc"
));
assert_eq!(
err.to_string(),
"Group by column host, (SELECT 1) not found on stream svc"
);
let response = axum::response::Response::from(err);
assert_eq!(response.status(), axum::http::StatusCode::BAD_REQUEST);
}
#[test]
fn test_sql_alerts_derive_group_by_elsewhere_and_are_not_checked_here() {
let mut alert = custom_alert(&["nonexistent"]);
alert.query_condition.query_type = QueryType::SQL;
assert!(validate_custom_group_by("svc", &schema(), &alert).is_ok());
}
}
#[cfg(test)]
mod send_path_tests {
use config::meta::destinations::{Endpoint, HTTPType, Template, TemplateKind};
+211 -94
View File
@@ -32,6 +32,7 @@ use config::{
utils::{
base64,
json::{Map, Value},
sql::quote_identifier,
},
};
use search::utils::is_permissable_function_error;
@@ -1561,13 +1562,19 @@ pub async fn build_sql(
))?
)
};
if query_condition.aggregation.is_none() {
let Some(agg) = query_condition.aggregation.as_ref() else {
return Ok(format!("SELECT * FROM \"{stream_name}\" {where_sql}"));
};
build_aggregation_sql(stream_name, &schema, agg, &where_sql)
}
// handle aggregation
let mut sql = String::new();
let agg = query_condition.aggregation.as_ref().unwrap();
/// Only columns of `schema` reach the SQL; any other `group_by`/`having` entry is refused.
fn build_aggregation_sql(
stream_name: &str,
schema: &Schema,
agg: &config::meta::alerts::Aggregation,
where_sql: &str,
) -> Result<String, anyhow::Error> {
let having_expr = {
let data_type = match schema.field_with_name(&agg.having.column) {
Ok(field) => field.data_type(),
@@ -1581,69 +1588,56 @@ pub async fn build_sql(
build_having_expr(agg, data_type)?
};
let col = quote_identifier(&agg.having.column);
let func_expr = match agg.function {
AggFunction::Avg => format!("AVG(\"{}\")", agg.having.column),
AggFunction::Max => format!("MAX(\"{}\")", agg.having.column),
AggFunction::Min => format!("MIN(\"{}\")", agg.having.column),
AggFunction::Sum => format!("SUM(\"{}\")", agg.having.column),
AggFunction::Count => format!("COUNT(\"{}\")", agg.having.column),
AggFunction::Median => format!("MEDIAN(\"{}\")", agg.having.column),
AggFunction::P50 => format!(
"approx_percentile_cont(0.5) WITHIN GROUP (ORDER BY \"{}\")",
agg.having.column
),
AggFunction::P75 => format!(
"approx_percentile_cont(0.75) WITHIN GROUP (ORDER BY \"{}\")",
agg.having.column
),
AggFunction::P90 => format!(
"approx_percentile_cont(0.9) WITHIN GROUP (ORDER BY \"{}\")",
agg.having.column
),
AggFunction::P95 => format!(
"approx_percentile_cont(0.95) WITHIN GROUP (ORDER BY \"{}\")",
agg.having.column
),
AggFunction::P99 => format!(
"approx_percentile_cont(0.99) WITHIN GROUP (ORDER BY \"{}\")",
agg.having.column
),
AggFunction::Avg => format!("AVG({col})"),
AggFunction::Max => format!("MAX({col})"),
AggFunction::Min => format!("MIN({col})"),
AggFunction::Sum => format!("SUM({col})"),
AggFunction::Count => format!("COUNT({col})"),
AggFunction::Median => format!("MEDIAN({col})"),
AggFunction::P50 => format!("approx_percentile_cont(0.5) WITHIN GROUP (ORDER BY {col})"),
AggFunction::P75 => format!("approx_percentile_cont(0.75) WITHIN GROUP (ORDER BY {col})"),
AggFunction::P90 => format!("approx_percentile_cont(0.9) WITHIN GROUP (ORDER BY {col})"),
AggFunction::P95 => format!("approx_percentile_cont(0.95) WITHIN GROUP (ORDER BY {col})"),
AggFunction::P99 => format!("approx_percentile_cont(0.99) WITHIN GROUP (ORDER BY {col})"),
};
if let Some(group) = agg.group_by.as_ref()
&& !group.is_empty()
{
let cols = group.join(", ");
if agg.multi_alert {
// Multi-alerts drop the HAVING filter. A group that falls back
// under the threshold must still be RETURNED: otherwise its
// recovery is indistinguishable from it vanishing, and it would
// only resolve later via the reaper timeout, with a NULL value
// where the real reading should be.
//
// The page stays bounded, so it is ordered worst-first. That
// ordering keeps the rollup level exact and lets the cap admit the
// true top of the distribution rather than an arbitrary slice. The
// group columns are the deterministic tiebreak within a severity
// band.
let severity_order =
config::meta::alerts::aggregation_level::severity_order_sql(agg, "alert_agg_value")
.map_err(|e| anyhow::anyhow!("Invalid aggregation threshold: {e}"))?;
sql = format!(
"SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} ORDER BY {severity_order}, {cols}"
);
} else {
sql = format!(
"SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} HAVING {having_expr}"
);
}
}
if sql.is_empty() {
sql = format!(
let group = agg.group_by.as_deref().unwrap_or(&[]);
if group.is_empty() {
return Ok(format!(
"SELECT {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} HAVING {having_expr}"
);
));
}
Ok(sql)
let cols = quoted_group_by(schema, group)
.map_err(|column| {
anyhow::anyhow!("Group by column {column} not found on stream {stream_name}")
})?
.join(", ");
if agg.multi_alert {
// No HAVING: a group back under the threshold must still be returned to show recovery.
// Worst-first order keeps the bounded page exact; group columns are the tiebreak.
let severity_order =
config::meta::alerts::aggregation_level::severity_order_sql(agg, "alert_agg_value")
.map_err(|e| anyhow::anyhow!("Invalid aggregation threshold: {e}"))?;
return Ok(format!(
"SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} ORDER BY {severity_order}, {cols}"
));
}
Ok(format!(
"SELECT {cols}, {func_expr} AS alert_agg_value, MIN({TIMESTAMP_COL_NAME}) as zo_sql_min_time, MAX({TIMESTAMP_COL_NAME}) AS zo_sql_max_time FROM \"{stream_name}\"{where_sql} GROUP BY {cols} HAVING {having_expr}"
))
}
/// Quoted, escaped identifiers; `Err` carries the first entry that is not a column of `schema`.
fn quoted_group_by(schema: &Schema, group_by: &[String]) -> Result<Vec<String>, String> {
group_by
.iter()
.map(|col| match schema.field_with_name(col) {
Ok(_) => Ok(quote_identifier(col)),
Err(_) => Err(col.clone()),
})
.collect()
}
/// `HAVING` clause filtering the `alert_agg_value` aggregate.
@@ -1707,11 +1701,11 @@ fn build_expr(
field_alias: &str,
field_type: &DataType,
) -> Result<String, anyhow::Error> {
let field_alias = if !field_alias.is_empty() {
field_alias
} else {
let field_alias = quote_identifier(if field_alias.is_empty() {
cond.column.as_str()
};
} else {
field_alias
});
// Null/empty checks take no value. On non-string columns "empty" can only
// mean null, so the empty checks degrade to the null checks there.
let is_string_type = matches!(
@@ -1719,20 +1713,20 @@ fn build_expr(
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View
);
match cond.operator {
Operator::IsNull => return Ok(format!("\"{field_alias}\" IS NULL")),
Operator::IsNotNull => return Ok(format!("\"{field_alias}\" IS NOT NULL")),
Operator::IsNull => return Ok(format!("{field_alias} IS NULL")),
Operator::IsNotNull => return Ok(format!("{field_alias} IS NOT NULL")),
Operator::IsEmpty => {
return Ok(if is_string_type {
format!("(\"{field_alias}\" IS NULL OR \"{field_alias}\" = '')")
format!("({field_alias} IS NULL OR {field_alias} = '')")
} else {
format!("\"{field_alias}\" IS NULL")
format!("{field_alias} IS NULL")
});
}
Operator::IsNotEmpty => {
return Ok(if is_string_type {
format!("(\"{field_alias}\" IS NOT NULL AND \"{field_alias}\" != '')")
format!("({field_alias} IS NOT NULL AND {field_alias} != '')")
} else {
format!("\"{field_alias}\" IS NOT NULL")
format!("{field_alias} IS NOT NULL")
});
}
_ => {}
@@ -1748,17 +1742,17 @@ fn build_expr(
// terminate the SQL string literal early.
let val = val.replace('\'', "''");
match cond.operator {
Operator::EqualTo => format!("\"{field_alias}\" = '{val}'"),
Operator::NotEqualTo => format!("\"{field_alias}\" != '{val}'"),
Operator::GreaterThan => format!("\"{field_alias}\" > '{val}'"),
Operator::EqualTo => format!("{field_alias} = '{val}'"),
Operator::NotEqualTo => format!("{field_alias} != '{val}'"),
Operator::GreaterThan => format!("{field_alias} > '{val}'"),
Operator::GreaterThanEquals => {
format!("\"{field_alias}\" >= '{val}'")
format!("{field_alias} >= '{val}'")
}
Operator::LessThan => format!("\"{field_alias}\" < '{val}'"),
Operator::LessThanEquals => format!("\"{field_alias}\" <= '{val}'"),
Operator::Contains => format!("str_match(\"{field_alias}\", '{val}')"),
Operator::LessThan => format!("{field_alias} < '{val}'"),
Operator::LessThanEquals => format!("{field_alias} <= '{val}'"),
Operator::Contains => format!("str_match({field_alias}, '{val}')"),
Operator::NotContains => {
format!("\"{field_alias}\" NOT LIKE '%{val}%'")
format!("{field_alias} NOT LIKE '%{val}%'")
}
// Null checks returned above; keep the dead arm panic-free.
_ => {
@@ -1807,15 +1801,15 @@ fn build_expr(
.to_string()
};
match cond.operator {
Operator::EqualTo => format!("\"{field_alias}\" = {val}"),
Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"),
Operator::GreaterThan => format!("\"{field_alias}\" > {val}"),
Operator::EqualTo => format!("{field_alias} = {val}"),
Operator::NotEqualTo => format!("{field_alias} != {val}"),
Operator::GreaterThan => format!("{field_alias} > {val}"),
Operator::GreaterThanEquals => {
format!("\"{field_alias}\" >= {val}")
format!("{field_alias} >= {val}")
}
Operator::LessThan => format!("\"{field_alias}\" < {val}"),
Operator::LessThan => format!("{field_alias} < {val}"),
Operator::LessThanEquals => {
format!("\"{field_alias}\" <= {val}")
format!("{field_alias} <= {val}")
}
_ => {
return Err(anyhow::anyhow!(
@@ -1843,15 +1837,15 @@ fn build_expr(
})?
};
match cond.operator {
Operator::EqualTo => format!("\"{field_alias}\" = {val}"),
Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"),
Operator::GreaterThan => format!("\"{field_alias}\" > {val}"),
Operator::EqualTo => format!("{field_alias} = {val}"),
Operator::NotEqualTo => format!("{field_alias} != {val}"),
Operator::GreaterThan => format!("{field_alias} > {val}"),
Operator::GreaterThanEquals => {
format!("\"{field_alias}\" >= {val}")
format!("{field_alias} >= {val}")
}
Operator::LessThan => format!("\"{field_alias}\" < {val}"),
Operator::LessThan => format!("{field_alias} < {val}"),
Operator::LessThanEquals => {
format!("\"{field_alias}\" <= {val}")
format!("{field_alias} <= {val}")
}
_ => {
return Err(anyhow::anyhow!(
@@ -1879,8 +1873,8 @@ fn build_expr(
})?
};
match cond.operator {
Operator::EqualTo => format!("\"{field_alias}\" = {val}"),
Operator::NotEqualTo => format!("\"{field_alias}\" != {val}"),
Operator::EqualTo => format!("{field_alias} = {val}"),
Operator::NotEqualTo => format!("{field_alias} != {val}"),
_ => {
return Err(anyhow::anyhow!(
"Column {} has data_type [{field_type}] and it does not supported operator [{:?}]",
@@ -2973,6 +2967,129 @@ mod tests {
assert_eq!(expr, "\"code\" < 1.5");
}
// ── Aggregation SQL: group_by identifiers ───────────────────────────────
fn agg_schema() -> Schema {
Schema::new(vec![
Field::new("host", DataType::Utf8, false),
Field::new("region", DataType::Utf8, false),
Field::new("latency", DataType::Int64, false),
])
}
fn group_agg(group_by: &[&str], multi_alert: bool) -> Aggregation {
Aggregation {
group_by: Some(group_by.iter().map(|s| s.to_string()).collect()),
function: AggFunction::Avg,
having: make_cond(
"latency",
Operator::GreaterThan,
Value::Number(serde_json::Number::from(100)),
),
warning_value: None,
multi_alert,
}
}
#[test]
fn test_build_aggregation_sql_quotes_group_by_columns() {
let agg = group_agg(&["host", "region"], false);
let sql = build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap();
assert_eq!(
sql,
"SELECT \"host\", \"region\", AVG(\"latency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" GROUP BY \"host\", \"region\" HAVING \"alert_agg_value\" > 100"
);
}
#[test]
fn test_build_aggregation_sql_multi_alert_quotes_group_by_in_order_by() {
let agg = group_agg(&["host"], true);
let sql = build_aggregation_sql("svc", &agg_schema(), &agg, " WHERE 1 = 1").unwrap();
assert_eq!(
sql,
"SELECT \"host\", AVG(\"latency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" WHERE 1 = 1 GROUP BY \"host\" ORDER BY CASE WHEN \"alert_agg_value\" > 100 THEN 2 ELSE 0 END DESC, \"host\""
);
}
#[test]
fn test_build_aggregation_sql_rejects_group_by_entry_not_in_schema() {
for entry in [
"nonexistent",
"host, (SELECT secret FROM \"other\" LIMIT 1)",
"host\" FROM \"other\" --",
"",
] {
let agg = group_agg(&["host", entry], false);
let err = build_aggregation_sql("svc", &agg_schema(), &agg, "")
.unwrap_err()
.to_string();
assert_eq!(
err,
format!("Group by column {entry} not found on stream svc"),
"entry {entry:?} must be rejected"
);
}
}
#[test]
fn test_build_aggregation_sql_escapes_quotes_inside_schema_backed_identifiers() {
let tricky = "host\" FROM \"other\" --";
let schema = Schema::new(vec![
Field::new(tricky, DataType::Utf8, false),
Field::new("lat\"ency", DataType::Int64, false),
]);
let mut agg = group_agg(&[tricky], false);
agg.having.column = "lat\"ency".to_string();
let sql = build_aggregation_sql("svc", &schema, &agg, "").unwrap();
assert_eq!(
sql,
"SELECT \"host\"\" FROM \"\"other\"\" --\", AVG(\"lat\"\"ency\") AS alert_agg_value, MIN(_timestamp) as zo_sql_min_time, MAX(_timestamp) AS zo_sql_max_time FROM \"svc\" GROUP BY \"host\"\" FROM \"\"other\"\" --\" HAVING \"alert_agg_value\" > 100"
);
}
#[test]
fn test_build_expr_escapes_quotes_inside_column_name() {
let cond = make_cond(
"lev\"el",
Operator::EqualTo,
Value::String("error".to_string()),
);
let expr = build_expr(&cond, "", &DataType::Utf8).unwrap();
assert_eq!(expr, "\"lev\"\"el\" = 'error'");
let expr = build_expr(&cond, "al\"ias", &DataType::Utf8).unwrap();
assert_eq!(expr, "\"al\"\"ias\" = 'error'");
}
#[test]
fn test_build_aggregation_sql_empty_group_by_collapses_to_single_row() {
let mut agg = group_agg(&[], false);
let sql = build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap();
assert!(!sql.contains("GROUP BY"), "{sql}");
agg.group_by = None;
assert_eq!(
build_aggregation_sql("svc", &agg_schema(), &agg, "").unwrap(),
sql
);
}
#[test]
fn test_quoted_group_by_reports_first_unknown_column() {
let schema = agg_schema();
assert_eq!(
quoted_group_by(&schema, &["host".into(), "region".into()]),
Ok(vec!["\"host\"".to_string(), "\"region\"".to_string()])
);
assert_eq!(
quoted_group_by(&schema, &["host".into(), "a".into(), "b".into()]),
Err("a".to_string())
);
let quoted = Schema::new(vec![Field::new("a\"b", DataType::Utf8, false)]);
assert_eq!(
quoted_group_by(&quoted, &["a\"b".into()]),
Ok(vec!["\"a\"\"b\"".to_string()])
);
}
// ── HAVING clause type resolution ───────────────────────────────────────
fn make_agg(function: AggFunction, operator: Operator, value: Value) -> Aggregation {
+2 -1
View File
@@ -105,7 +105,8 @@ impl From<AlertError> for Response {
| AlertError::RecoveryWithNotificationGrouping
| AlertError::RecoveryNeedsJsonPayload { .. }
| AlertError::RecoveryNeedsRoutingKey { .. }
| AlertError::MultiAlertGroupingError(_) => MetaHttpResponse::bad_request(value),
| AlertError::MultiAlertGroupingError(_)
| AlertError::GroupByColumnNotFound { .. } => MetaHttpResponse::bad_request(value),
// S-16 PR 4. A conflict, not a bad request: the alert being sent is
// fine on its own terms — it is the SLOs that already exist and
// measure from it that the request collides with. Both messages