fix: prevent history table's log loss during node startup (#18566)

* fix: prevent history table's log loss during node startup before operator initialization

* thanks copilot review

Author: dqhl76

.github/actions/test_logs/action.yml

@@ -25,7 +25,6 @@ runs:
         export AWS_EC2_METADATA_DISABLED=true
 
         aws --endpoint-url http://127.0.0.1:9900/ s3 mb s3://testbucket
-        aws --endpoint-url http://127.0.0.1:9900/ s3 cp tests/data s3://testbucket/data  --recursive --no-progress
 
     - name: Run logging Tests
       shell: bash

src/common/base/tests/it/time_series/profile.rs

@@ -139,12 +139,26 @@ fn test_finish_flush() {
     profile.record(TimeSeriesProfileName::OutputRows, 1);
 
     let batch = profile.flush(true, &mut 4);
-
     assert_eq!(batch.len(), 2);
-    // [[timestamp, 1000, 1]]
-    assert_eq!(batch[0][0].len(), 3);
-    // [[timestamp, 2000, 2]]
-    assert_eq!(batch[1][0].len(), 3);
+    // sleep(1) is not exactly 1 second sleep time, but **at least** 1 second.
+    if batch[0].len() == 1 {
+        // 99.9%: [[timestamp, 1000, 1]]
+        assert_eq!(batch[0][0][1], 1000);
+        assert_eq!(batch[0][0][2], 1);
+    } else {
+        // 0.1%: [[timestamp, 1000], [timestamp+2, 1]]
+        assert_eq!(batch[0][0][1], 1000);
+        assert_eq!(batch[0][1][1], 1);
+    }
+    if batch[1].len() == 1 {
+        // 99.9%: [[timestamp, 2000, 2]]
+        assert_eq!(batch[1][0][1], 2000);
+        assert_eq!(batch[1][0][2], 2);
+    } else {
+        // 0.1%: [[timestamp, 2000], [timestamp+2, 2]]
+        assert_eq!(batch[1][0][1], 2000);
+        assert_eq!(batch[1][1][1], 2);
+    }
 }
 
 #[test]

src/common/tracing/src/init.rs

@@ -14,6 +14,8 @@
 
 use std::borrow::Cow;
 use std::collections::BTreeMap;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use databend_common_base::base::tokio;
