Merge pull request #2201 from yangby-cryptape/pr/use-metric-for-memory-tracker

refactor: replace logs by metrics for memory tracker

Author: doitian

Cargo.lock

@@ -7,6 +7,7 @@ license = "MIT"
 
 [dependencies]
 ckb-logger = { path = "../logger" }
+ckb-metrics = { path = "../metrics" }
 ckb-db = { path = "../../db" }
 
 # TODO Why don't disable this crate by "target.*" in the crates which are dependent on this crate?

util/memory-tracker/Cargo.toml

@@ -1,11 +1,12 @@
 use std::{sync, thread, time};
 
-use ckb_logger::{debug, error, info};
+use ckb_logger::{error, info};
+use ckb_metrics::metrics;
 use futures::executor::block_on;
 use heim::units::information::byte;
 use jemalloc_ctl::{epoch, stats};
 
-use crate::{rocksdb::TrackRocksDBMemory, utils::HumanReadableSize};
+use crate::rocksdb::TrackRocksDBMemory;
 
 macro_rules! je_mib {
     ($key:ty) => {
@@ -21,7 +22,7 @@ macro_rules! je_mib {
 macro_rules! mib_read {
     ($mib:ident) => {
         if let Ok(value) = $mib.read() {
-            HumanReadableSize::from(value as u64)
+            value as i64
         } else {
             error!("failed to read jemalloc stats for {}", stringify!($mib));
             return;
@@ -58,17 +59,19 @@ pub fn track_current_process<Tracker: 'static + TrackRocksDBMemory + Sync + Send
             .name("MemoryTracker".to_string())
             .spawn(move || {
                 if let Ok(process) = block_on(heim::process::current()) {
-                    let pid = process.pid();
                     loop {
                         if je_epoch.advance().is_err() {
                             error!("failed to refresh the jemalloc stats");
                             return;
                         }
                         if let Ok(memory) = block_on(process.memory()) {
                             // Resident set size, amount of non-swapped physical memory.
-                            let rss: HumanReadableSize = memory.rss().get::<byte>().into();
+                            let rss = memory.rss().get::<byte>() as i64;
                             // Virtual memory size, total amount of memory.
-                            let virt: HumanReadableSize = memory.vms().get::<byte>().into();
+                            let vms = memory.vms().get::<byte>() as i64;
+
+                            metrics!(gauge, "ckb-sys.mem.process", rss, "type" => "rss");
+                            metrics!(gauge, "ckb-sys.mem.process", vms, "type" => "vms");
 
                             let allocated = mib_read!(allocated);
                             let resident = mib_read!(resident);
@@ -77,56 +80,15 @@ pub fn track_current_process<Tracker: 'static + TrackRocksDBMemory + Sync + Send
                             let retained = mib_read!(retained);
                             let metadata = mib_read!(metadata);
 
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", allocated, "type" => "allocated");
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", resident, "type" => "resident");
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", active, "type" => "active");
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", mapped, "type" => "mapped");
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", retained, "type" => "retained");
+                            metrics!(gauge, "ckb-sys.mem.jemalloc", metadata, "type" => "metadata");
+
                             if let Some(tracker) = tracker_opt.clone() {
-                                let stats = tracker.gather_memory_stats();
-                                debug!(
-                                    "CurrentProcess {{ \
-                                        pid: {}, rss: {}, virt: {}, \
-                                        Jemalloc: {{ \
-                                            allocated: {}, resident: {}, \
-                                            active: {}, mapped: {}, retained: {}, \
-                                            metadata: {} }}, \
-                                        RocksDB: {{ \
-                                            total: {}, cache: {}, readers: {}, \
-                                            memtables: {}, pinned: {}, \
-                                            cache-capacity: {} \
-                                        }} \
-                                    }}",
-                                    pid,
-                                    rss,
-                                    virt,
-                                    allocated,
-                                    resident,
-                                    active,
-                                    mapped,
-                                    retained,
-                                    metadata,
-                                    stats.total_memory,
-                                    stats.block_cache_usage,
-                                    stats.estimate_table_readers_mem,
-                                    stats.cur_size_all_mem_tables,
-                                    stats.block_cache_pinned_usage,
-                                    stats.block_cache_capacity,
-                                );
-                            } else {
-                                debug!(
-                                    "CurrentProcess {{ \
-                                        pid: {}, rss: {}, virt: {}, \
-                                        Jemalloc: {{ \
-                                            allocated: {}, resident: {}, \
-                                            active: {}, mapped: {}, retained: {}, \
-                                            metadata: {} }} \
-                                    }}",
-                                    pid,
-                                    rss,
-                                    virt,
-                                    allocated,
-                                    resident,
-                                    active,
-                                    mapped,
-                                    retained,
-                                    metadata,
-                                );
+                                let _ignored = tracker.gather_memory_stats();
                             }
                         } else {
                             error!("failed to fetch the memory information about current process");

util/memory-tracker/src/process.rs

@@ -1,39 +1,33 @@
 use ckb_db::internal::ops::{GetColumnFamilys, GetProperty, GetPropertyCF};
-use ckb_logger::trace;
+use ckb_metrics::metrics;
 
 use crate::utils::{sum_int_values, PropertyValue};
 
 // Ref: https://github.com/facebook/rocksdb/wiki/Memory-usage-in-RocksDB
 pub struct RocksDBMemoryStatistics {
-    pub total_memory: PropertyValue<u64>,
-    pub block_cache_usage: PropertyValue<u64>,
     pub estimate_table_readers_mem: PropertyValue<u64>,
+    pub size_all_mem_tables: PropertyValue<u64>,
     pub cur_size_all_mem_tables: PropertyValue<u64>,
-    pub block_cache_pinned_usage: PropertyValue<u64>,
     pub block_cache_capacity: PropertyValue<u64>,
+    pub block_cache_usage: PropertyValue<u64>,
+    pub block_cache_pinned_usage: PropertyValue<u64>,
 }
 
 pub trait TrackRocksDBMemory {
     fn gather_memory_stats(&self) -> RocksDBMemoryStatistics {
-        let block_cache_usage = self.gather_int_values("rocksdb.block-cache-usage");
-        let estimate_table_readers_mem =
-            self.gather_int_values("rocksdb.estimate-table-readers-mem");
-        let cur_size_all_mem_tables = self.gather_int_values("rocksdb.cur-size-all-mem-tables");
-        let block_cache_pinned_usage = self.gather_int_values("rocksdb.block-cache-pinned-usage");
-        let total_memory = sum_int_values(&[
-            block_cache_usage.clone(),
-            estimate_table_readers_mem.clone(),
-            cur_size_all_mem_tables.clone(),
-            block_cache_pinned_usage.clone(),
-        ]);
-        let block_cache_capacity = self.gather_int_values("rocksdb.block-cache-capacity");
+        let estimate_table_readers_mem = self.gather_int_values("estimate-table-readers-mem");
+        let size_all_mem_tables = self.gather_int_values("size-all-mem-tables");
+        let cur_size_all_mem_tables = self.gather_int_values("cur-size-all-mem-tables");
+        let block_cache_capacity = self.gather_int_values("block-cache-capacity");
+        let block_cache_usage = self.gather_int_values("block-cache-usage");
+        let block_cache_pinned_usage = self.gather_int_values("block-cache-pinned-usage");
         RocksDBMemoryStatistics {
-            total_memory,
-            block_cache_usage,
             estimate_table_readers_mem,
+            size_all_mem_tables,
             cur_size_all_mem_tables,
-            block_cache_pinned_usage,
             block_cache_capacity,
+            block_cache_usage,
+            block_cache_pinned_usage,
         }
     }
     fn gather_int_values(&self, key: &str) -> PropertyValue<u64>;
@@ -54,11 +48,11 @@ where
     fn gather_int_values(&self, key: &str) -> PropertyValue<u64> {
         let mut values = Vec::new();
         for (cf_name, cf) in self.get_cfs() {
-            let value_col = self
-                .property_int_value_cf(cf, key)
+            let value_col: PropertyValue<u64> = self
+                .property_int_value_cf(cf, &format!("rocksdb.{}", key))
                 .map_err(|err| format!("{}", err))
                 .into();
-            trace!("{}({}): {}", key, cf_name, value_col);
+            metrics!(gauge, "ckb-sys.mem.rocksdb", value_col.as_i64(), "type" => key.to_owned(), "cf" => cf_name.to_owned());
             values.push(value_col);
         }
         sum_int_values(&values)

util/memory-tracker/src/rocksdb.rs

@@ -36,6 +36,16 @@ pub enum PropertyValue<T> {
     Error(String),
 }
 
+impl PropertyValue<u64> {
+    pub(crate) fn as_i64(&self) -> i64 {
+        match self {
+            Self::Value(v) => *v as i64,
+            Self::Null => -1,
+            Self::Error(_) => -2,
+        }
+    }
+}
+
 impl fmt::Display for PropertyValue<u64> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {