mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-04 16:09:05 +00:00
fix: resolve "no querier online" when ZO_NODE_ROLE_GROUP=interactive by replacing is_http_req with SearchEventType-based RoleGroup (#12601)
This commit is contained in:
@@ -578,6 +578,8 @@ pub struct SearchPartitionRequest {
|
||||
pub histogram_interval: i64,
|
||||
#[serde(default)]
|
||||
pub sampling_ratio: Option<f64>,
|
||||
#[serde(default)]
|
||||
pub search_type: Option<SearchEventType>,
|
||||
}
|
||||
|
||||
impl SearchPartitionRequest {
|
||||
@@ -612,6 +614,7 @@ impl From<&Request> for SearchPartitionRequest {
|
||||
streaming_output: req.query.streaming_output,
|
||||
histogram_interval: req.query.histogram_interval,
|
||||
sampling_ratio: req.query.sampling_ratio,
|
||||
search_type: req.search_type,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2206,6 +2209,7 @@ mod tests {
|
||||
streaming_output: false,
|
||||
histogram_interval: 0,
|
||||
sampling_ratio: None,
|
||||
search_type: None,
|
||||
};
|
||||
|
||||
req.decode().unwrap();
|
||||
|
||||
@@ -242,7 +242,6 @@ impl Search for Searcher {
|
||||
stream_type,
|
||||
&request,
|
||||
req.skip_max_query_range,
|
||||
true,
|
||||
true, // allow streamings aggs cache for grpc search partition
|
||||
)
|
||||
.await;
|
||||
|
||||
@@ -1565,7 +1565,6 @@ pub async fn search_partition(
|
||||
stream_type,
|
||||
&req,
|
||||
false,
|
||||
true,
|
||||
true, // allow streamings aggs cache for http search partition handler
|
||||
)
|
||||
.instrument(http_span)
|
||||
|
||||
@@ -1366,6 +1366,7 @@ async fn process_latest_traces_stream(
|
||||
streaming_output: false,
|
||||
histogram_interval: 0,
|
||||
sampling_ratio: None,
|
||||
search_type: Some(config::meta::search::SearchEventType::UI),
|
||||
};
|
||||
|
||||
let partitions = match SearchService::search_partition(
|
||||
@@ -1375,7 +1376,6 @@ async fn process_latest_traces_stream(
|
||||
stream_type,
|
||||
&partition_req,
|
||||
false,
|
||||
false,
|
||||
use_cache,
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -23,6 +23,7 @@ use config::{
|
||||
cluster::LOCAL_NODE,
|
||||
get_config, ider,
|
||||
meta::{
|
||||
cluster::RoleGroup,
|
||||
function::RESULT_ARRAY,
|
||||
search::{self},
|
||||
self_reporting::usage::{RequestStats, UsageType},
|
||||
@@ -600,12 +601,13 @@ pub async fn search_partition(
|
||||
stream_type: StreamType,
|
||||
req: &search::SearchPartitionRequest,
|
||||
skip_max_query_range: bool,
|
||||
is_http_req: bool,
|
||||
use_cache: bool,
|
||||
) -> Result<search::SearchPartitionResponse, Error> {
|
||||
let cfg = get_config();
|
||||
|
||||
let ctx = PartitionSqlContext::new(req, org_id, stream_type, is_http_req).await?;
|
||||
let role_group = req.search_type.map(RoleGroup::from);
|
||||
|
||||
let ctx = PartitionSqlContext::new(req, org_id, stream_type).await?;
|
||||
|
||||
let stream_files = collect_stream_files(trace_id, user_id, &ctx).await?;
|
||||
|
||||
@@ -627,7 +629,7 @@ pub async fn search_partition(
|
||||
non_ts_order_by_cols: vec![],
|
||||
};
|
||||
|
||||
let total_secs = estimated_secs(trace_id, &ctx, is_http_req, resp.original_size).await?;
|
||||
let total_secs = estimated_secs(trace_id, &ctx, role_group, resp.original_size).await?;
|
||||
|
||||
if ctx.use_single_partition
|
||||
|| (ctx.is_streaming_aggregate && total_secs <= cfg.limit.aggs_min_num_partition_secs)
|
||||
@@ -1022,9 +1024,9 @@ pub async fn search_partition_multi(
|
||||
streaming_output: req.streaming_output,
|
||||
histogram_interval: req.histogram_interval,
|
||||
sampling_ratio: None,
|
||||
search_type: None,
|
||||
},
|
||||
false,
|
||||
true,
|
||||
false, // disable aggs cache
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -28,14 +28,9 @@ use super::sql_context::PartitionSqlContext;
|
||||
pub(crate) async fn estimated_secs(
|
||||
trace_id: &str,
|
||||
_ctx: &PartitionSqlContext,
|
||||
is_http_req: bool,
|
||||
role_group: Option<RoleGroup>,
|
||||
original_size: usize,
|
||||
) -> Result<usize, Error> {
|
||||
let role_group = if is_http_req {
|
||||
Some(RoleGroup::Interactive)
|
||||
} else {
|
||||
Some(RoleGroup::Background)
|
||||
};
|
||||
let nodes = get_cached_online_querier_nodes(role_group)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
|
||||
@@ -20,9 +20,7 @@ use config::{
|
||||
},
|
||||
utils::{
|
||||
base64,
|
||||
sql::{
|
||||
is_complex_query, is_eligible_for_histogram, is_explain_query, is_simple_distinct_query,
|
||||
},
|
||||
sql::{is_complex_query, is_eligible_for_histogram, is_explain_query},
|
||||
},
|
||||
};
|
||||
use infra::errors::Error;
|
||||
@@ -51,7 +49,6 @@ impl PartitionSqlContext {
|
||||
req: &SearchPartitionRequest,
|
||||
org_id: &str,
|
||||
stream_type: StreamType,
|
||||
is_http_req: bool,
|
||||
) -> Result<Self, Error> {
|
||||
let query = cluster_rpc::SearchQuery {
|
||||
start_time: req.start_time,
|
||||
@@ -64,15 +61,13 @@ impl PartitionSqlContext {
|
||||
|
||||
let is_explain = is_explain_query(&req.sql);
|
||||
let is_complex = is_complex_query(&req.sql).unwrap_or(false);
|
||||
let is_http_distinct = is_simple_distinct_query(&req.sql).unwrap_or(false) && is_http_req;
|
||||
let ts_column = get_ts_col_order_by(&sql, TIMESTAMP_COL_NAME, is_complex).map(|(v, _)| v);
|
||||
let is_streaming_agg = is_streaming_aggregate(&req.sql, ts_column.as_deref());
|
||||
let apply_over_hits = req.query_fn.as_ref().is_some_and(|v| {
|
||||
!v.is_empty() && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or(v.to_string()))
|
||||
});
|
||||
|
||||
let use_single_partition = is_http_distinct
|
||||
|| is_explain
|
||||
let use_single_partition = is_explain
|
||||
|| ((ts_column.is_none() || apply_over_hits)
|
||||
&& !(req.streaming_output && is_streaming_agg));
|
||||
|
||||
|
||||
@@ -463,6 +463,7 @@ pub async fn get_partitions(
|
||||
streaming_output: true,
|
||||
histogram_interval: req.query.histogram_interval,
|
||||
sampling_ratio: req.query.sampling_ratio,
|
||||
search_type: req.search_type,
|
||||
};
|
||||
|
||||
let res = SearchService::search_partition(
|
||||
@@ -472,7 +473,6 @@ pub async fn get_partitions(
|
||||
stream_type,
|
||||
&search_partition_req,
|
||||
false,
|
||||
false,
|
||||
req.use_cache,
|
||||
)
|
||||
.instrument(tracing::info_span!(
|
||||
|
||||
Reference in New Issue
Block a user