feat: metactl: add dump-raft-log-wal subcommand (databendlabs#18865)

Add `dump-raft-log-wal` subcommand to databend-metactl for dumping raft log WAL contents in human-readable format. The command accepts a `--raft-dir` parameter pointing to the meta directory (e.g., `.databend/meta1`) and automatically appends the `/df_meta/V004/log` path suffix to locate the WAL files. The output uses Display format to provide readable representation of raft log records including state changes, vote operations, log entries, and commit markers. Each record shows byte offsets, sizes, and formatted content with log IDs in compact notation (e.g., `T1-N1.2` for term 1, node 1, index 2).

Author: drmingdrmer

Cargo.lock

@@ -50,6 +50,7 @@ futures = { workspace = true }
 log = { workspace = true }
 mlua = { workspace = true }
 num_cpus = { workspace = true }
+raft-log = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

src/meta/binaries/Cargo.toml

@@ -28,6 +28,7 @@ use databend_common_meta_client::ClientHandle;
 use databend_common_meta_client::MetaGrpcClient;
 use databend_common_meta_control::admin::MetaAdminClient;
 use databend_common_meta_control::args::BenchArgs;
+use databend_common_meta_control::args::DumpRaftLogWalArgs;
 use databend_common_meta_control::args::ExportArgs;
 use databend_common_meta_control::args::GetArgs;
 use databend_common_meta_control::args::GlobalArgs;
@@ -309,6 +310,30 @@ return metrics, nil
  fn new_grpc_client(&self, addresses: Vec<String>) -> Result<Arc<ClientHandle>, CreationError> {
  lua_support::new_grpc_client(addresses, &BUILD_INFO)
  }
+
+ async fn dump_raft_log_wal(&self, args: &DumpRaftLogWalArgs) -> anyhow::Result<()> {
+ use std::path::PathBuf;
+
+ use raft_log::Config;
+ use raft_log::DumpApi;
+
+ let mut wal_dir = PathBuf::from(&args.raft_dir);
+ wal_dir.push("df_meta");
+ wal_dir.push("V004");
+ wal_dir.push("log");
+
+ let config = Arc::new(Config {
+ dir: wal_dir.to_string_lossy().to_string(),
+ ..Default::default()
+ });
+
+ let dump =
+ raft_log::Dump::<databend_common_meta_raft_store::raft_log_v004::RaftLogTypes>::new(
+ config,
+ )?;
+ dump.write_display(io::stdout())?;
+ Ok(())
+ }
 }
 
 #[derive(Debug, Clone, Deserialize, Subcommand)]
@@ -328,6 +353,7 @@ enum CtlCommand {
  Lua(LuaArgs),
  MemberList(MemberListArgs),
  Metrics(MetricsArgs),
+ DumpRaftLogWal(DumpRaftLogWalArgs),
 }
 
 /// Usage:
