starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 08:09:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: optimize compactor scheduler with multiple mpsc (#6269)

Browse Source
This commit is contained in:
Hengfei Yang
2025-03-15 15:03:28 +08:00
committed by GitHub
parent 26feda0628
commit 3977c9be42
14 changed files with 409 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -25
View File
@@ -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
src/config/src/config.rs
+17 -15
View File
@@ -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
src/handler/http/request/websocket/mod.rs
+4 -2
View File
@@ -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(())
}
src/handler/http/request/websocket/utils.rs
+2 -2
View File
@@ -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())
src/infra/src/storage/local.rs
+11 -2
View File
@@ -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();
src/infra/src/storage/remote.rs
+17 -4
View File
@@ -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();
src/job/compactor.rs
+36 -92
View File
@@ -13,20 +13,15 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
use std::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<compact::worker::MergeJob>) -> 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;
}
src/job/files/parquet.rs
+1 -1
View File
@@ -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::<Vec<_>>(),
start.elapsed().as_millis(),
src/main.rs
+15 -9
View File
@@ -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 {
src/service/compact/deleted.rs
+5 -5
View File
@@ -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::<Vec<_>>()).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());
}
src/service/compact/merge.rs
+23 -36
View File
@@ -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<FileKey>,
}
pub struct MergeResult {
pub batch_id: usize,
pub new_file: FileKey,
}
pub type MergeSender = mpsc::Sender<Result<(usize, Vec<FileKey>), 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::<i64>(),
@@ -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::<Vec<_>>();
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<Vec<String>, 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<Vec<String>, anyhow::Er
}
}
Err(e) => {
log::error!("[COMPACT] load file task err: {}", e);
log::error!("[COMPACTOR] load file task err: {}", e);
}
}
}
src/service/compact/mod.rs
+23 -37
View File
@@ -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<worker::MergeJob>) -> 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(())
src/service/compact/retention.rs
+30 -23
View File
@@ -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::<Vec<_>>();
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(())
src/service/compact/worker.rs
+222
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<FileKey>,
}
pub struct MergeResult {
pub batch_id: usize,
pub new_file: FileKey,
}
pub type MergeSender = mpsc::Sender<Result<(usize, Vec<FileKey>), 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<Mutex<mpsc::Receiver<MergeJob>>>,
tx: mpsc::Sender<MergeJob>,
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::<MergeJob>(num);
let rx = Arc::new(Mutex::new(rx));
Self {
num,
rx,
tx,
worker_tx,
}
}
pub fn tx(&self) -> mpsc::Sender<MergeJob> {
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<Mutex<mpsc::Receiver<(MergeSender, MergeBatch)>>>,
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(())
}
}