feat: meta: add I/O timing tracking for log entry application (#18854)

Implement RAII-based I/O timing to track state machine read operations
during Raft log entry application. The system now records timing for
all get, list, and expire_scan operations, logging detailed breakdowns
for observability and performance analysis.

Add `IoTimer` struct using Drop trait to automatically record timing
when it goes out of scope, eliminating manual timing code. Add
`IoTiming` struct to accumulate operation records with Display trait
for grouped output format like "get(key1:5ms, key2:3ms)".

Extend `CmdContext` with Arc<Mutex<IoTiming>> to store timing data
across async operations. Add `start_io_timer()` helper method that
returns IoTimer for convenient RAII usage.

Log entries now include I/O timing information at INFO level, with
WARN level for slow entries exceeding 100ms threshold. The log format
includes total I/O time, detailed operation breakdown, log timestamp,
and the full entry for debugging.

Instrument timing at 6 call sites: eval_one_condition (get and list),
txn_execute_get, txn_execute_delete_by_prefix, txn_execute_fetch_add_u64,
and clean_expired_kvs. Use block scopes to ensure timers drop before
mutable borrows where needed for borrow checker compliance.

Co-authored-by: drdrxp <[email protected]>

Author: drmingdrmer

src/meta/raft-store/src/applier/mod.rs

@@ -71,6 +71,9 @@ use crate::state_machine_api_ext::StateMachineApiExt;
 
 pub(crate) mod applier_data;
 
+/// Threshold in milliseconds for logging slow log entry application
+const SLOW_LOG_ENTRY_THRESHOLD_MS: u128 = 100;
+
 /// A helper that applies raft log `Entry` to the state machine.
 pub struct Applier<SM>
 where SM: StateMachineApi<SysData> + 'static
@@ -152,6 +155,31 @@ where SM: StateMachineApi<SysData> + 'static
             self.sm.on_change_applied(event);
         }
 
+        // Log I/O timing information
+        let io_time_ms = self.cmd_ctx.io_total_duration().as_millis();
+        let io_timing = self.cmd_ctx.io_timing();
+        let log_time = Duration::from_millis(log_time_ms);
+
+        if io_time_ms > SLOW_LOG_ENTRY_THRESHOLD_MS {
+            warn!(
+                "Slow log entry applied: {}, time: {}, io: {}ms, ops: {}, entry: {}",
+                log_id,
+                log_time.display_unix_timestamp_short(),
+                io_time_ms,
+                io_timing,
+                entry
+            );
+        } else {
+            info!(
+                "Log entry applied: {}, time: {}, io: {}ms, ops: {}, entry: {}",
+                log_id,
+                log_time.display_unix_timestamp_short(),
+                io_time_ms,
+                io_timing,
+                entry
+            );
+        }
+
         Ok(applied_state)
     }
 
@@ -427,7 +455,7 @@ where SM: StateMachineApi<SysData> + 'static
         // If the key expired, it should be treated as `None` value.
         // sm.get_kv() does not check expiration.
         // Expired keys are cleaned before applying a log, see: `clean_expired_kvs()`.
