mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 08:09:22 +00:00
fix: answer 400 for bad role updates and stop time-range overflow and PromQL time panics (#16190)
This commit is contained in:
@@ -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"
|
||||
)
|
||||
|
||||
@@ -396,6 +396,7 @@ pub async fn get_roles(Path(_org_id): Path<String>) -> 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<RoleRequest>,
|
||||
) -> 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"))]
|
||||
|
||||
@@ -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")]
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
|
||||
@@ -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<i64> {
|
||||
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)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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}\""
|
||||
|
||||
@@ -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}\""
|
||||
));
|
||||
|
||||
@@ -2433,9 +2433,10 @@ async fn clamp_start(
|
||||
}
|
||||
|
||||
fn clamped_start(stream_type: StreamType, start: i64, end: i64, max_hours: i64) -> Option<i64> {
|
||||
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.
|
||||
|
||||
@@ -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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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}",
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user