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(())
+ }
+}