starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 16:09:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add query ingester timeout for UI (#6060)

Browse Source
This commit is contained in:
Hengfei Yang
2025-02-23 10:18:37 +08:00
committed by GitHub
parent 1a150cf7ca
commit 5aeefbd52c
25 changed files with 342 additions and 380 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli/data/export.rs
+2 -9
View File
@@ -17,10 +17,7 @@ use std::{collections::HashMap, fs, path::Path};
use actix_web::web::Query;
use async_trait::async_trait;
use config::{
meta::{search, stream::StreamType},
TIMESTAMP_COL_NAME,
};
use config::{meta::search, TIMESTAMP_COL_NAME};
use crate::{
cli::data::{cli::Cli, Context},
@@ -38,11 +35,7 @@ impl Context for Export {
async fn operator(c: Cli) -> Result<bool, anyhow::Error> {
let map = HashMap::from([("type".to_string(), c.stream_type)]);
let query_map = Query(map);
let stream_type = match get_stream_type_from_request(&query_map) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(_) => return Ok(false),
};
let stream_type = get_stream_type_from_request(&query_map).unwrap_or_default();
let table = c.stream_name;
let search_type = match get_search_type_from_request(&query_map) {
src/common/utils/auth.rs
+2 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -210,10 +210,7 @@ impl FromRequest for AuthExtractor {
use crate::common::utils::http::{get_folder, get_stream_type_from_request};
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(_) => Some(StreamType::Logs),
};
let stream_type = get_stream_type_from_request(&query);
let folder = get_folder(&query);
src/common/utils/auth_tests.rs
+21 -6
View File
@@ -3009,7 +3009,6 @@ mod tests {
data_stream_dir: String::default(),
data_db_dir: String::default(),
data_cache_dir: String::default(),
column_timestamp: String::default(),
column_all: String::default(),
feature_per_thread_lock: bool::default(),
feature_fulltext_extra_fields: String::default(),
@@ -3026,6 +3025,7 @@ mod tests {
feature_query_streaming_aggs: bool::default(),
feature_join_match_one_enabled: bool::default(),
feature_join_right_side_max_rows: usize::default(),
feature_query_skip_wal: bool::default(),
ui_enabled: bool::default(),
ui_sql_base64_enabled: bool::default(),
metrics_dedup_enabled: bool::default(),
@@ -3034,6 +3034,8 @@ mod tests {
bloom_filter_default_fields: String::default(),
bloom_filter_ndv_ratio: u64::default(),
wal_fsync_disabled: bool::default(),
wal_write_queue_full_reject: bool::default(),
wal_write_queue_enabled: bool::default(),
tracing_enabled: bool::default(),
tracing_search_enabled: bool::default(),
otel_otlp_url: String::default(),
@@ -3075,6 +3077,7 @@ mod tests {
inverted_index_search_format: String::default(),
inverted_index_tantivy_mode: String::default(),
inverted_index_count_optimizer_enabled: bool::default(),
inverted_index_camel_case_tokenizer_disabled: bool::default(),
full_text_search_type: String::default(),
query_on_stream_selection: bool::default(),
show_stream_dates_doc_num: bool::default(),
@@ -3101,8 +3104,9 @@ mod tests {
metrics_cache_enabled: bool::default(),
swagger_enabled: bool::default(),
fake_es_version: String::default(),
websocket_enabled: bool::default(),
min_auto_refresh_interval: u32::default(),
websocket_enabled: bool::default(),
websocket_close_frame_delay: u64::default(),
},
limit: config::Limit {
cpu_num: usize::default(),
@@ -3122,6 +3126,7 @@ mod tests {
mem_table_bucket_num: 1,
mem_persist_interval: u64::default(),
wal_write_buffer_size: usize::default(),
wal_write_queue_size: usize::default(),
file_push_interval: u64::default(),
file_push_limit: usize::default(),
file_move_fields_limit: usize::default(),
@@ -3131,9 +3136,14 @@ mod tests {
usage_reporting_thread_num: usize::default(),
query_thread_num: usize::default(),
query_timeout: u64::default(),
query_ingester_timeout: u64::default(),
query_default_limit: i64::default(),
query_partition_by_secs: usize::default(),
query_group_base_speed: usize::default(),
circuit_breaker_enabled: bool::default(),
circuit_breaker_watching_window: i64::default(),
circuit_breaker_reset_window_num: i64::default(),
circuit_breaker_slow_request_threshold: u64::default(),
ingest_allowed_upto: i64::default(),
ingest_flatten_level: u32::default(),
ignore_file_retention_by_stream: bool::default(),
@@ -3309,6 +3319,8 @@ mod tests {
sync_to_cache_interval: u64::default(),
max_retries: usize::default(),
max_idle_per_host: usize::default(),
keepalive_timeout: u64::default(),
multi_part_upload_size: usize::default(),
},
sns: config::Sns {
endpoint: String::default(),
@@ -3323,10 +3335,13 @@ mod tests {
ha_cluster_label: String::default(),
ha_replica_label: String::default(),
},
profiling: config::Pyroscope {
enabled: bool::default(),
server_url: String::default(),
project_name: String::default(),
profiling: config::Profiling {
pprof_enabled: bool::default(),
pprof_protobuf_enabled: bool::default(),
pprof_flamegraph_path: String::default(),
pyroscope_enabled: bool::default(),
pyroscope_server_url: String::default(),
pyroscope_project_name: String::default(),
},
smtp: config::Smtp {
smtp_enabled: bool::default(),
src/common/utils/http.rs
+10 -35
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -33,26 +33,8 @@ use tracing_opentelemetry::OpenTelemetrySpanExt;
#[inline(always)]
pub(crate) fn get_stream_type_from_request(
query: &Query<HashMap<String, String>>,
) -> Result<Option<StreamType>, Error> {
let stream_type = match query.get("type") {
Some(s) => match s.to_lowercase().as_str() {
"logs" => Some(StreamType::Logs),
"metrics" => Some(StreamType::Metrics),
"traces" => Some(StreamType::Traces),
"enrichment_tables" => Some(StreamType::EnrichmentTables),
"metadata" => Some(StreamType::Metadata),
"index" => Some(StreamType::Index),
_ => {
return Err(Error::new(
ErrorKind::Other,
"'type' query param with value 'logs', 'metrics', 'traces', 'enrichment_table', 'metadata' or 'index' allowed",
));
}
},
None => None,
};
Ok(stream_type)
) -> Option<StreamType> {
query.get("type").map(|s| StreamType::from(s.as_str()))
}
#[inline(always)]
@@ -60,15 +42,8 @@ pub(crate) fn get_search_type_from_request(
query: &Query<HashMap<String, String>>,
) -> Result<Option<SearchEventType>, Error> {
let event_type = match query.get("search_type") {
Some(s) => match s.to_lowercase().as_str() {
"ui" => Some(SearchEventType::UI),
"dashboards" => Some(SearchEventType::Dashboards),
"reports" => Some(SearchEventType::Reports),
"alerts" => Some(SearchEventType::Alerts),
"values" => Some(SearchEventType::Values),
"rum" => Some(SearchEventType::RUM),
"derived_stream" => Some(SearchEventType::DerivedStream),
"search_job" => Some(SearchEventType::SearchJob),
Some(s) => match SearchEventType::try_from(s.as_str()) {
Ok(search_type) => Some(search_type),
_ => {
return Err(Error::new(
ErrorKind::Other,
@@ -218,26 +193,26 @@ mod tests {
use super::*;
#[test]
fn test_get_file_from_cache() {
fn test_get_stream_type_from_request() {
let key = "type".to_string();
let mut map: HashMap<String, String> = HashMap::default();
map.insert(key.clone(), key.clone());
let resp = get_stream_type_from_request(&Query(map.clone()));
assert!(resp.is_err());
assert_eq!(resp, Some(StreamType::default()));
map.insert(key.clone(), "LOGS".to_string());
let resp = get_stream_type_from_request(&Query(map.clone()));
assert_eq!(resp.unwrap(), Some(StreamType::Logs));
assert_eq!(resp, Some(StreamType::Logs));
map.insert(key.clone(), "METRICS".to_string());
let resp = get_stream_type_from_request(&Query(map.clone()));
assert_eq!(resp.unwrap(), Some(StreamType::Metrics));
assert_eq!(resp, Some(StreamType::Metrics));
map.insert(key.clone(), "TRACES".to_string());
let resp = get_stream_type_from_request(&Query(map.clone()));
assert_eq!(resp.unwrap(), Some(StreamType::Traces));
assert_eq!(resp, Some(StreamType::Traces));
}
/// Test logic for IP parsing
src/config/src/config.rs
+3
View File
@@ -1067,6 +1067,9 @@ pub struct Limit {
pub query_thread_num: usize,
#[env_config(name = "ZO_QUERY_TIMEOUT", default = 600)]
pub query_timeout: u64,
#[env_config(name = "ZO_QUERY_INGESTER_TIMEOUT", default = 0)]
// default equal to query_timeout
pub query_ingester_timeout: u64,
#[env_config(name = "ZO_QUERY_DEFAULT_LIMIT", default = 1000)]
pub query_default_limit: i64,
#[env_config(name = "ZO_QUERY_PARTITION_BY_SECS", default = 1)] // seconds
src/config/src/meta/search.rs
+4 -6
View File
@@ -13,8 +13,6 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::str::FromStr;
use proto::cluster_rpc;
use serde::{Deserialize, Deserializer, Serialize};
use utoipa::ToSchema;
@@ -767,7 +765,7 @@ impl<'de> Deserialize<'de> for SearchEventType {
where
E: serde::de::Error,
{
SearchEventType::from_str(value).map_err(serde::de::Error::custom)
SearchEventType::try_from(value).map_err(serde::de::Error::custom)
}
}
@@ -791,9 +789,9 @@ impl std::fmt::Display for SearchEventType {
}
}
impl FromStr for SearchEventType {
type Err = String;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
impl TryFrom<&str> for SearchEventType {
type Error = String;
fn try_from(s: &str) -> std::result::Result<Self, Self::Error> {
let s = s.to_lowercase();
match s.as_str() {
"ui" => Ok(SearchEventType::UI),
src/config/src/meta/stream.rs
+1 -1
View File
@@ -87,7 +87,7 @@ impl From<&str> for StreamType {
"file_list" => StreamType::Filelist,
"metadata" => StreamType::Metadata,
"index" => StreamType::Index,
_ => StreamType::Logs,
_ => StreamType::default(),
}
}
}
src/handler/grpc/flight/mod.rs
+16 -5
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -90,8 +90,12 @@ impl FlightService for FlightServiceImpl {
req.query_identifier.trace_id, req.query_identifier.job_id
);
let is_super_cluster = req.super_cluster_info.is_super_cluster;
log::info!("[trace_id {}] flight->search: do_get", trace_id);
let timeout = req.search_info.timeout as u64;
log::info!(
"[trace_id {}] flight->search: do_get, timeout: {}s",
trace_id,
timeout
);
#[cfg(feature = "enterprise")]
if is_super_cluster && !SEARCH_SERVER.contain_key(&trace_id).await {
@@ -165,6 +169,7 @@ impl FlightService for FlightServiceImpl {
})?,
defer,
start,
timeout,
))
.map_err(|err| Status::from_error(Box::new(err)));
@@ -242,6 +247,7 @@ struct FlightSenderStream {
stream: SendableRecordBatchStream,
defer: Option<AsyncDefer>,
start: std::time::Instant,
timeout: u64,
}
impl FlightSenderStream {
@@ -250,12 +256,14 @@ impl FlightSenderStream {
stream: SendableRecordBatchStream,
defer: Option<AsyncDefer>,
start: std::time::Instant,
timeout: u64,
) -> Self {
Self {
trace_id,
stream,
defer,
start,
timeout,
}
}
}
@@ -265,9 +273,12 @@ impl Stream for FlightSenderStream {
fn poll_next(
mut self: std::pin::Pin<&mut Self>,
cx: &mut Context<'_>,
ctx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
match self.stream.poll_next_unpin(cx) {
if self.start.elapsed().as_secs() > self.timeout {
return Poll::Ready(None);
}
match self.stream.poll_next_unpin(ctx) {
Poll::Ready(Some(Ok(batch))) => Poll::Ready(Some(Ok(batch))),
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
src/handler/http/request/alerts/deprecated.rs
+7 -32
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -142,12 +142,7 @@ pub async fn update_alert(
async fn list_stream_alerts(path: web::Path<(String, String)>, req: HttpRequest) -> HttpResponse {
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return MetaHttpResponse::bad_request(e);
}
};
let stream_type = get_stream_type_from_request(&query);
let user_filter = query.get("owner").map(|v| v.to_string());
let enabled_filter = query
.get("enabled")
@@ -233,7 +228,7 @@ async fn list_alerts(path: web::Path<String>, req: HttpRequest) -> HttpResponse
Ok(value) => Some(value),
Err(_) => None,
});
let stream_type_filter = get_stream_type_from_request(&query).unwrap_or_default();
let stream_type_filter = get_stream_type_from_request(&query);
let stream_name_filter = query.get("stream_name").map(|v| v.as_str());
let alert_filter = AlertListFilter {
@@ -312,12 +307,7 @@ async fn list_alerts(path: web::Path<String>, req: HttpRequest) -> HttpResponse
async fn get_alert(path: web::Path<(String, String, String)>, req: HttpRequest) -> HttpResponse {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or_default(),
Err(e) => {
return MetaHttpResponse::bad_request(e);
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
match alert::get_by_name(&org_id, stream_type, &stream_name, &name).await {
Ok(Some(mut data)) => {
if let Ok(scheduled_job) = scheduler::get(
@@ -367,12 +357,7 @@ async fn get_alert(path: web::Path<(String, String, String)>, req: HttpRequest)
async fn delete_alert(path: web::Path<(String, String, String)>, req: HttpRequest) -> HttpResponse {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or_default(),
Err(e) => {
return MetaHttpResponse::bad_request(e);
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
match alert::delete_by_name(&org_id, stream_type, &stream_name, &name).await {
Ok(_) => MetaHttpResponse::ok("Alert deleted"),
Err(e) => e.into(),
@@ -404,12 +389,7 @@ async fn delete_alert(path: web::Path<(String, String, String)>, req: HttpReques
async fn enable_alert(path: web::Path<(String, String, String)>, req: HttpRequest) -> HttpResponse {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or_default(),
Err(e) => {
return MetaHttpResponse::bad_request(e);
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let enable = match query.get("value") {
Some(v) => v.parse::<bool>().unwrap_or_default(),
None => false,
@@ -449,12 +429,7 @@ async fn trigger_alert(
) -> HttpResponse {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or_default(),
Err(e) => {
return MetaHttpResponse::bad_request(e);
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
match alert::trigger_by_name(&org_id, stream_type, &stream_name, &name).await {
Ok(_) => MetaHttpResponse::ok("Alert triggered"),
Err(e) => e.into(),
src/handler/http/request/search/mod.rs
+4 -16
View File
@@ -193,10 +193,7 @@ pub async fn search(
.to_string();
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let use_cache = cfg.common.result_cache_enabled && get_use_cache_from_request(&query);
// handle encoding for query and aggs
@@ -431,10 +428,7 @@ pub async fn around(
let mut uses_fn = false;
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let around_key = match query.get("key") {
Some(v) => v.parse::<i64>().unwrap_or(0),
@@ -742,10 +736,7 @@ pub async fn values(
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => return Ok(meta::http::HttpResponse::bad_request(e)),
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let user_id = in_req
.headers()
@@ -1185,10 +1176,7 @@ pub async fn search_partition(
.unwrap_or("")
.to_string();
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let mut req: config::meta::search::SearchPartitionRequest = match json::from_slice(&body) {
Ok(v) => v,
src/handler/http/request/search/multi_streams.rs
+3 -18
View File
@@ -135,12 +135,7 @@ pub async fn search_multi(
let trace_id = get_or_create_trace_id(in_req.headers(), &http_span);
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => {
return Ok(MetaHttpResponse::bad_request(e));
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let search_type = match get_search_type_from_request(&query) {
Ok(v) => v,
@@ -709,12 +704,7 @@ pub async fn _search_partition_multi(
.unwrap_or("")
.to_string();
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => {
return Ok(MetaHttpResponse::bad_request(e));
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let req: search::MultiSearchPartitionRequest = match json::from_slice(&body) {
Ok(v) => v,
@@ -856,12 +846,7 @@ pub async fn around_multi(
let stream_names = base64::decode_url(&stream_names)?;
let mut uses_fn = false;
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => {
return Ok(MetaHttpResponse::bad_request(e));
}
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let around_key = match query.get("key") {
Some(v) => v.parse::<i64>().unwrap_or(0),
src/handler/http/request/search/search_job.rs
+1 -4
View File
@@ -70,10 +70,7 @@ pub async fn submit_job(
.to_string();
let query = web::Query::<HashMap<String, String>>::from_query(in_req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v.unwrap_or(StreamType::Logs),
Err(e) => return Ok(MetaHttpResponse::bad_request(e)),
};
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let use_cache = cfg.common.result_cache_enabled && get_use_cache_from_request(&query);
// handle encoding for query and aggs
src/handler/http/request/stream/mod.rs
+23 -112
View File
@@ -61,18 +61,7 @@ async fn schema(
) -> Result<HttpResponse, Error> {
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = stream_type.unwrap_or(StreamType::Logs);
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
stream::get_stream(&org_id, &stream_name, stream_type).await
}
@@ -106,33 +95,15 @@ async fn settings(
stream_name = format_stream_name(&stream_name);
}
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => {
if let Some(s_type) = v {
if s_type == StreamType::EnrichmentTables || s_type == StreamType::Index {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
format!("Stream type '{}' not allowed", s_type),
)),
);
}
Some(s_type)
} else {
v
}
}
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = stream_type.unwrap_or(StreamType::Logs);
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
if stream_type == StreamType::EnrichmentTables || stream_type == StreamType::Index {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
format!("Stream type '{}' not allowed", stream_type),
)),
);
}
stream::save_stream_settings(&org_id, &stream_name, stream_type, settings.into_inner()).await
}
@@ -167,33 +138,15 @@ async fn update_settings(
stream_name = format_stream_name(&stream_name);
}
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => {
if let Some(s_type) = v {
if s_type == StreamType::EnrichmentTables || s_type == StreamType::Index {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
format!("Stream type '{}' not allowed", s_type),
)),
);
}
Some(s_type)
} else {
v
}
}
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = stream_type.unwrap_or(StreamType::Logs);
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
if stream_type == StreamType::EnrichmentTables || stream_type == StreamType::Index {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
format!("Stream type '{}' not allowed", stream_type),
)),
);
}
let stream_settings: UpdateStreamSettings = stream_settings.into_inner();
let main_stream_res =
stream::update_stream_settings(&org_id, &stream_name, stream_type, stream_settings.clone())
@@ -272,17 +225,7 @@ async fn delete_fields(
) -> Result<HttpResponse, Error> {
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = get_stream_type_from_request(&query);
match stream::delete_fields(
&org_id,
&stream_name,
@@ -327,18 +270,7 @@ async fn delete(
) -> Result<HttpResponse, Error> {
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = stream_type.unwrap_or(StreamType::Logs);
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
stream::delete_stream(&org_id, &stream_name, stream_type).await
}
@@ -362,17 +294,7 @@ async fn delete(
#[get("/{org_id}/streams")]
async fn list(org_id: web::Path<String>, req: HttpRequest) -> impl Responder {
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = get_stream_type_from_request(&query);
let fetch_schema = match query.get("fetchSchema") {
Some(s) => match s.to_lowercase().as_str() {
@@ -460,18 +382,7 @@ async fn delete_stream_cache(
}
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
let stream_type = match get_stream_type_from_request(&query) {
Ok(v) => v,
Err(e) => {
return Ok(
HttpResponse::BadRequest().json(meta::http::HttpResponse::error(
http::StatusCode::BAD_REQUEST.into(),
e.to_string(),
)),
);
}
};
let stream_type = stream_type.unwrap_or(StreamType::Logs);
let stream_type = get_stream_type_from_request(&query).unwrap_or_default();
let path = if stream_name.eq("_all") {
org_id
} else {
src/infra/src/table/alerts/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
src/ingester/src/partition.rs
+94 -91
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -142,104 +142,107 @@ impl Partition {
if data.data.is_empty() {
continue;
}
let mut file_meta = FileMeta::default();
data.data.iter().for_each(|r| {
file_meta.original_size += r.data_json_size as i64;
file_meta.records += r.data.num_rows() as i64;
if file_meta.min_ts == 0 || file_meta.min_ts > r.min_ts {
file_meta.min_ts = r.min_ts;
}
if file_meta.max_ts < r.max_ts {
file_meta.max_ts = r.max_ts;
}
});
let mut persist_stat = PersistStat {
json_size: file_meta.original_size,
arrow_size: 0,
file_num: 1,
batch_num: data.data.len(),
};
// write into parquet buf
let bloom_filter_fields =
if self.schema.fields().len() >= cfg.limit.file_move_fields_limit {
let settings = infra::schema::unwrap_stream_settings(self.schema.as_ref());
infra::schema::get_stream_setting_bloom_filter_fields(&settings)
} else {
vec![]
let chunks = data.data.chunks(2000);
for data in chunks {
let mut file_meta = FileMeta::default();
data.iter().for_each(|r| {
file_meta.original_size += r.data_json_size as i64;
file_meta.records += r.data.num_rows() as i64;
if file_meta.min_ts == 0 || file_meta.min_ts > r.min_ts {
file_meta.min_ts = r.min_ts;
}
if file_meta.max_ts < r.max_ts {
file_meta.max_ts = r.max_ts;
}
});
let mut persist_stat = PersistStat {
json_size: file_meta.original_size,
arrow_size: 0,
file_num: 1,
batch_num: data.len(),
};
// write into parquet buf
let bloom_filter_fields =
if self.schema.fields().len() >= cfg.limit.file_move_fields_limit {
let settings = infra::schema::unwrap_stream_settings(self.schema.as_ref());
infra::schema::get_stream_setting_bloom_filter_fields(&settings)
} else {
vec![]
};
let batches = data
.data
.iter()
.map(|r| {
persist_stat.arrow_size += r.data_arrow_size;
r.data.clone()
})
.collect::<Vec<_>>();
let (schema, batches) =
merge_record_batches("INGESTER:PERSIST", 0, self.schema.clone(), batches)
.context(MergeRecordBatchSnafu)?;
let batches = data
.iter()
.map(|r| {
persist_stat.arrow_size += r.data_arrow_size;
r.data.clone()
})
.collect::<Vec<_>>();
let (schema, batches) =
merge_record_batches("INGESTER:PERSIST", 0, self.schema.clone(), batches)
.context(MergeRecordBatchSnafu)?;
let mut buf_parquet = Vec::new();
let mut writer = new_parquet_writer(
&mut buf_parquet,
&schema,
&bloom_filter_fields,
&file_meta,
true,
);
let mut buf_parquet = Vec::new();
let mut writer = new_parquet_writer(
&mut buf_parquet,
&schema,
&bloom_filter_fields,
&file_meta,
true,
);
writer
.write(&batches)
.await
.context(WriteParquetRecordBatchSnafu)?;
writer.close().await.context(WriteParquetRecordBatchSnafu)?;
file_meta.compressed_size = buf_parquet.len() as i64;
writer
.write(&batches)
.await
.context(WriteParquetRecordBatchSnafu)?;
writer.close().await.context(WriteParquetRecordBatchSnafu)?;
file_meta.compressed_size = buf_parquet.len() as i64;
// write into local file
let file_name = generate_filename_with_time_range(file_meta.min_ts, file_meta.max_ts);
let mut path = path.clone();
path.push(hour.to_string());
path.push(file_name);
path.set_extension("par");
create_dir_all(path.parent().unwrap())
.context(CreateFileSnafu { path: path.clone() })?;
let mut f = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&path)
.await
.context(CreateFileSnafu { path: path.clone() })?;
f.write_all(&buf_parquet)
.await
.context(WriteFileSnafu { path: path.clone() })?;
drop(f);
// write into local file
let file_name =
generate_filename_with_time_range(file_meta.min_ts, file_meta.max_ts);
let mut path = path.clone();
path.push(hour.to_string());
path.push(file_name);
path.set_extension("par");
create_dir_all(path.parent().unwrap())
.context(CreateFileSnafu { path: path.clone() })?;
let mut f = OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.open(&path)
.await
.context(CreateFileSnafu { path: path.clone() })?;
f.write_all(&buf_parquet)
.await
.context(WriteFileSnafu { path: path.clone() })?;
drop(f);
// set parquet metadata cache
let mut file_key = path.clone();
file_key.set_extension("parquet");
let file_key = file_key
.strip_prefix(base_path.clone())
.unwrap()
.to_string_lossy()
.replace('\\', "/")
.trim_start_matches('/')
.to_string();
super::WAL_PARQUET_METADATA
.write()
.await
.insert(file_key, file_meta);
// set parquet metadata cache
let mut file_key = path.clone();
file_key.set_extension("parquet");
let file_key = file_key
.strip_prefix(base_path.clone())
.unwrap()
.to_string_lossy()
.replace('\\', "/")
.trim_start_matches('/')
.to_string();
super::WAL_PARQUET_METADATA
.write()
.await
.insert(file_key, file_meta);
// update metrics
metrics::INGEST_WAL_USED_BYTES
.with_label_values(&[org_id, stream_type])
.add(buf_parquet.len() as i64);
metrics::INGEST_WAL_WRITE_BYTES
.with_label_values(&[org_id, stream_type])
.inc_by(buf_parquet.len() as u64);
// update metrics
metrics::INGEST_WAL_USED_BYTES
.with_label_values(&[org_id, stream_type])
.add(buf_parquet.len() as i64);
metrics::INGEST_WAL_WRITE_BYTES
.with_label_values(&[org_id, stream_type])
.inc_by(buf_parquet.len() as u64);
paths.push((path, persist_stat));
paths.push((path, persist_stat));
}
}
Ok((self.schema.size(), paths))
}
src/service/search/cluster/flight.rs
+6 -2
View File
@@ -13,7 +13,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{str::FromStr, sync::Arc};
use std::sync::Arc;
use arrow::array::RecordBatch;
use async_recursion::async_recursion;
@@ -131,7 +131,11 @@ pub async fn search(
let node_group = req
.search_event_type
.as_ref()
.map(|v| SearchEventType::from_str(v).ok().map(RoleGroup::from))
.map(|v| {
SearchEventType::try_from(v.as_str())
.ok()
.map(RoleGroup::from)
})
.unwrap_or(None);
let nodes = get_online_querier_nodes(trace_id, node_group).await?;
let querier_num = nodes.iter().filter(|node| node.is_querier()).count();
src/service/search/datafusion/distributed_plan/codec.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
src/service/search/datafusion/distributed_plan/mod.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
src/service/search/datafusion/distributed_plan/node.rs
+15
View File
@@ -1,3 +1,18 @@
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::sync::Arc;
use config::{
src/service/search/datafusion/distributed_plan/remote_scan.rs
+103 -21
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -26,7 +26,10 @@ use arrow_flight::{
Ticket,
};
use arrow_schema::{Schema, SchemaRef};
use config::{meta::search::ScanStats, utils::rand::generate_random_string};
use config::{
meta::search::{ScanStats, SearchEventType},
utils::rand::generate_random_string,
};
use datafusion::{
common::{DataFusionError, Result, Statistics},
execution::{RecordBatchStream, SendableRecordBatchStream, TaskContext},
@@ -161,6 +164,7 @@ impl ExecutionPlan for RemoteScanExec {
let fut = get_remote_batch(
self.remote_scan_node.clone(),
partition,
self.input.schema().clone(),
self.scan_stats.clone(),
self.partial_err.clone(),
);
@@ -179,24 +183,41 @@ impl ExecutionPlan for RemoteScanExec {
async fn get_remote_batch(
remote_scan_node: RemoteScanNode,
partition: usize,
schema: SchemaRef,
scan_stats: Arc<Mutex<ScanStats>>,
partial_err: Arc<Mutex<String>>,
) -> Result<SendableRecordBatchStream> {
let cfg = config::get_config();
let trace_id = remote_scan_node.query_identifier.trace_id.clone();
let org_id = remote_scan_node.query_identifier.org_id.clone();
let context = remote_scan_node.opentelemetry_context.clone();
let node = remote_scan_node.nodes[partition].clone();
let is_querier = remote_scan_node.is_querier(partition);
let search_type = remote_scan_node
.super_cluster_info
.search_event_type
.as_ref()
.and_then(|s| s.as_str().try_into().ok());
let mut timeout = remote_scan_node.search_infos.timeout;
if matches!(search_type, Some(SearchEventType::UI)) && !is_querier {
timeout = std::cmp::min(timeout, cfg.limit.query_ingester_timeout);
}
if timeout == 0 {
timeout = cfg.limit.query_timeout;
}
let mut request = remote_scan_node.get_flight_search_request(partition);
request.set_job_id(generate_random_string(7));
request.set_partition(partition);
request.search_info.timeout = timeout as i64;
log::info!(
"[trace_id {}] flight->search: request node: {}, is_querier: {}, files: {}, idx_files: {}",
"[trace_id {}] flight->search: request node: {}, query_type: {}, is_querier: {}, timeout: {}, files: {}, idx_files: {}",
trace_id,
&node.get_grpc_addr(),
search_type.unwrap_or(SearchEventType::UI),
is_querier,
timeout,
request.search_info.file_id_list.len(),
request.search_info.idx_file_list.len(),
);
@@ -211,7 +232,6 @@ async fn get_remote_batch(
ticket: buf.clone().into(),
});
let cfg = config::get_config();
let org_id: MetadataValue<_> = org_id
.parse()
.map_err(|_| DataFusionError::Internal("invalid org_id".to_string()))?;
@@ -240,11 +260,13 @@ async fn get_remote_batch(
);
DataFusionError::Internal("connect search node error".to_string())
})?;
let mut client =
FlightServiceClient::with_interceptor(channel, move |mut req: tonic::Request<()>| {
req.metadata_mut().insert("authorization", token.clone());
req.metadata_mut()
.insert(org_header_key.clone(), org_id.clone());
req.set_timeout(std::time::Duration::from_secs(timeout));
Ok(req)
});
client = client
@@ -260,13 +282,25 @@ async fn get_remote_batch(
is_querier,
);
let mut stream = client
.do_get(request)
.await
.map_err(|e| DataFusionError::Execution(e.to_string()))?
.into_inner();
let start = std::time::Instant::now();
let mut stream = match client.do_get(request).await {
Ok(stream) => stream,
Err(e) => {
if e.code() == tonic::Code::Cancelled || e.code() == tonic::Code::DeadlineExceeded {
return Ok(get_empty_record_batch_stream(
trace_id,
schema,
node.get_grpc_addr(),
is_querier,
partial_err,
e,
start,
));
}
return Err(DataFusionError::Execution(e.to_string()));
}
}
.into_inner();
log::info!(
"[trace_id {}] flight->search: prepare to receive response from node: {}, is_querier: {}",
@@ -279,7 +313,20 @@ async fn get_remote_batch(
let flight_data = match stream.message().await {
Ok(Some(flight_data)) => flight_data,
Ok(None) => return Err(DataFusionError::Execution("No schema returned".to_string())),
Err(e) => return Err(DataFusionError::Execution(e.to_string())),
Err(e) => {
if e.code() == tonic::Code::Cancelled || e.code() == tonic::Code::DeadlineExceeded {
return Ok(get_empty_record_batch_stream(
trace_id,
schema,
node.get_grpc_addr(),
is_querier,
partial_err,
e,
start,
));
}
return Err(DataFusionError::Execution(e.to_string()));
}
};
// convert FlightData to a stream
let schema = Arc::new(Schema::try_from(&flight_data)?);
@@ -303,9 +350,42 @@ async fn get_remote_batch(
scan_size,
partial_err,
start,
timeout,
)))
}
fn get_empty_record_batch_stream(
trace_id: String,
schema: SchemaRef,
node_addr: String,
is_querier: bool,
partial_err: Arc<Mutex<String>>,
e: tonic::Status,
start: std::time::Instant,
) -> SendableRecordBatchStream {
log::info!(
"[trace_id {}] flight->search: response node: {}, is_querier: {}, took: {} ms, err: {}",
trace_id,
node_addr,
is_querier,
start.elapsed().as_millis(),
e.to_string()
);
process_partial_err(partial_err, e);
let stream = futures::stream::empty::<Result<RecordBatch>>();
Box::pin(RecordBatchStreamAdapter::new(schema, stream))
}
fn process_partial_err(partial_err: Arc<Mutex<String>>, e: tonic::Status) {
let mut guard = partial_err.lock();
let partial_err = guard.clone();
if partial_err.is_empty() {
guard.push_str(e.to_string().as_str());
} else {
guard.push_str(format!(" \n {}", e).as_str());
}
}
struct FlightStream {
trace_id: String,
schema: SchemaRef,
@@ -316,6 +396,7 @@ struct FlightStream {
scan_size: i64,
partial_err: Arc<Mutex<String>>,
start: std::time::Instant,
timeout: u64,
}
impl FlightStream {
@@ -330,6 +411,7 @@ impl FlightStream {
scan_size: i64,
partial_err: Arc<Mutex<String>>,
start: std::time::Instant,
timeout: u64,
) -> Self {
Self {
trace_id,
@@ -341,6 +423,7 @@ impl FlightStream {
scan_size,
partial_err,
start,
timeout,
}
}
}
@@ -352,8 +435,15 @@ impl Stream for FlightStream {
mut self: std::pin::Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Self::Item>> {
let dictionaries_by_field = HashMap::new();
if self.start.elapsed().as_secs() > self.timeout {
process_partial_err(
self.partial_err.clone(),
tonic::Status::new(tonic::Code::DeadlineExceeded, "timeout"),
);
return Poll::Ready(None);
}
let dictionaries_by_field = HashMap::new();
match self.stream.poll_next_unpin(cx) {
Poll::Ready(Some(Ok(flight_data))) => {
let record_batch = flight_data_to_arrow_batch(
@@ -366,15 +456,7 @@ impl Stream for FlightStream {
Poll::Ready(None) => Poll::Ready(None),
Poll::Pending => Poll::Pending,
Poll::Ready(Some(Err(e))) => {
{
let mut guard = self.partial_err.lock();
let partial_err = guard.clone();
if partial_err.is_empty() {
guard.push_str(e.to_string().as_str());
} else {
guard.push_str(format!(" \n {}", e).as_str());
}
}
process_partial_err(self.partial_err.clone(), e);
Poll::Ready(None)
}
}
src/service/search/datafusion/distributed_plan/rewrite.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
src/service/search/datafusion/distributed_plan/streaming_aggs_exec.rs
+1 -1
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
src/service/search/grpc/storage.rs
+10 -4
View File
@@ -327,9 +327,8 @@ async fn cache_files(
scan_stats.querier_disk_cached_files += 1;
}
}
if files.len() as i64
== scan_stats.querier_memory_cached_files + scan_stats.querier_disk_cached_files
{
let files_num = files.len() as i64;
if files_num == scan_stats.querier_memory_cached_files + scan_stats.querier_disk_cached_files {
// all files are cached
return Ok(file_data::CacheType::None);
}
@@ -379,7 +378,14 @@ async fn cache_files(
}
}
});
Ok(cache_type)
// if cached file less than 50% of the total files, return None
if scan_stats.querier_memory_cached_files + scan_stats.querier_disk_cached_files < files_num / 2
{
Ok(file_data::CacheType::None)
} else {
Ok(cache_type)
}
}
#[tracing::instrument(name = "service:search:grpc:storage:cache_files_inner", skip_all)]
src/service/search/mod.rs
+5 -5
View File
@@ -55,8 +55,7 @@ use tracing_opentelemetry::OpenTelemetrySpanExt;
#[cfg(feature = "enterprise")]
use {
crate::service::grpc::make_grpc_search_client, o2_enterprise::enterprise::search::TaskStatus,
o2_enterprise::enterprise::search::WorkGroup, std::collections::HashSet, std::str::FromStr,
tracing::info_span,
o2_enterprise::enterprise::search::WorkGroup, std::collections::HashSet, tracing::info_span,
};
use super::self_reporting::report_request_usage_stats;
@@ -892,9 +891,10 @@ pub async fn query_status() -> Result<search::QueryStatusResponse, Error> {
} else {
"Long"
};
let search_type: Option<search::SearchEventType> = result
.search_type
.map(|s_event_type| search::SearchEventType::from_str(&s_event_type).unwrap());
let search_type: Option<search::SearchEventType> = result.search_type.map(|s_event_type| {
search::SearchEventType::try_from(s_event_type.as_str())
.unwrap_or(search::SearchEventType::UI)
});
status.push(search::QueryStatus {
trace_id: result.trace_id,
created_at: result.created_at,
src/service/search/super_cluster/follower.rs
+7 -3
View File
@@ -1,4 +1,4 @@
// Copyright 2024 OpenObserve Inc.
// Copyright 2025 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
@@ -13,7 +13,7 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::{str::FromStr, sync::Arc};
use std::sync::Arc;
use config::{
meta::{
@@ -153,7 +153,11 @@ pub async fn search(
let node_group = req
.search_event_type
.as_ref()
.map(|v| SearchEventType::from_str(v).ok().map(RoleGroup::from))
.map(|v| {
SearchEventType::try_from(v.as_str())
.ok()
.map(RoleGroup::from)
})
.unwrap_or(None);
let nodes = get_online_querier_nodes(&trace_id, node_group).await?;
let querier_num = nodes.iter().filter(|node| node.is_querier()).count();