chore: adjust storage engine related metrics (#3261)

* chore: adjust metrics to metric engine and mito engine

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* adjust more mito bucket

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix compile

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

src/metric-engine/src/engine/put.rs

@@ -28,7 +28,7 @@ use crate::engine::MetricEngineInner;
 use crate::error::{
     ColumnNotFoundSnafu, ForbiddenPhysicalAlterSnafu, LogicalRegionNotFoundSnafu, Result,
 };
-use crate::metrics::FORBIDDEN_OPERATION_COUNT;
+use crate::metrics::{FORBIDDEN_OPERATION_COUNT, MITO_OPERATION_ELAPSED};
 use crate::utils::{to_data_region_id, to_metadata_region_id};
 
 // A random number
@@ -65,6 +65,10 @@ impl MetricEngineInner {
         logical_region_id: RegionId,
         mut request: RegionPutRequest,
     ) -> Result<AffectedRows> {
+        let _timer = MITO_OPERATION_ELAPSED
+            .with_label_values(&["put"])
+            .start_timer();
+
         let physical_region_id = *self
             .state
             .read()

src/metric-engine/src/engine/read.rs

@@ -30,6 +30,7 @@ use crate::engine::MetricEngineInner;
 use crate::error::{
     InvalidMetadataSnafu, LogicalRegionNotFoundSnafu, MitoReadOperationSnafu, Result,
 };
+use crate::metrics::MITO_OPERATION_ELAPSED;
 use crate::utils;
 
 impl MetricEngineInner {
@@ -57,6 +58,10 @@ impl MetricEngineInner {
         region_id: RegionId,
         request: ScanRequest,
     ) -> Result<SendableRecordBatchStream> {
+        let _timer = MITO_OPERATION_ELAPSED
+            .with_label_values(&["read_physical"])
+            .start_timer();
+
         self.mito
             .handle_query(region_id, request)
             .await
@@ -68,6 +73,10 @@ impl MetricEngineInner {
         logical_region_id: RegionId,
         request: ScanRequest,
     ) -> Result<SendableRecordBatchStream> {
+        let _timer = MITO_OPERATION_ELAPSED
+            .with_label_values(&["read"])
+            .start_timer();
+
         let physical_region_id = self.get_physical_region_id(logical_region_id).await?;
         let data_region_id = utils::to_data_region_id(physical_region_id);
         let request = self

src/metric-engine/src/metrics.rs

@@ -17,6 +17,9 @@
 use lazy_static::lazy_static;
 use prometheus::*;
 
+/// Stage label.
+pub const OPERATION_LABEL: &str = "operation";
+
 lazy_static! {
     /// Gauge for opened regions
     pub static ref PHYSICAL_REGION_COUNT: IntGauge =
@@ -37,4 +40,13 @@ lazy_static! {
     /// Counter for forbidden operations
     pub static ref FORBIDDEN_OPERATION_COUNT: IntCounter =
         register_int_counter!("greptime_metric_engine_forbidden_request", "metric forbidden request").unwrap();
+
+    /// Histogram for underlying mito operations
+    pub static ref MITO_OPERATION_ELAPSED: HistogramVec = register_histogram_vec!(
+        "greptime_metric_engine_mito_op_elapsed",
+        "metric engine's mito operation elapsed",
+        &[OPERATION_LABEL],
+        vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
+    )
+    .unwrap();
 }

src/mito2/src/metrics.rs

@@ -35,7 +35,8 @@ lazy_static! {
     pub static ref HANDLE_REQUEST_ELAPSED: HistogramVec = register_histogram_vec!(
             "greptime_mito_handle_request_elapsed",
             "mito handle request elapsed",
-            &[TYPE_LABEL]
+            &[TYPE_LABEL],
+            vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 60.0, 300.0]
         )
         .unwrap();
 
@@ -55,7 +56,8 @@ lazy_static! {
     pub static ref FLUSH_ELAPSED: HistogramVec = register_histogram_vec!(
             "greptime_mito_flush_elapsed",
             "mito flush elapsed",
-            &[TYPE_LABEL]
+            &[TYPE_LABEL],
+            vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
         )
         .unwrap();
     /// Histogram of flushed bytes.
@@ -75,7 +77,8 @@ lazy_static! {
     pub static ref WRITE_STAGE_ELAPSED: HistogramVec = register_histogram_vec!(
             "greptime_mito_write_stage_elapsed",
             "mito write stage elapsed",
-            &[STAGE_LABEL]
+            &[STAGE_LABEL],
+            vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
         )
         .unwrap();
     /// Counter of rows to write.
@@ -93,7 +96,8 @@ lazy_static! {
     pub static ref COMPACTION_STAGE_ELAPSED: HistogramVec = register_histogram_vec!(
         "greptime_mito_compaction_stage_elapsed",
         "mito compaction stage elapsed",
-        &[STAGE_LABEL]
+        &[STAGE_LABEL],
+        vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
     )
     .unwrap();
     /// Timer of whole compaction task.
@@ -112,7 +116,8 @@ lazy_static! {
     pub static ref READ_STAGE_ELAPSED: HistogramVec = register_histogram_vec!(
         "greptime_mito_read_stage_elapsed",
         "mito read stage elapsed",
-        &[STAGE_LABEL]
+        &[STAGE_LABEL],
+        vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
     )
     .unwrap();
     /// Counter of rows read.

src/servers/src/http/prom_store.rs

@@ -59,12 +59,12 @@ pub async fn route_write_without_metric_engine(
     Extension(query_ctx): Extension<QueryContextRef>,
     RawBody(body): RawBody,
 ) -> Result<(StatusCode, ())> {
-    let request = decode_remote_write_request(body).await?;
     let db = params.db.clone().unwrap_or_default();
     let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
         .with_label_values(&[db.as_str()])
         .start_timer();
 
+    let request = decode_remote_write_request(body).await?;
     // reject if physical table is specified when metric engine is disabled
     if params.physical_table.is_some() {
         return UnexpectedPhysicalTableSnafu {}.fail();
@@ -81,18 +81,18 @@ pub async fn remote_write(
     Extension(mut query_ctx): Extension<QueryContextRef>,
     RawBody(body): RawBody,
 ) -> Result<(StatusCode, ())> {
-    let request = decode_remote_write_request(body).await?;
     let db = params.db.clone().unwrap_or_default();
+    let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
+        .with_label_values(&[db.as_str()])
+        .start_timer();
+
+    let request = decode_remote_write_request(body).await?;
     if let Some(physical_table) = params.physical_table {
         let mut new_query_ctx = query_ctx.as_ref().clone();
         new_query_ctx.set_extension(PHYSICAL_TABLE_PARAM, physical_table);
         query_ctx = Arc::new(new_query_ctx);
     }
 
-    let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_WRITE_ELAPSED
-        .with_label_values(&[db.as_str()])
-        .start_timer();
-
     handler.write(request, query_ctx, true).await?;
     Ok((StatusCode::NO_CONTENT, ()))
 }
@@ -117,12 +117,13 @@ pub async fn remote_read(
     Extension(query_ctx): Extension<QueryContextRef>,
     RawBody(body): RawBody,
 ) -> Result<PromStoreResponse> {
-    let request = decode_remote_read_request(body).await?;
     let db = params.db.clone().unwrap_or_default();
-
     let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_READ_ELAPSED
         .with_label_values(&[db.as_str()])
         .start_timer();
+
+    let request = decode_remote_read_request(body).await?;
+
     handler.read(request, query_ctx).await
 }