[Feat] runtime metrics & [Chore] update dependencies

Cargo.toml

@@ -26,12 +26,13 @@ path = "benches/concurrent_insert/mod.rs"
 anyhow = "1.0"
 log = "0.4"
 toml = "0.7.3"
-prometheus-client = "0.20"
+prometheus-client = "0.21"
+tokio = { version = "1", features = ["full"] }
 
 [dependencies]
 anyhow = { workspace = true }
 byteorder = "1.2"
-bytes = "0.4"
+bytes = "1.4"
 chrono = "0.4"
 crossbeam = "0.8.2"
 futures = "0.1"
@@ -49,22 +50,23 @@ reqwest = { version = "0.11.13", default-features = false, features = ["rustls-t
 rust-crypto = "0.2"
 scheduled-thread-pool = "0.2"
 serde = "1.0"
-serde_bytes = "0.10"
+serde_bytes = "0.11"
 serde_derive = "1.0"
 serde_json = "1.0"
-spin = "0.5.0"
-tokio-codec = "0.1"
+spin = "0.9"
+tokio = { workspace = true }
+tokio-util = "0.7"
 uuid = { version = "1.3.0", default-features = false, features = ["v4", "fast-rng", "macro-diagnostics"] }
-zstd = "0.11"
+zstd = "0.12"
 
 [dev-dependencies]
 env_logger = "0.10"
 scoped_threadpool = "0.1"
-serial_test = "0.2"
-serial_test_derive = "0.2"
+serial_test = "2.0"
+serial_test_derive = "2.0"
 tempfile = "3.0"
 test-log = "0.2"
-time = "0.1"
+time = "0.3"
 
 [profile.release]
 debug = true

src/lib.rs

@@ -23,7 +23,7 @@ extern crate mysql;
 #[macro_use]
 extern crate serde;
 extern crate serde_bytes;
-extern crate tokio_codec;
+extern crate tokio_util;
 #[macro_use]
 extern crate serde_derive;
 #[macro_use]
@@ -51,6 +51,7 @@ pub mod error;
 mod location;
 pub mod monitors;
 mod rpc;
+mod runtime;
 pub mod serde_obkv;
 mod util;
 pub use self::{

src/monitors/mod.rs

@@ -19,3 +19,4 @@ pub mod client_metrics;
 pub mod prometheus;
 pub mod proxy_metrics;
 pub mod rpc_metrics;
+pub mod runtime_metrics;

src/monitors/runtime_metrics.rs

@@ -0,0 +1,88 @@
+/*-
+ * #%L
+ * OBKV Table Client Framework
+ * %%
+ * Copyright (C) 2021 OceanBase
+ * %%
+ * OBKV Table Client Framework is licensed under Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the
+ * Mulan PSL v2. You may obtain a copy of Mulan PSL v2 at:
+ *          http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY
+ * KIND, EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+ * NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * #L%
+ */
+
+use prometheus_client::{
+    encoding::{EncodeLabelSet, EncodeLabelValue},
+    metrics::{family::Family, gauge},
+    registry::Registry,
+};
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelValue)]
+pub enum ObClientRuntimeGaugeType {
+    Default = 0,
+    ConnWriter = 1,
+    ConnReader = 2,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct RuntimeThreadLabels {
+    pub rt_type: ObClientRuntimeGaugeType,
+}
+
+#[derive(Default)]
+pub struct RuntimeGaugeMetrics {
+    runtime_thread_alive_gauges: Family<RuntimeThreadLabels, gauge::Gauge>,
+    runtime_thread_idle_gauges: Family<RuntimeThreadLabels, gauge::Gauge>,
+}
+
+impl RuntimeGaugeMetrics {
+    pub fn register(&self, registry: &mut Registry) {
+        let sub_registry = registry.sub_registry_with_prefix("runtime");
+        sub_registry.register(
+            "Runtime Alive Threads Gauges ",
+            "Client alive threads num.",
+            self.runtime_thread_alive_gauges.clone(),
+        );
+        sub_registry.register(
+            "Runtime Idle Threads Gauges ",
+            "Client idle threads num.",
+            self.runtime_thread_idle_gauges.clone(),
+        );
+    }
+
+    pub fn on_thread_start(&self, rt_type: ObClientRuntimeGaugeType) {
+        self.runtime_thread_alive_gauges
+            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .inc();
+    }
+
+    pub fn on_thread_stop(&self, rt_type: ObClientRuntimeGaugeType) {
+        self.runtime_thread_alive_gauges
+            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .dec();
+    }
+
+    pub fn on_thread_park(&self, rt_type: ObClientRuntimeGaugeType) {
+        self.runtime_thread_alive_gauges
+            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .inc();
+    }
+
+    pub fn on_thread_unpark(&self, rt_type: ObClientRuntimeGaugeType) {
+        self.runtime_thread_alive_gauges
+            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .dec();
+    }
+
+    pub fn get_runtime_thread_alive_gauges(&self) -> &Family<RuntimeThreadLabels, gauge::Gauge> {
+        &self.runtime_thread_alive_gauges
+    }
+
+    pub fn get_runtime_thread_idle_gauges(&self) -> &Family<RuntimeThreadLabels, gauge::Gauge> {
+        &self.runtime_thread_idle_gauges
+    }
+}

src/rpc/mod.rs

@@ -38,7 +38,7 @@ use byteorder::{BigEndian, ByteOrder};
 use bytes::BytesMut;
 use crossbeam::channel::{bounded, unbounded, Receiver, Sender};
 use net2::{TcpBuilder, TcpStreamExt};
-use tokio_codec::{Decoder, Encoder};
+use tokio_util::codec::{Decoder, Encoder};
 use uuid::Uuid;
 
 use self::protocol::{