From 3977c9be423f98a4bd1e005c87c50e07ea28cfe8 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Sat, 15 Mar 2025 15:03:28 +0800 Subject: [PATCH] perf: optimize compactor scheduler with multiple mpsc (#6269) --- .gitignore | 28 +-- src/config/src/config.rs | 32 +-- src/handler/http/request/websocket/mod.rs | 6 +- src/handler/http/request/websocket/utils.rs | 4 +- src/infra/src/storage/local.rs | 13 +- src/infra/src/storage/remote.rs | 21 +- src/job/compactor.rs | 128 ++++------- src/job/files/parquet.rs | 2 +- src/main.rs | 24 ++- src/service/compact/deleted.rs | 10 +- src/service/compact/merge.rs | 59 ++---- src/service/compact/mod.rs | 60 ++---- src/service/compact/retention.rs | 53 +++-- src/service/compact/worker.rs | 222 ++++++++++++++++++++ 14 files changed, 409 insertions(+), 253 deletions(-) create mode 100644 src/service/compact/worker.rs diff --git a/.gitignore b/.gitignore index 7a9e677154..0ee4555176 100644 --- a/.gitignore +++ b/.gitignore @@ -11,18 +11,10 @@ flamegraph.html /target/ /bin/ - -# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries -# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html -# Cargo.lock - -# These are backup files generated by rustfmt -**/*.rs.bk - - - +# data /tmp/ /data/ +/data2/ /.idea/ /.vscode /.env @@ -30,20 +22,6 @@ flamegraph.html /integration-tests /download/ -#web distribution +# web distribution /web/dist /web/.env - - -/tmp/ -/data/ -/.idea/ -/.vscode -/.env -/.env* -/integration-tests -/download/ - -#web distribution -/web/dist -/web/.env \ No newline at end of file diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 24929b6c0b..430a75d110 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -1189,6 +1189,8 @@ pub struct Limit { pub http_request_timeout: u64, #[env_config(name = "ZO_ACTIX_KEEP_ALIVE", default = 5)] // seconds pub http_keep_alive: u64, + #[env_config(name = "ZO_ACTIX_KEEP_ALIVE_DISABLED", default = false)] + pub http_keep_alive_disabled: bool, #[env_config(name = "ZO_ACTIX_SHUTDOWN_TIMEOUT", default = 5)] // seconds pub http_shutdown_timeout: u64, #[env_config(name = "ZO_ACTIX_SLOW_LOG_THRESHOLD", default = 5)] // seconds @@ -1450,12 +1452,12 @@ pub struct MemoryCache { // max_size #[env_config(name = "ZO_MEMORY_CACHE_SKIP_SIZE", default = 0)] pub skip_size: usize, - // MB, when cache is full will release how many data once time, default is 1% of max_size + // MB, when cache is full will release how many data once time, default is 10% of max_size #[env_config(name = "ZO_MEMORY_CACHE_RELEASE_SIZE", default = 0)] pub release_size: usize, - #[env_config(name = "ZO_MEMORY_CACHE_GC_SIZE", default = 50)] // MB + #[env_config(name = "ZO_MEMORY_CACHE_GC_SIZE", default = 100)] // MB pub gc_size: usize, - #[env_config(name = "ZO_MEMORY_CACHE_GC_INTERVAL", default = 0)] // seconds + #[env_config(name = "ZO_MEMORY_CACHE_GC_INTERVAL", default = 60)] // seconds pub gc_interval: u64, #[env_config(name = "ZO_MEMORY_CACHE_SKIP_DISK_CHECK", default = false)] pub skip_disk_check: bool, @@ -1486,12 +1488,12 @@ pub struct DiskCache { // max_size #[env_config(name = "ZO_DISK_CACHE_SKIP_SIZE", default = 0)] pub skip_size: usize, - // MB, when cache is full will release how many data once time, default is 1% of max_size + // MB, when cache is full will release how many data once time, default is 10% of max_size #[env_config(name = "ZO_DISK_CACHE_RELEASE_SIZE", default = 0)] pub release_size: usize, #[env_config(name = "ZO_DISK_CACHE_GC_SIZE", default = 100)] // MB pub gc_size: usize, - #[env_config(name = "ZO_DISK_CACHE_GC_INTERVAL", default = 0)] // seconds + #[env_config(name = "ZO_DISK_CACHE_GC_INTERVAL", default = 60)] // seconds pub gc_interval: u64, #[env_config(name = "ZO_DISK_CACHE_MULTI_DIR", default = "")] // dir1,dir2,dir3... pub multi_dir: String, @@ -2230,14 +2232,14 @@ fn check_memory_config(cfg: &mut Config) -> Result<(), anyhow::Error> { cfg.memory_cache.skip_size *= 1024 * 1024; } if cfg.memory_cache.release_size == 0 { - // when cache is full will release how many data once time, default is 1% of + // when cache is full will release how many data once time, default is 10% of // max_size - cfg.memory_cache.release_size = cfg.memory_cache.max_size / 100; + cfg.memory_cache.release_size = cfg.memory_cache.max_size / 10; } else { cfg.memory_cache.release_size *= 1024 * 1024; } if cfg.memory_cache.gc_size == 0 { - cfg.memory_cache.gc_size = 10 * 1024 * 1024; // 10 MB + cfg.memory_cache.gc_size = 100 * 1024 * 1024; // 100 MB } else { cfg.memory_cache.gc_size *= 1024 * 1024; } @@ -2257,7 +2259,7 @@ fn check_memory_config(cfg: &mut Config) -> Result<(), anyhow::Error> { } if cfg.memory_cache.bucket_num == 0 { - cfg.memory_cache.bucket_num = cfg.limit.real_cpu_num; + cfg.memory_cache.bucket_num = cfg.limit.cpu_num; } cfg.memory_cache.max_size /= cfg.memory_cache.bucket_num; cfg.memory_cache.release_size /= cfg.memory_cache.bucket_num; @@ -2348,14 +2350,14 @@ fn check_disk_cache_config(cfg: &mut Config) -> Result<(), anyhow::Error> { cfg.disk_cache.skip_size *= 1024 * 1024; } if cfg.disk_cache.release_size == 0 { - // when cache is full will release how many data once time, default is 1% of + // when cache is full will release how many data once time, default is 10% of // max_size - cfg.disk_cache.release_size = cfg.disk_cache.max_size / 100; + cfg.disk_cache.release_size = cfg.disk_cache.max_size / 10; } else { cfg.disk_cache.release_size *= 1024 * 1024; } if cfg.disk_cache.gc_size == 0 { - cfg.disk_cache.gc_size = 10 * 1024 * 1024; // 10 MB + cfg.disk_cache.gc_size = 100 * 1024 * 1024; // 100 MB } else { cfg.disk_cache.gc_size *= 1024 * 1024; } @@ -2367,18 +2369,18 @@ fn check_disk_cache_config(cfg: &mut Config) -> Result<(), anyhow::Error> { } if cfg.disk_cache.bucket_num == 0 { - // because we validate thread_query_num before this + // because we validate cpu_num before this // we can be sure here that that value is sane. // following numbers are imperically decided, users can set the value // directly if they know better, otherwise this was the best numbers // for bucket_num based on thread count. - let threads = cfg.limit.query_thread_num; + let threads = cfg.limit.cpu_num; if threads <= 16 { // for less than 16 threads, same buckets would be good enough // with 16 files in parallel we should not run into that many // files going into same bucket, so ok. - cfg.disk_cache.bucket_num = cfg.limit.query_thread_num; + cfg.disk_cache.bucket_num = threads; } else if threads > 16 && threads <= 64 { // for 32 -> 64 ish range, there can be a lot of collisions // so we set it to double the threads to avoid any collisions diff --git a/src/handler/http/request/websocket/mod.rs b/src/handler/http/request/websocket/mod.rs index ecf2a3c75f..94050e7325 100644 --- a/src/handler/http/request/websocket/mod.rs +++ b/src/handler/http/request/websocket/mod.rs @@ -19,7 +19,7 @@ pub mod sort; pub mod utils; use actix_web::{Error, HttpRequest, HttpResponse, get, web}; -use config::get_config; +use config::{cluster::LOCAL_NODE, get_config}; use session::WsSession; use utils::sessions_cache_utils; @@ -71,6 +71,8 @@ pub async fn websocket( /// Initialize the job init for websocket pub async fn init() -> Result<(), anyhow::Error> { // Run the garbage collector for websocket sessions - sessions_cache_utils::run_gc_ws_sessions().await; + if !LOCAL_NODE.is_compactor() && !LOCAL_NODE.is_alert_manager() { + sessions_cache_utils::run_gc_ws_sessions().await; + } Ok(()) } diff --git a/src/handler/http/request/websocket/utils.rs b/src/handler/http/request/websocket/utils.rs index 4236b35ee9..2811dc23ea 100644 --- a/src/handler/http/request/websocket/utils.rs +++ b/src/handler/http/request/websocket/utils.rs @@ -97,7 +97,7 @@ pub mod sessions_cache_utils { }; pub async fn run_gc_ws_sessions() { - log::info!("[WS_GC] Running garbage collector for websocket sessions"); + log::debug!("[WS_GC] Running garbage collector for websocket sessions"); let cfg = get_config(); let interval_secs = cfg.websocket.session_gc_interval_secs; @@ -107,7 +107,7 @@ pub mod sessions_cache_utils { tokio::spawn(async move { loop { interval.tick().await; - log::info!("[WS_GC] Running garbage collector for websocket sessions"); + log::debug!("[WS_GC] Running garbage collector for websocket sessions"); // Use catch_unwind to prevent task from crashing if let Err(e) = std::panic::AssertUnwindSafe(cleanup_expired_sessions()) diff --git a/src/infra/src/storage/local.rs b/src/infra/src/storage/local.rs index 3bed7688f9..3c04550241 100644 --- a/src/infra/src/storage/local.rs +++ b/src/infra/src/storage/local.rs @@ -180,8 +180,17 @@ impl ObjectStore for Local { let file = location.to_string(); let data = self .client - .get_range(&(format_key(&file, self.with_prefix).into()), range) - .await?; + .get_range(&(format_key(&file, self.with_prefix).into()), range.clone()) + .await + .map_err(|e| { + log::error!( + "[STORAGE] get_range local file: {}, range: {:?}, error: {:?}", + file, + range, + e + ); + e + })?; // metrics let data_len = data.len(); diff --git a/src/infra/src/storage/remote.rs b/src/infra/src/storage/remote.rs index def9b20cad..ca946dba6e 100644 --- a/src/infra/src/storage/remote.rs +++ b/src/infra/src/storage/remote.rs @@ -87,7 +87,7 @@ impl ObjectStore for Remote { }) } Err(err) => { - log::error!("s3 File upload error: {:?}", err); + log::error!("[STORAGE] put_opts remote file: {}, error: {:?}", file, err); Err(err) } } @@ -106,7 +106,11 @@ impl ObjectStore for Remote { { Ok(r) => Ok(r), Err(err) => { - log::error!("s3 multipart File upload error: {:?}", err); + log::error!( + "[STORAGE] put_multipart_opts remote file: {}, error: {:?}", + file, + err + ); Err(err) } } @@ -169,8 +173,17 @@ impl ObjectStore for Remote { let file = location.to_string(); let data = self .client - .get_range(&(format_key(&file, true).into()), range) - .await?; + .get_range(&(format_key(&file, true).into()), range.clone()) + .await + .map_err(|e| { + log::error!( + "[STORAGE] get_range remote file: {}, range: {:?}, error: {:?}", + file, + range, + e + ); + e + })?; // metrics let data_len = data.len(); diff --git a/src/job/compactor.rs b/src/job/compactor.rs index 164189264e..d1cd3b5499 100644 --- a/src/job/compactor.rs +++ b/src/job/compactor.rs @@ -13,20 +13,15 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::sync::Arc; - use config::{ cluster::LOCAL_NODE, get_config, - meta::{ - cluster::CompactionJobType, - stream::{ALL_STREAM_TYPES, FileKey}, - }, + meta::{cluster::CompactionJobType, stream::ALL_STREAM_TYPES}, metrics, }; #[cfg(feature = "enterprise")] use o2_enterprise::enterprise::common::infra::config::get_config as get_o2_config; -use tokio::sync::{Mutex, mpsc}; +use tokio::sync::mpsc; use crate::service::compact; @@ -40,71 +35,22 @@ pub async fn run() -> Result<(), anyhow::Error> { return Ok(()); } - let (tx, rx) = mpsc::channel::<(compact::merge::MergeSender, compact::merge::MergeBatch)>( - cfg.limit.file_merge_thread_num * 2, - ); - let rx = Arc::new(Mutex::new(rx)); - // start merge workers - for thread_id in 0..cfg.limit.file_merge_thread_num { - let rx = rx.clone(); - tokio::spawn(async move { - loop { - let ret = rx.lock().await.recv().await; - match ret { - None => { - log::debug!("[COMPACTOR:JOB] Receiving files channel is closed"); - break; - } - Some((tx, msg)) => { - match compact::merge::merge_files( - thread_id, - &msg.org_id, - msg.stream_type, - &msg.stream_name, - &msg.prefix, - &msg.files, - ) - .await - { - Ok((file, meta, _)) => { - let mut new_file_keys = Vec::with_capacity(file.len()); - for (file, meta) in file.into_iter().zip(meta.into_iter()) { - new_file_keys.push(FileKey::new(file, meta, false)); - } - if let Err(e) = tx.send(Ok((msg.batch_id, new_file_keys))).await { - log::error!( - "[COMPACTOR:JOB] Error sending file to merge_job: {}", - e - ); - } - } - Err(e) => { - log::error!( - "[COMPACTOR:JOB] Error merging files: stream: {}/{}/{}, err: {}", - msg.org_id, - msg.stream_type, - msg.stream_name, - e - ); - if let Err(e) = tx.send(Err(e)).await { - log::error!( - "[COMPACTOR:JOB] Error sending error to merge_job: {}", - e - ); - } - } - } - } - } - } - }); + let mut worker = compact::worker::MergeWorker::new(cfg.limit.file_merge_thread_num); + if let Err(e) = worker.run() { + log::error!("[COMPACTOR::JOB] start merge worker error: {e}"); + } + + let mut scheduler = + compact::worker::JobScheduler::new(cfg.limit.file_merge_thread_num, worker.tx()); + if let Err(e) = scheduler.run() { + log::error!("[COMPACTOR::JOB] start merge job scheduler error: {e}"); } tokio::task::spawn(async move { run_generate_job().await }); tokio::task::spawn(async move { run_generate_old_data_job().await }); #[cfg(feature = "enterprise")] tokio::task::spawn(async move { run_generate_downsampling_job().await }); - tokio::task::spawn(async move { run_merge(tx).await }); + tokio::task::spawn(async move { run_merge(scheduler.tx()).await }); tokio::task::spawn(async move { run_retention().await }); tokio::task::spawn(async move { run_delay_deletion().await }); tokio::task::spawn(async move { run_sync_to_db().await }); @@ -121,16 +67,16 @@ pub async fn run() -> Result<(), anyhow::Error> { async fn run_compactor_pending_jobs_metric() -> Result<(), anyhow::Error> { let interval = get_config().compact.pending_jobs_metric_interval; - log::info!("[COMPACTOR] start run_compactor_pending_jobs_metric job"); + log::info!("[COMPACTOR::JOB] start run_compactor_pending_jobs_metric job"); loop { tokio::time::sleep(tokio::time::Duration::from_secs(interval)).await; - log::debug!("[COMPACTOR] Running compactor pending jobs to report metric"); + log::debug!("[COMPACTOR::JOB] Running compactor pending jobs to report metric"); let job_status = match infra::file_list::get_pending_jobs_count().await { Ok(status) => status, Err(e) => { - log::error!("[COMPACTOR] run compactor pending jobs metric error: {e}"); + log::error!("[COMPACTOR::JOB] run compactor pending jobs metric error: {e}"); continue; } }; @@ -169,9 +115,9 @@ async fn run_generate_job() -> Result<(), anyhow::Error> { get_config().compact.interval, )) .await; - log::debug!("[COMPACTOR] Running generate merge job"); + log::debug!("[COMPACTOR::JOB] Running generate merge job"); if let Err(e) = compact::run_generate_job(CompactionJobType::Current).await { - log::error!("[COMPACTOR] run generate merge job error: {e}"); + log::error!("[COMPACTOR::JOB] run generate merge job error: {e}"); } } } @@ -184,9 +130,9 @@ async fn run_generate_old_data_job() -> Result<(), anyhow::Error> { get_config().compact.old_data_interval + 1, )) .await; - log::debug!("[COMPACTOR] Running generate merge job for old data"); + log::debug!("[COMPACTOR::JOB] Running generate merge job for old data"); if let Err(e) = compact::run_generate_job(CompactionJobType::Historical).await { - log::error!("[COMPACTOR] run generate merge job for old data error: {e}"); + log::error!("[COMPACTOR::JOB] run generate merge job for old data error: {e}"); } } } @@ -206,25 +152,23 @@ async fn run_generate_downsampling_job() -> Result<(), anyhow::Error> { get_o2_config().downsampling.downsampling_interval, )) .await; - log::debug!("[COMPACTOR] Running generate downsampling job"); + log::debug!("[COMPACTOR::JOB] Running generate downsampling job"); if let Err(e) = compact::run_generate_downsampling_job().await { - log::error!("[COMPACTOR] run generate downsampling job error: {e}"); + log::error!("[COMPACTOR::JOB] run generate downsampling job error: {e}"); } } } /// Merge small files -async fn run_merge( - tx: mpsc::Sender<(compact::merge::MergeSender, compact::merge::MergeBatch)>, -) -> Result<(), anyhow::Error> { +async fn run_merge(tx: mpsc::Sender) -> Result<(), anyhow::Error> { loop { tokio::time::sleep(tokio::time::Duration::from_secs( get_config().compact.interval + 2, )) .await; - log::debug!("[COMPACTOR] Running data merge"); + log::debug!("[COMPACTOR::JOB] Running data merge"); if let Err(e) = compact::run_merge(tx.clone()).await { - log::error!("[COMPACTOR] run data merge error: {e}"); + log::error!("[COMPACTOR::JOB] run data merge error: {e}"); } } } @@ -236,9 +180,9 @@ async fn run_retention() -> Result<(), anyhow::Error> { get_config().compact.interval + 3, )) .await; - log::debug!("[COMPACTOR] Running data retention"); + log::debug!("[COMPACTOR::JOB] Running data retention"); if let Err(e) = compact::run_retention().await { - log::error!("[COMPACTOR] run data retention error: {e}"); + log::error!("[COMPACTOR::JOB] run data retention error: {e}"); } } } @@ -250,9 +194,9 @@ async fn run_delay_deletion() -> Result<(), anyhow::Error> { get_config().compact.interval + 4, )) .await; - log::debug!("[COMPACTOR] Running data delay deletion"); + log::debug!("[COMPACTOR::JOB] Running data delay deletion"); if let Err(e) = compact::run_delay_deletion().await { - log::error!("[COMPACTOR] run data delay deletion error: {e}"); + log::error!("[COMPACTOR::JOB] run data delay deletion error: {e}"); } } } @@ -263,9 +207,9 @@ async fn run_sync_to_db() -> Result<(), anyhow::Error> { get_config().compact.sync_to_db_interval, )) .await; - log::debug!("[COMPACTOR] Running sync cached compact offset to db"); + log::debug!("[COMPACTOR::JOB] Running sync cached compact offset to db"); if let Err(e) = crate::service::db::compact::files::sync_cache_to_db().await { - log::error!("[COMPACTOR] run sync cached compact offset to db error: {e}"); + log::error!("[COMPACTOR::JOB] run sync cached compact offset to db error: {e}"); } } } @@ -284,9 +228,9 @@ async fn run_downsampling_sync_to_db() -> Result<(), anyhow::Error> { get_config().compact.sync_to_db_interval, )) .await; - log::debug!("[COMPACTOR] Running sync cached downsampling offset to db"); + log::debug!("[COMPACTOR::JOB] Running sync cached downsampling offset to db"); if let Err(e) = crate::service::db::compact::downsampling::sync_cache_to_db().await { - log::error!("[COMPACTOR] run sync cached downsampling offset to db error: {e}"); + log::error!("[COMPACTOR::JOB] run sync cached downsampling offset to db error: {e}"); } } } @@ -294,10 +238,10 @@ async fn run_downsampling_sync_to_db() -> Result<(), anyhow::Error> { async fn run_check_running_jobs() -> Result<(), anyhow::Error> { loop { let time = get_config().compact.job_run_timeout; - log::debug!("[COMPACTOR] Running check running jobs"); + log::debug!("[COMPACTOR::JOB] Running check running jobs"); let updated_at = config::utils::time::now_micros() - (time * 1000 * 1000); if let Err(e) = infra::file_list::check_running_jobs(updated_at).await { - log::error!("[COMPACTOR] run check running jobs error: {e}"); + log::error!("[COMPACTOR::JOB] run check running jobs error: {e}"); } tokio::time::sleep(tokio::time::Duration::from_secs(time as u64)).await; } @@ -306,10 +250,10 @@ async fn run_check_running_jobs() -> Result<(), anyhow::Error> { async fn run_clean_done_jobs() -> Result<(), anyhow::Error> { loop { let time = get_config().compact.job_clean_wait_time; - log::debug!("[COMPACTOR] Running clean done jobs"); + log::debug!("[COMPACTOR::JOB] Running clean done jobs"); let updated_at = config::utils::time::now_micros() - (time * 1000 * 1000); if let Err(e) = infra::file_list::clean_done_jobs(updated_at).await { - log::error!("[COMPACTOR] run clean done jobs error: {e}"); + log::error!("[COMPACTOR::JOB] run clean done jobs error: {e}"); } tokio::time::sleep(tokio::time::Duration::from_secs(time as u64)).await; } diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index 938b9047b3..2843ce7fcf 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -1166,7 +1166,7 @@ pub(crate) async fn generate_index_on_compactor( .await?; log::info!( - "[COMPACT:JOB] generated parquet index file: {}, index files: {:?}, took: {} ms", + "[COMPACTOR:JOB] generated parquet index file: {}, index files: {:?}, took: {} ms", new_file_key, files.iter().map(|(k, _)| k).collect::>(), start.elapsed().as_millis(), diff --git a/src/main.rs b/src/main.rs index 09d785cba6..fca99a9002 100644 --- a/src/main.rs +++ b/src/main.rs @@ -719,9 +719,11 @@ async fn init_http_server() -> Result<(), anyhow::Error> { )) .wrap(RequestTracing::new()) }) - .keep_alive(KeepAlive::Timeout(Duration::from_secs( - cfg.limit.http_keep_alive, - ))) + .keep_alive(if cfg.limit.http_keep_alive_disabled { + KeepAlive::Disabled + } else { + KeepAlive::Timeout(Duration::from_secs(max(1, cfg.limit.http_keep_alive))) + }) .client_request_timeout(Duration::from_secs(max(1, cfg.limit.http_request_timeout))) .shutdown_timeout(max(1, cfg.limit.http_shutdown_timeout)); let server = if cfg.http.tls_enabled { @@ -828,9 +830,11 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> { r#"%a "%r" %s %b "%{Content-Length}i" "%{Referer}i" "%{User-Agent}i" %T"#, )) }) - .keep_alive(KeepAlive::Timeout(Duration::from_secs( - cfg.limit.http_keep_alive, - ))) + .keep_alive(if cfg.limit.http_keep_alive_disabled { + KeepAlive::Disabled + } else { + KeepAlive::Timeout(Duration::from_secs(max(1, cfg.limit.http_keep_alive))) + }) .client_request_timeout(Duration::from_secs(max(1, cfg.limit.http_request_timeout))) .shutdown_timeout(max(1, cfg.limit.http_shutdown_timeout)); let server = if cfg.http.tls_enabled { @@ -1065,9 +1069,11 @@ async fn init_script_server() -> Result<(), anyhow::Error> { )) .wrap(RequestTracing::new()) }) - .keep_alive(KeepAlive::Timeout(Duration::from_secs( - cfg.limit.http_keep_alive, - ))) + .keep_alive(if cfg.limit.http_keep_alive_disabled { + KeepAlive::Disabled + } else { + KeepAlive::Timeout(Duration::from_secs(max(1, cfg.limit.http_keep_alive))) + }) .client_request_timeout(Duration::from_secs(max(1, cfg.limit.http_request_timeout))) .shutdown_timeout(max(1, cfg.limit.http_shutdown_timeout)); let server = if cfg.http.tls_enabled { diff --git a/src/service/compact/deleted.rs b/src/service/compact/deleted.rs index bb3f4f95d1..8c11d5a2ca 100644 --- a/src/service/compact/deleted.rs +++ b/src/service/compact/deleted.rs @@ -44,7 +44,7 @@ pub async fn delete( { // maybe the file already deleted, so we just skip the `not found` error if !e.to_string().to_lowercase().contains("not found") { - log::error!("[COMPACT] delete files from storage failed: {}", e); + log::error!("[COMPACTOR] delete files from storage failed: {}", e); return Err(e.into()); } } @@ -73,7 +73,7 @@ pub async fn delete( // maybe the file already deleted or there's not related index files, // so we just skip the `not found` error if !e.to_string().to_lowercase().contains("not found") { - log::error!("[COMPACT] delete files from storage failed: {}", e); + log::error!("[COMPACTOR] delete files from storage failed: {}", e); return Err(e.into()); } } @@ -106,7 +106,7 @@ pub async fn delete( { // maybe the file already deleted, so we just skip the `not found` error if !e.to_string().to_lowercase().contains("not found") { - log::error!("[COMPACT] delete files from storage failed: {}", e); + log::error!("[COMPACTOR] delete files from storage failed: {}", e); return Err(e.into()); } } @@ -117,7 +117,7 @@ pub async fn delete( if let Err(e) = storage::del(&files.keys().map(|file| file.as_str()).collect::>()).await { - log::error!("[COMPACT] delete files from storage failed: {}", e); + log::error!("[COMPACTOR] delete files from storage failed: {}", e); return Err(e.into()); } } @@ -132,7 +132,7 @@ pub async fn delete( ) .await { - log::error!("[COMPACT] delete files from table failed: {}", e); + log::error!("[COMPACTOR] delete files from table failed: {}", e); return Err(e.into()); } diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index c364bd7db8..81dff06995 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -56,6 +56,7 @@ use tokio::{ task::JoinHandle, }; +use super::worker::{MergeBatch, MergeSender}; use crate::{ common::infra::cluster::get_node_by_uuid, job::files::parquet::{create_tantivy_index, generate_index_on_compactor}, @@ -70,23 +71,6 @@ use crate::{ }, }; -#[derive(Clone)] -pub struct MergeBatch { - pub batch_id: usize, - pub org_id: String, - pub stream_type: StreamType, - pub stream_name: String, - pub prefix: String, - pub files: Vec, -} - -pub struct MergeResult { - pub batch_id: usize, - pub new_file: FileKey, -} - -pub type MergeSender = mpsc::Sender), anyhow::Error>>; - /// Generate merging job by stream /// 1. get offset from db /// 2. check if other node is processing @@ -180,7 +164,7 @@ pub async fn generate_job_by_stream( // generate merging job if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { return Err(anyhow::anyhow!( - "[COMPACT] add file_list_jobs failed: {}", + "[COMPACTOR] add file_list_jobs failed: {}", e )); } @@ -297,7 +281,7 @@ pub async fn generate_old_data_job_by_stream( ); if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { return Err(anyhow::anyhow!( - "[COMPACT] add file_list_jobs for old data failed: {}", + "[COMPACTOR] add file_list_jobs for old data failed: {}", e )); } @@ -449,7 +433,7 @@ pub async fn merge_by_stream( if schema == Schema::empty() { // the stream was deleted, mark the job as done if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACT] set_job_done failed: {e}"); + log::error!("[COMPACTOR] set_job_done failed: {e}"); } return Ok(()); } @@ -496,7 +480,7 @@ pub async fn merge_by_stream( if files.is_empty() { // update job status if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACT] set_job_done failed: {e}"); + log::error!("[COMPACTOR] set_job_done failed: {e}"); } return Ok(()); } @@ -606,7 +590,7 @@ pub async fn merge_by_stream( let (inner_tx, mut inner_rx) = mpsc::channel(batch_group_len); for batch in batch_groups.iter() { if let Err(e) = worker_tx.send((inner_tx.clone(), batch.clone())).await { - log::error!("[COMPACT] send batch to worker failed: {}", e); + log::error!("[COMPACTOR] send batch to worker failed: {}", e); return Err(anyhow::Error::msg("send batch to worker failed")); } } @@ -622,7 +606,7 @@ pub async fn merge_by_stream( let (batch_id, new_files) = match ret { Ok(v) => v, Err(e) => { - log::error!("[COMPACT] merge files failed: {}", e); + log::error!("[COMPACTOR] merge files failed: {}", e); last_error = Some(e); continue; } @@ -630,7 +614,7 @@ pub async fn merge_by_stream( if check_guard.contains(&batch_id) { log::warn!( - "[COMPACT] merge files for stream: [{}/{}/{}] found error files, batch_id: {} duplicate", + "[COMPACTOR] merge files for stream: [{}/{}/{}] found error files, batch_id: {} duplicate", org_id, stream_type, stream_name, @@ -667,7 +651,7 @@ pub async fn merge_by_stream( // write file list to storage if let Err(e) = write_file_list(&org_id, &events).await { - log::error!("[COMPACT] write file list failed: {}", e); + log::error!("[COMPACTOR] write file list failed: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } @@ -687,7 +671,7 @@ pub async fn merge_by_stream( // update job status if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACT] set_job_done failed: {e}"); + log::error!("[COMPACTOR] set_job_done failed: {e}"); } // metrics @@ -810,7 +794,7 @@ pub async fn merge_files( let mut fi = 0; for file in new_file_list.iter() { fi += 1; - log::info!("[COMPACT:{thread_id}:{fi}] merge small file: {}", &file.key); + log::info!("[COMPACTOR:WORKER:{thread_id}:{fi}] merge small file: {}", &file.key); let buf = file_data::get(&file.key, None).await?; let schema = read_schema_from_bytes(&buf).await?; let schema = schema.as_ref().clone().with_metadata(Default::default()); @@ -946,7 +930,7 @@ pub async fn merge_files( let id = ider::generate(); let new_file_key = format!("{prefix}/{id}{}", FILE_EXT_PARQUET); log::info!( - "[COMPACT:{thread_id}] merged {} files into a new file: {}, original_size: {}, compressed_size: {}, took: {} ms", + "[COMPACTOR:WORKER:{thread_id}] merged {} files into a new file: {}, original_size: {}, compressed_size: {}, took: {} ms", retain_file_list.len(), new_file_key, new_file_meta.original_size, @@ -1013,7 +997,7 @@ pub async fn merge_files( new_file_metas.push(new_file_meta); } log::info!( - "[COMPACT:{thread_id}] merged {} files into a new file: {:?}, original_size: {}, compressed_size: {}, took: {} ms", + "[COMPACTOR:WORKER:{thread_id}] merged {} files into a new file: {:?}, original_size: {}, compressed_size: {}, took: {} ms", retain_file_list.len(), new_file_keys, new_file_metas.iter().map(|m| m.original_size).sum::(), @@ -1155,20 +1139,23 @@ async fn write_file_list(org_id: &str, events: &[FileKey]) -> Result<(), anyhow: if !del_items.is_empty() { if let Err(e) = infra_file_list::batch_add_deleted(org_id, created_at, &del_items).await { - log::error!("[COMPACT] batch_add_deleted to db failed, retrying: {}", e); + log::error!( + "[COMPACTOR] batch_add_deleted to db failed, retrying: {}", + e + ); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } } if let Err(e) = infra_file_list::batch_add(&put_items).await { - log::error!("[COMPACT] batch_add to db failed, retrying: {}", e); + log::error!("[COMPACTOR] batch_add to db failed, retrying: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } if !del_items.is_empty() { let del_files = del_items.iter().map(|v| v.file.clone()).collect::>(); if let Err(e) = infra_file_list::batch_remove(&del_files).await { - log::error!("[COMPACT] batch_delete to db failed, retrying: {}", e); + log::error!("[COMPACTOR] batch_delete to db failed, retrying: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } @@ -1176,7 +1163,7 @@ async fn write_file_list(org_id: &str, events: &[FileKey]) -> Result<(), anyhow: // send broadcast to other nodes if get_config().memory_cache.cache_latest_files { if let Err(e) = db::file_list::broadcast::send(events, None).await { - log::error!("[COMPACT] send broadcast for file_list failed: {}", e); + log::error!("[COMPACTOR] send broadcast for file_list failed: {}", e); } } // broadcast success @@ -1301,12 +1288,12 @@ async fn cache_remote_files(files: &[FileKey]) -> Result, anyhow::Er // delete file from file list log::error!("found invalid file: {}", file_name); if let Err(e) = file_list::delete_parquet_file(&file_name, true).await { - log::error!("[COMPACT] delete from file_list err: {}", e); + log::error!("[COMPACTOR] delete from file_list err: {}", e); } Some(file_name) } else { log::warn!( - "[COMPACT] download file to cache err: {}, file: {}", + "[COMPACTOR] download file to cache err: {}, file: {}", e, file_name ); @@ -1332,7 +1319,7 @@ async fn cache_remote_files(files: &[FileKey]) -> Result, anyhow::Er } } Err(e) => { - log::error!("[COMPACT] load file task err: {}", e); + log::error!("[COMPACTOR] load file task err: {}", e); } } } diff --git a/src/service/compact/mod.rs b/src/service/compact/mod.rs index 7001ae4420..0ad729e3a1 100644 --- a/src/service/compact/mod.rs +++ b/src/service/compact/mod.rs @@ -28,7 +28,7 @@ use infra::{ }; #[cfg(feature = "enterprise")] use o2_enterprise::enterprise::common::downsampling::get_matching_downsampling_rules; -use tokio::sync::{Semaphore, mpsc}; +use tokio::sync::mpsc; use crate::{common::infra::cluster::get_node_from_consistent_hash, service::db}; @@ -37,6 +37,7 @@ pub mod flatten; pub mod merge; pub mod retention; pub mod stats; +pub mod worker; /// compactor retention run steps: pub async fn run_retention() -> Result<(), anyhow::Error> { @@ -323,9 +324,7 @@ pub async fn run_generate_downsampling_job() -> Result<(), anyhow::Error> { } /// compactor merging -pub async fn run_merge( - worker_tx: mpsc::Sender<(merge::MergeSender, merge::MergeBatch)>, -) -> Result<(), anyhow::Error> { +pub async fn run_merge(job_tx: mpsc::Sender) -> Result<(), anyhow::Error> { let cfg = get_config(); let mut jobs = infra_file_list::get_pending_jobs(&LOCAL_NODE.uuid, cfg.compact.batch_size).await?; @@ -377,7 +376,7 @@ pub async fn run_merge( if !need_done_ids.is_empty() { // set those jobs to done if let Err(e) = infra_file_list::set_job_done(&need_done_ids).await { - log::error!("[COMPACT] set_job_done failed: {}", e); + log::error!("[COMPACTOR] set_job_done failed: {}", e); } jobs.retain(|job| !need_done_ids.contains(&job.id)); } @@ -385,7 +384,7 @@ pub async fn run_merge( if !need_release_ids.is_empty() { // release those jobs if let Err(e) = infra_file_list::set_job_pending(&need_release_ids).await { - log::error!("[COMPACT] set_job_pending failed: {}", e); + log::error!("[COMPACTOR] set_job_pending failed: {}", e); } jobs.retain(|job| !need_release_ids.contains(&job.id)); } @@ -405,20 +404,18 @@ pub async fn run_merge( tokio::select! { _ = tokio::time::sleep(tokio::time::Duration::from_secs(ttl)) => {} _ = rx.recv() => { - log::debug!("[COMPACT] update_running_jobs done"); + log::debug!("[COMPACTOR] update_running_jobs done"); return; } } for id in job_ids.iter() { if let Err(e) = infra_file_list::update_running_jobs(*id).await { - log::error!("[COMPACT] update_job_status failed: {}", e); + log::error!("[COMPACTOR] update_job_status failed: {}", e); } } } }); - let mut tasks = Vec::with_capacity(jobs.len()); - let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.file_merge_thread_num)); for job in jobs { if job.offsets == 0 { log::error!("[COMPACTOR] merge job offset error: {}", job.offsets); @@ -442,35 +439,24 @@ pub async fn run_merge( continue; } - let org_id = org_id.clone(); - let permit = semaphore.clone().acquire_owned().await.unwrap(); - let worker_tx = worker_tx.clone(); - let task = tokio::task::spawn(async move { - if let Err(e) = merge::merge_by_stream( - worker_tx, - &org_id, + if let Err(e) = job_tx + .send(worker::MergeJob { + org_id: org_id.to_string(), stream_type, - &stream_name, - job.id, - job.offsets, - ) + stream_name: stream_name.to_string(), + job_id: job.id, + offset: job.offsets, + }) .await - { - log::error!( - "[COMPACTOR] merge_by_stream [{}/{}/{}] error: {}", - org_id, - stream_type, - stream_name, - e - ); - } - drop(permit); - }); - tasks.push(task); - } - - for task in tasks { - task.await?; + { + log::error!( + "[COMPACTOR] send merge job to worker failed [{}/{}/{}] error: {}", + org_id, + stream_type, + stream_name, + e + ); + } } Ok(()) diff --git a/src/service/compact/retention.rs b/src/service/compact/retention.rs index 5116b6ef2c..d736c68e50 100644 --- a/src/service/compact/retention.rs +++ b/src/service/compact/retention.rs @@ -52,7 +52,7 @@ fn generate_time_ranges_for_deletion( TimeRange::flatten_overlapping_ranges(extended_retention_ranges).into_iter(); log::debug!( - "[COMPACT] populate_time_ranges_for_deletion exclude_range: {}, original_time_range: {}", + "[COMPACTOR] populate_time_ranges_for_deletion exclude_range: {}, original_time_range: {}", extended_retention_ranges_iter.clone().join(", "), original_time_range.clone() ); @@ -129,7 +129,7 @@ fn generate_time_ranges_for_deletion( } log::debug!( - "[COMPACT] populate_time_ranges_for_deletion time_ranges_for_deletion: {}", + "[COMPACTOR] populate_time_ranges_for_deletion time_ranges_for_deletion: {}", time_ranges_for_deletion.iter().join(", ") ); @@ -157,7 +157,7 @@ pub async fn delete_by_stream( } log::debug!( - "[COMPACT] delete_by_stream {}/{}/{}/{},{}", + "[COMPACTOR] delete_by_stream {}/{}/{}/{},{}", org_id, stream_type, stream_name, @@ -179,17 +179,21 @@ pub async fn delete_by_stream( lifecycle_end.timestamp_micros(), ); - let final_deletion_time_ranges = generate_time_ranges_for_deletion( - extended_retentions.to_vec(), - original_deletion_time_range, - last_retained_time, - ); - - log::debug!( - "[COMPACT] extended_retentions: {}, final_deletion_time_ranges: {}", - extended_retentions.iter().join(", "), - final_deletion_time_ranges.iter().join(", ") - ); + let final_deletion_time_ranges = if extended_retentions.is_empty() { + vec![original_deletion_time_range] + } else { + let ranges = generate_time_ranges_for_deletion( + extended_retentions.to_vec(), + original_deletion_time_range, + last_retained_time, + ); + log::debug!( + "[COMPACTOR] extended_retentions: {}, final_deletion_time_ranges: {}", + extended_retentions.iter().join(", "), + ranges.iter().join(", ") + ); + ranges + }; let job_nos = final_deletion_time_ranges.len(); @@ -207,7 +211,7 @@ pub async fn delete_by_stream( } log::debug!( - "[COMPACT] delete_by_stream {}/{}/{}/{},{}", + "[COMPACTOR] delete_by_stream {}/{}/{}/{},{}", org_id, stream_type, stream_name, @@ -236,7 +240,7 @@ pub async fn delete_all( let locker = dist_lock::lock(&lock_key, 0).await?; let node = db::compact::retention::get_stream(org_id, stream_type, stream_name, None).await; if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { - log::warn!("[COMPACT] stream {org_id}/{stream_type}/{stream_name} is deleting by {node}"); + log::warn!("[COMPACTOR] stream {org_id}/{stream_type}/{stream_name} is deleting by {node}"); dist_lock::unlock(&locker).await?; return Ok(()); // not this node, just skip } @@ -284,7 +288,7 @@ pub async fn delete_all( if cfg.compact.data_retention_history { // only store the file_list into history, don't delete files if let Err(e) = infra_file_list::batch_add_history(&files).await { - log::error!("[COMPACT] file_list batch_add_history failed: {}", e); + log::error!("[COMPACTOR] file_list batch_add_history failed: {}", e); return Err(e.into()); } } @@ -324,7 +328,7 @@ pub async fn delete_by_date( .await; if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { log::warn!( - "[COMPACT] stream {org_id}/{stream_type}/{stream_name}/{:?} is deleting by {node}", + "[COMPACTOR] stream {org_id}/{stream_type}/{stream_name}/{:?} is deleting by {node}", date_range ); dist_lock::unlock(&locker).await?; @@ -398,7 +402,7 @@ pub async fn delete_by_date( if cfg.compact.data_retention_history { // only store the file_list into history, don't delete files if let Err(e) = infra_file_list::batch_add_history(&files).await { - log::error!("[COMPACT] file_list batch_add_history failed: {}", e); + log::error!("[COMPACTOR] file_list batch_add_history failed: {}", e); return Err(e.into()); } } @@ -521,19 +525,22 @@ async fn write_file_list( for _ in 0..5 { if let Err(e) = infra_file_list::batch_add_deleted(org_id, created_at, &del_items).await { - log::error!("[COMPACT] batch_add_deleted to db failed, retrying: {}", e); + log::error!( + "[COMPACTOR] batch_add_deleted to db failed, retrying: {}", + e + ); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } if let Err(e) = infra_file_list::batch_add(&put_items).await { - log::error!("[COMPACT] batch_add to db failed, retrying: {}", e); + log::error!("[COMPACTOR] batch_add to db failed, retrying: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } if !del_items.is_empty() { let del_files = del_items.iter().map(|v| v.file.clone()).collect::>(); if let Err(e) = infra_file_list::batch_remove(&del_files).await { - log::error!("[COMPACT] batch_delete to db failed, retrying: {}", e); + log::error!("[COMPACTOR] batch_delete to db failed, retrying: {}", e); tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; continue; } @@ -542,7 +549,7 @@ async fn write_file_list( break; } if !success { - return Err(anyhow::anyhow!("[COMPACT] batch_write to db failed")); + return Err(anyhow::anyhow!("[COMPACTOR] batch_write to db failed")); } } Ok(()) diff --git a/src/service/compact/worker.rs b/src/service/compact/worker.rs new file mode 100644 index 0000000000..886117c25f --- /dev/null +++ b/src/service/compact/worker.rs @@ -0,0 +1,222 @@ +// Copyright 2025 OpenObserve 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 std::sync::Arc; + +use config::{ + cluster::is_offline, + meta::stream::{FileKey, StreamType}, +}; +use tokio::sync::{Mutex, mpsc}; + +#[derive(Clone)] +pub struct MergeBatch { + pub batch_id: usize, + pub org_id: String, + pub stream_type: StreamType, + pub stream_name: String, + pub prefix: String, + pub files: Vec, +} + +pub struct MergeResult { + pub batch_id: usize, + pub new_file: FileKey, +} + +pub type MergeSender = mpsc::Sender), anyhow::Error>>; + +pub struct MergeJob { + pub org_id: String, + pub stream_type: StreamType, + pub stream_name: String, + pub job_id: i64, + pub offset: i64, +} + +/// JobScheduler is a worker that processes jobs +pub struct JobScheduler { + num: usize, + rx: Arc>>, + tx: mpsc::Sender, + worker_tx: mpsc::Sender<(MergeSender, MergeBatch)>, +} + +impl JobScheduler { + pub fn new(num: usize, worker_tx: mpsc::Sender<(MergeSender, MergeBatch)>) -> Self { + let (tx, rx) = mpsc::channel::(num); + let rx = Arc::new(Mutex::new(rx)); + Self { + num, + rx, + tx, + worker_tx, + } + } + + pub fn tx(&self) -> mpsc::Sender { + self.tx.clone() + } + + pub fn run(&mut self) -> Result<(), anyhow::Error> { + let cfg = config::get_config(); + let ttl = std::cmp::max(60, cfg.compact.job_run_timeout / 4) as u64; + for thread_id in 0..self.num { + let rx = self.rx.clone(); + let worker_tx = self.worker_tx.clone(); + tokio::spawn(async move { + loop { + if is_offline() { + break; + } + let ret = rx.lock().await.recv().await; + match ret { + None => { + log::debug!( + "[COMPACTOR:SCHEDULER:{thread_id}] Receiving job channel is closed" + ); + break; + } + Some(job) => { + let (_tx, mut rx) = mpsc::channel::<()>(1); + tokio::task::spawn(async move { + loop { + tokio::select! { + _ = tokio::time::sleep(tokio::time::Duration::from_secs(ttl)) => {} + _ = rx.recv() => { + log::debug!("[COMPACTOR:SCHEDULER:{thread_id}] update_running_jobs[{}] done", job.job_id); + return; + } + } + if let Err(e) = + infra::file_list::update_running_jobs(job.job_id).await + { + log::error!( + "[COMPACTOR:SCHEDULER:{thread_id}] update_job_status[{}] failed: {}", + job.job_id, + e + ); + } + } + }); + if let Err(e) = super::merge::merge_by_stream( + worker_tx.clone(), + &job.org_id, + job.stream_type, + &job.stream_name, + job.job_id, + job.offset, + ) + .await + { + log::error!( + "[COMPACTOR:SCHEDULER:{thread_id}] merge_by_stream [{}/{}/{}] error: {}", + job.org_id, + job.stream_type, + job.stream_name, + e + ); + } + } + } + } + }); + } + Ok(()) + } +} + +/// MergeWorker is a worker that merges files +pub struct MergeWorker { + num: usize, + rx: Arc>>, + tx: mpsc::Sender<(MergeSender, MergeBatch)>, +} + +impl MergeWorker { + pub fn new(num: usize) -> Self { + let (tx, rx) = mpsc::channel::<(MergeSender, MergeBatch)>(num * 2); + let rx = Arc::new(Mutex::new(rx)); + Self { num, rx, tx } + } + + pub fn tx(&self) -> mpsc::Sender<(MergeSender, MergeBatch)> { + self.tx.clone() + } + + pub fn run(&mut self) -> Result<(), anyhow::Error> { + for thread_id in 0..self.num { + let rx = self.rx.clone(); + tokio::spawn(async move { + loop { + if is_offline() { + break; + } + let ret = rx.lock().await.recv().await; + match ret { + None => { + log::debug!( + "[COMPACTOR:WORKER:{thread_id}] Receiving files channel is closed" + ); + break; + } + Some((tx, msg)) => { + match super::merge::merge_files( + thread_id, + &msg.org_id, + msg.stream_type, + &msg.stream_name, + &msg.prefix, + &msg.files, + ) + .await + { + Ok((file, meta, _)) => { + let mut new_file_keys = Vec::with_capacity(file.len()); + for (file, meta) in file.into_iter().zip(meta.into_iter()) { + new_file_keys.push(FileKey::new(file, meta, false)); + } + if let Err(e) = tx.send(Ok((msg.batch_id, new_file_keys))).await + { + log::error!( + "[COMPACTOR:WORKER:{thread_id}] Error sending file to merge_job: {}", + e + ); + } + } + Err(e) => { + log::error!( + "[COMPACTOR:WORKER:{thread_id}] Error merging files: stream: {}/{}/{}, err: {}", + msg.org_id, + msg.stream_type, + msg.stream_name, + e + ); + if let Err(e) = tx.send(Err(e)).await { + log::error!( + "[COMPACTOR:WORKER:{thread_id}] Error sending error to merge_job: {}", + e + ); + } + } + } + } + } + } + }); + } + Ok(()) + } +}