chore: add metrics for write logs in wal

wal/Cargo.toml

@@ -40,9 +40,11 @@ codec = { workspace = true }
 common_types = { workspace = true }
 futures = { workspace = true, features = ["async-await"], optional = true }
 generic_error = { workspace = true }
+lazy_static = { workspace = true }
 log = { workspace = true }
 macros = { workspace = true }
 message_queue = { workspace = true }
+prometheus = { workspace = true }
 prost = { workspace = true }
 runtime = { workspace = true }
 serde = { workspace = true }

wal/src/lib.rs

@@ -18,6 +18,7 @@ pub mod kv_encoder;
 pub mod log_batch;
 pub mod manager;
 pub mod message_queue_impl;
+pub(crate) mod metrics;
 pub mod rocks_impl;
 pub mod table_kv_impl;
 

wal/src/manager.rs

@@ -27,7 +27,10 @@ use generic_error::BoxError;
 use runtime::Runtime;
 use snafu::ResultExt;
 
-use crate::log_batch::{LogEntry, LogWriteBatch, PayloadDecoder};
+use crate::{
+    log_batch::{LogEntry, LogWriteBatch, PayloadDecoder},
+    metrics::WAL_WRITE_BYTES_HISTOGRAM,
+};
 
 pub mod error {
     use generic_error::GenericError;
@@ -333,6 +336,12 @@ pub trait WalManager: Send + Sync + fmt::Debug + 'static {
     async fn get_statistics(&self) -> Option<String>;
 }
 
+/// Used to collect the metrics about the write logs.
+pub(crate) fn collect_write_log_metrics(batch: &LogWriteBatch) {
+    let total_bytes: usize = batch.entries.iter().map(|v| v.payload.len()).sum();
+    WAL_WRITE_BYTES_HISTOGRAM.observe(total_bytes as f64);
+}
+
 #[derive(Debug)]
 enum LogIterator {
     Sync {

wal/src/message_queue_impl/wal.rs

@@ -26,8 +26,8 @@ use snafu::ResultExt;
 use crate::{
     log_batch::{LogEntry, LogWriteBatch},
     manager::{
-        error::*, AsyncLogIterator, BatchLogIteratorAdapter, ReadContext, ReadRequest, RegionId,
-        ScanContext, ScanRequest, WalLocation, WalManager, WriteContext,
+        self, error::*, AsyncLogIterator, BatchLogIteratorAdapter, ReadContext, ReadRequest,
+        RegionId, ScanContext, ScanRequest, WalLocation, WalManager, WriteContext,
     },
     message_queue_impl::{
         config::Config,
@@ -105,6 +105,7 @@ impl<M: MessageQueue> WalManager for MessageQueueImpl<M> {
     }
 
     async fn write(&self, ctx: &WriteContext, batch: &LogWriteBatch) -> Result<SequenceNumber> {
+        manager::collect_write_log_metrics(batch);
         self.0.write(ctx, batch).await.box_err().context(Write)
     }
 

wal/src/metrics.rs

@@ -0,0 +1,25 @@
+// Copyright 2023 The CeresDB Authors
+//
+// 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 lazy_static::lazy_static;
+use prometheus::{exponential_buckets, register_histogram, Histogram};
+
+lazy_static! {
+    pub static ref WAL_WRITE_BYTES_HISTOGRAM: Histogram = register_histogram!(
+        "wal_write_bytes_distribution",
+        "Bucketed histogram of wal write bytes",
+        exponential_buckets(64.0, 4.0, 10).unwrap()
+    )
+    .unwrap();
+}

wal/src/rocks_impl/manager.rs

@@ -42,7 +42,7 @@ use crate::{
     kv_encoder::{CommonLogEncoding, CommonLogKey, MaxSeqMetaEncoding, MaxSeqMetaValue, MetaKey},
     log_batch::{LogEntry, LogWriteBatch},
     manager::{
-        error::*, BatchLogIteratorAdapter, ReadContext, ReadRequest, RegionId, ScanContext,
+        self, error::*, BatchLogIteratorAdapter, ReadContext, ReadRequest, RegionId, ScanContext,
         ScanRequest, SyncLogIterator, WalLocation, WalManager, WriteContext,
     },
 };
@@ -204,6 +204,8 @@ impl TableUnit {
             batch.entries.len()
         );
 
+        manager::collect_write_log_metrics(batch);
+
         let entries_num = batch.len() as u64;
         let (wb, max_sequence_num) = {
             let wb = WriteBatch::default();

wal/src/table_kv_impl/wal.rs

@@ -172,6 +172,8 @@ impl<T: TableKv> WalManager for WalNamespaceImpl<T> {
         ctx: &manager::WriteContext,
         batch: &LogWriteBatch,
     ) -> Result<SequenceNumber> {
+        manager::collect_write_log_metrics(batch);
+
         self.namespace
             .write_log(ctx, batch)
             .await