perf: some improve (openobserve#1008)

Co-authored-by: oasisk <[email protected]>

Author: hengfeiyang

src/handler/http/auth/mod.rs

@@ -201,7 +201,7 @@ pub async fn validator_gcp(
     let query =
         web::Query::<std::collections::HashMap<String, String>>::from_query(req.query_string())
             .unwrap();
-    match query.get("API_Key") {
+    match query.get("API-Key") {
         Some(val) => {
             let gcp_creds = common::base64::decode(val).unwrap();
             let creds = gcp_creds

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

@@ -137,7 +137,7 @@ pub async fn search(
             let time = start.elapsed().as_secs_f64();
             metrics::HTTP_RESPONSE_TIME
                 .with_label_values(&[
-                    "/_search",
+                    "/api/org/_search",
                     "200",
                     &org_id,
                     "",
@@ -146,7 +146,7 @@ pub async fn search(
                 .observe(time);
             metrics::HTTP_INCOMING_REQUESTS
                 .with_label_values(&[
-                    "/_search",
+                    "/api/org/_search",
                     "200",
                     &org_id,
                     "",
@@ -160,7 +160,7 @@ pub async fn search(
             let time = start.elapsed().as_secs_f64();
             metrics::HTTP_RESPONSE_TIME
                 .with_label_values(&[
-                    "/_search",
+                    "/api/org/_search",
                     "500",
                     &org_id,
                     "",
@@ -169,7 +169,7 @@ pub async fn search(
                 .observe(time);
             metrics::HTTP_INCOMING_REQUESTS
                 .with_label_values(&[
-                    "/_search",
+                    "/api/org/_search",
                     "500",
                     &org_id,
                     "",
@@ -311,7 +311,7 @@ pub async fn around(
             let time = start.elapsed().as_secs_f64();
             metrics::HTTP_RESPONSE_TIME
                 .with_label_values(&[
-                    "/_around",
+                    "/api/org/_around",
                     "500",
                     &org_id,
                     &stream_name,
@@ -320,7 +320,7 @@ pub async fn around(
                 .observe(time);
             metrics::HTTP_INCOMING_REQUESTS
                 .with_label_values(&[
-                    "/_around",
+                    "/api/org/_around",
                     "500",
                     &org_id,
                     &stream_name,
@@ -363,7 +363,7 @@ pub async fn around(
             let time = start.elapsed().as_secs_f64();
             metrics::HTTP_RESPONSE_TIME
                 .with_label_values(&[
-                    "/_around",
+                    "/api/org/_around",
                     "500",
                     &org_id,
                     &stream_name,
@@ -372,7 +372,7 @@ pub async fn around(
                 .observe(time);
             metrics::HTTP_INCOMING_REQUESTS
                 .with_label_values(&[
-                    "/_around",
+                    "/api/org/_around",
                     "500",
                     &org_id,
                     &stream_name,
@@ -410,7 +410,7 @@ pub async fn around(
     let time = start.elapsed().as_secs_f64();
     metrics::HTTP_RESPONSE_TIME
         .with_label_values(&[
-            "/_around",
+            "/api/org/_around",
             "200",
             &org_id,
             &stream_name,
@@ -419,7 +419,7 @@ pub async fn around(
         .observe(time);
     metrics::HTTP_INCOMING_REQUESTS
         .with_label_values(&[
-            "/_around",
+            "/api/org/_around",
             "200",
             &org_id,
             &stream_name,
@@ -548,7 +548,7 @@ pub async fn values(
             let time = start.elapsed().as_secs_f64();
             metrics::HTTP_RESPONSE_TIME
                 .with_label_values(&[
-                    "/_values",
+                    "/api/org/_values",
                     "500",
                     &org_id,
                     &stream_name,
@@ -557,7 +557,7 @@ pub async fn values(
                 .observe(time);
             metrics::HTTP_INCOMING_REQUESTS
                 .with_label_values(&[
-                    "/_values",
+                    "/api/org/_values",
                     "500",
                     &org_id,
                     &stream_name,
@@ -593,7 +593,7 @@ pub async fn values(
     let time = start.elapsed().as_secs_f64();
     metrics::HTTP_RESPONSE_TIME
         .with_label_values(&[
-            "/_values",
+            "/api/org/_values",
             "200",
             &org_id,
             &stream_name,
@@ -602,7 +602,7 @@ pub async fn values(
         .observe(time);
     metrics::HTTP_INCOMING_REQUESTS
         .with_label_values(&[
-            "/_values",
+            "/api/org/_values",
             "200",
             &org_id,
             &stream_name,

src/infra/cluster.rs

@@ -98,8 +98,8 @@ impl std::fmt::Display for Role {
 /// Register and keepalive the node to cluster
 pub async fn register_and_keepalive() -> Result<()> {
     if CONFIG.common.local_mode {
-        let nodes = load_local_node_role();
-        if !is_single_node(nodes) {
+        let roles = load_local_node_role();
+        if !is_single_node(&roles) {
             panic!("For local mode only single node deployment is allowed !");
         }
         // cache local node
@@ -439,7 +439,7 @@ pub fn is_alert_manager(role: &[Role]) -> bool {
 }
 
 #[inline(always)]
-pub fn is_single_node(role: Vec<Role>) -> bool {
+pub fn is_single_node(role: &[Role]) -> bool {
     role.contains(&Role::All)
 }
 

src/infra/config.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use dashmap::DashMap;
+use dashmap::{DashMap, DashSet};
 use datafusion::arrow::datatypes::Schema;
 use dotenv_config::EnvConfig;
 use dotenvy::dotenv;
@@ -32,6 +32,7 @@ use crate::meta::user::User;
 use crate::service::enrichment::StreamTable;
 
 pub type RwHashMap<K, V> = DashMap<K, V, ahash::RandomState>;
+pub type RwHashSet<K> = DashSet<K, ahash::RandomState>;
 
 pub static VERSION: &str = env!("GIT_VERSION");
 pub static COMMIT_HASH: &str = env!("GIT_COMMIT_HASH");
@@ -184,6 +185,8 @@ pub struct Common {
     pub column_timestamp: String,
     #[env_config(name = "ZO_WIDENING_SCHEMA_EVOLUTION", default = false)]
     pub widening_schema_evolution: bool,
+    #[env_config(name = "ZO_SKIP_SCHEMA_VALIDATION", default = false)]
+    pub skip_schema_validation: bool,
     #[env_config(name = "ZO_FEATURE_PER_THREAD_LOCK", default = false)]
     pub feature_per_thread_lock: bool,
     #[env_config(name = "ZO_FEATURE_FULLTEXT_ON_ALL_FIELDS", default = false)]
@@ -216,8 +219,6 @@ pub struct Common {
     pub prometheus_enabled: bool,
     #[env_config(name = "ZO_PRINT_KEY_CONFIG", default = false)]
     pub print_key_config: bool,
-    #[env_config(name = "ZO_SKIP_SCHEMA_VALIDATION", default = false)]
-    pub skip_schema_validation: bool,
 }
 
 #[derive(EnvConfig)]
@@ -265,6 +266,8 @@ pub struct Compact {
     pub max_file_size: u64,
     #[env_config(name = "ZO_COMPACT_DATA_RETENTION_DAYS", default = 3650)] // in days
     pub data_retention_days: i64,
+    #[env_config(name = "ZO_COMPACT_BLOCKED_ORGS", default = "")] // use comma to split
+    pub blocked_orgs: String,
 }
 
 #[derive(EnvConfig)]
@@ -314,7 +317,7 @@ pub struct Etcd {
     pub connect_timeout: u64,
     #[env_config(name = "ZO_ETCD_COMMAND_TIMEOUT", default = 5)]
     pub command_timeout: u64,
-    #[env_config(name = "ZO_ETCD_LOCK_WAIT_TIMEOUT", default = 600)]
+    #[env_config(name = "ZO_ETCD_LOCK_WAIT_TIMEOUT", default = 3600)]
     pub lock_wait_timeout: u64,
     #[env_config(name = "ZO_ETCD_USER", default = "")]
     pub user: String,
@@ -360,6 +363,8 @@ pub struct S3 {
     pub bucket_prefix: String,
     #[env_config(name = "ZO_S3_CONNECT_TIMEOUT", default = 10)] // seconds
     pub connect_timeout: u64,
+    #[env_config(name = "ZO_S3_REQUEST_TIMEOUT", default = 3600)] // seconds
+    pub request_timeout: u64,
     #[env_config(name = "ZO_S3_FEATURE_FORCE_PATH_STYLE", default = false)]
     pub feature_force_path_style: bool,
 }

src/infra/storage/remote.rs

@@ -183,6 +183,7 @@ fn init_aws_config() -> object_store::Result<object_store::aws::AmazonS3> {
         .with_client_options(
             object_store::ClientOptions::default()
                 .with_connect_timeout(std::time::Duration::from_secs(CONFIG.s3.connect_timeout))
+                .with_timeout(std::time::Duration::from_secs(CONFIG.s3.request_timeout))
                 .with_allow_http(true),
         )
         .with_profile("default")
@@ -207,7 +208,8 @@ fn init_azure_config() -> object_store::Result<object_store::azure::MicrosoftAzu
     let mut builder = object_store::azure::MicrosoftAzureBuilder::from_env()
         .with_client_options(
             object_store::ClientOptions::default()
-                .with_connect_timeout(std::time::Duration::from_secs(CONFIG.s3.connect_timeout)),
+                .with_connect_timeout(std::time::Duration::from_secs(CONFIG.s3.connect_timeout))
+                .with_timeout(std::time::Duration::from_secs(CONFIG.s3.request_timeout)),
         )
         .with_container_name(&CONFIG.s3.bucket_name);
     if !CONFIG.s3.access_key.is_empty() {
@@ -223,7 +225,8 @@ fn init_gcp_config() -> object_store::Result<object_store::gcp::GoogleCloudStora
     let mut builder = object_store::gcp::GoogleCloudStorageBuilder::from_env()
         .with_client_options(
             object_store::ClientOptions::default()
-                .with_connect_timeout(std::time::Duration::from_secs(CONFIG.s3.connect_timeout)),
+                .with_connect_timeout(std::time::Duration::from_secs(CONFIG.s3.connect_timeout))
+                .with_timeout(std::time::Duration::from_secs(CONFIG.s3.request_timeout)),
         )
         .with_bucket_name(&CONFIG.s3.bucket_name);
     if !CONFIG.s3.access_key.is_empty() {

src/job/files/disk.rs

@@ -26,9 +26,6 @@ use crate::infra::{cluster, config::CONFIG, metrics, storage, wal};
 use crate::meta::{common::FileMeta, StreamType};
 use crate::service::{db, schema::schema_evolution, search::datafusion::new_writer};
 
-/// TaskResult (local file path, remote file key, file meta, stream type)
-type TaskResult = (String, String, FileMeta, StreamType);
-
 pub async fn run() -> Result<(), anyhow::Error> {
     if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) {
         return Ok(()); // not an ingester, no need to init job
@@ -59,8 +56,7 @@ async fn move_files_to_storage() -> Result<(), anyhow::Error> {
     let files = scan_files(&pattern);
 
     // use multiple threads to upload files
-    let mut tasks: Vec<task::JoinHandle<Result<TaskResult, anyhow::Error>>> = Vec::new();
-
+    let mut tasks = Vec::new();
     let semaphore = std::sync::Arc::new(Semaphore::new(CONFIG.limit.file_move_thread_num));
     for file in files {
         let local_file = file.to_owned();
@@ -117,33 +113,22 @@ async fn move_files_to_storage() -> Result<(), anyhow::Error> {
         }
 
         let permit = semaphore.clone().acquire_owned().await.unwrap();
-        let task: task::JoinHandle<Result<(String, String, FileMeta, StreamType), anyhow::Error>> =
-            task::spawn(async move {
-                let ret = upload_file(
-                    &org_id,
-                    &stream_name,
-                    stream_type,
-                    &local_file,
-                    partition_key,
-                )
-                .await;
-                drop(permit);
-                match ret {
-                    Ok((key, meta, stream_type)) => Ok((local_file, key, meta, stream_type)),
-                    Err(e) => Err(e),
-                }
-            });
-        tasks.push(task);
-        task::yield_now().await;
-    }
-
-    for task in tasks {
-        match task.await {
-            Ok(ret) => match ret {
-                Ok((path, key, meta, _stream_type)) => {
+        let task: task::JoinHandle<Result<(), anyhow::Error>> = task::spawn(async move {
+            let ret = upload_file(
+                &org_id,
+                &stream_name,
+                stream_type,
+                &local_file,
+                partition_key,
+            )
+            .await;
+            drop(permit);
+            match ret {
+                Err(e) => log::error!("[JOB] Error while uploading disk file to storage {}", e),
+                Ok((key, meta, _stream_type)) => {
                     match db::file_list::local::set(&key, meta, false).await {
                         Ok(_) => {
-                            match fs::remove_file(&path) {
+                            match fs::remove_file(&local_file) {
                                 Ok(_) => {
                                     // metrics
                                     let columns = key.split('/').collect::<Vec<&str>>();
@@ -158,22 +143,28 @@ async fn move_files_to_storage() -> Result<(), anyhow::Error> {
                                 Err(e) => {
                                     log::error!(
                                         "[JOB] Failed to remove disk file from disk: {}, {}",
-                                        path,
+                                        local_file,
                                         e.to_string()
                                     )
                                 }
                             }
                         }
                         Err(e) => log::error!(
                             "[JOB] Failed write disk file meta:{}, error: {}",
-                            path,
+                            local_file,
                             e.to_string()
                         ),
                     }
                 }
-                Err(e) => log::error!("[JOB] Error while uploading disk file to storage {}", e),
-            },
-            Err(e) => log::error!("[JOB] Error while uploading disk file to storage {}", e),
+            };
+            Ok(())
+        });
+        tasks.push(task);
+    }
+
+    for task in tasks {
+        if let Err(e) = task.await {
+            log::error!("[JOB] Error while uploading disk file to storage {}", e);
         };
     }
     Ok(())