From 9136398d3ae0741b8fb8ac79edcd83de0e6fc8a2 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Wed, 25 Feb 2026 21:08:47 +0800 Subject: [PATCH] fix: do not move file when db not able to write (#10651) --- src/config/src/utils/file.rs | 5 ++--- src/handler/http/auth/validator.rs | 4 +--- src/infra/src/cache/file_data/disk.rs | 11 ++++------- src/infra/src/db/nats.rs | 10 +++++----- src/infra/src/file_list/mod.rs | 5 +++++ src/infra/src/file_list/postgres.rs | 27 +++++++++++++++++++++++++-- src/infra/src/file_list/sqlite.rs | 6 +++++- src/infra/src/storage/remote.rs | 2 +- src/infra/src/table/sessions.rs | 8 ++------ src/job/files/parquet.rs | 9 +++++++++ src/job/flatten_compactor.rs | 5 ++--- src/router/http/mod.rs | 11 ++++------- src/service/schema.rs | 22 +++------------------- 13 files changed, 68 insertions(+), 57 deletions(-) diff --git a/src/config/src/utils/file.rs b/src/config/src/utils/file.rs index 9133597561..19261384b6 100644 --- a/src/config/src/utils/file.rs +++ b/src/config/src/utils/file.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -133,9 +133,8 @@ pub async fn scan_files_with_channel( let mut files = Vec::with_capacity(std::cmp::max(16, limit)); let dir = std::fs::read_dir(root).map_err(|e| { std::io::Error::other(format!( - "Error reading directory: {}, err: {}", + "Error reading directory: {}, err: {e}", root.display(), - e )) })?; for entry in dir { diff --git a/src/handler/http/auth/validator.rs b/src/handler/http/auth/validator.rs index 21dd165e57..341a8b3545 100644 --- a/src/handler/http/auth/validator.rs +++ b/src/handler/http/auth/validator.rs @@ -1198,10 +1198,8 @@ fn _handle_auth_failure_for_redirect(req: &Request, error: &AuthError) -> AuthEr .with_query_param("short_url", &full_url) .build(); log::warn!( - "Authentication failed for path: {}, err: {:?}, {}", + "Authentication failed for path: {}, err: {error:?}, {redirect_http}", req.uri().path(), - error, - &redirect_http, ); AuthError::Unauthorized(redirect_http.to_string()) } diff --git a/src/infra/src/cache/file_data/disk.rs b/src/infra/src/cache/file_data/disk.rs index a7574f2d90..0831e5c8b5 100644 --- a/src/infra/src/cache/file_data/disk.rs +++ b/src/infra/src/cache/file_data/disk.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -298,16 +298,13 @@ impl FileData { // delete file from local disk let file_path = self.get_file_path(key.as_str()); log::debug!( - "[CacheType:{}] File disk cache gc remove file: {}", + "[CacheType:{}] File disk cache gc remove file: {key}", self.file_type, - key ); if let Err(e) = tokio::fs::remove_file(&file_path).await { log::error!( - "[CacheType:{}] File disk cache gc remove file: {}, error: {}", + "[CacheType:{}] File disk cache gc remove file: {file_path}, error: {e}", self.file_type, - file_path, - e ); } @@ -731,7 +728,7 @@ async fn load(root_dir: &PathBuf, scan_dir: &PathBuf) -> Result<(), anyhow::Erro fp.display() ); if let Err(e) = tokio::fs::remove_file(&fp).await { - log::warn!("Failed to remove tmp file: {}, error: {}", fp.display(), e); + log::warn!("Failed to remove tmp file: {}, error: {e}", fp.display()); } continue; } diff --git a/src/infra/src/db/nats.rs b/src/infra/src/db/nats.rs index cc4b01409d..2d0d6a5efe 100644 --- a/src/infra/src/db/nats.rs +++ b/src/infra/src/db/nats.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -721,8 +721,8 @@ impl Locker { ))); } else { return Err(Error::Message(format!( - "nats lock for key: {}, error: {}", - self.key, err + "nats lock for key: {}, error: {err}", + self.key ))); } } @@ -762,10 +762,10 @@ impl Locker { } self.state.store(2, Ordering::SeqCst); if let Err(e) = self.tx.as_ref().unwrap().send(()).await { - log::error!("nats unlock sender for key: {}, error: {}", self.key, e); + log::error!("nats unlock sender for key: {}, error: {e}", self.key); } if let Err(e) = bucket.purge(&key).await { - log::error!("nats unlock for key: {}, error: {}", self.key, e); + log::error!("nats unlock for key: {}, error: {e}", self.key); return Err(Error::Message("nats unlock error".to_string())); }; Ok(()) diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index c467f234b2..bbbafec722 100644 --- a/src/infra/src/file_list/mod.rs +++ b/src/infra/src/file_list/mod.rs @@ -48,6 +48,7 @@ pub fn connect_local_cache() -> Box { #[async_trait] pub trait FileList: Sync + Send + 'static { + async fn health_check(&self) -> Result<()>; async fn create_table(&self) -> Result<()>; async fn create_table_index(&self) -> Result<()>; async fn add(&self, account: &str, file: &str, meta: &FileMeta) -> Result; @@ -202,6 +203,10 @@ pub trait FileList: Sync + Send + 'static { ) -> Result; } +pub async fn health_check() -> Result<()> { + CLIENT.health_check().await +} + pub async fn create_table() -> Result<()> { CLIENT.create_table().await } diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 9245ed9ba8..a066639788 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -68,6 +68,25 @@ impl Default for PostgresFileList { #[async_trait] impl super::FileList for PostgresFileList { + async fn health_check(&self) -> Result<()> { + let pool = CLIENT.clone(); + let is_writable: bool = sqlx::query_scalar( + r#"SELECT NOT pg_is_in_recovery() + AND current_setting('transaction_read_only')::bool = false + AND current_setting('default_transaction_read_only')::bool = false"#, + ) + .fetch_one(&pool) + .await + .map_err(|e| Error::Message(format!("PostgreSQL health check failed: {e}")))?; + if !is_writable { + return Err(Error::Message( + "PostgreSQL health check: database is not writable (in recovery or read-only mode)" + .to_string(), + )); + } + Ok(()) + } + async fn create_table(&self) -> Result<()> { create_table().await } @@ -135,8 +154,12 @@ impl super::FileList for PostgresFileList { DB_QUERY_NUMS .with_label_values(&["insert", "file_list"]) .inc(); - if let Err(e) = sqlx::query(r#"INSERT INTO file_list (account, org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) ON CONFLICT DO NOTHING;"# + if let Err(e) = sqlx::query( + r#"INSERT INTO file_list + (account, org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened, updated_at) + VALUES + ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) + ON CONFLICT DO NOTHING"# ) .bind(&dump_file.account) .bind(org_id) diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index fcdd821fea..6a0df9b88d 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -54,6 +54,10 @@ impl Default for SqliteFileList { #[async_trait] impl super::FileList for SqliteFileList { + async fn health_check(&self) -> Result<()> { + Ok(()) + } + async fn create_table(&self) -> Result<()> { create_table().await } diff --git a/src/infra/src/storage/remote.rs b/src/infra/src/storage/remote.rs index c2cd139581..2d422411d6 100644 --- a/src/infra/src/storage/remote.rs +++ b/src/infra/src/storage/remote.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 diff --git a/src/infra/src/table/sessions.rs b/src/infra/src/table/sessions.rs index ba04d1787b..eef86a6c96 100644 --- a/src/infra/src/table/sessions.rs +++ b/src/infra/src/table/sessions.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -67,11 +67,7 @@ pub async fn set_with_expiry( match result { Ok(_) => Ok(()), Err(e) => { - log::error!( - "[DB] Failed to insert/update session: id={}, error={}", - session_id, - e - ); + log::error!("[DB] Failed to insert/update session: id={session_id}, error={e}"); Err(e.into()) } } diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index fefc0f45b6..70c2be881f 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -129,6 +129,15 @@ pub async fn run() -> Result<(), anyhow::Error> { .await; } + // check if db is available, skip this iteration to avoid generating + // orphaned files in object store when db is down + if let Err(e) = infra::file_list::health_check().await { + log::error!( + "[INGESTER:JOB] DB health check failed, skip uploading to avoid orphaned files in object store: {e}" + ); + continue; + } + // check pending delete files if let Err(e) = scan_pending_delete_files().await { log::error!("[INGESTER:JOB] Error scan pending delete files: {e}"); diff --git a/src/job/flatten_compactor.rs b/src/job/flatten_compactor.rs index bb347b8c5c..bda5a94b8b 100644 --- a/src/job/flatten_compactor.rs +++ b/src/job/flatten_compactor.rs @@ -1,4 +1,4 @@ -// Copyright 2025 OpenObserve Inc. +// Copyright 2026 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 @@ -47,9 +47,8 @@ pub async fn run() -> Result<(), anyhow::Error> { Ok(_) => {} Err(e) => { log::error!( - "[FLATTEN_COMPACTOR:JOB] Error generate flatten file: {}, err: {}", + "[FLATTEN_COMPACTOR:JOB] Error generate flatten file: {}, err: {e}", file.key, - e ); } }, diff --git a/src/router/http/mod.rs b/src/router/http/mod.rs index a3bed70961..1fc00c9a6b 100644 --- a/src/router/http/mod.rs +++ b/src/router/http/mod.rs @@ -470,10 +470,9 @@ async fn send_and_respond( Ok(r) => r, Err(e) => { log::error!( - "proxy request failed: {} -> {}, error: {:?}, took: {} ms", + "proxy request failed: {} -> {}, error: {e:?}, took: {} ms", target.path, target.node_addr, - e, start.elapsed().as_millis() ); return ( @@ -521,14 +520,13 @@ async fn send_and_respond( Ok(bytes) => Ok(bytes), Err(e) => { log::error!( - "proxy streaming error: {} -> {}, error: {:?}", + "proxy streaming error: {} -> {}, error: {e:?}", path_for_log, node_for_log, - e ); let error_event = serde_json::json!({ "type": "error", - "error": format!("Stream interrupted: {}", e) + "error": format!("Stream interrupted: {e}") }); Ok(Bytes::from(format!("data: {}\n\n", error_event))) } @@ -547,10 +545,9 @@ async fn send_and_respond( Ok(b) => b, Err(e) => { log::error!( - "proxy response failed: {} -> {}, error: {:?}, took: {} ms", + "proxy response failed: {} -> {}, error: {e:?}, took: {} ms", target.path, target.node_addr, - e, start.elapsed().as_millis() ); return ( diff --git a/src/service/schema.rs b/src/service/schema.rs index dd73a5f4a1..a76fd8611b 100644 --- a/src/service/schema.rs +++ b/src/service/schema.rs @@ -301,14 +301,8 @@ pub(crate) async fn handle_diff_schema( { Err(e) => { log::error!( - "handle_diff_schema [{}/{}/{}] with hash {}, start_dt {}, error: {}, retrying...{}", - org_id, - stream_type, - stream_name, + "handle_diff_schema [{org_id}/{stream_type}/{stream_name}] with hash {}, start_dt {record_ts}, error: {e}, retrying...{retries}", inferred_schema.hash_key(), - record_ts, - e, - retries ); err = Some(e); retries += 1; @@ -323,14 +317,8 @@ pub(crate) async fn handle_diff_schema( } if let Some(e) = err { log::error!( - "handle_diff_schema [{}/{}/{}] with hash {}, start_dt {}, abort after retry {} times, error: {}", - org_id, - stream_type, - stream_name, + "handle_diff_schema [{org_id}/{stream_type}/{stream_name}] with hash {}, start_dt {record_ts}, abort after retry {retries} times, error: {e}", inferred_schema.hash_key(), - record_ts, - retries, - e ); return Err(e); } @@ -475,11 +463,7 @@ pub(crate) async fn handle_diff_schema( stream_schema_map.insert(stream_name.to_string(), final_schema); log::debug!( - "handle_diff_schema end for [{}/{}/{}] start_dt: {}, elapsed: {} ms", - org_id, - stream_type, - stream_name, - record_ts, + "handle_diff_schema end for [{org_id}/{stream_type}/{stream_name}] start_dt: {record_ts}, elapsed: {} ms", start.elapsed().as_millis() );