diff --git a/src/api/management/src/request/alerts/history.rs b/src/api/management/src/request/alerts/history.rs index 7752b3058d..213a0686ee 100644 --- a/src/api/management/src/request/alerts/history.rs +++ b/src/api/management/src/request/alerts/history.rs @@ -224,8 +224,10 @@ pub async fn get_alert_history( Some(&user_email.user_id), ) .await; - if max_query_range > 0 && (end_time - start_time) > max_query_range * 3600 * 1_000_000 { - start_time = end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(start_time, end_time, max_query_range) + { + start_time = clamped; log::warn!( "Start time for alert History api for org {org_id} updated as per max query range set for triggers stream" ) diff --git a/src/api/management/src/request/authz/fga.rs b/src/api/management/src/request/authz/fga.rs index be69ac38a5..59ec6ebb1f 100644 --- a/src/api/management/src/request/authz/fga.rs +++ b/src/api/management/src/request/authz/fga.rs @@ -396,6 +396,7 @@ pub async fn get_roles(Path(_org_id): Path) -> Response { request_body(content = inline(RoleRequest), description = "RoleRequest", content_type = "application/json"), responses( (status = 200, description = "Success", content_type = "application/json", body = Object), + (status = 400, description = "Invalid role name, permission or user id; nothing is applied if any entry is invalid", content_type = "application/json", body = ()), (status = 500, description = "Failure", content_type = "application/json", body = ()), ), extensions( @@ -407,6 +408,16 @@ pub async fn update_role( Path((org_id, role_id)): Path<(String, String)>, Json(update_role): Json, ) -> Response { + if let Err(msg) = o2_openfga::authorizer::roles::validate_role_update( + &org_id, + &role_id, + &update_role.add, + &update_role.remove, + update_role.add_users.as_ref(), + update_role.remove_users.as_ref(), + ) { + return MetaHttpResponse::bad_request(msg); + } match o2_openfga::authorizer::roles::update_role( &org_id, &role_id, @@ -1478,28 +1489,34 @@ mod tests { #[cfg(feature = "enterprise")] { init_ofga_test(); - let role_request = o2_dex::meta::auth::RoleRequest { - add: vec![o2_dex::meta::auth::O2EntityAuthorization { - object: "permission1".to_string(), - permission: o2_dex::meta::auth::Permission::AllowAll, - }], - remove: vec![o2_dex::meta::auth::O2EntityAuthorization { - object: "permission2".to_string(), - permission: o2_dex::meta::auth::Permission::AllowAll, - }], + let grant = |object: &str| o2_dex::meta::auth::O2EntityAuthorization { + object: object.to_string(), + permission: o2_dex::meta::auth::Permission::AllowAll, + }; + let role_request = |object: &str| o2_dex::meta::auth::RoleRequest { + add: vec![grant(object)], + remove: vec![grant("stream:permission2")], add_users: Some(HashSet::from_iter(vec!["user1".to_string()])), remove_users: Some(HashSet::from_iter(vec!["user2".to_string()])), }; let resp = update_role( Path(("test_org".to_string(), "test_role".to_string())), - Json(role_request), + Json(role_request("stream:permission1")), ) .await; assert_eq!( extract_status_code(&resp), StatusCode::INTERNAL_SERVER_ERROR ); + + // A type the model lacks is the client's error, not OpenFGA's 400 surfaced as a 500. + let resp = update_role( + Path(("test_org".to_string(), "test_role".to_string())), + Json(role_request("no_such_type:permission1")), + ) + .await; + assert_eq!(extract_status_code(&resp), StatusCode::BAD_REQUEST); } #[cfg(not(feature = "enterprise"))] diff --git a/src/api/management/src/request/dashboards/reports_history.rs b/src/api/management/src/request/dashboards/reports_history.rs index 48379d7bef..662acc5c6e 100644 --- a/src/api/management/src/request/dashboards/reports_history.rs +++ b/src/api/management/src/request/dashboards/reports_history.rs @@ -568,11 +568,7 @@ async fn clamp_to_max_query_range(org_id: &str, user_id: &str, start: i64, end: }; let max_hours = get_settings_max_query_range(settings.max_query_range, org_id, Some(user_id)).await; - if max_hours > 0 && (end - start) > max_hours * 3600 * 1_000_000 { - end - max_hours * 3600 * 1_000_000 - } else { - start - } + config::utils::query_range::clamp_start_time(start, end, max_hours).unwrap_or(start) } #[cfg(feature = "enterprise")] diff --git a/src/api/management/src/request/workflows/mod.rs b/src/api/management/src/request/workflows/mod.rs index 439c83c613..04abc74eff 100644 --- a/src/api/management/src/request/workflows/mod.rs +++ b/src/api/management/src/request/workflows/mod.rs @@ -1096,8 +1096,10 @@ pub async fn get_workflow_history( Some(&user_email.user_id), ) .await; - if max_query_range > 0 && (end_time - start_time) > max_query_range * 3600 * 1_000_000 { - start_time = end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(start_time, end_time, max_query_range) + { + start_time = clamped; log::warn!( "Start time for alert History api for org {org_id} updated as per max query range set for triggers stream" ) diff --git a/src/api/pipelines/src/request/pipelines/history.rs b/src/api/pipelines/src/request/pipelines/history.rs index 6901298c55..89473e30ca 100644 --- a/src/api/pipelines/src/request/pipelines/history.rs +++ b/src/api/pipelines/src/request/pipelines/history.rs @@ -154,8 +154,10 @@ pub async fn get_pipeline_history( Some(&user_email.user_id), ) .await; - if max_query_range > 0 && (end_time - start_time) > max_query_range * 3600 * 1_000_000 { - start_time = end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(start_time, end_time, max_query_range) + { + start_time = clamped; log::warn!( "Start time for pipeline History api for org {org_id} updated as per max query range set for triggers stream" ) diff --git a/src/api/search/src/promql/mod.rs b/src/api/search/src/promql/mod.rs index d1b71439ad..fa5a365cf5 100644 --- a/src/api/search/src/promql/mod.rs +++ b/src/api/search/src/promql/mod.rs @@ -541,6 +541,16 @@ async fn query_range( } }, }; + if let Err(e) = validate_range_window(start, end) { + return ( + StatusCode::BAD_REQUEST, + axum::Json(config::meta::promql::ApiFuncResponse::<()>::err_bad_data( + e, + Some(trace_id), + )), + ) + .into_response(); + } if step == 0 { step = promql::round_step((end - start) / promql::MAX_DATA_POINTS); } @@ -1201,6 +1211,19 @@ fn validate_metadata_params( Ok((selector, start, end)) } +fn validate_range_window(start: i64, end: i64) -> Result<(), String> { + // `end - start` and the gRPC epoch conversion both overflow on a negative or reversed window. + if start < 0 { + return Err(format!( + "start must not be before the Unix epoch, got {start}" + )); + } + if end < start { + return Err(format!("end {end} must not be before start {start}")); + } + Ok(()) +} + /// prometheus formatting query expressions // refer: https://prometheus.io/docs/prometheus/latest/querying/api/#formatting-query-expressions @@ -1750,6 +1773,15 @@ impl promql_parser::util::ExprVisitor for MaxLookbackWindowVisitor { mod tests { use super::*; + #[test] + fn validate_range_window_rejects_negative_and_inverted_windows() { + assert!(validate_range_window(0, 0).is_ok()); + assert!(validate_range_window(1_000, 2_000).is_ok()); + assert!(validate_range_window(-1, 10).is_err()); + assert!(validate_range_window(i64::MIN, 10).is_err()); + assert!(validate_range_window(10, 9).is_err()); + } + #[test] fn metric_reads_check_permission_for_every_non_root_user() { let text = include_str!("mod.rs"); diff --git a/src/api/search/src/search/mod.rs b/src/api/search/src/search/mod.rs index 0bae914991..48026343c7 100644 --- a/src/api/search/src/search/mod.rs +++ b/src/api/search/src/search/mod.rs @@ -320,10 +320,13 @@ pub async fn search( .await; // Enrichment tables have no retention and must be searched in full; skip the clamp. if stream_type != StreamType::EnrichmentTables - && max_query_range > 0 - && (req.query.end_time - req.query.start_time) > max_query_range * 3600 * 1_000_000 + && let Some(clamped) = config::utils::query_range::clamp_start_time( + req.query.start_time, + req.query.end_time, + max_query_range, + ) { - req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; + req.query.start_time = clamped; range_error = format!( "Query duration is modified due to query range restriction of {max_query_range} hours" ); @@ -1807,10 +1810,13 @@ pub async fn result_schema( .await; // Enrichment tables have no retention and must be searched in full; skip the clamp. if stream_type != StreamType::EnrichmentTables - && max_query_range > 0 - && (req.query.end_time - req.query.start_time) > max_query_range * 3600 * 1_000_000 + && let Some(clamped) = config::utils::query_range::clamp_start_time( + req.query.start_time, + req.query.end_time, + max_query_range, + ) { - req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; + req.query.start_time = clamped; } } } diff --git a/src/api/search/src/search/multi_streams.rs b/src/api/search/src/search/multi_streams.rs index 4e706b2219..a18ce8d762 100644 --- a/src/api/search/src/search/multi_streams.rs +++ b/src/api/search/src/search/multi_streams.rs @@ -256,10 +256,13 @@ pub async fn search_multi( .await; // Enrichment tables have no retention and must be searched in full; skip the clamp. if stream_type != StreamType::EnrichmentTables - && max_query_range > 0 - && (req.query.end_time - req.query.start_time) > max_query_range * 3600 * 1_000_000 + && let Some(clamped) = config::utils::query_range::clamp_start_time( + req.query.start_time, + req.query.end_time, + max_query_range, + ) { - req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; + req.query.start_time = clamped; range_error = format!( "{} Query duration for stream {} is modified due to query range restriction of {} hours", range_error, stream_name, max_query_range diff --git a/src/api/search/src/search/search_inspector.rs b/src/api/search/src/search/search_inspector.rs index 9449786f3c..5f635abaa8 100644 --- a/src/api/search/src/search/search_inspector.rs +++ b/src/api/search/src/search/search_inspector.rs @@ -181,10 +181,12 @@ pub async fn get_search_profile( let max_query_range = get_settings_max_query_range(settings.max_query_range, meta_org_id, Some(&user_id)) .await; - if max_query_range > 0 - && (req.query.end_time - req.query.start_time) > max_query_range * 3600 * 1_000_000 - { - req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = config::utils::query_range::clamp_start_time( + req.query.start_time, + req.query.end_time, + max_query_range, + ) { + req.query.start_time = clamped; range_error = format!( "Query duration is modified due to query range restriction of {max_query_range} hours" ); diff --git a/src/api/search/src/traces/list/plan.rs b/src/api/search/src/traces/list/plan.rs index 19d6d5ba55..36c0f1ab43 100644 --- a/src/api/search/src/traces/list/plan.rs +++ b/src/api/search/src/traces/list/plan.rs @@ -198,8 +198,10 @@ impl TraceListOptions { } pub fn restrict_range(&mut self, max_hours: i64) { - if max_hours > 0 && (self.range.1 - self.range.0) > max_hours * 3600 * 1_000_000 { - self.range.0 = self.range.1 - max_hours * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(self.range.0, self.range.1, max_hours) + { + self.range.0 = clamped; self.range_error = format!( "Query duration is modified due to query range restriction of {max_hours} hours" ); diff --git a/src/api/search/src/traces/session.rs b/src/api/search/src/traces/session.rs index abfc9e43d9..0ea3399874 100644 --- a/src/api/search/src/traces/session.rs +++ b/src/api/search/src/traces/session.rs @@ -188,8 +188,10 @@ pub async fn get_latest_sessions( ) .await; let mut range_error = String::new(); - if max_query_range > 0 && (end_time - start_time) > max_query_range * 3600 * 1_000_000 { - start_time = end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(start_time, end_time, max_query_range) + { + start_time = clamped; range_error = format!( "Query duration is modified due to query range restriction of {max_query_range} hours" ); diff --git a/src/api/search/src/traces/user.rs b/src/api/search/src/traces/user.rs index 6697bcd943..65cd303a8d 100644 --- a/src/api/search/src/traces/user.rs +++ b/src/api/search/src/traces/user.rs @@ -155,8 +155,10 @@ pub async fn get_latest_users( ) .await; let mut range_error = String::new(); - if max_query_range > 0 && (end_time - start_time) > max_query_range * 3600 * 1_000_000 { - start_time = end_time - max_query_range * 3600 * 1_000_000; + if let Some(clamped) = + config::utils::query_range::clamp_start_time(start_time, end_time, max_query_range) + { + start_time = clamped; range_error = format!( "Query duration is modified due to query range restriction of {max_query_range} hours" ); diff --git a/src/config/src/utils/query_range.rs b/src/config/src/utils/query_range.rs index 3a68f44748..dcba5c41fb 100644 --- a/src/config/src/utils/query_range.rs +++ b/src/config/src/utils/query_range.rs @@ -26,3 +26,47 @@ pub fn get_default_max_query_range(stream_max_query_range: i64) -> i64 { default_max_query_range } } + +/// Start time limited to `max_hours` before `end`, or `None` when no clamp applies. +pub fn clamp_start_time(start: i64, end: i64, max_hours: i64) -> Option { + if max_hours <= 0 { + return None; + } + // Stream settings accept any i64 hours; a plain multiply wraps and pushes start past end. + let max_micros = max_hours.saturating_mul(3600 * 1_000_000); + (end.saturating_sub(start) > max_micros).then(|| end - max_micros) +} + +#[cfg(test)] +mod tests { + use super::*; + + const HOUR: i64 = 3600 * 1_000_000; + + #[test] + fn clamp_start_time_limits_the_range() { + assert_eq!(clamp_start_time(0, 10 * HOUR, 2), Some(8 * HOUR)); + assert_eq!(clamp_start_time(9 * HOUR, 10 * HOUR, 2), None); + assert_eq!(clamp_start_time(8 * HOUR, 10 * HOUR, 2), None); + assert_eq!(clamp_start_time(0, 10 * HOUR, 0), None); + assert_eq!(clamp_start_time(0, 10 * HOUR, -5), None); + } + + #[test] + fn clamp_start_time_survives_a_huge_max_query_range() { + let (start, end) = (1_792_919_976_430_732, 1_793_524_776_430_732); + assert_eq!(clamp_start_time(start, end, 4_746_838_768), None); + assert_eq!(clamp_start_time(start, end, i64::MAX), None); + assert_eq!(clamp_start_time(i64::MIN, end, i64::MAX / HOUR + 1), None); + } + + #[test] + fn clamp_start_time_survives_extreme_start_times() { + let end = 1_793_524_776_430_732; + assert_eq!(clamp_start_time(i64::MIN, end, 2), Some(end - 2 * HOUR)); + assert_eq!( + clamp_start_time(i64::MIN, i64::MAX, 2), + Some(i64::MAX - 2 * HOUR) + ); + } +} diff --git a/src/core/src/alerts/alert.rs b/src/core/src/alerts/alert.rs index 298040a56b..5e74990e1e 100644 --- a/src/core/src/alerts/alert.rs +++ b/src/core/src/alerts/alert.rs @@ -892,7 +892,7 @@ async fn prepare_alert( if max_query_range > 0 && !alert.query_condition.promql_rule_mode && !alert.is_real_time - && alert.trigger_condition.period > max_query_range * 60 + && alert.trigger_condition.period > max_query_range.saturating_mul(60) { return Err(AlertError::PeriodExceedsMaxQueryRange { max_query_range_hours: max_query_range, @@ -1093,7 +1093,7 @@ async fn prepare_alert( let max_query_range = settings.max_query_range; if max_query_range > 0 && !alert.is_real_time - && alert.trigger_condition.period > max_query_range * 60 + && alert.trigger_condition.period > max_query_range.saturating_mul(60) { return Err(AlertError::PeriodExceedsMaxQueryRange { max_query_range_hours: max_query_range, diff --git a/src/core/src/alerts/derived_streams.rs b/src/core/src/alerts/derived_streams.rs index 3fda351a8f..db30a6d240 100644 --- a/src/core/src/alerts/derived_streams.rs +++ b/src/core/src/alerts/derived_streams.rs @@ -83,7 +83,8 @@ pub async fn save( { let max_query_range = settings.max_query_range; if max_query_range > 0 - && derived_stream.trigger_condition.period > max_query_range * 60 + && derived_stream.trigger_condition.period + > max_query_range.saturating_mul(60) { return Err(anyhow::anyhow!( "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\"" diff --git a/src/core/src/alerts/mod.rs b/src/core/src/alerts/mod.rs index a0c9691c1c..3aaa01a376 100644 --- a/src/core/src/alerts/mod.rs +++ b/src/core/src/alerts/mod.rs @@ -892,7 +892,7 @@ async fn check_query_streams( 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 { + if max_query_range > 0 && period > max_query_range.saturating_mul(60) { return Err(anyhow::anyhow!( "Query period is greater than max query range of {max_query_range} hours for stream \"{stream}\"" )); diff --git a/src/core/src/public_dashboards/mod.rs b/src/core/src/public_dashboards/mod.rs index d4a792a42d..b3ed5d8aa0 100644 --- a/src/core/src/public_dashboards/mod.rs +++ b/src/core/src/public_dashboards/mod.rs @@ -2433,9 +2433,10 @@ async fn clamp_start( } fn clamped_start(stream_type: StreamType, start: i64, end: i64, max_hours: i64) -> Option { - let max = max_hours * 3600 * 1_000_000; - (stream_type != StreamType::EnrichmentTables && max_hours > 0 && end - start > max) - .then_some(end - max) + if stream_type == StreamType::EnrichmentTables { + return None; + } + config::utils::query_range::clamp_start_time(start, end, max_hours) } /// The shifts the live renderer runs; PromQL takes fixed units only and none when instant. diff --git a/src/promql_service/src/search/grpc/mod.rs b/src/promql_service/src/search/grpc/mod.rs index 0ceaa90ba0..190653ee32 100644 --- a/src/promql_service/src/search/grpc/mod.rs +++ b/src/promql_service/src/search/grpc/mod.rs @@ -15,7 +15,7 @@ use std::{ sync::Arc, - time::{Duration, UNIX_EPOCH}, + time::{Duration, SystemTime, UNIX_EPOCH}, }; use async_trait::async_trait; @@ -400,16 +400,13 @@ pub async fn search_inner( let org_id = &req.org_id; let query = req.query.as_ref().unwrap(); let prom_expr = promql::parse(&query.query).map_err(DataFusionError::Execution)?; + let (eval_start, eval_end, interval) = eval_window(query)?; let eval_stmt = parser::EvalStmt { expr: prom_expr, - start: UNIX_EPOCH - .checked_add(Duration::from_micros(query.start as _)) - .unwrap(), - end: UNIX_EPOCH - .checked_add(Duration::from_micros(query.end as _)) - .unwrap(), - interval: Duration::from_micros(query.step as _), + start: eval_start, + end: eval_end, + interval, lookback_delta: DEFAULT_LOOKBACK, }; @@ -461,6 +458,33 @@ pub async fn search_inner( Ok((value, result_type, scan_stats, took, warnings)) } +/// The request's microsecond window as evaluator times, rejecting values they cannot hold. +fn eval_window( + query: &cluster_rpc::MetricsQueryStmt, +) -> Result<(SystemTime, SystemTime, Duration)> { + let to_duration = |name: &str, v: i64| { + u64::try_from(v).map(Duration::from_micros).map_err(|_| { + DataFusionError::Execution(format!("promql {name} must not be negative, got {v}")) + }) + }; + let start = to_duration("start", query.start)?; + let end = to_duration("end", query.end)?; + let step = to_duration("step", query.step)?; + if end < start { + return Err(DataFusionError::Execution(format!( + "promql end {} must not be before start {}", + query.end, query.start + )) + .into()); + } + let to_time = |name: &str, d: Duration| { + UNIX_EPOCH.checked_add(d).ok_or_else(|| { + DataFusionError::Execution(format!("promql {name} is out of range, got {d:?}")) + }) + }; + Ok((to_time("start", start)?, to_time("end", end)?, step)) +} + async fn get_max_file_list( trace_id: &str, org_id: &str, @@ -942,4 +966,45 @@ mod tests { ); } } + + fn stmt(start: i64, end: i64, step: i64) -> cluster_rpc::MetricsQueryStmt { + cluster_rpc::MetricsQueryStmt { + start, + end, + step, + ..Default::default() + } + } + + #[test] + fn eval_window_converts_a_valid_window() { + let (start, end, step) = eval_window(&stmt(1_000_000, 61_000_000, 15_000_000)).unwrap(); + assert_eq!(start, UNIX_EPOCH + Duration::from_secs(1)); + assert_eq!(end, UNIX_EPOCH + Duration::from_secs(61)); + assert_eq!(step, Duration::from_secs(15)); + let (start, end, step) = eval_window(&stmt(5, 5, 0)).unwrap(); + assert_eq!(start, end); + assert_eq!(step, Duration::ZERO); + // Whether a max end fits SystemTime is platform dependent; it must error, never panic. + let max_end_fits = UNIX_EPOCH + .checked_add(Duration::from_micros(i64::MAX as u64)) + .is_some(); + assert_eq!(eval_window(&stmt(0, i64::MAX, 1)).is_ok(), max_end_fits); + } + + #[test] + fn eval_window_rejects_negative_times_and_inverted_ranges() { + for (start, end, step, needle) in [ + (-1, 10, 1, "start must not be negative"), + (i64::MIN, 10, 1, "start must not be negative"), + (0, -1, 1, "end must not be negative"), + (0, 10, -1, "step must not be negative"), + (10, 5, 1, "end 5 must not be before start 10"), + ] { + let err = eval_window(&stmt(start, end, step)) + .unwrap_err() + .to_string(); + assert!(err.contains(needle), "{start},{end},{step}: {err}"); + } + } } diff --git a/src/search_service/src/partition/stream_files.rs b/src/search_service/src/partition/stream_files.rs index f8a05409f2..ebdf77e038 100644 --- a/src/search_service/src/partition/stream_files.rs +++ b/src/search_service/src/partition/stream_files.rs @@ -80,7 +80,10 @@ pub async fn collect_stream_files( let settings_max_query_range = get_settings_max_query_range(stream_settings.max_query_range, org_id, user_id) .await; - max_query_range = max(max_query_range, settings_max_query_range * 3600 * 1_000_000); + max_query_range = max( + max_query_range, + settings_max_query_range.saturating_mul(3600 * 1_000_000), + ); max_query_range_in_hour = max(max_query_range_in_hour, settings_max_query_range); files.extend(stream_files); } else { diff --git a/src/search_service/src/streaming/cache.rs b/src/search_service/src/streaming/cache.rs index 939db11101..3fdfdb5fc7 100644 --- a/src/search_service/src/streaming/cache.rs +++ b/src/search_service/src/streaming/cache.rs @@ -226,7 +226,8 @@ pub async fn handle_cache_responses_and_deltas( let cache_start_time = all_timestamps.iter().min().copied().unwrap_or_default(); let cache_end_time = all_timestamps.iter().max().copied().unwrap_or_default(); let cache_duration = cache_end_time - cache_start_time; // microseconds - let cached_search_duration = cache_duration + (max_query_range * 3600 * 1_000_000); // microseconds + let cached_search_duration = + cache_duration.saturating_add(max_query_range.saturating_mul(3600 * 1_000_000)); // microseconds log::info!( "[HTTP2_STREAM trace_id {trace_id}] Handling cache response and deltas, curr_res_size: {curr_res_size}, cached_search_duration: {cached_search_duration}, remaining_query_duration: {remaining_query_range}, deltas_len: {}, cache_start_time: {cache_start_time}, cache_end_time: {cache_end_time}", diff --git a/src/search_service/src/streaming/execution.rs b/src/search_service/src/streaming/execution.rs index d8e5e525a7..a46ef9aee9 100644 --- a/src/search_service/src/streaming/execution.rs +++ b/src/search_service/src/streaming/execution.rs @@ -86,10 +86,13 @@ pub async fn do_partitioned_search( // [creation_time .. now]); clamping the range would drop older enrichment data, so skip it. let mut range_error = String::new(); if stream_type != StreamType::EnrichmentTables - && max_query_range > 0 - && (req.query.end_time - req.query.start_time) > max_query_range * 3600 * 1_000_000 + && let Some(clamped) = config::utils::query_range::clamp_start_time( + req.query.start_time, + req.query.end_time, + max_query_range, + ) { - req.query.start_time = req.query.end_time - max_query_range * 3600 * 1_000_000; + req.query.start_time = clamped; log::info!( "[HTTP2_STREAM trace_id {trace_id}] Query duration is modified due to query range restriction of {max_query_range} hours, new start_time: {}", req.query.start_time