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

fix: do not move file when db not able to write (#10651)

Browse Source
This commit is contained in:
Huaijin
2026-02-25 21:08:47 +08:00
committed by GitHub
parent bb7103930a
commit 9136398d3a
13 changed files with 68 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/utils/file.rs
+2 -3
View File
@@ -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 {
src/handler/http/auth/validator.rs
+1 -3
View File
@@ -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())
}
src/infra/src/cache/file_data/disk.rs
Vendored
+4 -7
View File
@@ -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;
}
src/infra/src/db/nats.rs
+5 -5
View File
@@ -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(())
src/infra/src/file_list/mod.rs
+5
View File
@@ -48,6 +48,7 @@ pub fn connect_local_cache() -> Box<dyn FileList> {
#[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<i64>;
@@ -202,6 +203,10 @@ pub trait FileList: Sync + Send + 'static {
) -> Result<StreamStats>;
}
pub async fn health_check() -> Result<()> {
CLIENT.health_check().await
}
pub async fn create_table() -> Result<()> {
CLIENT.create_table().await
}
src/infra/src/file_list/postgres.rs
+25 -2
View File
@@ -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)
src/infra/src/file_list/sqlite.rs
+5 -1
View File
@@ -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
}
src/infra/src/storage/remote.rs
+1 -1
View File
@@ -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
src/infra/src/table/sessions.rs
+2 -6
View File
@@ -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())
}
}
src/job/files/parquet.rs
+9
View File
@@ -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}");
src/job/flatten_compactor.rs
+2 -3
View File
@@ -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
);
}
},
src/router/http/mod.rs
+4 -7
View File
@@ -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 (
src/service/schema.rs
+3 -19
View File
@@ -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()
);