From 8a73f2cf05e71ca55a4a4f672eb169cfa64fefc6 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Sun, 28 Apr 2024 22:30:07 +0800 Subject: [PATCH] fix: value API is always waiting (#3364) fixed #3334 --- Cargo.lock | 4 +- Cargo.toml | 2 +- src/common/infra/mod.rs | 6 --- .../http/request/alerts/destinations.rs | 3 +- src/handler/http/request/alerts/mod.rs | 4 +- src/handler/http/request/alerts/templates.rs | 3 +- src/handler/http/request/search/mod.rs | 48 +++++++++++++++---- src/handler/http/request/traces/mod.rs | 13 ++++- src/service/alerts/destinations.rs | 1 + src/service/alerts/mod.rs | 3 +- src/service/alerts/templates.rs | 1 + src/service/search/cluster/mod.rs | 6 +++ src/service/search/mod.rs | 9 ++-- 13 files changed, 73 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 9652ee5c5f..a20d9f9f04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5081,7 +5081,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openobserve" -version = "0.10.0" +version = "0.10.4" dependencies = [ "actix-cors", "actix-multipart", @@ -9009,4 +9009,4 @@ checksum = "c253a4914af5bafc8fa8c86ee400827e83cf6ec01195ec1f1ed8441bf00d65aa" dependencies = [ "cc", "pkg-config", -] \ No newline at end of file +] diff --git a/Cargo.toml b/Cargo.toml index ce5750ef1a..95d1066f37 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ keywords = [ license = "AGPL-3.0-only" name = "openobserve" repository = "https://github.com/openobserve/openobserve/" -version = "0.10.0" +version = "0.10.4" publish = false # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/src/common/infra/mod.rs b/src/common/infra/mod.rs index 2274950988..d0c05f6a3f 100644 --- a/src/common/infra/mod.rs +++ b/src/common/infra/mod.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 . -#[cfg(feature = "enterprise")] -use ::config::CONFIG; use ::config::{ider, INSTANCE_ID}; use crate::service::db::instance; @@ -40,9 +38,5 @@ pub async fn init() -> Result<(), anyhow::Error> { // because of asynchronous, we need to wait for a while tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - // check incomplete work group - #[cfg(feature = "enterprise")] - o2_enterprise::enterprise::search::queue::clean(CONFIG.limit.query_timeout as i64).await?; - Ok(()) } diff --git a/src/handler/http/request/alerts/destinations.rs b/src/handler/http/request/alerts/destinations.rs index b2f8eaa393..e1ed0a399b 100644 --- a/src/handler/http/request/alerts/destinations.rs +++ b/src/handler/http/request/alerts/destinations.rs @@ -80,8 +80,7 @@ pub async fn update_destination( ) -> Result { let (org_id, name) = path.into_inner(); let dest = dest.into_inner(); - let name = name.trim(); - match destinations::save(&org_id, name, dest, false).await { + match destinations::save(&org_id, &name, dest, false).await { Ok(_) => Ok(MetaHttpResponse::ok("Alert destination saved")), Err(e) => match e { (http::StatusCode::BAD_REQUEST, e) => Ok(MetaHttpResponse::bad_request(e)), diff --git a/src/handler/http/request/alerts/mod.rs b/src/handler/http/request/alerts/mod.rs index 5fad55c81d..603dcf532f 100644 --- a/src/handler/http/request/alerts/mod.rs +++ b/src/handler/http/request/alerts/mod.rs @@ -92,9 +92,7 @@ pub async fn update_alert( // Hack for frequency: convert minutes to seconds let mut alert = alert.into_inner(); alert.trigger_condition.frequency *= 60; - - let name = name.trim(); - match alerts::save(&org_id, &stream_name, name, alert, false).await { + match alerts::save(&org_id, &stream_name, &name, alert, false).await { Ok(_) => Ok(MetaHttpResponse::ok("Alert Updated")), Err(e) => Ok(MetaHttpResponse::bad_request(e)), } diff --git a/src/handler/http/request/alerts/templates.rs b/src/handler/http/request/alerts/templates.rs index af09b67c06..ecb27bfa6d 100644 --- a/src/handler/http/request/alerts/templates.rs +++ b/src/handler/http/request/alerts/templates.rs @@ -77,8 +77,7 @@ pub async fn update_template( ) -> Result { let (org_id, name) = path.into_inner(); let tmpl = tmpl.into_inner(); - let name = name.trim(); - match templates::save(&org_id, name, tmpl, false).await { + match templates::save(&org_id, &name, tmpl, false).await { Ok(_) => Ok(MetaHttpResponse::ok("Alert template updated")), Err(e) => Ok(MetaHttpResponse::bad_request(e)), } diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 3e46ac7476..e8d4fc2b09 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -230,6 +230,7 @@ pub async fn search( let took_wait = start.elapsed().as_millis() as usize; #[cfg(feature = "enterprise")] let took_wait = 0; + log::info!("http search API wait in queue took: {}", took_wait); let search_fut = SearchService::search( &trace_id, @@ -439,15 +440,20 @@ pub async fn around( .map_or(10, |v| v.parse::().unwrap_or(10)); // get a local search queue lock + #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); - let _locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + let locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + if !CONFIG.common.feature_query_queue_enabled { + drop(locker); + } + #[cfg(not(feature = "enterprise"))] + let took_wait = start.elapsed().as_millis() as usize; + #[cfg(feature = "enterprise")] + let took_wait = 0; + log::info!("http search around API wait in queue took: {}", took_wait); - // We don't need query_context now - // let query_context = if uses_fn { - // Some(around_sql.clone()) - // } else { - // None - // }; let query_context: Option = None; let timeout = query @@ -900,8 +906,19 @@ async fn values_v1( .map_or(0, |v| v.parse::().unwrap_or(0)); // get a local search queue lock + #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); - let _locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + let locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + if !CONFIG.common.feature_query_queue_enabled { + drop(locker); + } + #[cfg(not(feature = "enterprise"))] + let took_wait = start.elapsed().as_millis() as usize; + #[cfg(feature = "enterprise")] + let took_wait = 0; + log::info!("http search value_v1 API wait in queue took: {}", took_wait); // search let mut req = config::meta::search::Request { @@ -1116,6 +1133,21 @@ async fn values_v2( .get("timeout") .map_or(0, |v| v.parse::().unwrap_or(0)); + // get a local search queue lock + #[cfg(not(feature = "enterprise"))] + let locker = SearchService::QUEUE_LOCKER.clone(); + #[cfg(not(feature = "enterprise"))] + let locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + if !CONFIG.common.feature_query_queue_enabled { + drop(locker); + } + #[cfg(not(feature = "enterprise"))] + let took_wait = start.elapsed().as_millis() as usize; + #[cfg(feature = "enterprise")] + let took_wait = 0; + log::info!("http search value_v2 API wait in queue took: {}", took_wait); + // search let req = config::meta::search::Request { query: config::meta::search::Query { diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 3d82c8ca9d..e93fc03e55 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -222,8 +222,19 @@ pub async fn get_latest_traces( .map_or(0, |v| v.parse::().unwrap_or(0)); // get a local search queue lock + #[cfg(not(feature = "enterprise"))] let locker = SearchService::QUEUE_LOCKER.clone(); - let _locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + let locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + if !CONFIG.common.feature_query_queue_enabled { + drop(locker); + } + #[cfg(not(feature = "enterprise"))] + let took_wait = start.elapsed().as_millis() as usize; + #[cfg(feature = "enterprise")] + let took_wait = 0; + log::info!("http traces latest API wait in queue took: {}", took_wait); // search let query_sql = format!( diff --git a/src/service/alerts/destinations.rs b/src/service/alerts/destinations.rs index a365a398f7..3841b4924b 100644 --- a/src/service/alerts/destinations.rs +++ b/src/service/alerts/destinations.rs @@ -63,6 +63,7 @@ pub async fn save( if !name.is_empty() { destination.name = name.to_string(); } + destination.name = destination.name.trim().to_string(); if destination.name.is_empty() { return Err(( http::StatusCode::BAD_REQUEST, diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index adf1aca842..7e55935dd4 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -61,8 +61,9 @@ pub async fn save( create: bool, ) -> Result<(), anyhow::Error> { if !name.is_empty() { - alert.name = name.trim().to_string(); + alert.name = name.to_string(); } + alert.name = alert.name.trim().to_string(); alert.org_id = org_id.to_string(); let stream_type = alert.stream_type; alert.stream_name = stream_name.to_string(); diff --git a/src/service/alerts/templates.rs b/src/service/alerts/templates.rs index 0fc1b70ba9..dea1e6db28 100644 --- a/src/service/alerts/templates.rs +++ b/src/service/alerts/templates.rs @@ -36,6 +36,7 @@ pub async fn save( if !name.is_empty() { template.name = name.to_owned(); } + template.name = template.name.trim().to_string(); if template.name.is_empty() { return Err(anyhow::anyhow!("Alert template name is required")); } diff --git a/src/service/search/cluster/mod.rs b/src/service/search/cluster/mod.rs index 35d505d66a..d2e6f9bd99 100644 --- a/src/service/search/cluster/mod.rs +++ b/src/service/search/cluster/mod.rs @@ -437,6 +437,12 @@ pub async fn search( log::info!( "[trace_id {trace_id}] search->grpc: search canceled before call search->grpc" ); + work_group + .as_ref() + .unwrap() + .done(&trace_id) + .await + .map_err(|e| Error::Message(e.to_string()))?; return Err(Error::ErrorCode(ErrorCodes::SearchCancelQuery(format!( "[trace_id {trace_id}] search->grpc: search canceled before call search->grpc" )))); diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index ad9ff9d558..1d0625ab07 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.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::{cmp::max, sync::Arc}; +use std::cmp::max; use chrono::Duration; use config::{ @@ -33,16 +33,16 @@ use once_cell::sync::Lazy; use opentelemetry::trace::TraceContextExt; use proto::cluster_rpc; use regex::Regex; -use tokio::sync::Mutex; use tracing_opentelemetry::OpenTelemetrySpanExt; #[cfg(feature = "enterprise")] use { hashbrown::HashSet, - o2_enterprise::enterprise::common::infra::config::O2_CONFIG, - o2_enterprise::enterprise::search::TaskStatus, + o2_enterprise::enterprise::{common::infra::config::O2_CONFIG, search::TaskStatus}, tonic::{codec::CompressionEncoding, metadata::MetadataValue, transport::Channel, Request}, tracing::{info_span, Instrument}, }; +#[cfg(not(feature = "enterprise"))] +use {std::sync::Arc, tokio::sync::Mutex}; use crate::{ common::{infra::cluster as infra_cluster, meta::stream::StreamParams}, @@ -57,6 +57,7 @@ pub(crate) mod sql; pub static SEARCH_SERVER: Lazy = Lazy::new(Searcher::new); +#[cfg(not(feature = "enterprise"))] pub(crate) static QUEUE_LOCKER: Lazy>> = Lazy::new(|| Arc::new(Mutex::const_new(false)));