-        let seqv = self.sm.get_maybe_expired_kv(key).await?;
+        let seqv = self.get_maybe_expired_kv_with_timing(key).await?;
 
         debug!(
             "txn_execute_one_condition: key: {} curr: seq:{} value:{:?}",
@@ -461,7 +489,7 @@ where SM: StateMachineApi<SysData> + 'static
             Target::KeysWithPrefix(against_n) => {
                 let against_n = *against_n;
 
-                let strm = self.sm.list_kv(key).await?;
+                let strm = self.list_kv_with_timing(key).await?;
                 // Taking at most `against_n + 1` keys is just enough for every predicate.
                 let strm = strm.take((against_n + 1) as usize);
                 let count: u64 = strm.try_fold(0, |acc, _| ready(Ok(acc + 1))).await?;
@@ -525,7 +553,8 @@ where SM: StateMachineApi<SysData> + 'static
     }
 
     async fn txn_execute_get(&self, get: &TxnGetRequest) -> Result<pb::TxnGetResponse, io::Error> {
-        let sv = self.sm.get_maybe_expired_kv(&get.key).await?;
+        let sv = self.get_maybe_expired_kv_with_timing(&get.key).await?;
+
         let get_resp = pb::TxnGetResponse {
             key: get.key.clone(),
             value: sv.map(pb::SeqV::from),
@@ -580,9 +609,9 @@ where SM: StateMachineApi<SysData> + 'static
         &mut self,
         delete_by_prefix: &TxnDeleteByPrefixRequest,
     ) -> Result<TxnDeleteByPrefixResponse, io::Error> {
-        let mut strm = self.sm.list_kv(&delete_by_prefix.prefix).await?;
-        let mut count = 0;
+        let mut strm = self.list_kv_with_timing(&delete_by_prefix.prefix).await?;
 
+        let mut count = 0;
         while let Some((key, _seq_v)) = strm.try_next().await? {
             let (prev, res) = self.upsert_kv(&UpsertKV::delete(&key)).await?;
             self.push_change(key, prev, res);
@@ -601,7 +630,7 @@ where SM: StateMachineApi<SysData> + 'static
         &mut self,
         req: &FetchIncreaseU64,
     ) -> Result<pb::FetchIncreaseU64Response, io::Error> {
-        let before_seqv = self.sm.get_maybe_expired_kv(&req.key).await?;
+        let before_seqv = self.get_maybe_expired_kv_with_timing(&req.key).await?;
 
         let before_seq = before_seqv.seq();
 
@@ -703,24 +732,29 @@ where SM: StateMachineApi<SysData> + 'static
             Duration::from_millis(log_time_ms).display_unix_timestamp_short()
         );
 
-        let mut to_clean = vec![];
-        let mut strm = self.sm.list_expire_index(log_time_ms).await?;
+        let to_clean = {
+            let _timer = self.cmd_ctx.start_io_timer("expire_scan", "expired_keys");
+            let mut to_clean = vec![];
+            let mut strm = self.sm.list_expire_index(log_time_ms).await?;
 
-        // Save the log time for next cleaning.
-        // Avoid listing tombstone records.
-        self.sm
-            .set_cleanup_start_timestamp(Duration::from_millis(log_time_ms));
+            // Save the log time for next cleaning.
+            // Avoid listing tombstone records.
+            self.sm
+                .set_cleanup_start_timestamp(Duration::from_millis(log_time_ms));
 
-        {
-            let mut strm = std::pin::pin!(strm);
-            while let Some((expire_key, key)) = strm.try_next().await? {
-                if !expire_key.is_expired(log_time_ms) {
-                    break;
-                }
+            {
+                let mut strm = std::pin::pin!(strm);
+                while let Some((expire_key, key)) = strm.try_next().await? {
+                    if !expire_key.is_expired(log_time_ms) {
+                        break;
+                    }
 
-                to_clean.push((expire_key, key));
+                    to_clean.push((expire_key, key));
+                }
             }
-        }
+
+            to_clean
+        };
 
         for (expire_key, key) in to_clean {
             let upsert = UpsertKV::delete(key);
@@ -742,6 +776,23 @@ where SM: StateMachineApi<SysData> + 'static
         self.changes.push((key.to_string(), prev, result))
     }
 
+    /// Get KV with I/O timing tracking.
+    ///
+    /// Does not check expiration - may return expired entries.
+    async fn get_maybe_expired_kv_with_timing(&self, key: &str) -> Result<Option<SeqV>, io::Error> {
+        let _timer = self.cmd_ctx.start_io_timer("get", key);
+        self.sm.get_maybe_expired_kv(key).await
+    }
+
+    /// List KV with I/O timing tracking.
+    async fn list_kv_with_timing(
+        &self,
+        prefix: &str,
+    ) -> Result<map_api::IOResultStream<(String, SeqV)>, io::Error> {
+        let _timer = self.cmd_ctx.start_io_timer("list", format!("{}*", prefix));
+        self.sm.list_kv(prefix).await
+    }
+
     /// Retrieve the proposing time from a raft-log.
     ///
     /// Only `Normal` log has a time embedded.

src/meta/types/src/cmd/cmd_context.rs

@@ -13,21 +13,29 @@
 // limitations under the License.
 
 use std::fmt;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
 
 use display_more::DisplayUnixTimeStampExt;
 
+use crate::cmd::io_timing::IoTimer;
+use crate::cmd::io_timing::IoTiming;
 use crate::raft_types::LogId;
 use crate::Time;
 
 /// A context used when executing a [`Cmd`], to provide additional environment information.
 ///
 /// [`Cmd`]: crate::Cmd
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone)]
 pub struct CmdContext {
     time: Time,
 
     /// Current log id to apply
     log_id: LogId,
+
+    /// I/O timing information for tracking read operations
+    io_timing: Arc<Mutex<IoTiming>>,
 }
 
 impl fmt::Display for CmdContext {
@@ -46,6 +54,7 @@ impl CmdContext {
         Self {
             time: Time::from_millis(millis),
             log_id,
+            io_timing: Arc::new(Mutex::new(IoTiming::new())),
         }
     }
 
@@ -57,11 +66,51 @@ impl CmdContext {
         CmdContext {
             time,
             log_id: LogId::default(),
+            io_timing: Arc::new(Mutex::new(IoTiming::new())),
         }
     }
 
     /// Returns the time since 1970-01-01 when this log is proposed by the leader.
     pub fn time(&self) -> Time {
         self.time
     }
+
+    /// Record an I/O operation with its timing.
+    pub fn record_io(
+        &self,
+        op_type: impl Into<String>,
+        details: impl Into<String>,
+        duration: Duration,
+    ) {
+        self.io_timing
+            .lock()
+            .unwrap()
+            .record(op_type, details, duration);
+    }
+
+    /// Get I/O timing total duration.
+    pub fn io_total_duration(&self) -> Duration {
+        self.io_timing.lock().unwrap().total_duration()
+    }
+
+    /// Get I/O timing details (cloned for logging).
+    pub fn io_timing(&self) -> IoTiming {
+        self.io_timing.lock().unwrap().clone()
+    }
+
+    /// Start an RAII-based I/O timer that automatically records timing on drop.
+    ///
+    /// # Example
+    /// ```ignore
+    /// let _timer = self.cmd_ctx.start_io_timer("get", key);
+    /// let result = self.sm.get_maybe_expired_kv(key).await?;
+    /// // Timer automatically records on scope exit
+    /// ```
+    pub fn start_io_timer(
+        &self,
+        op_type: impl Into<String>,
+        details: impl Into<String>,
+    ) -> IoTimer {
+        IoTimer::new(self, op_type, details)
+    }
 }

src/meta/types/src/cmd/io_timing.rs

@@ -0,0 +1,141 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+use std::time::Duration;
+use std::time::Instant;
+
+use crate::CmdContext;
+
+/// Tracks I/O operation timing during log entry application.
+#[derive(Debug, Clone, Default)]
+pub struct IoTiming {
+    /// Records of I/O operations: (operation_type, details, duration)
+    operations: Vec<(String, String, Duration)>,
+}
+
+/// RAII-based timer for automatic I/O timing recording.
+///
+/// Records timing when dropped, using Rust's RAII pattern.
+/// This ensures timing is recorded even if the function returns early.
+pub struct IoTimer<'a> {
+    cmd_ctx: &'a CmdContext,
+    op_type: String,
+    details: String,
+    start: Instant,
+}
+
+impl<'a> IoTimer<'a> {
+    pub fn new(
+        cmd_ctx: &'a CmdContext,
+        op_type: impl Into<String>,
+        details: impl Into<String>,
+    ) -> Self {
+        Self {
+            cmd_ctx,
+            op_type: op_type.into(),
+            details: details.into(),
+            start: Instant::now(),
+        }
+    }
+}
+
+impl Drop for IoTimer<'_> {
+    fn drop(&mut self) {
+        self.cmd_ctx
+            .record_io(&self.op_type, &self.details, self.start.elapsed());
+    }
+}
+
+impl IoTiming {
+    pub fn new() -> Self {
+        Self {
+            operations: Vec::new(),
+        }
+    }
+
+    /// Record an I/O operation with its timing.
+    pub fn record(
+        &mut self,
+        op_type: impl Into<String>,
+        details: impl Into<String>,
+        duration: Duration,
+    ) {
+        self.operations
+            .push((op_type.into(), details.into(), duration));
+    }
+
+    /// Calculate total I/O time across all operations.
+    pub fn total_duration(&self) -> Duration {
+        self.operations.iter().map(|(_, _, d)| *d).sum()
+    }
+}
+
+impl fmt::Display for IoTiming {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if self.operations.is_empty() {
+            return write!(f, "no_io");
+        }
+
+        let mut last_op_type = "";
+        let mut first = true;
+
+        for (op_type, details, duration) in &self.operations {
+            if op_type != last_op_type {
+                if !first {
+                    write!(f, "), ")?;
+                }
+                write!(f, "{}(", op_type)?;
+                last_op_type = op_type;
+                first = false;
+            } else {
+                write!(f, ", ")?;
+            }
+
+            write!(f, "{}:{}ms", details, duration.as_millis())?;
+        }
+
+        if !first {
+            write!(f, ")")?;
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_io_timing_basic() {
+        let mut timing = IoTiming::new();
+        timing.record("get", "key1", Duration::from_millis(5));
+        timing.record("get", "key2", Duration::from_millis(3));
+        timing.record("list", "prefix:10", Duration::from_millis(20));
+
+        assert_eq!(timing.total_duration(), Duration::from_millis(28));
+        assert_eq!(
+            timing.to_string(),
+            "get(key1:5ms, key2:3ms), list(prefix:10:20ms)"
+        );
+    }
+
+    #[test]
+    fn test_io_timing_empty() {
+        let timing = IoTiming::new();
+        assert_eq!(timing.total_duration(), Duration::ZERO);
+        assert_eq!(timing.to_string(), "no_io");
+    }
+}