diff --git a/src/cli/data/export.rs b/src/cli/data/export.rs index 7795963a37..381f63c878 100644 --- a/src/cli/data/export.rs +++ b/src/cli/data/export.rs @@ -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 { 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) { diff --git a/src/common/utils/auth.rs b/src/common/utils/auth.rs index 31e0b76feb..8f187fc2eb 100644 --- a/src/common/utils/auth.rs +++ b/src/common/utils/auth.rs @@ -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::>::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); diff --git a/src/common/utils/auth_tests.rs b/src/common/utils/auth_tests.rs index d62eb40471..3e977e7556 100644 --- a/src/common/utils/auth_tests.rs +++ b/src/common/utils/auth_tests.rs @@ -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(), diff --git a/src/common/utils/http.rs b/src/common/utils/http.rs index ae7b885825..81e6947acc 100644 --- a/src/common/utils/http.rs +++ b/src/common/utils/http.rs @@ -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>, -) -> Result, 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 { + 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>, ) -> Result, 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 = 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 diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 771f58940c..407a92b13a 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -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 diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index c9156fc71d..5fa7ff4ae3 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -13,8 +13,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -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 { +impl TryFrom<&str> for SearchEventType { + type Error = String; + fn try_from(s: &str) -> std::result::Result { let s = s.to_lowercase(); match s.as_str() { "ui" => Ok(SearchEventType::UI), diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index e4a3e27c3f..c0d7f84467 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -87,7 +87,7 @@ impl From<&str> for StreamType { "file_list" => StreamType::Filelist, "metadata" => StreamType::Metadata, "index" => StreamType::Index, - _ => StreamType::Logs, + _ => StreamType::default(), } } } diff --git a/src/handler/grpc/flight/mod.rs b/src/handler/grpc/flight/mod.rs index 4ec667b205..63d3baf640 100644 --- a/src/handler/grpc/flight/mod.rs +++ b/src/handler/grpc/flight/mod.rs @@ -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, start: std::time::Instant, + timeout: u64, } impl FlightSenderStream { @@ -250,12 +256,14 @@ impl FlightSenderStream { stream: SendableRecordBatchStream, defer: Option, 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> { - 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, diff --git a/src/handler/http/request/alerts/deprecated.rs b/src/handler/http/request/alerts/deprecated.rs index 43f3cea657..2954e94b9f 100644 --- a/src/handler/http/request/alerts/deprecated.rs +++ b/src/handler/http/request/alerts/deprecated.rs @@ -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::>::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, 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, 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::>::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::>::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::>::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::().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::>::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(), diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 77c5a7676d..481598458d 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -193,10 +193,7 @@ pub async fn search( .to_string(); let query = web::Query::>::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::>::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::().unwrap_or(0), @@ -742,10 +736,7 @@ pub async fn values( let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::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::>::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, diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index 06dd61561f..f70f1bcadb 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -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::>::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::>::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::>::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::().unwrap_or(0), diff --git a/src/handler/http/request/search/search_job.rs b/src/handler/http/request/search/search_job.rs index 0541c777d3..a5ac36b26b 100644 --- a/src/handler/http/request/search/search_job.rs +++ b/src/handler/http/request/search/search_job.rs @@ -70,10 +70,7 @@ pub async fn submit_job( .to_string(); let query = web::Query::>::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 diff --git a/src/handler/http/request/stream/mod.rs b/src/handler/http/request/stream/mod.rs index 4e0a44fc8b..c7b644b7e0 100644 --- a/src/handler/http/request/stream/mod.rs +++ b/src/handler/http/request/stream/mod.rs @@ -61,18 +61,7 @@ async fn schema( ) -> Result { let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::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::>::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::>::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 { let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::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 { let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::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, req: HttpRequest) -> impl Responder { let query = web::Query::>::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::>::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 { diff --git a/src/infra/src/table/alerts/mod.rs b/src/infra/src/table/alerts/mod.rs index e01b0f4e75..1f3e0b4eef 100644 --- a/src/infra/src/table/alerts/mod.rs +++ b/src/infra/src/table/alerts/mod.rs @@ -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 diff --git a/src/ingester/src/partition.rs b/src/ingester/src/partition.rs index ba80de1081..ddedf820c2 100644 --- a/src/ingester/src/partition.rs +++ b/src/ingester/src/partition.rs @@ -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::>(); - 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::>(); + 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)) } diff --git a/src/service/search/cluster/flight.rs b/src/service/search/cluster/flight.rs index bd6ae69b19..02f476dff4 100644 --- a/src/service/search/cluster/flight.rs +++ b/src/service/search/cluster/flight.rs @@ -13,7 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -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(); diff --git a/src/service/search/datafusion/distributed_plan/codec.rs b/src/service/search/datafusion/distributed_plan/codec.rs index 12727d64ec..c97ad9200e 100644 --- a/src/service/search/datafusion/distributed_plan/codec.rs +++ b/src/service/search/datafusion/distributed_plan/codec.rs @@ -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 diff --git a/src/service/search/datafusion/distributed_plan/mod.rs b/src/service/search/datafusion/distributed_plan/mod.rs index a6804d12cb..67268e66a0 100644 --- a/src/service/search/datafusion/distributed_plan/mod.rs +++ b/src/service/search/datafusion/distributed_plan/mod.rs @@ -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 diff --git a/src/service/search/datafusion/distributed_plan/node.rs b/src/service/search/datafusion/distributed_plan/node.rs index c5ee261bf0..be5132e454 100644 --- a/src/service/search/datafusion/distributed_plan/node.rs +++ b/src/service/search/datafusion/distributed_plan/node.rs @@ -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 . + use std::sync::Arc; use config::{ diff --git a/src/service/search/datafusion/distributed_plan/remote_scan.rs b/src/service/search/datafusion/distributed_plan/remote_scan.rs index cdf866b18a..16bc9b56b8 100644 --- a/src/service/search/datafusion/distributed_plan/remote_scan.rs +++ b/src/service/search/datafusion/distributed_plan/remote_scan.rs @@ -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>, partial_err: Arc>, ) -> Result { + 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>, + 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::>(); + Box::pin(RecordBatchStreamAdapter::new(schema, stream)) +} + +fn process_partial_err(partial_err: Arc>, 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>, start: std::time::Instant, + timeout: u64, } impl FlightStream { @@ -330,6 +411,7 @@ impl FlightStream { scan_size: i64, partial_err: Arc>, 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> { - 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) } } diff --git a/src/service/search/datafusion/distributed_plan/rewrite.rs b/src/service/search/datafusion/distributed_plan/rewrite.rs index e9e3c8e4aa..5769e08c69 100644 --- a/src/service/search/datafusion/distributed_plan/rewrite.rs +++ b/src/service/search/datafusion/distributed_plan/rewrite.rs @@ -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 diff --git a/src/service/search/datafusion/distributed_plan/streaming_aggs_exec.rs b/src/service/search/datafusion/distributed_plan/streaming_aggs_exec.rs index 760ecb6e3d..4ccad320ca 100644 --- a/src/service/search/datafusion/distributed_plan/streaming_aggs_exec.rs +++ b/src/service/search/datafusion/distributed_plan/streaming_aggs_exec.rs @@ -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 diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 38255efb1e..e866311dbf 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -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)] diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index af73e8240e..e5cc71de3b 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -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 { } else { "Long" }; - let search_type: Option = result - .search_type - .map(|s_event_type| search::SearchEventType::from_str(&s_event_type).unwrap()); + let search_type: Option = 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, diff --git a/src/service/search/super_cluster/follower.rs b/src/service/search/super_cluster/follower.rs index f762452213..e751a939bd 100644 --- a/src/service/search/super_cluster/follower.rs +++ b/src/service/search/super_cluster/follower.rs @@ -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 . -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();