diff --git a/src/handler/http/request/search/utils.rs b/src/handler/http/request/search/utils.rs index f37066049e..0d1b6c7ec5 100644 --- a/src/handler/http/request/search/utils.rs +++ b/src/handler/http/request/search/utils.rs @@ -127,59 +127,57 @@ pub async fn validate_query_fields( stream_type: StreamType, sql: &str, ) -> Result<(), Error> { - // Step 1: Parse SQL to get columns (lightweight parsing) let search_query = proto::cluster_rpc::SearchQuery { sql: sql.to_string(), ..Default::default() }; - let sql = Sql::new_with_options(&search_query, org_id, stream_type, None, false).await?; - // Step 2: Resolve which stream and fields to validate. - // - // For multi-stream CTE queries sql.stream_names lists every real stream referenced - // (e.g. ["default", "k8s_events"]). Each stream has its own set of fields in - // sql.columns. Look up the entry matching stream_name so that fields belonging to - // a different stream are not incorrectly checked against stream_name's schema. - let (target_stream, target_fields) = sql + // A bare stream name can refer to multiple types within the same query. + let mut targets: Vec<_> = sql .stream_names .iter() - .find(|r| r.stream_name() == stream_name) + .filter(|r| r.stream_name() == stream_name) .map(|r| { - let fields = sql.columns.get(r).cloned().unwrap_or_default(); - (stream_name.to_string(), fields) + ( + r.get_stream_type(stream_type), + sql.columns.get(r).cloned().unwrap_or_default(), + ) }) - .unwrap_or_else(|| { - // stream_name not found in sql.stream_names (aliased or unresolved table). - // Fall back to original behaviour: all tracked columns against stream_name. - let fields = sql.columns.values().flatten().cloned().collect(); - (stream_name.to_string(), fields) - }); + .collect(); + if targets.is_empty() { + targets.push(( + stream_type, + sql.columns.values().flatten().cloned().collect(), + )); + } - // Step 3: Validate target stream's fields against its own schema and UDS. - let schema = infra::schema::get(org_id, &target_stream, stream_type) - .await - .map_err(|_| Error::ErrorCode(ErrorCodes::SearchStreamNotFound(target_stream.clone())))?; + for (target_stream_type, target_fields) in targets { + let schema = infra::schema::get(org_id, stream_name, target_stream_type) + .await + .map_err(|_| { + Error::ErrorCode(ErrorCodes::SearchStreamNotFound(stream_name.to_string())) + })?; + let settings = infra::schema::get_settings(org_id, stream_name, target_stream_type).await; + let uds_fields = infra::schema::get_stream_setting_defined_schema_fields(&settings); - let settings = infra::schema::get_settings(org_id, &target_stream, stream_type).await; - let uds_fields = infra::schema::get_stream_setting_defined_schema_fields(&settings); + for field in target_fields { + if is_system_field(&field) { + continue; + } - for field in target_fields { - if is_system_field(&field) { - continue; - } + if schema.field_with_name(&field).is_err() { + return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!( + "{}. Field not found in stream schema.", + field + )))); + } - if schema.field_with_name(&field).is_err() { - return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!( - "{}. Field not found in stream schema.", - field - )))); - } - - if !uds_fields.is_empty() && !uds_fields.contains(&field) { - return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!( - "{field}. Field exists but not in User-Defined Schema (UDS)", - )))); + if !uds_fields.is_empty() && !uds_fields.contains(&field) { + return Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(format!( + "{field}. Field exists but not in User-Defined Schema (UDS)", + )))); + } } } @@ -380,6 +378,135 @@ JOIN "test1" AS b"#; err_msg ); } + + #[tokio::test] + async fn test_validate_query_fields_join_enrichment_table() { + use arrow_schema::{DataType, Field, Schema}; + use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache}; + + let org_id = "test_org_join_enrich"; + let logs_key = format!("{org_id}/{}/app", StreamType::Logs); + let enrich_key = format!("{org_id}/{}/geo", StreamType::EnrichmentTables); + { + let mut w = STREAM_SCHEMAS_LATEST.write().await; + w.insert( + logs_key.clone(), + SchemaCache::new(Schema::new(vec![ + Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false), + Field::new("ip", DataType::Utf8, true), + ])), + ); + w.insert( + enrich_key.clone(), + SchemaCache::new(Schema::new(vec![ + Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false), + Field::new("addr", DataType::Utf8, true), + Field::new("country", DataType::Utf8, true), + ])), + ); + } + + let sql = r#"SELECT a.ip, b.country FROM "app" AS a JOIN "enrichment_tables"."geo" AS b ON a.ip = b.addr"#; + let logs_res = validate_query_fields(org_id, "app", StreamType::Logs, sql).await; + let enrich_res = validate_query_fields(org_id, "geo", StreamType::Logs, sql).await; + + { + let mut w = STREAM_SCHEMAS_LATEST.write().await; + w.remove(&logs_key); + w.remove(&enrich_key); + } + + assert!(logs_res.is_ok(), "logs side failed: {logs_res:?}"); + assert!(enrich_res.is_ok(), "enrichment side failed: {enrich_res:?}"); + } + + #[tokio::test] + async fn test_validate_query_fields_same_name_logs_uds() { + check_same_name_join_uds( + StreamType::Logs, + r#"SELECT l.secret FROM "logs"."app" l JOIN "enrichment_tables"."app" e ON l.id = e.id"#, + ) + .await; + } + + #[tokio::test] + async fn test_validate_query_fields_same_name_enrichment_uds() { + check_same_name_join_uds( + StreamType::EnrichmentTables, + r#"SELECT e.secret FROM "app" l JOIN "enrichment_tables"."app" e ON l.id = e.id"#, + ) + .await; + } + + async fn check_same_name_join_uds(restricted_type: StreamType, sql: &str) { + use arrow_schema::{DataType, Field, Schema}; + use config::meta::stream::StreamSettings; + use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache}; + + let org_id = format!("test_org_same_name_{restricted_type}_uds"); + let logs_key = format!("{org_id}/{}/app", StreamType::Logs); + let enrich_key = format!("{org_id}/{}/app", StreamType::EnrichmentTables); + let schema = Schema::new(vec![ + Field::new(config::TIMESTAMP_COL_NAME, DataType::Int64, false), + Field::new("id", DataType::Utf8, true), + Field::new("secret", DataType::Utf8, true), + ]); + { + let mut cache = STREAM_SCHEMAS_LATEST.write().await; + for (stream_type, key) in [ + (StreamType::Logs, &logs_key), + (StreamType::EnrichmentTables, &enrich_key), + ] { + let settings = StreamSettings { + defined_schema_fields: if stream_type == restricted_type { + vec!["id".to_string()] + } else { + Vec::new() + }, + ..Default::default() + }; + let metadata = std::collections::HashMap::from([( + "settings".to_string(), + serde_json::to_string(&settings).unwrap(), + )]); + cache.insert( + key.clone(), + SchemaCache::new(schema.clone().with_metadata(metadata)), + ); + } + } + + let invalid = validate_query_fields(&org_id, "app", StreamType::Logs, sql).await; + let valid = validate_query_fields( + &org_id, + "app", + StreamType::Logs, + &sql.replace("secret", "id"), + ) + .await; + + { + let mut cache = STREAM_SCHEMAS_LATEST.write().await; + cache.remove(&logs_key); + cache.remove(&enrich_key); + } + { + let mut settings = infra::schema::STREAM_SETTINGS.write().await; + settings.remove(&logs_key); + settings.remove(&enrich_key); + infra::schema::set_stream_settings_atomic(settings.clone()); + } + + assert!( + matches!( + &invalid, + Err(Error::ErrorCode(ErrorCodes::SearchFieldNotFound(message))) + if message.contains("secret") && message.contains("User-Defined Schema") + ), + "{invalid:?}" + ); + assert!(valid.is_ok(), "{valid:?}"); + } } // ============================================================================ diff --git a/src/service/alerts/alert.rs b/src/service/alerts/alert.rs index 7b778292d3..8ef70fe23b 100644 --- a/src/service/alerts/alert.rs +++ b/src/service/alerts/alert.rs @@ -34,7 +34,7 @@ use config::{ }, folder::{DEFAULT_FOLDER, Folder, FolderType}, search::{SearchEventContext, SearchEventType}, - sql::resolve_stream_names, + sql::{TableReferenceExt, resolve_stream_names_with_type}, stream::StreamType, }, utils::{ @@ -391,8 +391,8 @@ async fn prepare_alert( } let sql = alert.query_condition.sql.as_ref().unwrap(); - let stream_names = match resolve_stream_names(sql) { - Ok(stream_names) => stream_names, + let stream_refs = match resolve_stream_names_with_type(sql) { + Ok(stream_refs) => stream_refs, Err(e) => { return Err(AlertError::ResolveStreamNameError(e)); } @@ -400,10 +400,12 @@ async fn prepare_alert( // SQL may contain multiple stream names, check for each stream // if the alert period is greater than the max query range - for stream in stream_names.iter() { - if !stream.eq(stream_name) + for stream_ref in stream_refs.iter() { + let stream = stream_ref.stream_name(); + let ref_type = stream_ref.get_stream_type(stream_type); + if !(stream == stream_name && ref_type == stream_type) && let Some(settings) = - infra::schema::get_settings(org_id, stream, stream_type).await + infra::schema::get_settings(org_id, &stream, ref_type).await { let max_query_range = settings.max_query_range; if max_query_range > 0 @@ -412,7 +414,7 @@ async fn prepare_alert( { return Err(AlertError::PeriodExceedsMaxQueryRange { max_query_range_hours: max_query_range, - stream_name: stream_name.to_owned(), + stream_name: stream, }); } } @@ -1885,6 +1887,75 @@ mod tests { use super::*; use crate::service::alerts::{Condition, build_expr}; + #[tokio::test] + async fn prepare_alert_checks_a_joined_stream_range_by_its_own_type() { + use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache}; + + use crate::common::infra::config::DESTINATIONS; + + let org_id = "test_org_alert_save_join_range"; + let logs_key = format!("{org_id}/{}/app", StreamType::Logs); + let traces_key = format!("{org_id}/{}/spans", StreamType::Traces); + let dest_key = format!("{org_id}/dest"); + STREAM_SCHEMAS_LATEST.write().await.insert( + logs_key.clone(), + SchemaCache::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new( + "id", + DataType::Utf8, + true, + )])), + ); + { + let mut w = infra::schema::STREAM_SETTINGS.write().await; + w.insert( + traces_key.clone(), + config::meta::stream::StreamSettings { + max_query_range: 1, + ..Default::default() + }, + ); + infra::schema::set_stream_settings_atomic( + w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + ); + } + DESTINATIONS.insert( + dest_key.clone(), + config::meta::destinations::Destination { + org_id: org_id.to_string(), + name: "dest".to_string(), + ..Default::default() + }, + ); + + let mut alert = Alert::default(); + alert.destinations = vec!["dest".to_string()]; + alert.trigger_condition.frequency = 60; + alert.trigger_condition.period = 120; + alert.query_condition.query_type = QueryType::SQL; + alert.query_condition.sql = Some( + r#"SELECT a.id FROM "app" AS a JOIN "traces"."spans" AS b ON a.id = b.id"#.to_string(), + ); + let res = prepare_alert(org_id, "app", "join", &mut alert, true, false).await; + + STREAM_SCHEMAS_LATEST.write().await.remove(&logs_key); + { + let mut w = infra::schema::STREAM_SETTINGS.write().await; + w.remove(&traces_key); + infra::schema::set_stream_settings_atomic( + w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + ); + } + DESTINATIONS.remove(&dest_key); + + assert!( + matches!( + &res, + Err(AlertError::PeriodExceedsMaxQueryRange { stream_name, .. }) if stream_name == "spans" + ), + "{res:?}" + ); + } + #[test] fn test_format_variable_value() { // Test common control characters diff --git a/src/service/alerts/derived_streams.rs b/src/service/alerts/derived_streams.rs index e9b9ae40c1..536f5baed9 100644 --- a/src/service/alerts/derived_streams.rs +++ b/src/service/alerts/derived_streams.rs @@ -23,7 +23,7 @@ use config::{ alerts::{FrequencyType, QueryType, TriggerEvalResults}, pipeline::components::DerivedStream, search::{SearchEventContext, SearchEventType}, - sql::resolve_stream_names, + sql::{TableReferenceExt, resolve_stream_names_with_type}, }, utils::sql::is_timestamp_selected, }; @@ -57,8 +57,8 @@ pub async fn save( } // Check the max_query_range of streams in the sql query - let stream_names = match resolve_stream_names(sql) { - Ok(stream_names) => stream_names, + let stream_refs = match resolve_stream_names_with_type(sql) { + Ok(stream_refs) => stream_refs, Err(e) => { return Err(anyhow::anyhow!( "Error resolving stream names in SQL query: {e}" @@ -66,9 +66,14 @@ pub async fn save( } }; let (org_id, stream_type) = (&derived_stream.org_id, derived_stream.stream_type); - for stream in stream_names.iter() { - if let Some(settings) = - infra::schema::get_settings(org_id, stream, stream_type).await + for stream_ref in stream_refs.iter() { + let stream = stream_ref.stream_name(); + if let Some(settings) = infra::schema::get_settings( + org_id, + &stream, + stream_ref.get_stream_type(stream_type), + ) + .await { let max_query_range = settings.max_query_range; if max_query_range > 0 @@ -239,3 +244,65 @@ impl DerivedStreamExt for DerivedStream { .await } } + +#[cfg(test)] +mod tests { + use super::*; + + /// Runs the real `save`; every rejection here happens before the DB is touched. + async fn save_err(mutate: impl FnOnce(&mut DerivedStream)) -> String { + let mut derived_stream = DerivedStream::default(); + mutate(&mut derived_stream); + save(derived_stream, "p", "p1", true) + .await + .unwrap_err() + .to_string() + } + + #[tokio::test] + async fn save_checks_a_joined_stream_range_by_its_own_type() { + use config::meta::stream::{StreamSettings, StreamType}; + + let org_id = "test_org_ds_save_join_range"; + let logs_key = format!("{org_id}/{}/app", StreamType::Logs); + let traces_key = format!("{org_id}/{}/spans", StreamType::Traces); + let limited = StreamSettings { + max_query_range: 1, + ..Default::default() + }; + { + let mut w = infra::schema::STREAM_SETTINGS.write().await; + w.insert(logs_key.clone(), StreamSettings::default()); + w.insert(traces_key.clone(), limited); + infra::schema::set_stream_settings_atomic( + w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + ); + } + + let err = save_err(|d| { + d.org_id = org_id.to_string(); + d.stream_type = StreamType::Logs; + d.query_condition.query_type = QueryType::SQL; + d.query_condition.sql = Some( + r#"SELECT a._timestamp AS _timestamp FROM "app" AS a JOIN "traces"."spans" AS b ON a.id = b.id"# + .to_string(), + ); + d.trigger_condition.period = 120; + }) + .await; + + { + let mut w = infra::schema::STREAM_SETTINGS.write().await; + w.remove(&logs_key); + w.remove(&traces_key); + infra::schema::set_stream_settings_atomic( + w.iter().map(|(k, v)| (k.clone(), v.clone())).collect(), + ); + } + + assert!( + err.contains("max query range") && err.contains("\"spans\""), + "{err}" + ); + } +} diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index 3c33acd2ee..bf82fac275 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -26,7 +26,7 @@ use config::{ }, cluster::RoleGroup, search::{SearchEventContext, SearchEventType, SqlQuery}, - sql::resolve_stream_names, + sql::{TableReferenceExt, resolve_stream_names_with_type}, stream::StreamType, }, utils::{ @@ -244,30 +244,7 @@ impl QueryConditionExt for QueryCondition { } }; - let stream_names = resolve_stream_names(&sql) - .map_err(|e| anyhow::anyhow!("Error resolving stream names in SQL query: {e}"))?; - - // SQL may contain multiple stream names, check for each stream - // if the query period is greater than the max query range - for stream in stream_names.iter() { - if infra::schema::get_stream_schema_from_cache(org_id, stream, stream_type) - .await - .is_none() - { - return Err(anyhow::anyhow!( - "Stream \"{stream}\" not found in schema, skipping alert evaluation" - )); - }; - - if let Some(settings) = infra::schema::get_settings(org_id, stream, stream_type).await { - let max_query_range = settings.max_query_range; - if max_query_range > 0 && trigger_condition.period > max_query_range * 60 { - return Err(anyhow::anyhow!( - "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\"" - )); - } - } - } + check_query_streams(org_id, stream_type, &sql, trigger_condition.period).await?; let mut time_diff = Duration::try_minutes(trigger_condition.period) .unwrap() @@ -845,6 +822,41 @@ impl ConditionGroupExt for config::meta::alerts::ConditionItem { } } +/// Checks that every stream in the alert SQL exists and allows the alert period. +async fn check_query_streams( + org_id: &str, + stream_type: StreamType, + sql: &str, + period: i64, +) -> Result<(), anyhow::Error> { + let stream_refs = resolve_stream_names_with_type(sql) + .map_err(|e| anyhow::anyhow!("Error resolving stream names in SQL query: {e}"))?; + + for stream_ref in stream_refs.iter() { + let stream = stream_ref.stream_name(); + // A JOIN may name another type, e.g. "enrichment_tables".tbl in a logs alert. + let stream_type = stream_ref.get_stream_type(stream_type); + if infra::schema::get_stream_schema_from_cache(org_id, &stream, stream_type) + .await + .is_none() + { + return Err(anyhow::anyhow!( + "Stream \"{stream}\" not found in schema, skipping alert evaluation" + )); + }; + + if let Some(settings) = infra::schema::get_settings(org_id, &stream, stream_type).await { + let max_query_range = settings.max_query_range; + if max_query_range > 0 && period > max_query_range * 60 { + return Err(anyhow::anyhow!( + "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\"" + )); + } + } + } + Ok(()) +} + /// Evaluates a list of condition items with left-to-right logical operator application /// /// Algorithm: @@ -2012,4 +2024,40 @@ mod tests { println!("✓ All deeply nested group tests with operator precedence passed!"); } + + #[tokio::test] + async fn test_check_query_streams_join_enrichment_table() { + use infra::schema::{STREAM_SCHEMAS_LATEST, SchemaCache}; + + let org_id = "test_org_alert_join_enrich"; + let logs_key = format!("{org_id}/{}/app", StreamType::Logs); + let enrich_key = format!("{org_id}/{}/geo", StreamType::EnrichmentTables); + let schema = Schema::new(vec![Field::new("ip", DataType::Utf8, true)]); + { + let mut w = STREAM_SCHEMAS_LATEST.write().await; + w.insert(logs_key.clone(), SchemaCache::new(schema.clone())); + w.insert(enrich_key.clone(), SchemaCache::new(schema)); + } + + let sql = + r#"SELECT a.ip FROM "app" AS a JOIN "enrichment_tables"."geo" AS b ON a.ip = b.ip"#; + let missing_sql = + r#"SELECT a.ip FROM "app" AS a JOIN "enrichment_tables"."nope" AS b ON a.ip = b.ip"#; + let ok = check_query_streams(org_id, StreamType::Logs, sql, 10).await; + let missing = check_query_streams(org_id, StreamType::Logs, missing_sql, 10).await; + + { + let mut w = STREAM_SCHEMAS_LATEST.write().await; + w.remove(&logs_key); + w.remove(&enrich_key); + } + + assert!(ok.is_ok(), "join with enrichment table rejected: {ok:?}"); + assert!( + missing + .unwrap_err() + .to_string() + .contains("\"nope\" not found") + ); + } }