feat(meta-service): add timing monitoring for Raft-Log IO (#18509)

Enhance `IODesc` with timing fields to track operation phases:
- Add `start_time`, `flush_time`, `flush_done_time`, `done_time` fields
- Update Callback constructors to accept `IODesc` instead of string context
- Add slow operation warnings for flush operations > 50ms
- Improve logging with detailed timing information display

This provides better observability for raft log performance debugging
and helps identify slow IO operations in production systems.

Author: drmingdrmer

src/meta/raft-store/src/raft_log_v004/callback.rs

@@ -14,56 +14,72 @@
 
 use std::io;
 use std::sync::mpsc::SyncSender;
+use std::time::Duration;
 
 use databend_common_meta_types::raft_types;
 use log::info;
 use log::warn;
 use tokio::sync::oneshot;
 
 use crate::raft_log_v004::callback_data::CallbackData;
+use crate::raft_log_v004::IODesc;
 
 /// The callback to be called when the IO is completed.
 ///
 /// This is used as a wrapper of Openraft callback or used directly internally in RaftLog.
 pub struct Callback {
-    context: String,
+    io_desc: IODesc,
     data: CallbackData,
 }
 
 impl Callback {
-    pub fn new_io_flushed(io_flushed: raft_types::IOFlushed, context: impl ToString) -> Self {
+    pub fn new_io_flushed(io_flushed: raft_types::IOFlushed, mut io_desc: IODesc) -> Self {
+        io_desc.set_flush_time();
+
         Callback {
-            context: context.to_string(),
+            io_desc,
             data: CallbackData::IOFlushed(io_flushed),
         }
     }
 
-    pub fn new_oneshot(tx: oneshot::Sender<Result<(), io::Error>>, context: impl ToString) -> Self {
+    pub fn new_oneshot(tx: oneshot::Sender<Result<(), io::Error>>, mut io_desc: IODesc) -> Self {
+        io_desc.set_flush_time();
+
         Callback {
-            context: context.to_string(),
+            io_desc,
             data: CallbackData::Oneshot(tx),
         }
     }
 
-    pub fn new_sync_oneshot(tx: SyncSender<Result<(), io::Error>>, context: impl ToString) -> Self {
+    pub fn new_sync_oneshot(tx: SyncSender<Result<(), io::Error>>, mut io_desc: IODesc) -> Self {
+        io_desc.set_flush_time();
+
         Callback {
-            context: context.to_string(),
+            io_desc,
             data: CallbackData::SyncOneshot(tx),
         }
     }
 }
 
 impl raft_log::Callback for Callback {
-    fn send(self, res: Result<(), io::Error>) {
-        info!("{}: Callback is called with: {:?}", self.context, res);
+    fn send(mut self, res: Result<(), io::Error>) {
+        self.io_desc.set_flush_done_time();
+
+        info!("{}: Callback is called with: {:?}", self.io_desc, res);
+
+        if let Some(duration) = self.io_desc.total_duration() {
+            if duration > Duration::from_millis(50) {
+                warn!("{}: Slow IO operation: {:?}", self.io_desc, res);
+            }
+        }
 
         match self.data {
             CallbackData::Oneshot(tx) => {
                 let send_res = tx.send(res);
                 if send_res.is_err() {
                     warn!(
                         "{}: Callback failed to send Oneshot result back to caller",
-                        self.context
+                        self.io_desc
                     );
                 }
             }

src/meta/raft-store/src/raft_log_v004/io_desc.rs

@@ -14,26 +14,66 @@
 
 use std::error::Error;
 use std::fmt;
+use std::time::Duration;
+use std::time::SystemTime;
 
 use databend_common_meta_types::anyerror::AnyError;
 use databend_common_meta_types::raft_types::ErrorSubject;
 use databend_common_meta_types::raft_types::ErrorVerb;
 use databend_common_meta_types::raft_types::StorageError;
+use display_more::DisplayUnixTimeStampExt;
 use log::debug;
 use log::error;
 
 use crate::raft_log_v004::io_phase::IOPhase;
 
 /// Describe an IO operation.
+#[derive(Clone)]
 pub struct IODesc {
     pub subject: ErrorSubject,
     pub verb: ErrorVerb,
     pub ctx: String,
+
+    /// Unix timestamp when this IO operation started.
+    pub start_time: Duration,
+
+    /// Unix timestamp when the flush request is sent.
+    pub flush_time: Option<Duration>,
+
+    /// Unix timestamp when the flush request is completed.
+    pub flush_done_time: Option<Duration>,
+
+    /// Unix timestamp when the IO operation is done.
+    pub done_time: Option<Duration>,
 }
 
 impl fmt::Display for IODesc {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "{}: ({}-{:?})", self.ctx, self.verb, self.subject)
+        write!(
+            f,
+            "IODesc({}): ({}-{:?})",
+            self.ctx, self.verb, self.subject
+        )?;
+
+        let start_time = self.start_time;
+        write!(f, "[start: {}", start_time.display_unix_timestamp_short())?;
+
+        if let Some(flush_time) = self.flush_time {
+            let elapsed = flush_time.saturating_sub(start_time);
+            write!(f, ", flush: +{:?}", elapsed,)?;
+
+            if let Some(flushed_time) = self.flush_done_time {
+                let elapsed = flushed_time.saturating_sub(flush_time);
+                write!(f, ", flushed: +{:?}", elapsed,)?;
+            }
+        }
+
+        if let Some(done_time) = self.done_time {
+            let elapsed = done_time.saturating_sub(self.start_time);
+            write!(f, ", total: {:?}", elapsed)?;
+        }
+
+        write!(f, "]")
     }
 }
 
@@ -45,9 +85,40 @@ impl IODesc {
             subject,
             verb,
             ctx: s,
+            start_time: Self::since_epoch(),
+            flush_time: None,
+            flush_done_time: None,
+            done_time: None,
         }
     }
 
+    pub fn set_flush_time(&mut self) {
+        self.flush_time = Some(Self::since_epoch());
+    }
+
+    pub fn set_flush_done_time(&mut self) {
+        self.flush_done_time = Some(Self::since_epoch());
+    }
+
+    pub fn set_done_time(&mut self) {
+        self.done_time = Some(Self::since_epoch());
+    }
+
+    pub fn total_duration(&self) -> Option<Duration> {
+        self.flush_done_time
+            .map(|flushed_time| flushed_time.saturating_sub(self.start_time))
+    }
+
+    fn since_epoch() -> Duration {
+        SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap()
+    }
+
+    pub fn unknown(ctx: impl ToString) -> Self {
+        Self::start(ErrorSubject::None, ErrorVerb::Write, ctx)
+    }
+
     pub fn save_vote(ctx: impl ToString) -> Self {
         Self::start(ErrorSubject::Vote, ErrorVerb::Write, ctx)
     }
@@ -114,3 +185,40 @@ impl IODesc {
         self.to_storage_error(IOPhase::ReceiveFlushCallback, err)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_display() {
+        let now_us = 1754708580036171;
+
+        let mut desc = IODesc::start(ErrorSubject::Logs, ErrorVerb::Write, "test");
+        assert!(desc.start_time.as_secs() > 0);
+        desc.start_time = Duration::from_micros(now_us);
+
+        assert_eq!(
+            format!("{}", desc),
+            "IODesc(test): (Write-Logs)[start: 2025-08-09T03:03:00.036]"
+        );
+
+        desc.flush_time = Some(Duration::from_micros(now_us + 1_000_000));
+        assert_eq!(
+            format!("{}", desc),
+            "IODesc(test): (Write-Logs)[start: 2025-08-09T03:03:00.036, flush: +1s]"
+        );
+
+        desc.flush_done_time = Some(Duration::from_micros(now_us + 2_000_000));
+        assert_eq!(
+            format!("{}", desc),
+            "IODesc(test): (Write-Logs)[start: 2025-08-09T03:03:00.036, flush: +1s, flushed: +1s]"
+        );
+
+        desc.done_time = Some(Duration::from_micros(now_us + 3_000_000));
+        assert_eq!(
+            format!("{}", desc),
+            "IODesc(test): (Write-Logs)[start: 2025-08-09T03:03:00.036, flush: +1s, flushed: +1s, total: 3s]"
+        );
+    }
+}

src/meta/raft-store/src/raft_log_v004/util.rs

@@ -18,11 +18,12 @@ use raft_log::api::raft_log_writer::RaftLogWriter;
 use tokio::sync::oneshot;
 
 use crate::raft_log_v004::callback::Callback;
+use crate::raft_log_v004::IODesc;
 use crate::raft_log_v004::RaftLogV004;
 
 pub async fn blocking_flush(rl: &mut RaftLogV004) -> Result<(), io::Error> {
     let (tx, rx) = oneshot::channel();
-    let callback = Callback::new_oneshot(tx, "blocking_flush");
+    let callback = Callback::new_oneshot(tx, IODesc::unknown("blocking_flush(RaftLogV004)"));
 
     rl.flush(callback)?;
     rx.await

src/meta/service/src/store/raft_log_storage_impl.rs

@@ -103,7 +103,7 @@ impl RaftLogReader<TypeConfig> for RaftStore {
     ) -> Result<Vec<Entry>, StorageError> {
         let (start, end) = range_boundary(range);
 
-        let io = IODesc::read_logs(format!(
+        let mut io = IODesc::read_logs(format!(
             "RaftStore(id={})::try_get_log_entries([{},{})",
             self.id, start, end
         ));
@@ -119,7 +119,8 @@ impl RaftLogReader<TypeConfig> for RaftStore {
             .collect::<Result<Vec<_>, _>>()
             .map_err(|e| io.err_submit(e))?;
 
-        debug!("{}", io.ok_done());
+        io.set_done_time();
+        info!("{}", io.ok_done());
         Ok(entries)
     }
 
@@ -193,7 +194,7 @@ impl RaftLogStorage<TypeConfig> for RaftStore {
             let mut log = self.log.write().await;
 
             log.save_vote(Cw(*vote)).map_err(|e| io.err_submit(e))?;
-            log.flush(raft_log_v004::Callback::new_oneshot(tx, &io))
+            log.flush(raft_log_v004::Callback::new_oneshot(tx, io.clone()))
                 .map_err(|e| io.err_submit_flush(e))?;
         }
 
@@ -230,8 +231,11 @@ impl RaftLogStorage<TypeConfig> for RaftStore {
 
         debug!("{}", io.ok_submit());
 
-        log.flush(raft_log_v004::Callback::new_io_flushed(callback, &io))
-            .map_err(|e| io.err_submit_flush(e))?;
+        log.flush(raft_log_v004::Callback::new_io_flushed(
+            callback,
+            io.clone(),
+        ))
+        .map_err(|e| io.err_submit_flush(e))?;
 
         info!("{}", io.ok_submit_flush());