@@ -411,6 +437,9 @@ async fn main() -> anyhow::Result<()> {
  CtlCommand::Metrics(args) => {
  app.get_metrics(args).await?;
  }
+ CtlCommand::DumpRaftLogWal(args) => {
+ app.dump_raft_log_wal(args).await?;
+ }
  },
  // for backward compatibility
  None => {

src/meta/binaries/metactl/main.rs

@@ -295,3 +295,10 @@ pub struct KeysLayoutArgs {
  #[clap(long)]
  pub depth: Option<u32>,
 }
+
+#[derive(Debug, Clone, Deserialize, Args)]
+pub struct DumpRaftLogWalArgs {
+ /// The dir to store persisted meta state, e.g., `.databend/meta1`
+ #[clap(long)]
+ pub raft_dir: String,
+}

src/meta/control/src/args.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::io;
 
 use databend_common_meta_types::raft_types;
@@ -25,6 +26,12 @@ pub struct LogStoreMeta {
  pub node_id: Option<raft_types::NodeId>,
 }
 
+impl fmt::Display for LogStoreMeta {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "LogStoreMeta{{ node_id: {:?} }}", self.node_id)
+ }
+}
+
 impl raft_log::codeq::Encode for LogStoreMeta {
  fn encode<W: io::Write>(&self, mut w: W) -> Result<usize, io::Error> {
  let mut ow = OffsetWriter::new(&mut w);

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

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+
+import re
+import shutil
+import time
+from metactl_utils import metactl_bin, metactl_upsert, metactl_trigger_snapshot
+from utils import run_command, kill_databend_meta, start_meta_node, print_title
+
+
+def test_dump_raft_log_wal():
+ """Test dump-raft-log-wal subcommand from raft directory."""
+ print_title("Test dump-raft-log-wal from raft directory")
+ kill_databend_meta()
+ shutil.rmtree(".databend", ignore_errors=True)
+ start_meta_node(1, False)
+
+ grpc_addr = "127.0.0.1:9191"
+ admin_addr = "127.0.0.1:28101"
+
+ # Insert test data
+ test_keys = [
+ ("app/db/host", "localhost"),
+ ("app/db/port", "5432"),
+ ("app/config/timeout", "30"),
+ ]
+
+ for key, value in test_keys:
+ metactl_upsert(grpc_addr, key, value)
+ print("✓ Test data inserted")
+
+ # Trigger snapshot to ensure raft log has data
+ metactl_trigger_snapshot(admin_addr)
+ print("✓ Snapshot triggered")
+
+ # Wait for snapshot to complete
+ time.sleep(2)
+
+ # Stop meta service before accessing raft directory
+ kill_databend_meta()
+
+ # Test dump-raft-log-wal from raft directory
+ result = run_command([metactl_bin, "dump-raft-log-wal", "--raft-dir", "./.databend/meta1"])
+
+ print("Output:")
+ print(result)
+
+ # Expected output with time field that will be masked
+ expected = """RaftLog:
+ChunkId(00_000_000_000_000_000_000)
+ R-00000: [000_000_000, 000_000_018) Size(18): RaftLogState(RaftLogState(vote: None, last: None, committed: None, purged: None, user_data: None))
+ R-00001: [000_000_018, 000_000_046) Size(28): RaftLogState(RaftLogState(vote: None, last: None, committed: None, purged: None, user_data: LogStoreMeta{ node_id: Some(1) }))
+ R-00002: [000_000_046, 000_000_125) Size(79): Append(log_id: T0-N0.0, payload: membership:{voters:[{1:EmptyNode}], learners:[]})
+ R-00003: [000_000_125, 000_000_175) Size(50): SaveVote(<T1-N1:->)
+ R-00004: [000_000_175, 000_000_225) Size(50): SaveVote(<T1-N1:Q>)
+ R-00005: [000_000_225, 000_000_277) Size(52): Append(log_id: T1-N1.1, payload: blank)
+ R-00006: [000_000_277, 000_000_472) Size(195): Append(log_id: T1-N1.2, payload: normal:time: <TIME> cmd: add_node(no-override):1=id=1 raft=localhost:28103 grpc=127.0.0.1:9191)
+ R-00007: [000_000_472, 000_000_518) Size(46): Commit(T1-N1.1)
+ R-00008: [000_000_518, 000_000_564) Size(46): Commit(T1-N1.2)
+ R-00009: [000_000_564, 000_000_643) Size(79): Append(log_id: T1-N1.3, payload: membership:{voters:[{1:EmptyNode}], learners:[]})
+ R-00010: [000_000_643, 000_000_689) Size(46): Commit(T1-N1.3)
+ R-00011: [000_000_689, 000_000_884) Size(195): Append(log_id: T1-N1.4, payload: normal:time: <TIME> cmd: add_node(override):1=id=1 raft=localhost:28103 grpc=127.0.0.1:9191)
+ R-00012: [000_000_884, 000_000_930) Size(46): Commit(T1-N1.4)
+ R-00013: [000_000_930, 000_001_192) Size(262): Append(log_id: T1-N1.5, payload: normal:time: <TIME> cmd: txn:TxnRequest{if:[app/db/host >= seq(0)] then:[Put(Put key=app/db/host)] else:[Get(Get key=app/db/host)]})
+ R-00014: [000_001_192, 000_001_238) Size(46): Commit(T1-N1.5)
+ R-00015: [000_001_238, 000_001_495) Size(257): Append(log_id: T1-N1.6, payload: normal:time: <TIME> cmd: txn:TxnRequest{if:[app/db/port >= seq(0)] then:[Put(Put key=app/db/port)] else:[Get(Get key=app/db/port)]})
+ R-00016: [000_001_495, 000_001_541) Size(46): Commit(T1-N1.6)
+ R-00017: [000_001_541, 000_001_817) Size(276): Append(log_id: T1-N1.7, payload: normal:time: <TIME> cmd: txn:TxnRequest{if:[app/config/timeout >= seq(0)] then:[Put(Put key=app/config/timeout)] else:[Get(Get key=app/config/timeout)]})
+ R-00018: [000_001_817, 000_001_863) Size(46): Commit(T1-N1.7)"""
+
+ def mask_time(text):
+ """Mask time fields in the output for comparison."""
+ # Mask timestamp in format "time: 2025-10-19T15:03:55.193"
+ text = re.sub(r'time: \d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+', 'time: <TIME>', text)
+ return text
+
+ # Normalize both actual and expected output
+ actual_lines = result.strip().split("\n")
+ expected_lines = expected.strip().split("\n")
+
+ # Mask time in actual output
+ actual_masked = [mask_time(line) for line in actual_lines]
+
+ # Verify line count matches
+ assert len(actual_masked) == len(expected_lines), (
+ f"Line count mismatch: got {len(actual_masked)} lines, expected {len(expected_lines)} lines"
+ )
+
+ # Compare line by line
+ for i, (actual_line, expected_line) in enumerate(zip(actual_masked, expected_lines)):
+ assert actual_line == expected_line, (
+ f"Line {i} mismatch:\n"
+ f" Actual : {actual_line}\n"
+ f" Expected: {expected_line}"
+ )
+
+ print(f"✓ All {len(actual_masked)} lines match expected output (time fields masked)")
+
+ # Clean up only on success
+ shutil.rmtree(".databend", ignore_errors=True)
+
+
+def main():
+ """Main function to run all dump-raft-log-wal tests."""
+ test_dump_raft_log_wal()
+
+
+if __name__ == "__main__":
+ main()

tests/metactl/subcommands/cmd_dump_raft_log_wal.py

@@ -1,4 +1,4 @@
-#!/usr/bin/env python3
+#!/usr/bin/env python2
 """
 Comprehensive test runner for all metactl subcommands.