refactor: add grpc query nums metrics (#1090)

## Rationale
See title

## Detailed Changes
- Add query nums metric in grpc proxy.
- Add query nums metric in grpc remote engine service.

## Test Plan
Existing tests

Author: baojinri

proxy/src/grpc/sql_query.rs

@@ -37,6 +37,7 @@ const STREAM_QUERY_CHANNEL_LEN: usize = 20;
 
 impl<Q: QueryExecutor + 'static> Proxy<Q> {
     pub async fn handle_sql_query(&self, ctx: Context, req: SqlQueryRequest) -> SqlQueryResponse {
+        GRPC_HANDLER_COUNTER_VEC.query.inc();
         self.hotspot_recorder.inc_sql_query_reqs(&req).await;
         match self.handle_sql_query_internal(ctx, req).await {
             Err(e) => {
@@ -80,6 +81,7 @@ impl<Q: QueryExecutor + 'static> Proxy<Q> {
         ctx: Context,
         req: SqlQueryRequest,
     ) -> BoxStream<'static, SqlQueryResponse> {
+        GRPC_HANDLER_COUNTER_VEC.stream_query.inc();
         self.hotspot_recorder.inc_sql_query_reqs(&req).await;
         match self.clone().handle_stream_query_internal(ctx, req).await {
             Err(e) => stream::once(async {

proxy/src/metrics.rs

@@ -12,8 +12,10 @@ make_auto_flush_static_metric! {
         write_failed,
         query_succeeded,
         query_failed,
+        query,
         stream_query_succeeded,
         stream_query_failed,
+        stream_query,
         write_succeeded_row,
         write_failed_row,
         query_succeeded_row,

server/src/grpc/metrics.rs

@@ -38,8 +38,7 @@ make_auto_flush_static_metric! {
     pub label_enum RemoteEngineGrpcTypeKind {
         write_succeeded,
         write_failed,
-        query_succeeded,
-        query_failed,
+        stream_query,
         stream_query_succeeded,
         stream_query_failed,
         write_succeeded_row,

server/src/grpc/remote_engine_service/mod.rs

@@ -235,6 +235,7 @@ impl<Q: QueryExecutor + 'static> RemoteEngineService for RemoteEngineServiceImpl
         &self,
         request: Request<ReadRequest>,
     ) -> std::result::Result<Response<Self::ReadStream>, Status> {
+        REMOTE_ENGINE_GRPC_HANDLER_COUNTER_VEC.stream_query.inc();
         match self.stream_read_internal(request).await {
             Ok(stream) => {
                 let new_stream: Self::ReadStream = Box::pin(stream.map(|res| match res {