*: directly report statistics to pd

Cargo.toml

@@ -80,7 +80,7 @@ git = "https://github.com/pingcap/kvproto.git"
 git = "https://github.com/pingcap/tipb.git"
 
 [dependencies.prometheus]
-version = "0.3"
+git = "https://github.com/pingcap/rust-prometheus.git"
 default-features = false
 features = ["nightly", "push", "process"]
 

src/pd/metrics.rs

@@ -61,4 +61,18 @@ lazy_static! {
             "Histogram of bytes written for regions",
             exponential_buckets(256.0, 2.0, 20).unwrap()
         ).unwrap();
+
+    pub static ref REGION_WRITTEN_BYTES_HISTOGRAM: Histogram =
+        register_histogram!(
+            "tikv_region_written_bytes",
+            "Histogram of bytes written for regions",
+             exponential_buckets(256.0, 2.0, 20).unwrap()
+        ).unwrap();
+
+    pub static ref REGION_WRITTEN_KEYS_HISTOGRAM: Histogram =
+        register_histogram!(
+            "tikv_region_written_keys",
+            "Histogram of keys written for regions",
+             exponential_buckets(1.0, 2.0, 20).unwrap()
+        ).unwrap();
 }

src/pd/pd.rs

@@ -36,6 +36,7 @@ use raftstore::store::store::StoreInfo;
 use raftstore::store::Callback;
 use storage::FlowStatistics;
 use util::collections::HashMap;
+use prometheus::local::LocalHistogram;
 use super::metrics::*;
 
 // Use an asynchronous thread to tell pd something.
@@ -71,12 +72,32 @@ pub enum Task {
     ReadStats { read_stats: HashMap<u64, FlowStatistics>, },
 }
 
-#[derive(Default)]
 pub struct StoreReadStat {
     pub engine_total_bytes_read: u64,
     pub engine_total_keys_read: u64,
     pub engine_last_total_bytes_read: u64,
     pub engine_last_total_keys_read: u64,
+
+    pub region_bytes_read: LocalHistogram,
+    pub region_keys_read: LocalHistogram,
+    pub region_bytes_written: LocalHistogram,
+    pub region_keys_written: LocalHistogram,
+}
+
+impl Default for StoreReadStat {
+    fn default() -> StoreReadStat {
+        StoreReadStat {
+            region_bytes_read: REGION_READ_BYTES_HISTOGRAM.local(),
+            region_keys_read: REGION_READ_KEYS_HISTOGRAM.local(),
+            region_bytes_written: REGION_WRITTEN_BYTES_HISTOGRAM.local(),
+            region_keys_written: REGION_WRITTEN_KEYS_HISTOGRAM.local(),
+
+            engine_total_bytes_read: 0,
+            engine_total_keys_read: 0,
+            engine_last_total_bytes_read: 0,
+            engine_last_total_keys_read: 0,
+        }
+    }
 }
 
 #[derive(Default)]
@@ -207,6 +228,18 @@ impl<T: PdClient> Runner<T> {
         PD_REQ_COUNTER_VEC
             .with_label_values(&["heartbeat", "all"])
             .inc();
+        self.store_stat
+            .region_bytes_written
+            .observe(region_stat.written_bytes as f64);
+        self.store_stat
+            .region_keys_written
+            .observe(region_stat.written_keys as f64);
+        self.store_stat
+            .region_bytes_read
+            .observe(region_stat.read_bytes as f64);
+        self.store_stat
+            .region_keys_read
+            .observe(region_stat.read_keys as f64);
 
         // Now we use put region protocol for heartbeat.
         let f = self.pd_client
@@ -274,10 +307,10 @@ impl<T: PdClient> Runner<T> {
         self.store_stat.engine_last_total_bytes_read = self.store_stat.engine_total_bytes_read;
         self.store_stat.engine_last_total_keys_read = self.store_stat.engine_total_keys_read;
 
-        for peer in self.region_peers.values() {
-            REGION_READ_BYTES_HISTOGRAM.observe((peer.read_bytes - peer.last_read_bytes) as f64);
-            REGION_READ_KEYS_HISTOGRAM.observe((peer.read_keys - peer.last_read_keys) as f64);
-        }
+        self.store_stat.region_bytes_written.flush();
+        self.store_stat.region_keys_written.flush();
+        self.store_stat.region_bytes_read.flush();
+        self.store_stat.region_keys_read.flush();
 
         STORE_SIZE_GAUGE_VEC
             .with_label_values(&["capacity"])

src/raftstore/store/metrics.rs

@@ -115,20 +115,6 @@ lazy_static! {
                     512.0, 1024.0, 5120.0, 10240.0]
         ).unwrap();
 
-    pub static ref REGION_WRITTEN_BYTES_HISTOGRAM: Histogram =
-        register_histogram!(
-            "tikv_region_written_bytes",
-            "Histogram of bytes written for regions",
-             exponential_buckets(256.0, 2.0, 20).unwrap()
-        ).unwrap();
-
-    pub static ref REGION_WRITTEN_KEYS_HISTOGRAM: Histogram =
-        register_histogram!(
-            "tikv_region_written_keys",
-            "Histogram of keys written for regions",
-             exponential_buckets(1.0, 2.0, 20).unwrap()
-        ).unwrap();
-
     pub static ref REQUEST_WAIT_TIME_HISTOGRAM: Histogram =
         register_histogram!(
             "tikv_raftstore_request_wait_time_duration_secs",

src/raftstore/store/store.rs

@@ -62,7 +62,6 @@ use super::cmd_resp::{bind_term, new_error};
 use super::transport::Transport;
 use super::metrics::*;
 use super::local_metrics::RaftMetrics;
-use prometheus::local::LocalHistogram;
 
 type Key = Vec<u8>;
 
@@ -91,8 +90,6 @@ pub struct StoreChannel {
 }
 
 pub struct StoreStat {
-    pub region_bytes_written: LocalHistogram,
-    pub region_keys_written: LocalHistogram,
     pub lock_cf_bytes_written: u64,
 
     pub engine_total_bytes_written: u64,
@@ -105,8 +102,6 @@ pub struct StoreStat {
 impl Default for StoreStat {
     fn default() -> StoreStat {
         StoreStat {
-            region_bytes_written: REGION_WRITTEN_BYTES_HISTOGRAM.local(),
-            region_keys_written: REGION_WRITTEN_KEYS_HISTOGRAM.local(),
             lock_cf_bytes_written: 0,
             engine_total_bytes_written: 0,
             engine_total_keys_written: 0,
@@ -1679,18 +1674,9 @@ impl<T: Transport, C: PdClient> Store<T, C> {
 
     fn on_update_region_flow(&mut self) {
         for peer in self.region_peers.values_mut() {
-            self.store_stat.region_bytes_written.observe(
-                (peer.peer_stat.written_bytes - peer.peer_stat.last_written_bytes) as f64,
-            );
-            self.store_stat.region_keys_written.observe(
-                (peer.peer_stat.written_keys - peer.peer_stat.last_written_keys) as f64,
-            );
-
             peer.peer_stat.last_written_bytes = peer.peer_stat.written_bytes;
             peer.peer_stat.last_written_keys = peer.peer_stat.written_keys;
         }
-        self.store_stat.region_bytes_written.flush();
-        self.store_stat.region_keys_written.flush();
     }
 
     fn on_update_store_flow(&mut self) {