@@ -47,6 +49,7 @@ const HEADER_TRACE_PARENT: &str = "traceparent";
 pub struct GlobalLogger {
     _drop_guards: Vec<Box<dyn Send + Sync + 'static>>,
     pub remote_log_operator: RwLock<Option<Operator>>,
+    pub ready: AtomicBool,
 }
 
 impl GlobalLogger {
@@ -59,10 +62,19 @@ impl GlobalLogger {
         let instance = Arc::new(Self {
             _drop_guards,
             remote_log_operator,
+            ready: AtomicBool::new(false),
         });
         GlobalInstance::set(instance);
     }
 
+    pub fn dummy() -> Arc<GlobalLogger> {
+        Arc::new(Self {
+            _drop_guards: Vec::new(),
+            remote_log_operator: RwLock::new(None),
+            ready: AtomicBool::new(true),
+        })
+    }
+
     pub fn instance() -> Arc<GlobalLogger> {
         GlobalInstance::get()
     }
@@ -80,6 +92,7 @@ impl GlobalLogger {
     pub async fn set_operator(&self, operator: Operator) {
         let mut remote_log_operator = self.remote_log_operator.write().await;
         *remote_log_operator = Some(operator);
+        self.ready.store(true, Ordering::SeqCst);
     }
 }
 

src/common/tracing/src/remote_log.rs

@@ -289,6 +289,12 @@ impl LogBuffer {
                 Ordering::SeqCst,
             ) {
                 Ok(_) => {
+                    // Ensure operator is ready before collecting logs to prevent data loss during startup.
+                    // If operator is not ready, skip collection this round and keep log entries in the queue
+                    // MAX_BUFFER_SIZE limit can help to prevent unbounded queue growth.
+                    if !GlobalLogger::instance().ready.load(Ordering::SeqCst) {
+                        break;
+                    }
                     self.collect()?;
                     break;
                 }

src/common/tracing/tests/it/remote_log.rs

@@ -18,9 +18,11 @@ use std::time::Duration;
 
 use async_channel::bounded;
 use databend_common_base::base::tokio;
+use databend_common_base::base::GlobalInstance;
 use databend_common_exception::Result;
 use databend_common_tracing::convert_to_batch;
 use databend_common_tracing::Config;
+use databend_common_tracing::GlobalLogger;
 use databend_common_tracing::LogMessage;
 use databend_common_tracing::RemoteLog;
 use databend_common_tracing::RemoteLogBuffer;
@@ -103,17 +105,18 @@ async fn test_buffer_flush_with_buffer_limit() -> Result<()> {
     Ok(())
 }
 
-#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-async fn test_buffer_flush_with_buffer_interval() -> Result<()> {
+#[test]
+fn test_buffer_flush_with_buffer_interval() -> Result<()> {
+    init_global_logger()?;
     let (tx, rx) = bounded(10);
     let interval = Duration::from_secs(1).as_micros() as u64;
     let buffer = Arc::new(RemoteLogBuffer::new(tx.clone(), interval));
     for _i in 0..5 {
         buffer.log(get_remote_log_elements())?
     }
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    std::thread::sleep(Duration::from_secs(1));
     buffer.log(get_remote_log_elements())?;
-    let res = rx.recv().await.unwrap();
+    let res = rx.recv_blocking().unwrap();
     if let LogMessage::Flush(elements) = res {
         assert_eq!(elements.len(), 6);
     }
@@ -152,3 +155,11 @@ async fn test_do_flush() -> Result<()> {
 
     Ok(())
 }
+
+fn init_global_logger() -> Result<()> {
+    let thread = std::thread::current();
+    GlobalInstance::init_testing(thread.name().unwrap());
+    let instance = GlobalLogger::dummy();
+    GlobalInstance::set(instance);
+    Ok(())
+}

tests/logging/check_logs_table.sh

@@ -29,7 +29,7 @@ function check_query_log() {
   fi
 }
 
-response=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d '{"sql": "set global timezone='Asia/Shanghai'"}')
+response=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d "{\"sql\": \"set global timezone='Asia/Shanghai'\"}")
 
 # Execute the initial query and get the query_id
 response=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d '{"sql": "select 123"}')

tests/logging/history_table.toml

@@ -1,5 +1,6 @@
 [log.history]
 on = true
+interval = 1
 
 [[log.history.tables]]
 table_name = "query_history"
@@ -11,4 +12,4 @@ table_name = "profile_history"
 table_name = "login_history"
 
 [[log.history.tables]]
-table_name = "access_history"
+table_name = "access_history"

tests/logging/test-history-tables.sh

@@ -25,7 +25,7 @@ echo "Cleaning up previous runs"
 killall -9 databend-query || true
 killall -9 databend-meta || true
 killall -9 vector || true
-rm -rf .databend
+rm -rf ./.databend
 
 echo "Starting Databend Query cluster with 2 nodes enable history tables"
 
@@ -98,6 +98,18 @@ else
     echo "✓ meta node logs are collected as expected"
 fi
 
+startup_check_response=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d "{\"sql\": \"select count(*) from system_history.log_history where message like 'Ready for connections after%'\"}")
+startup_check_response_data=$(echo "$startup_check_response" | jq -r '.data')
+if [[ "$startup_check_response_data" != *"2"* ]]; then
+    echo "ERROR: startup check failed"
+    debug_info=$(curl -s -u root: -XPOST "http://localhost:8000/v1/query" -H 'Content-Type: application/json' -d "{\"sql\": \"select * from system_history.log_history where message like 'Ready for connections after%'\"}")
+    echo "$debug_info"
+    echo "startup_check_response_data: $startup_check_response_data"
+    exit 1
+else
+    echo "✓ node startup test passed"
+fi
+
 # **Internal -> External**: should reset
 
 echo "Add a node with external history table enabled"