From 33343bbab37d97cbdc35c2f2c8ca5e370630870f Mon Sep 17 00:00:00 2001 From: Huaijin Date: Thu, 4 Jul 2024 14:13:25 +0800 Subject: [PATCH] feat: add metrics for query lifetime (#3893) --- src/config/src/metrics.rs | 52 ++++ .../grpc/request/search/intra_cluster.rs | 285 ------------------ src/handler/grpc/request/search/mod.rs | 282 ++++++++++++++++- src/handler/http/request/search/mod.rs | 27 ++ .../http/request/search/multi_streams.rs | 12 + src/handler/http/request/traces/mod.rs | 6 + src/infra/src/errors/mod.rs | 9 +- src/service/metrics/prom.rs | 22 +- src/service/search/cluster/mod.rs | 39 ++- src/service/search/grpc/storage.rs | 11 +- src/service/search/grpc/wal.rs | 18 +- src/service/search/mod.rs | 7 +- 12 files changed, 471 insertions(+), 299 deletions(-) delete mode 100644 src/handler/grpc/request/search/intra_cluster.rs diff --git a/src/config/src/metrics.rs b/src/config/src/metrics.rs index 39acd3e604..5c4ded0c33 100644 --- a/src/config/src/metrics.rs +++ b/src/config/src/metrics.rs @@ -564,6 +564,44 @@ pub static MEMORY_USAGE: Lazy = Lazy::new(|| { .expect("Metric created") }); +// metrics for query manager +pub static QUERY_RUNNING_NUMS: Lazy = Lazy::new(|| { + IntGaugeVec::new( + Opts::new("query_running_nums", "Running query numbers") + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization"], + ) + .expect("Metric created") +}); +pub static QUERY_PENDING_NUMS: Lazy = Lazy::new(|| { + IntGaugeVec::new( + Opts::new("query_pending_nums", "Pending query numbers") + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization"], + ) + .expect("Metric created") +}); +pub static QUERY_TIMEOUT_NUMS: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new("query_timeout_nums", "Timeout query numbers") + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization"], + ) + .expect("Metric created") +}); +pub static QUERY_CANCELED_NUMS: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new("query_canceled_nums", "Cancel query numbers") + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization"], + ) + .expect("Metric created") +}); + fn register_metrics(registry: &Registry) { // http latency registry @@ -633,6 +671,20 @@ fn register_metrics(registry: &Registry) { .register(Box::new(QUERY_DISK_CACHE_FILES.clone())) .expect("Metric registered"); + // query manager + registry + .register(Box::new(QUERY_RUNNING_NUMS.clone())) + .expect("Metric registered"); + registry + .register(Box::new(QUERY_PENDING_NUMS.clone())) + .expect("Metric registered"); + registry + .register(Box::new(QUERY_TIMEOUT_NUMS.clone())) + .expect("Metric registered"); + registry + .register(Box::new(QUERY_CANCELED_NUMS.clone())) + .expect("Metric registered"); + // compactor stats registry .register(Box::new(COMPACT_USED_TIME.clone())) diff --git a/src/handler/grpc/request/search/intra_cluster.rs b/src/handler/grpc/request/search/intra_cluster.rs deleted file mode 100644 index eb1b611597..0000000000 --- a/src/handler/grpc/request/search/intra_cluster.rs +++ /dev/null @@ -1,285 +0,0 @@ -// Copyright 2024 Zinc Labs 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 config::metrics; -use infra::errors; -#[cfg(feature = "enterprise")] -use o2_enterprise::enterprise::{ - common::infra::config::O2_CONFIG, - search::{QueryManager, TaskStatus}, -}; -use proto::cluster_rpc::{ - search_server::Search, CancelQueryRequest, CancelQueryResponse, QueryStatusRequest, - QueryStatusResponse, SearchRequest, SearchResponse, -}; -use tonic::{Request, Response, Status}; -use tracing_opentelemetry::OpenTelemetrySpanExt; - -use crate::service::search as SearchService; - -#[derive(Clone, Debug)] -#[cfg(feature = "enterprise")] -pub struct Searcher { - pub query_manager: std::sync::Arc, -} - -#[cfg(feature = "enterprise")] -impl Searcher { - pub fn new() -> Self { - Self { - query_manager: std::sync::Arc::new(QueryManager::new()), - } - } - - // check is the trace_id in the query_manager - pub async fn contain_key(&self, trace_id: &str) -> bool { - self.query_manager.contain_key(trace_id).await - } - - // insert the trace_id and task_status into the query_manager - pub async fn insert(&self, trace_id: String, task_status: TaskStatus) { - self.query_manager.insert(trace_id, task_status).await; - } - - // remove the trace_id from the query_manager - pub async fn remove(&self, trace_id: &str) -> Option<(String, TaskStatus)> { - self.query_manager.remove(trace_id).await - } - - // check is the trace_id in the query_manager and is_leader - pub async fn is_leader(&self, trace_id: &str) -> bool { - self.query_manager.is_leader(trace_id).await - } - - // insert the sender into the query_manager by trace_id - pub async fn insert_sender( - &self, - trace_id: &str, - sender: tokio::sync::oneshot::Sender<()>, - ) -> Result<(), infra::errors::Error> { - self.query_manager.insert_sender(trace_id, sender).await - } - - // get all task status that is leader - pub async fn get_task_status(&self) -> Vec { - self.query_manager.get_task_status().await - } - - // add file stats - pub async fn add_file_stats( - &self, - trace_id: &str, - files: i64, - records: i64, - original_size: i64, - compressed_size: i64, - ) { - self.query_manager - .add_file_stats(trace_id, files, records, original_size, compressed_size) - .await; - } -} - -#[derive(Clone, Debug)] -#[cfg(not(feature = "enterprise"))] -pub struct Searcher; - -#[cfg(not(feature = "enterprise"))] -impl Searcher { - pub fn new() -> Self { - Self - } -} - -impl Default for Searcher { - fn default() -> Self { - Self::new() - } -} - -#[tonic::async_trait] -impl Search for Searcher { - #[tracing::instrument(name = "grpc:search:enter", skip_all, fields(trace_id=req.get_ref().job.as_ref().unwrap().trace_id, org_id = req.get_ref().org_id))] - async fn search( - &self, - req: Request, - ) -> Result, Status> { - let start = std::time::Instant::now(); - let parent_cx = opentelemetry::global::get_text_map_propagator(|prop| { - prop.extract(&crate::handler::grpc::request::MetadataMap(req.metadata())) - }); - tracing::Span::current().set_parent(parent_cx); - - let req = req.get_ref().to_owned(); - let org_id = req.org_id.clone(); - let stream_type = req.stream_type.clone(); - - // set search task - #[cfg(feature = "enterprise")] - let trace_id = req.job.as_ref().unwrap().trace_id.to_string(); - #[cfg(feature = "enterprise")] - if !self.contain_key(&trace_id).await { - self.insert( - trace_id.clone(), - TaskStatus::new(vec![], false, None, None, None, None, None, None), - ) - .await; - } - - let result = SearchService::grpc::search(&req).await; - - // remove task - #[cfg(feature = "enterprise")] - if !O2_CONFIG.super_cluster.enabled && !self.is_leader(&trace_id).await { - self.remove(&trace_id).await; - } - - match result { - Ok(res) => { - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) - .inc(); - - Ok(Response::new(res)) - } - Err(err) => { - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) - .inc(); - let message = if let errors::Error::ErrorCode(code) = err { - code.to_json() - } else { - err.to_string() - }; - Err(Status::internal(message)) - } - } - } - - #[tracing::instrument(name = "grpc:cluster_search:enter", skip_all, fields(trace_id=req.get_ref().job.as_ref().unwrap().trace_id, org_id = req.get_ref().org_id))] - async fn cluster_search( - &self, - req: Request, - ) -> Result, Status> { - let start = std::time::Instant::now(); - let parent_cx = opentelemetry::global::get_text_map_propagator(|prop| { - prop.extract(&crate::handler::grpc::request::MetadataMap(req.metadata())) - }); - tracing::Span::current().set_parent(parent_cx); - - let req = req.get_ref().to_owned(); - let org_id = req.org_id.clone(); - let stream_type = req.stream_type.clone(); - - // set search task - #[cfg(feature = "enterprise")] - let trace_id = req.job.as_ref().unwrap().trace_id.to_string(); - #[cfg(feature = "enterprise")] - if !self.contain_key(&trace_id).await { - self.insert( - trace_id.clone(), - TaskStatus::new(vec![], false, None, None, None, None, None, None), - ) - .await; - } - - let result = SearchService::cluster::grpc::search(req).await; - - // remove task - #[cfg(feature = "enterprise")] - if !self.is_leader(&trace_id).await { - self.remove(&trace_id).await; - } - - match result { - Ok(res) => { - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) - .inc(); - - Ok(Response::new(res)) - } - Err(err) => { - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) - .inc(); - let message = if let errors::Error::ErrorCode(code) = err { - code.to_json() - } else { - err.to_string() - }; - Err(Status::internal(message)) - } - } - } - - #[cfg(feature = "enterprise")] - async fn query_status( - &self, - _req: Request, - ) -> Result, Status> { - let status = self.get_task_status().await; - Ok(Response::new(QueryStatusResponse { status })) - } - - #[cfg(not(feature = "enterprise"))] - async fn query_status( - &self, - _req: Request, - ) -> Result, Status> { - Err(Status::unimplemented("Not Supported")) - } - - #[cfg(feature = "enterprise")] - async fn cancel_query( - &self, - req: Request, - ) -> Result, Status> { - let trace_id = req.into_inner().trace_id; - match self.remove(&trace_id).await { - Some((_, senders)) => { - for sender in senders.abort_senders.into_iter().rev() { - let _ = sender.send(()); - } - Ok(Response::new(CancelQueryResponse { is_success: true })) - } - None => Ok(Response::new(CancelQueryResponse { is_success: false })), - } - } - - #[cfg(not(feature = "enterprise"))] - async fn cancel_query( - &self, - _req: Request, - ) -> Result, Status> { - Err(Status::unimplemented("Not Supported")) - } -} diff --git a/src/handler/grpc/request/search/mod.rs b/src/handler/grpc/request/search/mod.rs index ab4ba4ac52..99311751de 100644 --- a/src/handler/grpc/request/search/mod.rs +++ b/src/handler/grpc/request/search/mod.rs @@ -13,4 +13,284 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -pub mod intra_cluster; +use config::metrics; +use infra::errors::{Error::ErrorCode, ErrorCodes}; +#[cfg(feature = "enterprise")] +use o2_enterprise::enterprise::{ + common::infra::config::O2_CONFIG, + search::{QueryManager, TaskStatus}, +}; +use proto::cluster_rpc::{ + search_server::Search, CancelQueryRequest, CancelQueryResponse, QueryStatusRequest, + QueryStatusResponse, SearchRequest, SearchResponse, +}; +use tonic::{Request, Response, Status}; +use tracing_opentelemetry::OpenTelemetrySpanExt; + +use crate::service::search as SearchService; + +#[derive(Clone, Debug)] +#[cfg(feature = "enterprise")] +pub struct Searcher { + pub query_manager: std::sync::Arc, +} + +#[cfg(feature = "enterprise")] +impl Searcher { + pub fn new() -> Self { + Self { + query_manager: std::sync::Arc::new(QueryManager::new()), + } + } + + // check is the trace_id in the query_manager + pub async fn contain_key(&self, trace_id: &str) -> bool { + self.query_manager.contain_key(trace_id).await + } + + // insert the trace_id and task_status into the query_manager + pub async fn insert(&self, trace_id: String, task_status: TaskStatus) { + self.query_manager.insert(trace_id, task_status).await; + } + + // remove the trace_id from the query_manager + pub async fn remove(&self, trace_id: &str) -> Option<(String, TaskStatus)> { + self.query_manager.remove(trace_id).await + } + + // check is the trace_id in the query_manager and is_leader + pub async fn is_leader(&self, trace_id: &str) -> bool { + self.query_manager.is_leader(trace_id).await + } + + // insert the sender into the query_manager by trace_id + pub async fn insert_sender( + &self, + trace_id: &str, + sender: tokio::sync::oneshot::Sender<()>, + ) -> Result<(), infra::errors::Error> { + self.query_manager.insert_sender(trace_id, sender).await + } + + // get all task status that is leader + pub async fn get_task_status(&self) -> Vec { + self.query_manager.get_task_status().await + } + + // add file stats + pub async fn add_file_stats( + &self, + trace_id: &str, + files: i64, + records: i64, + original_size: i64, + compressed_size: i64, + ) { + self.query_manager + .add_file_stats(trace_id, files, records, original_size, compressed_size) + .await; + } +} + +#[derive(Clone, Debug)] +#[cfg(not(feature = "enterprise"))] +pub struct Searcher; + +#[cfg(not(feature = "enterprise"))] +impl Searcher { + pub fn new() -> Self { + Self + } +} + +impl Default for Searcher { + fn default() -> Self { + Self::new() + } +} + +#[tonic::async_trait] +impl Search for Searcher { + #[tracing::instrument(name = "grpc:search:enter", skip_all, fields(trace_id=req.get_ref().job.as_ref().unwrap().trace_id, org_id = req.get_ref().org_id))] + async fn search( + &self, + req: Request, + ) -> Result, Status> { + let start = std::time::Instant::now(); + let parent_cx = opentelemetry::global::get_text_map_propagator(|prop| { + prop.extract(&crate::handler::grpc::request::MetadataMap(req.metadata())) + }); + tracing::Span::current().set_parent(parent_cx); + + let req = req.get_ref().to_owned(); + let org_id = req.org_id.clone(); + let stream_type = req.stream_type.clone(); + + // set search task + #[cfg(feature = "enterprise")] + let trace_id = req.job.as_ref().unwrap().trace_id.to_string(); + #[cfg(feature = "enterprise")] + if !self.contain_key(&trace_id).await { + self.insert( + trace_id.clone(), + TaskStatus::new(vec![], false, None, None, None, None, None, None), + ) + .await; + } + + let result = SearchService::grpc::search(&req).await; + + // remove task + #[cfg(feature = "enterprise")] + if !O2_CONFIG.super_cluster.enabled && !self.is_leader(&trace_id).await { + self.remove(&trace_id).await; + } + + match result { + Ok(res) => { + let time = start.elapsed().as_secs_f64(); + metrics::GRPC_RESPONSE_TIME + .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) + .observe(time); + metrics::GRPC_INCOMING_REQUESTS + .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) + .inc(); + + Ok(Response::new(res)) + } + Err(err) => { + let time = start.elapsed().as_secs_f64(); + metrics::GRPC_RESPONSE_TIME + .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) + .observe(time); + metrics::GRPC_INCOMING_REQUESTS + .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) + .inc(); + let message = if let ErrorCode(ref code) = err { + code.to_json() + } else { + err.to_string() + }; + if let ErrorCode(ErrorCodes::SearchTimeout(_)) = err { + Err(Status::deadline_exceeded(message)) + } else { + Err(Status::internal(message)) + } + } + } + } + + #[tracing::instrument(name = "grpc:cluster_search:enter", skip_all, fields(trace_id=req.get_ref().job.as_ref().unwrap().trace_id, org_id = req.get_ref().org_id))] + async fn cluster_search( + &self, + req: Request, + ) -> Result, Status> { + let start = std::time::Instant::now(); + let parent_cx = opentelemetry::global::get_text_map_propagator(|prop| { + prop.extract(&crate::handler::grpc::request::MetadataMap(req.metadata())) + }); + tracing::Span::current().set_parent(parent_cx); + + let req = req.get_ref().to_owned(); + let org_id = req.org_id.clone(); + let stream_type = req.stream_type.clone(); + + // set search task + #[cfg(feature = "enterprise")] + let trace_id = req.job.as_ref().unwrap().trace_id.to_string(); + #[cfg(feature = "enterprise")] + if !self.contain_key(&trace_id).await { + self.insert( + trace_id.clone(), + TaskStatus::new(vec![], false, None, None, None, None, None, None), + ) + .await; + } + + let result = SearchService::cluster::grpc::search(req).await; + + // remove task + #[cfg(feature = "enterprise")] + if !self.is_leader(&trace_id).await { + self.remove(&trace_id).await; + } + + metrics::QUERY_RUNNING_NUMS + .with_label_values(&[&org_id]) + .dec(); + + match result { + Ok(res) => { + let time = start.elapsed().as_secs_f64(); + metrics::GRPC_RESPONSE_TIME + .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) + .observe(time); + metrics::GRPC_INCOMING_REQUESTS + .with_label_values(&["/_search", "200", &org_id, "", &stream_type]) + .inc(); + + Ok(Response::new(res)) + } + Err(err) => { + let time = start.elapsed().as_secs_f64(); + metrics::GRPC_RESPONSE_TIME + .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) + .observe(time); + metrics::GRPC_INCOMING_REQUESTS + .with_label_values(&["/_search", "500", &org_id, "", &stream_type]) + .inc(); + let message = if let ErrorCode(code) = err { + code.to_json() + } else { + err.to_string() + }; + Err(Status::internal(message)) + } + } + } + + #[cfg(feature = "enterprise")] + async fn query_status( + &self, + _req: Request, + ) -> Result, Status> { + let status = self.get_task_status().await; + Ok(Response::new(QueryStatusResponse { status })) + } + + #[cfg(not(feature = "enterprise"))] + async fn query_status( + &self, + _req: Request, + ) -> Result, Status> { + Err(Status::unimplemented("Not Supported")) + } + + #[cfg(feature = "enterprise")] + async fn cancel_query( + &self, + req: Request, + ) -> Result, Status> { + let trace_id = req.into_inner().trace_id; + match self.remove(&trace_id).await { + Some((_, senders)) => { + for sender in senders.abort_senders.into_iter().rev() { + let _ = sender.send(()); + } + metrics::QUERY_CANCELED_NUMS + .with_label_values(&[&senders.org_id.unwrap()]) + .inc(); + Ok(Response::new(CancelQueryResponse { is_success: true })) + } + None => Ok(Response::new(CancelQueryResponse { is_success: false })), + } + } + + #[cfg(not(feature = "enterprise"))] + async fn cancel_query( + &self, + _req: Request, + ) -> Result, Status> { + Err(Status::unimplemented("Not Supported")) + } +} diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 94a3be9cd3..2e912ff3d6 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -297,6 +297,10 @@ pub async fn search( } } + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .inc(); + let cfg = get_config(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] @@ -313,6 +317,11 @@ pub async fn search( let took_wait = 0; ext_took_wait = took_wait; log::info!("http search API wait in queue took: {} ms", took_wait); + + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .dec(); + let search_tracing = !cfg.common.tracing_enabled && cfg.common.tracing_search_enabled; let mut tasks = Vec::new(); @@ -598,6 +607,9 @@ pub async fn around( .collect::>() }); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .inc(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); @@ -615,6 +627,9 @@ pub async fn around( "http search around API wait in queue took: {} ms", took_wait ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .dec(); let query_context: Option = None; @@ -1049,6 +1064,9 @@ async fn values_v1( .get("timeout") .map_or(0, |v| v.parse::().unwrap_or(0)); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .inc(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); @@ -1066,6 +1084,9 @@ async fn values_v1( "http search value_v1 API wait in queue took: {} ms", took_wait ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .dec(); // search let mut req = config::meta::search::Request { @@ -1273,6 +1294,9 @@ async fn values_v2( .get("timeout") .map_or(0, |v| v.parse::().unwrap_or(0)); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .inc(); let cfg = get_config(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] @@ -1291,6 +1315,9 @@ async fn values_v2( "http search value_v2 API wait in queue took: {} ms", took_wait ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .dec(); // search let req = config::meta::search::Request { diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index d84014519f..074faa8890 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -224,6 +224,9 @@ pub async fn search_multi( } } + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .inc(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); @@ -238,6 +241,9 @@ pub async fn search_multi( #[cfg(feature = "enterprise")] let took_wait = 0; log::info!("http search multi API wait in queue took: {}", took_wait); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .dec(); let trace_id = trace_id.clone(); // do search @@ -642,6 +648,9 @@ pub async fn around_multi( .map(|v| v.to_string()); for around_sql in around_sqls.iter() { + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .inc(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); @@ -659,6 +668,9 @@ pub async fn around_multi( "http search around multi API wait in queue took: {}", took_wait ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .dec(); // search forward let req = config::meta::search::Request { diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 02f566e76a..71a2e56b63 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -201,6 +201,9 @@ pub async fn get_latest_traces( .get("timeout") .map_or(0, |v| v.parse::().unwrap_or(0)); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .inc(); let cfg = get_config(); // get a local search queue lock #[cfg(not(feature = "enterprise"))] @@ -219,6 +222,9 @@ pub async fn get_latest_traces( "http traces latest API wait in queue took: {} ms", took_wait ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&org_id]) + .dec(); // search let query_sql = format!( diff --git a/src/infra/src/errors/mod.rs b/src/infra/src/errors/mod.rs index 538e8bc1bb..027250c3e1 100644 --- a/src/infra/src/errors/mod.rs +++ b/src/infra/src/errors/mod.rs @@ -98,6 +98,7 @@ pub enum ErrorCodes { SearchFieldHasNoCompatibleDataType(String), SearchSQLExecuteError(String), SearchCancelQuery(String), + SearchTimeout(String), } impl std::fmt::Display for ErrorCodes { @@ -123,7 +124,8 @@ impl ErrorCodes { ErrorCodes::SearchParquetFileNotFound => 20006, ErrorCodes::SearchFieldHasNoCompatibleDataType(_) => 20007, ErrorCodes::SearchSQLExecuteError(_) => 20008, - ErrorCodes::SearchCancelQuery(_) => 429, + ErrorCodes::SearchCancelQuery(_) => 20009, + ErrorCodes::SearchTimeout(_) => 20010, } } @@ -149,6 +151,7 @@ impl ErrorCodes { ErrorCodes::SearchCancelQuery(_) => { "Search query was cancelled by the administrator".to_string() } + ErrorCodes::SearchTimeout(_) => "Search query timed out".to_string(), } } @@ -164,6 +167,7 @@ impl ErrorCodes { ErrorCodes::SearchFieldHasNoCompatibleDataType(field) => field.to_owned(), ErrorCodes::SearchSQLExecuteError(msg) => msg.to_owned(), ErrorCodes::SearchCancelQuery(msg) => msg.to_owned(), + ErrorCodes::SearchTimeout(msg) => msg.to_owned(), } } @@ -179,6 +183,7 @@ impl ErrorCodes { ErrorCodes::SearchFieldHasNoCompatibleDataType(_) => "".to_string(), ErrorCodes::SearchSQLExecuteError(msg) => msg.to_owned(), ErrorCodes::SearchCancelQuery(msg) => msg.to_string(), + ErrorCodes::SearchTimeout(msg) => msg.to_owned(), } } @@ -226,6 +231,8 @@ impl ErrorCodes { 20006 => Ok(ErrorCodes::SearchParquetFileNotFound), 20007 => Ok(ErrorCodes::SearchFieldHasNoCompatibleDataType(message)), 20008 => Ok(ErrorCodes::SearchSQLExecuteError(message)), + 20009 => Ok(ErrorCodes::SearchCancelQuery(message)), + 20010 => Ok(ErrorCodes::SearchTimeout(message)), _ => Ok(ErrorCodes::ServerInternalError(json.to_string())), } } diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index 377f781897..a53a5f8388 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -28,6 +28,7 @@ use config::{ FxIndexMap, }; use datafusion::arrow::datatypes::Schema; +use hashbrown::HashSet; use infra::{ cache::stats, errors::{Error, Result}, @@ -331,7 +332,24 @@ pub async fn remote_write( let value_str = config::utils::json::to_string(&val_map).unwrap(); // check for schema evolution - if !schema_evolved.contains_key(&metric_name) + let schema_fields = match metric_schema_map.get(&metric_name) { + Some(schema) => schema + .schema() + .fields() + .iter() + .map(|f| f.name()) + .collect::>(), + None => HashSet::default(), + }; + let mut need_schema_check = !schema_evolved.contains_key(&metric_name); + for key in val_map.keys() { + if !schema_fields.contains(&key) { + need_schema_check = true; + break; + } + } + drop(schema_fields); + if need_schema_check && check_for_schema( org_id, &metric_name, @@ -343,7 +361,7 @@ pub async fn remote_write( .await .is_ok() { - schema_evolved.insert(metric_name.to_owned(), true); + schema_evolved.insert(metric_name.clone(), true); } let schema = metric_schema_map diff --git a/src/service/search/cluster/mod.rs b/src/service/search/cluster/mod.rs index 153ed611be..b691f292cc 100644 --- a/src/service/search/cluster/mod.rs +++ b/src/service/search/cluster/mod.rs @@ -27,6 +27,7 @@ use config::{ FileKey, PartitionTimeLevel, QueryPartitionStrategy, StreamPartition, StreamType, }, }, + metrics, utils::{inverted_index::split_token, json}, }; use hashbrown::{HashMap, HashSet}; @@ -266,6 +267,10 @@ pub async fn search( file_list_took, ); + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .inc(); + #[cfg(not(feature = "enterprise"))] let work_group: Option = None; // 1. get work group @@ -285,7 +290,14 @@ pub async fn search( let locker = if cfg.common.local_mode || !cfg.common.feature_query_queue_enabled { None } else { - dist_lock::lock(&locker_key, req.timeout as u64).await? + dist_lock::lock(&locker_key, req.timeout as u64) + .await + .map_err(|e| { + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .dec(); + Error::Message(e.to_string()) + })? }; // check global concurrency @@ -306,11 +318,17 @@ pub async fn search( .process(trace_id, user_id) .await { + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .dec(); dist_lock::unlock(&locker).await?; return Err(Error::Message(e.to_string())); } #[cfg(feature = "enterprise")] if let Err(e) = dist_lock::unlock(&locker).await { + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .dec(); work_group .as_ref() .unwrap() @@ -385,6 +403,13 @@ pub async fn search( .await; } + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .dec(); + metrics::QUERY_RUNNING_NUMS + .with_label_values(&[&req.org_id]) + .inc(); + // make cluster request let mut tasks = Vec::new(); let mut offset_start: usize = 0; @@ -470,6 +495,7 @@ pub async fn search( .org_id .parse() .map_err(|_| Error::Message("invalid org_id".to_string()))?; + let org_id_string = req.org_id.clone(); let mut request = tonic::Request::new(req); // request.set_timeout(Duration::from_secs(cfg.grpc.timeout)); @@ -520,6 +546,11 @@ pub async fn search( Ok(res) => response = res.into_inner(), Err(err) => { log::error!("[trace_id {trace_id}] search->grpc: node: {}, search err: {:?}", &node.grpc_addr, err); + if err.code() == tonic::Code::DeadlineExceeded { + metrics::QUERY_TIMEOUT_NUMS + .with_label_values(&[&org_id_string]) + .inc(); + } if err.code() == tonic::Code::Internal { let err = ErrorCodes::from_json(err.message())?; return Err(Error::ErrorCode(err)); @@ -703,6 +734,12 @@ async fn work_group_need_wait( ) -> Result<()> { loop { if start.elapsed().as_millis() as u64 >= req.timeout as u64 * 1000 { + metrics::QUERY_PENDING_NUMS + .with_label_values(&[&req.org_id]) + .dec(); + metrics::QUERY_TIMEOUT_NUMS + .with_label_values(&[&req.org_id]) + .inc(); dist_lock::unlock(locker).await?; return Err(Error::Message(format!( "[trace_id {trace_id}] search: request timeout in queue" diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 2f745473bb..c1229dae6c 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -303,7 +303,7 @@ pub async fn search( }, _ = tokio::time::sleep(Duration::from_secs(timeout)) => { log::error!("[trace_id {}] search->storage: search timeout", session.id); - Err(datafusion::error::DataFusionError::Execution(format!( + Err(datafusion::error::DataFusionError::ResourcesExhausted(format!( "[trace_id {}] search->storage: task timeout", session.id ))) }, @@ -344,9 +344,12 @@ pub async fn search( } } } - Err(err) => { - return Err(err.into()); - } + Err(err) => match err { + datafusion::error::DataFusionError::ResourcesExhausted(e) => { + return Err(Error::ErrorCode(ErrorCodes::SearchTimeout(e))); + } + _ => return Err(err.into()), + }, }; } diff --git a/src/service/search/grpc/wal.rs b/src/service/search/grpc/wal.rs index 0b7ab9459f..00c7c3c2b2 100644 --- a/src/service/search/grpc/wal.rs +++ b/src/service/search/grpc/wal.rs @@ -318,7 +318,7 @@ pub async fn search_parquet( ) => ret, _ = tokio::time::sleep(Duration::from_secs(timeout)) => { log::error!("[trace_id {}] wal->parquet->search: search timeout", session.id); - Err(datafusion::error::DataFusionError::Execution(format!( + Err(datafusion::error::DataFusionError::ResourcesExhausted(format!( "[trace_id {}] wal->parquet->search: task timeout", session.id ))) }, @@ -373,7 +373,12 @@ pub async fn search_parquet( "[trace_id {trace_id}] wal->parquet->search: datafusion execute error: {}", err ); - return Err(err.into()); + match err { + datafusion::error::DataFusionError::ResourcesExhausted(e) => { + return Err(Error::ErrorCode(ErrorCodes::SearchTimeout(e))); + } + _ => return Err(err.into()), + } } }; } @@ -535,7 +540,7 @@ pub async fn search_memtable( ) => ret, _ = tokio::time::sleep(Duration::from_secs(timeout)) => { log::error!("[trace_id {}] wal->mem->search: search timeout", session.id); - Err(datafusion::error::DataFusionError::Execution(format!( + Err(datafusion::error::DataFusionError::ResourcesExhausted(format!( "[trace_id {}] wal->mem->search: task timeout", session.id ))) }, @@ -581,7 +586,12 @@ pub async fn search_memtable( "[trace_id {trace_id}] wal->mem->search: datafusion execute error: {}", err ); - return Err(err.into()); + match err { + datafusion::error::DataFusionError::ResourcesExhausted(e) => { + return Err(Error::ErrorCode(ErrorCodes::SearchTimeout(e))); + } + _ => return Err(err.into()), + } } }; } diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index 375fb6d472..66aeb22b8a 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -23,6 +23,7 @@ use config::{ stream::{FileKey, StreamType}, usage::{RequestStats, UsageType}, }, + metrics, utils::str::find, }; use infra::{ @@ -47,7 +48,7 @@ use {std::sync::Arc, tokio::sync::Mutex}; use super::usage::report_request_usage_stats; use crate::{ common::{infra::cluster as infra_cluster, meta::stream::StreamParams}, - handler::grpc::request::search::intra_cluster::Searcher, + handler::grpc::request::search::Searcher, service::format_partition_key, }; @@ -146,6 +147,10 @@ pub async fn search( #[cfg(feature = "enterprise")] SEARCH_SERVER.remove(&trace_id).await; + metrics::QUERY_RUNNING_NUMS + .with_label_values(&[org_id]) + .dec(); + // do this because of clippy warning match res { Ok(res) => {