fix: value API is always waiting (#3364)

fixed #3334

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",
-]
+]

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
 

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 <http://www.gnu.org/licenses/>.
 
-#[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(())
 }

src/handler/http/request/alerts/destinations.rs

@@ -80,8 +80,7 @@ pub async fn update_destination(
 ) -> Result<HttpResponse, Error> {
     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)),

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)),
     }

src/handler/http/request/alerts/templates.rs

@@ -77,8 +77,7 @@ pub async fn update_template(
 ) -> Result<HttpResponse, Error> {
     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)),
     }

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::<usize>().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<String> = None;
 
     let timeout = query
@@ -900,8 +906,19 @@ async fn values_v1(
         .map_or(0, |v| v.parse::<i64>().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::<i64>().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 {

src/handler/http/request/traces/mod.rs

@@ -222,8 +222,19 @@ pub async fn get_latest_traces(
         .map_or(0, |v| v.parse::<i64>().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!(

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,

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();

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"));
     }

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"
             ))));

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 <http://www.gnu.org/licenses/>.
 
-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<Searcher> = Lazy::new(Searcher::new);
 
+#[cfg(not(feature = "enterprise"))]
 pub(crate) static QUEUE_LOCKER: Lazy<Arc<Mutex<bool>>> =
     Lazy::new(|| Arc::new(Mutex::const_new(false)));