feat: add obkv operation metrics (#1122)

## Rationale
In one of our deployment, we see lots of obkv timeout requests, so we
need to add some metrics for trouble shooting.

## Detailed Changes
Add metrics for different obkv operations.

## Test Plan

components/table_kv/Cargo.toml

@@ -11,9 +11,11 @@ workspace = true
 workspace = true
 
 [dependencies]
+lazy_static = { workspace = true }
 log = { workspace = true }
 macros = { workspace = true }
 obkv-table-client-rs = { git = "https://github.com/oceanbase/obkv-table-client-rs.git", rev = "f7c7a2c866fc83af4862b02447538f2abc5e8839" }
+prometheus = { workspace = true }
 serde = { workspace = true }
 snafu = { workspace = true }
 time_ext = { workspace = true }

components/table_kv/src/lib.rs

@@ -6,6 +6,7 @@ use std::{fmt, time::Duration};
 
 pub mod config;
 pub mod memory;
+mod metrics;
 pub mod obkv;
 #[cfg(test)]
 mod tests;

components/table_kv/src/metrics.rs

@@ -0,0 +1,15 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+use lazy_static::lazy_static;
+use prometheus::{exponential_buckets, register_histogram_vec, HistogramVec};
+
+lazy_static! {
+    // Buckets: 0.001, .., 0.001 * 2^15 = 32.7s
+    pub static ref OBKV_OP_DURATION_HISTOGRAM: HistogramVec = register_histogram_vec!(
+        "obkv_op_duration",
+        "Histogram for duration of different obkv operations",
+        &["type"],
+        exponential_buckets(0.001, 2.0, 15).unwrap()
+    )
+    .unwrap();
+}

components/table_kv/src/obkv.rs

@@ -13,8 +13,8 @@ use obkv::{
 use snafu::{ensure, Backtrace, OptionExt, ResultExt, Snafu};
 
 use crate::{
-    config::ObkvConfig, KeyBoundary, ScanContext, ScanIter, ScanRequest, SeekKey, TableError,
-    TableKv, WriteBatch, WriteContext,
+    config::ObkvConfig, metrics::OBKV_OP_DURATION_HISTOGRAM, KeyBoundary, ScanContext, ScanIter,
+    ScanRequest, SeekKey, TableError, TableKv, WriteBatch, WriteContext,
 };
 
 #[cfg(test)]
@@ -484,6 +484,10 @@ impl TableKv for ObkvImpl {
         table_name: &str,
         write_batch: ObkvWriteBatch,
     ) -> Result<()> {
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["write"])
+            .start_timer();
+
         let results = self
             .client
             .execute_batch(table_name, write_batch.batch_op)
@@ -506,6 +510,10 @@ impl TableKv for ObkvImpl {
     }
 
     fn get(&self, table_name: &str, key: &[u8]) -> Result<Option<Vec<u8>>> {
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["get"])
+            .start_timer();
+
         let mut values = self
             .client
             .get(
@@ -519,6 +527,10 @@ impl TableKv for ObkvImpl {
     }
 
     fn get_batch(&self, table_name: &str, keys: Vec<&[u8]>) -> Result<Vec<Option<Vec<u8>>>> {
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["get_batch"])
+            .start_timer();
+
         let mut batch_ops = ObTableBatchOperation::with_ops_num_raw(keys.len());
         let mut batch_res = Vec::with_capacity(keys.len());
 
@@ -543,6 +555,10 @@ impl TableKv for ObkvImpl {
     }
 
     fn delete(&self, table_name: &str, key: &[u8]) -> std::result::Result<(), Self::Error> {
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["delete"])
+            .start_timer();
+
         self.client
             .delete(table_name, bytes_to_values(key))
             .context(DeleteData { table_name })?;
@@ -555,6 +571,10 @@ impl TableKv for ObkvImpl {
         table_name: &str,
         keys: Vec<Vec<u8>>,
     ) -> std::result::Result<(), Self::Error> {
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["delete_batch"])
+            .start_timer();
+
         let mut batch_ops = ObTableBatchOperation::with_ops_num_raw(keys.len());
         for key in keys {
             batch_ops.delete(bytes_to_values(&key));
@@ -742,6 +762,10 @@ impl ScanIter for ObkvScanIter {
     fn next(&mut self) -> Result<bool> {
         assert!(self.valid());
 
+        let _timer = OBKV_OP_DURATION_HISTOGRAM
+            .with_label_values(&["scan_next"])
+            .start_timer();
+
         // Try to fetch next key-value from current result set.
         if self.step_result_set()? {
             return Ok(true);