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

feat: add metrics for query lifetime (#3893)

Browse Source
This commit is contained in:
Huaijin
2024-07-04 14:13:25 +08:00
committed by GitHub
parent 6157ef5331
commit 33343bbab3
12 changed files with 471 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/metrics.rs
+52
View File
@@ -564,6 +564,44 @@ pub static MEMORY_USAGE: Lazy<IntGaugeVec> = Lazy::new(|| {
.expect("Metric created")
});
// metrics for query manager
pub static QUERY_RUNNING_NUMS: Lazy<IntGaugeVec> = 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<IntGaugeVec> = 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<IntCounterVec> = 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<IntCounterVec> = 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()))
src/handler/grpc/request/search/intra_cluster.rs
-285
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<QueryManager>,
}
#[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<proto::cluster_rpc::QueryStatus> {
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<SearchRequest>,
) -> Result<Response<SearchResponse>, 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<SearchRequest>,
) -> Result<Response<SearchResponse>, 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<QueryStatusRequest>,
) -> Result<Response<QueryStatusResponse>, Status> {
let status = self.get_task_status().await;
Ok(Response::new(QueryStatusResponse { status }))
}
#[cfg(not(feature = "enterprise"))]
async fn query_status(
&self,
_req: Request<QueryStatusRequest>,
) -> Result<Response<QueryStatusResponse>, Status> {
Err(Status::unimplemented("Not Supported"))
}
#[cfg(feature = "enterprise")]
async fn cancel_query(
&self,
req: Request<CancelQueryRequest>,
) -> Result<Response<CancelQueryResponse>, 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<CancelQueryRequest>,
) -> Result<Response<CancelQueryResponse>, Status> {
Err(Status::unimplemented("Not Supported"))
}
}
src/handler/grpc/request/search/mod.rs
+281 -1
View File
@@ -13,4 +13,284 @@
// 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/>.
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<QueryManager>,
}
#[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<proto::cluster_rpc::QueryStatus> {
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<SearchRequest>,
) -> Result<Response<SearchResponse>, 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<SearchRequest>,
) -> Result<Response<SearchResponse>, 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<QueryStatusRequest>,
) -> Result<Response<QueryStatusResponse>, Status> {
let status = self.get_task_status().await;
Ok(Response::new(QueryStatusResponse { status }))
}
#[cfg(not(feature = "enterprise"))]
async fn query_status(
&self,
_req: Request<QueryStatusRequest>,
) -> Result<Response<QueryStatusResponse>, Status> {
Err(Status::unimplemented("Not Supported"))
}
#[cfg(feature = "enterprise")]
async fn cancel_query(
&self,
req: Request<CancelQueryRequest>,
) -> Result<Response<CancelQueryResponse>, 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<CancelQueryRequest>,
) -> Result<Response<CancelQueryResponse>, Status> {
Err(Status::unimplemented("Not Supported"))
}
}
src/handler/http/request/search/mod.rs
+27
View File
@@ -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::<Vec<_>>()
});
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<String> = None;
@@ -1049,6 +1064,9 @@ async fn values_v1(
.get("timeout")
.map_or(0, |v| v.parse::<i64>().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::<i64>().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 {
src/handler/http/request/search/multi_streams.rs
+12
View File
@@ -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 {
src/handler/http/request/traces/mod.rs
+6
View File
@@ -201,6 +201,9 @@ pub async fn get_latest_traces(
.get("timeout")
.map_or(0, |v| v.parse::<i64>().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!(
src/infra/src/errors/mod.rs
+8 -1
View File
@@ -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())),
}
}
src/service/metrics/prom.rs
+20 -2
View File
@@ -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::<HashSet<_>>(),
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
src/service/search/cluster/mod.rs
+38 -1
View File
@@ -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<String> = 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"
src/service/search/grpc/storage.rs
+7 -4
View File
@@ -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()),
},
};
}
src/service/search/grpc/wal.rs
+14 -4
View File
@@ -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()),
}
}
};
}
src/service/search/mod.rs
+6 -1
View File
@@ -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) => {