refactor(query): record physical plan level metrics into profiling log (#15092)

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): tracking plan metrics

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): support scoped metrics registry

* refactor(metrics): keep metric name suffix

* refactor(metrics): fix unit test failure

Author: zhang2014

Cargo.lock

@@ -46,6 +46,8 @@ pprof = { version = "0.11.1", features = [
     "protobuf-codec",
     "protobuf",
 ] }
+prometheus-client = { workspace = true }
+prometheus-parse = "0.2.3"
 regex = { workspace = true }
 replace_with = "0.1.7"
 semver = { workspace = true }

src/common/base/Cargo.toml

@@ -25,6 +25,7 @@
 #![feature(slice_swap_unchecked)]
 #![feature(lint_reasons)]
 #![feature(variant_count)]
+#![feature(lazy_cell)]
 
 pub mod base;
 pub mod containers;

src/common/base/src/lib.rs

@@ -21,35 +21,49 @@ use prometheus_client::encoding::MetricEncoder;
 use prometheus_client::metrics::MetricType;
 use prometheus_client::metrics::TypedMetric;
 
+use crate::runtime::metrics::registry::DatabendMetric;
+use crate::runtime::metrics::registry::ScopedRegistry;
+use crate::runtime::metrics::sample::MetricSample;
+use crate::runtime::metrics::sample::MetricValue;
+
 #[derive(Debug)]
 pub struct Counter {
     value: Arc<AtomicU64>,
+    index: usize,
 }
 
 impl Clone for Counter {
     fn clone(&self) -> Self {
         Self {
+            index: self.index,
             value: self.value.clone(),
         }
     }
 }
 
-impl Default for Counter {
-    fn default() -> Self {
+impl Counter {
+    pub fn create(index: usize) -> Counter {
         Counter {
+            index,
             value: Arc::new(AtomicU64::new(0)),
         }
     }
-}
 
-impl Counter {
     /// Increase the [`Counter`] by 1, returning the previous value.
     pub fn inc(&self) -> u64 {
+        ScopedRegistry::op(self.index, |m: &Self| {
+            m.value.fetch_add(1, Ordering::Relaxed);
+        });
+
         self.value.fetch_add(1, Ordering::Relaxed)
     }
 
     /// Increase the [`Counter`] by `v`, returning the previous value.
     pub fn inc_by(&self, v: u64) -> u64 {
+        ScopedRegistry::op(self.index, |m: &Self| {
+            m.value.fetch_add(v, Ordering::Relaxed);
+        });
+
         self.value.fetch_add(v, Ordering::Relaxed)
     }
 
@@ -77,3 +91,17 @@ impl EncodeMetric for Counter {
         Self::TYPE
     }
 }
+
+impl DatabendMetric for Counter {
+    fn reset_metric(&self) {
+        self.value.store(0, Ordering::Release);
+    }
+
+    fn sample(&self, name: &str, samples: &mut Vec<MetricSample>) {
+        samples.push(MetricSample {
+            name: format!("{}_total", name),
+            labels: Default::default(),
+            value: MetricValue::Counter(self.get() as f64),
+        });
+    }
+}