refactor: add grpc write failed counter metric (#963)

## Rationale
add grpc write failed counter metric

## Detailed Changes
- add counter metric in grpc proxy
- add counter metric in grpc remote engine service

## Test Plan 
Existing tests

Author: baojinri

Cargo.lock

@@ -34,6 +34,7 @@ meta_client = { workspace = true }
 paste = { workspace = true }
 prom-remote-api = { workspace = true, features = ["warp"] }
 prometheus = { workspace = true }
+prometheus-static-metric = { workspace = true }
 prost = { workspace = true }
 query_engine = { workspace = true }
 query_frontend = { workspace = true }

proxy/Cargo.toml

@@ -0,0 +1,28 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+// Grpc proxy metrics
+
+use lazy_static::lazy_static;
+use prometheus::{register_int_counter_vec, IntCounterVec};
+use prometheus_static_metric::{auto_flush_from, make_auto_flush_static_metric};
+
+make_auto_flush_static_metric! {
+    pub label_enum GrpcTypeKind {
+        write_failed,
+    }
+
+    pub struct GrpcHandlerCounterVec: LocalIntCounter {
+        "type" => GrpcTypeKind,
+    }
+}
+
+lazy_static! {
+    pub static ref GRPC_HANDLER_COUNTER_VEC_GLOBAL: IntCounterVec =
+        register_int_counter_vec!("grpc_handler_counter", "Grpc handler counter", &["type"])
+            .unwrap();
+}
+
+lazy_static! {
+    pub static ref GRPC_HANDLER_COUNTER_VEC: GrpcHandlerCounterVec =
+        auto_flush_from!(GRPC_HANDLER_COUNTER_VEC_GLOBAL, GrpcHandlerCounterVec);
+}

proxy/src/grpc/metrics.rs

@@ -1,5 +1,6 @@
 // Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
+mod metrics;
 mod prom_query;
 mod route;
 mod sql_query;

proxy/src/grpc/mod.rs

@@ -3,24 +3,42 @@
 use ceresdbproto::storage::{WriteRequest, WriteResponse};
 use query_engine::executor::Executor as QueryExecutor;
 
-use crate::{error, error::build_ok_header, Context, Proxy};
+use crate::{
+    error, error::build_ok_header, grpc::metrics::GRPC_HANDLER_COUNTER_VEC, Context, Proxy,
+};
 
 impl<Q: QueryExecutor + 'static> Proxy<Q> {
     pub async fn handle_write(&self, ctx: Context, req: WriteRequest) -> WriteResponse {
         self.hotspot_recorder.inc_write_reqs(&req).await;
+
+        let mut num_rows = 0;
+        for table_request in &req.table_requests {
+            for entry in &table_request.entries {
+                num_rows += entry.field_groups.len();
+            }
+        }
+
         match self.handle_write_internal(ctx, req).await {
             Err(e) => {
                 error!("Failed to handle write, err:{e}");
+                GRPC_HANDLER_COUNTER_VEC
+                    .write_failed
+                    .inc_by(num_rows as u64);
                 WriteResponse {
                     header: Some(error::build_err_header(e)),
                     ..Default::default()
                 }
             }
-            Ok(v) => WriteResponse {
-                header: Some(build_ok_header()),
-                success: v.success,
-                failed: v.failed,
-            },
+            Ok(v) => {
+                GRPC_HANDLER_COUNTER_VEC
+                    .write_failed
+                    .inc_by(v.failed as u64);
+                WriteResponse {
+                    header: Some(build_ok_header()),
+                    success: v.success,
+                    failed: v.failed,
+                }
+            }
         }
     }
 }

proxy/src/grpc/write.rs

@@ -3,7 +3,10 @@
 // Grpc server metrics
 
 use lazy_static::lazy_static;
-use prometheus::{exponential_buckets, register_histogram_vec, HistogramVec};
+use prometheus::{
+    exponential_buckets, register_histogram_vec, register_int_counter_vec, HistogramVec,
+    IntCounterVec,
+};
 use prometheus_static_metric::{auto_flush_from, make_auto_flush_static_metric};
 
 // Register auto flush static metrics.
@@ -32,6 +35,14 @@ make_auto_flush_static_metric! {
         "type" => RemoteEngineTypeKind,
     }
 
+    pub label_enum RemoteEngineGrpcTypeKind {
+        write_failed,
+    }
+
+    pub struct RemoteEngineGrpcHandlerCounterVec: LocalIntCounter {
+        "type" => RemoteEngineGrpcTypeKind,
+    }
+
     pub label_enum MetaEventTypeKind {
         open_shard,
         close_shard,
@@ -64,6 +75,13 @@ lazy_static! {
             exponential_buckets(0.0005, 2.0, 20).unwrap()
         )
         .unwrap();
+    pub static ref REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC_GLOBAL: IntCounterVec =
+        register_int_counter_vec!(
+            "remote_engine_grpc_handler_counter",
+            "Remote engine grpc handler counter",
+            &["type"]
+        )
+        .unwrap();
     pub static ref META_EVENT_GRPC_HANDLER_DURATION_HISTOGRAM_VEC_GLOBAL: HistogramVec =
         register_histogram_vec!(
             "meta_event_grpc_handler_duration",
@@ -84,6 +102,10 @@ lazy_static! {
         REMOTE_ENGINE_GRPC_HANDLER_DURATION_HISTOGRAM_VEC_GLOBAL,
         RemoteEngineGrpcHandlerDurationHistogramVec
     );
+    pub static ref REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC: RemoteEngineGrpcHandlerCounterVec = auto_flush_from!(
+        REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC_GLOBAL,
+        RemoteEngineGrpcHandlerCounterVec
+    );
     pub static ref META_EVENT_GRPC_HANDLER_DURATION_HISTOGRAM_VEC: MetaEventGrpcHandlerDurationHistogramVec = auto_flush_from!(
         META_EVENT_GRPC_HANDLER_DURATION_HISTOGRAM_VEC_GLOBAL,
         MetaEventGrpcHandlerDurationHistogramVec

server/src/grpc/metrics.rs

@@ -31,7 +31,9 @@ use tokio_stream::wrappers::ReceiverStream;
 use tonic::{Request, Response, Status};
 
 use crate::grpc::{
-    metrics::REMOTE_ENGINE_GRPC_HANDLER_DURATION_HISTOGRAM_VEC,
+    metrics::{
+        REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC, REMOTE_ENGINE_GRPC_HANDLER_DURATION_HISTOGRAM_VEC,
+    },
     remote_engine_service::error::{ErrNoCause, ErrWithCause, Result, StatusCode},
 };
 
@@ -353,20 +355,29 @@ async fn handle_write(ctx: HandlerContext, request: WriteRequest) -> Result<Writ
             msg: "fail to convert write request",
         })?;
 
+    let num_rows = write_request.write_request.row_group.num_rows();
     let table = find_table_by_identifier(&ctx, &write_request.table)?;
 
-    let affected_rows = table
+    let res = table
         .write(write_request.write_request)
         .await
         .box_err()
         .context(ErrWithCause {
             code: StatusCode::Internal,
             msg: format!("fail to write table, table:{:?}", write_request.table),
-        })?;
-    Ok(WriteResponse {
-        header: None,
-        affected_rows: affected_rows as u64,
-    })
+        });
+    match res {
+        Ok(affected_rows) => Ok(WriteResponse {
+            header: None,
+            affected_rows: affected_rows as u64,
+        }),
+        Err(e) => {
+            REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC
+                .write_failed
+                .inc_by(num_rows as u64);
+            Err(e)
+        }
+    }
 }
 
 async fn handle_get_table_info(