[Feat] impl runtime (#49)

* [Feat] impl runtime

* [Fix] cargo sort

Cargo.toml

@@ -42,6 +42,7 @@ log = { workspace = true }
 murmur2 = "0.1"
 mysql = "16.1"
 net2 = "0.2"
+pin-project-lite = "0.2"
 prometheus-client = { workspace = true }
 quick-error = "1.2"
 r2d2 = "0.8.3"
@@ -67,6 +68,7 @@ serial_test_derive = "2.0"
 tempfile = "3.0"
 test-log = "0.2"
 time = "0.3"
+tokio-test = "0.4"
 
 [profile.release]
 debug = true

src/error.rs

@@ -97,6 +97,12 @@ quick_error! {
             display("Query database error, err:{}", e)
             cause(e)
         }
+        JoinTask(e: tokio::task::JoinError) {
+            from()
+            description("Tokio join error")
+            display("Tokio join error, err:{}", e)
+            cause(e)
+        }
     }
 }
 

src/monitors/runtime_metrics.rs

@@ -16,27 +16,30 @@
  */
 
 use prometheus_client::{
-    encoding::{EncodeLabelSet, EncodeLabelValue},
+    encoding::EncodeLabelSet,
     metrics::{family::Family, gauge},
     registry::Registry,
 };
 
-#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelValue)]
-pub enum ObClientRuntimeGaugeType {
-    Default = 0,
-    ConnWriter = 1,
-    ConnReader = 2,
+use crate::monitors::prometheus::OBKV_CLIENT_REGISTRY;
+
+lazy_static! {
+    pub static ref OBKV_RUNTIME_GAUGE_METRICS: RuntimeGaugeMetrics = {
+        let runtime_metrics = RuntimeGaugeMetrics::default();
+        runtime_metrics.register(&mut OBKV_CLIENT_REGISTRY.lock().unwrap().registry);
+        runtime_metrics
+    };
 }
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
-pub struct RuntimeThreadLabels {
-    pub rt_type: ObClientRuntimeGaugeType,
+pub struct RuntimeNameLabels {
+    pub string_type: String,
 }
 
 #[derive(Default)]
 pub struct RuntimeGaugeMetrics {
-    runtime_thread_alive_gauges: Family<RuntimeThreadLabels, gauge::Gauge>,
-    runtime_thread_idle_gauges: Family<RuntimeThreadLabels, gauge::Gauge>,
+    runtime_thread_alive_gauges: Family<RuntimeNameLabels, gauge::Gauge>,
+    runtime_thread_idle_gauges: Family<RuntimeNameLabels, gauge::Gauge>,
 }
 
 impl RuntimeGaugeMetrics {
@@ -54,35 +57,95 @@ impl RuntimeGaugeMetrics {
         );
     }
 
-    pub fn on_thread_start(&self, rt_type: ObClientRuntimeGaugeType) {
+    pub fn on_thread_start(&self, rt_name: &str) {
         self.runtime_thread_alive_gauges
-            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .get_or_create(&RuntimeNameLabels {
+                string_type: rt_name.to_string(),
+            })
             .inc();
     }
 
-    pub fn on_thread_stop(&self, rt_type: ObClientRuntimeGaugeType) {
+    pub fn on_thread_stop(&self, rt_name: &str) {
         self.runtime_thread_alive_gauges
-            .get_or_create(&RuntimeThreadLabels { rt_type })
+            .get_or_create(&RuntimeNameLabels {
+                string_type: rt_name.to_string(),
+            })
             .dec();
     }
 
-    pub fn on_thread_park(&self, rt_type: ObClientRuntimeGaugeType) {
-        self.runtime_thread_alive_gauges
-            .get_or_create(&RuntimeThreadLabels { rt_type })
+    pub fn on_thread_park(&self, rt_name: &str) {
+        self.runtime_thread_idle_gauges
+            .get_or_create(&RuntimeNameLabels {
+                string_type: rt_name.to_string(),
+            })
             .inc();
     }
 
-    pub fn on_thread_unpark(&self, rt_type: ObClientRuntimeGaugeType) {
-        self.runtime_thread_alive_gauges
-            .get_or_create(&RuntimeThreadLabels { rt_type })
+    pub fn on_thread_unpark(&self, rt_name: &str) {
+        self.runtime_thread_idle_gauges
+            .get_or_create(&RuntimeNameLabels {
+                string_type: rt_name.to_string(),
+            })
             .dec();
     }
 
-    pub fn get_runtime_thread_alive_gauges(&self) -> &Family<RuntimeThreadLabels, gauge::Gauge> {
+    pub fn get_runtime_thread_alive_gauges(&self) -> &Family<RuntimeNameLabels, gauge::Gauge> {
         &self.runtime_thread_alive_gauges
     }
 
-    pub fn get_runtime_thread_idle_gauges(&self) -> &Family<RuntimeThreadLabels, gauge::Gauge> {
+    pub fn get_runtime_thread_idle_gauges(&self) -> &Family<RuntimeNameLabels, gauge::Gauge> {
         &self.runtime_thread_idle_gauges
     }
 }
+
+/// Runtime metrics.
+#[derive(Debug)]
+pub struct RuntimeMetrics {
+    pub runtime_name: String,
+}
+
+impl RuntimeMetrics {
+    pub fn new(runtime_name: &str) -> Self {
+        Self {
+            runtime_name: runtime_name.to_owned(),
+        }
+    }
+
+    pub fn set_runtime_name(&mut self, runtime_name: String) {
+        self.runtime_name = runtime_name;
+    }
+
+    pub fn on_thread_start(&self) {
+        OBKV_RUNTIME_GAUGE_METRICS.on_thread_start(&self.runtime_name);
+    }
+
+    pub fn on_thread_stop(&self) {
+        OBKV_RUNTIME_GAUGE_METRICS.on_thread_stop(&self.runtime_name);
+    }
+
+    pub fn on_thread_park(&self) {
+        OBKV_RUNTIME_GAUGE_METRICS.on_thread_park(&self.runtime_name);
+    }
+
+    pub fn on_thread_unpark(&self) {
+        OBKV_RUNTIME_GAUGE_METRICS.on_thread_unpark(&self.runtime_name);
+    }
+
+    pub fn get_runtime_thread_alive_gauges(&self) -> i64 {
+        OBKV_RUNTIME_GAUGE_METRICS
+            .get_runtime_thread_alive_gauges()
+            .get_or_create(&RuntimeNameLabels {
+                string_type: self.runtime_name.clone(),
+            })
+            .get()
+    }
+
+    pub fn get_runtime_thread_idle_gauges(&self) -> i64 {
+        OBKV_RUNTIME_GAUGE_METRICS
+            .get_runtime_thread_idle_gauges()
+            .get_or_create(&RuntimeNameLabels {
+                string_type: self.runtime_name.clone(),
+            })
+            .get()
+    }
+}