fix: add debug for task panic & etcd lock error (#3040)

This commit is contained in:
Ashish Kolhe
2024-03-19 12:22:33 +05:30
committed by GitHub
parent 5300cb1f22
commit 0d86eabdcb
2 changed files with 19 additions and 3 deletions
+18 -2
View File
@@ -15,6 +15,7 @@
use std::{collections::HashMap, io::Cursor, path::Path, sync::Arc, time::UNIX_EPOCH};
use anyhow::Context;
use arrow::{
array::{ArrayRef, BooleanArray, Int64Array, StringArray},
record_batch::RecordBatch,
@@ -135,7 +136,11 @@ pub async fn move_files_to_storage() -> Result<(), anyhow::Error> {
}
let wal_dir = wal_dir.clone();
let permit = semaphore.clone().acquire_owned().await.unwrap();
let permit = semaphore
.clone()
.acquire_owned()
.await
.context("[INGESTER:JOB] Failed to acquire semaphore for file uploading")?;
let task: JoinHandle<Result<(), anyhow::Error>> = tokio::task::spawn(async move {
// sort by created time
let mut files_with_size = files_with_size.to_owned();
@@ -175,7 +180,18 @@ pub async fn move_files_to_storage() -> Result<(), anyhow::Error> {
}
// get latest schema
let latest_schema = db::schema::get(&org_id, &stream_name, stream_type).await?;
let latest_schema = db::schema::get(&org_id, &stream_name, stream_type)
.await
.map_err(|e| {
log::error!(
"[INGESTER:JOB] Failed to get latest schema for stream [{}/{}/{}]: {}",
&org_id,
stream_type,
&stream_name,
e
);
e
})?;
// start merge files and upload to s3
loop {
+1 -1
View File
@@ -596,7 +596,7 @@ async fn handle_diff_schema_cluster_mode(
let locker = match dist_lock::lock(&lock_key, 0).await {
Ok(v) => v,
Err(e) => {
log::error!("dist_lock for handle diff schema error: {}", e);
log::error!("[SCHEMA] dist_lock schema {} error : {}", lock_key, e);
return None;
}
};