refactor: add new raft-log metrics to "metactl status" response (#16899)

These new metrics are added to gRPC API `GetClusterStatus` and
corresponding databend-metactl CLI command.

Newly added metrics are:

```
$ databend-metactl status
# ...
RaftLog:
  - CacheItems: 287
  - CacheUsedSize: 4959439
  - WALTotalSize: 4378076
  - WALOpenChunkSize: 4378058
  - WALOffset: 4378076
  - WALClosedChunkCount: 1
  - WALClosedChunkTotalSize: 18
  - WALClosedChunkSizes:
    - ChunkId(00_000_000_000_000_000_000): 18
```

src/meta/binaries/metactl/main.rs

@@ -237,6 +237,23 @@ impl App {
         println!("BinaryVersion: {}", res.binary_version);
         println!("DataVersion: {}", res.data_version);
         println!("RaftLogSize: {}", res.raft_log_size);
+        if let Some(s) = res.raft_log_status {
+            println!("RaftLog:");
+            println!("  - CacheItems: {}", s.cache_items);
+            println!("  - CacheUsedSize: {}", s.cache_used_size);
+            println!("  - WALTotalSize: {}", s.wal_total_size);
+            println!("  - WALOpenChunkSize: {}", s.wal_open_chunk_size);
+            println!("  - WALOffset: {}", s.wal_offset);
+            println!("  - WALClosedChunkCount: {}", s.wal_closed_chunk_count);
+            println!(
+                "  - WALClosedChunkTotalSize: {}",
+                s.wal_closed_chunk_total_size
+            );
+            println!("  - WALClosedChunkSizes:");
+            for (k, v) in s.wal_closed_chunk_sizes {
+                println!("    - {}: {}", k, v);
+            }
+        }
         println!("SnapshotKeyCount: {}", res.snapshot_key_count);
         println!("Node: id={} raft={}", res.id, res.endpoint);
         println!("State: {}", res.state);

src/meta/service/src/api/grpc/grpc_service.rs

@@ -462,7 +462,20 @@ impl MetaService for MetaServiceImpl {
             binary_version: status.binary_version,
             data_version: status.data_version.to_string(),
             endpoint: status.endpoint,
-            raft_log_size: status.raft_log_size,
+
+            raft_log_size: status.raft_log.wal_total_size,
+
+            raft_log_status: Some(pb::RaftLogStatus {
+                cache_items: status.raft_log.cache_items,
+                cache_used_size: status.raft_log.cache_used_size,
+                wal_total_size: status.raft_log.wal_total_size,
+                wal_open_chunk_size: status.raft_log.wal_open_chunk_size,
+                wal_offset: status.raft_log.wal_offset,
+                wal_closed_chunk_count: status.raft_log.wal_closed_chunk_count,
+                wal_closed_chunk_total_size: status.raft_log.wal_closed_chunk_total_size,
+                wal_closed_chunk_sizes: status.raft_log.wal_closed_chunk_sizes,
+            }),
+
             snapshot_key_count: status.snapshot_key_count as u64,
             state: status.state,
             is_leader: status.is_leader,

src/meta/service/src/meta_service/meta_node.rs

@@ -852,8 +852,8 @@ impl MetaNode {
 
         let endpoint = self.sto.get_node_raft_endpoint(&self.sto.id).await?;
 
-        let raft_log_size = self.get_raft_log_size().await;
-        let key_count = self.get_snapshot_key_count().await;
+        let raft_log_status = self.get_raft_log_stat().await.into();
+        let snapshot_key_count = self.get_snapshot_key_count().await;
 
         let metrics = self.raft.metrics().borrow().clone();
 
@@ -870,8 +870,8 @@ impl MetaNode {
             binary_version: METASRV_COMMIT_VERSION.as_str().to_string(),
             data_version: DATA_VERSION,
             endpoint: endpoint.to_string(),
-            raft_log_size,
-            snapshot_key_count: key_count,
+            raft_log: raft_log_status,
+            snapshot_key_count,
             state: format!("{:?}", metrics.state),
             is_leader: metrics.state == openraft::ServerState::Leader,
             current_term: metrics.current_term,

src/meta/service/src/meta_service/meta_node_status.rs

@@ -15,6 +15,7 @@
 use std::collections::BTreeMap;
 
 use databend_common_meta_raft_store::ondisk::DataVersion;
+use databend_common_meta_raft_store::raft_log_v004::RaftLogStat;
 use databend_common_meta_types::raft_types::LogId;
 use databend_common_meta_types::raft_types::NodeId;
 use databend_common_meta_types::Node;
@@ -32,8 +33,8 @@ pub struct MetaNodeStatus {
     /// The raft service endpoint for internal communication
     pub endpoint: String,
 
-    /// The size in bytes of the raft-log on disk data.
-    pub raft_log_size: u64,
+    /// The status about local raft-log
+    pub raft_log: RaftLogStatus,
 
     /// Total number of keys in current snapshot
     pub snapshot_key_count: u64,
@@ -78,3 +79,39 @@ pub struct MetaNodeStatus {
     /// `seq` is a monotonically incremental integer for every value that is inserted or updated.
     pub last_seq: u64,
 }
+
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
+pub struct RaftLogStatus {
+    pub cache_items: u64,
+    pub cache_used_size: u64,
+    pub wal_total_size: u64,
+    pub wal_open_chunk_size: u64,
+    pub wal_offset: u64,
+    pub wal_closed_chunk_count: u64,
+    pub wal_closed_chunk_total_size: u64,
+    pub wal_closed_chunk_sizes: BTreeMap<String, u64>,
+}
+
+impl From<RaftLogStat> for RaftLogStatus {
+    fn from(s: RaftLogStat) -> Self {
+        let closed_sizes = s
+            .closed_chunks
+            .iter()
+            .map(|c| (c.chunk_id.to_string(), c.size))
+            .collect();
+
+        let closed_total_size = s.closed_chunks.iter().map(|c| c.size).sum::<u64>();
+        let wal_total_size = closed_total_size + s.open_chunk.size;
+
+        Self {
+            cache_items: s.payload_cache_item_count,
+            cache_used_size: s.payload_cache_size,
+            wal_total_size,
+            wal_open_chunk_size: s.open_chunk.size,
+            wal_offset: s.open_chunk.global_end,
+            wal_closed_chunk_count: s.closed_chunks.len() as u64,
+            wal_closed_chunk_total_size: closed_total_size,
+            wal_closed_chunk_sizes: closed_sizes,
+        }
+    }
+}

src/meta/types/build.rs

@@ -45,6 +45,7 @@ fn build_proto() {
 
     let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());
     tonic_build::configure()
+        .btree_map(["RaftLogStatus.wal_closed_chunk_sizes"])
         .file_descriptor_set_path(out_dir.join("meta_descriptor.bin"))
         .type_attribute(
             "SeqV",

src/meta/types/proto/meta.proto

@@ -36,9 +36,9 @@ message TransferLeaderRequest {
   LogId last_log_id = 3;
 }
 
-message MemberListRequest { string data = 1; }
+message MemberListRequest {string data = 1;}
 
-message MemberListReply { repeated string data = 1; }
+message MemberListReply {repeated string data = 1;}
 
 message HandshakeRequest {
   uint64 protocol_version = 1;
@@ -94,7 +94,7 @@ message Event {
   optional SeqV prev = 3;
 }
 
-message WatchResponse { Event event = 1; }
+message WatchResponse {Event event = 1;}
 
 // messages for txn
 message TxnCondition {
@@ -181,6 +181,19 @@ message ClusterStatus {
   repeated string non_voters = 16;
   uint64 last_seq = 17;
   uint64 snapshot_key_count = 18;
+  RaftLogStatus raft_log_status = 19;
+}
+
+// Status about local raft-log storage
+message RaftLogStatus {
+  uint64 cache_items = 1;
+  uint64 cache_used_size = 2;
+  uint64 wal_total_size = 3;
+  uint64 wal_open_chunk_size = 4;
+  uint64 wal_offset = 5;
+  uint64 wal_closed_chunk_count = 6;
+  uint64 wal_closed_chunk_total_size = 7;
+  map<string, uint64> wal_closed_chunk_sizes = 8;
 }
 
 message ClientInfo {