Prometheus metrics

Cargo.toml

@@ -4,6 +4,7 @@ name = "openethereum"
 # NOTE Make sure to update util/version/Cargo.toml as well
 version = "3.0.0"
 license = "GPL-3.0"
+edition = "2018"
 authors = [
  "OpenEthereum developers",
  "Parity Technologies <[email protected]>"
@@ -40,6 +41,7 @@ ethkey = { path = "accounts/ethkey" }
 ethstore = { path = "accounts/ethstore" }
 fdlimit = "0.1"
 futures = "0.1"
+hyper = { version = "0.13" }
 journaldb = { path = "util/journaldb" }
 jsonrpc-core = "14.0.3"
 keccak-hash = "0.5.0"
@@ -58,7 +60,7 @@ parity-hash-fetch = { path = "updater/hash-fetch" }
 parity-local-store = { path = "miner/local-store" }
 parity-path = "0.1"
 parity-rpc = { path = "rpc" }
-parity-runtime = "0.1.1"
+parity-runtime = "0.1.2"
 parity-secretstore = { git = "https://github.com/paritytech/secret-store", branch = "v1.x", optional = true }
 parity-updater = { path = "updater" }
 parity-util-mem = { version = "0.6.0", features = ["jemalloc-global"] }
@@ -75,10 +77,12 @@ serde_derive = "1.0"
 serde_json = "1.0"
 snapshot = { path = "ethcore/snapshot" }
 spec = { path = "ethcore/spec" }
+stats = { path = "util/stats" }
 term_size = "0.3"
 textwrap = "0.11.0"
 toml = "0.5.6"
 verification = { path = "ethcore/verification" }
+prometheus = "0.9.0"
 
 [build-dependencies]
 rustc_version = "0.2"

ethcore/Cargo.toml

@@ -56,6 +56,7 @@ serde_derive = "1.0"
 snapshot = { path = "snapshot" }
 spec = { path = "spec" }
 state-db = { path = "state-db" }
+stats = { path = "../util/stats" }
 tempfile = { version = "3.1", optional = true }
 trace = { path = "trace" }
 trace-time = "0.1"
@@ -81,9 +82,8 @@ kvdb-memorydb = "0.5.0"
 kvdb-rocksdb = "0.7.0"
 lazy_static = "1.3"
 machine = { path = "./machine", features = ["test-helpers"] }
-parity-runtime = "0.1.1"
+parity-runtime = "0.1.2"
 serde_json = "1.0"
-stats = { path = "../util/stats" }
 pod = { path = "pod" }
 tempfile = "3.1"
 trie-standardmap = "0.15.0"

ethcore/service/src/service.rs

@@ -240,7 +240,7 @@ where
  match timer {
  CLIENT_TICK_TIMER => {
  use snapshot::SnapshotService;
- let snapshot_restoration = if let RestorationStatus::Ongoing{..} = self.snapshot.status() { true } else { false };
+ let snapshot_restoration = if let RestorationStatus::Ongoing{..} = self.snapshot.restoration_status() { true } else { false };
  self.client.tick(snapshot_restoration)
  },
  SNAPSHOT_TICK_TIMER => self.snapshot.tick(),

ethcore/snapshot/snapshot-tests/src/service.rs

@@ -71,7 +71,7 @@ fn sends_async_messages() {
 
  assert!(service.manifest().is_none());
  assert!(service.chunk(Default::default()).is_none());
- assert_eq!(service.status(), RestorationStatus::Inactive);
+ assert_eq!(service.restoration_status(), RestorationStatus::Inactive);
 
  let manifest = ManifestData {
  version: 2,
@@ -190,7 +190,7 @@ fn restored_is_equivalent() {
  service.feed_block_chunk(hash, &chunk);
  }
 
- assert_eq!(service.status(), RestorationStatus::Inactive);
+ assert_eq!(service.restoration_status(), RestorationStatus::Inactive);
 
  for x in 0..NUM_BLOCKS {
  let block1 = client.block(BlockId::Number(x as u64)).unwrap();
@@ -347,7 +347,7 @@ fn keep_ancient_blocks() {
  service.feed_state_chunk(*hash, &chunk);
  }
 
- match service.status() {
+ match service.restoration_status() {
  RestorationStatus::Inactive => (),
  RestorationStatus::Failed => panic!("Snapshot Restoration has failed."),
  RestorationStatus::Ongoing { .. } => panic!("Snapshot Restoration should be done."),
@@ -409,7 +409,7 @@ fn recover_aborted_recovery() {
  service.feed_state_chunk(*hash, &chunk);
  }
 
- match service.status() {
+ match service.restoration_status() {
  RestorationStatus::Ongoing { block_chunks_done, state_chunks_done, .. } => {
  assert_eq!(state_chunks_done, manifest.state_hashes.len() as u32);
  assert_eq!(block_chunks_done, 0);
@@ -423,7 +423,7 @@ fn recover_aborted_recovery() {
  // And try again!
  service.init_restore(manifest.clone(), true).unwrap();
 
- match service.status() {
+ match service.restoration_status() {
  RestorationStatus::Ongoing { block_chunks_done, state_chunks_done, .. } => {
  assert_eq!(state_chunks_done, manifest.state_hashes.len() as u32);
  assert_eq!(block_chunks_done, 0);
@@ -438,7 +438,7 @@ fn recover_aborted_recovery() {
  // And try again!
  service.init_restore(manifest.clone(), true).unwrap();
 
- match service.status() {
+ match service.restoration_status() {
  RestorationStatus::Ongoing { block_chunks_done, state_chunks_done, .. } => {
  assert_eq!(block_chunks_done, 0);
  assert_eq!(state_chunks_done, 0);

ethcore/snapshot/src/service.rs

@@ -30,7 +30,7 @@ use common_types::{
  io_message::ClientIoMessage,
  errors::{EthcoreError as Error, SnapshotError, SnapshotError::UnlinkedAncientBlockChain},
  ids::BlockId,
- snapshot::{ManifestData, Progress, RestorationStatus},
+ snapshot::{ManifestData, Progress, CreationStatus, RestorationStatus},
 };
 use client_traits::ChainInfo;
 use engine::Engine;
@@ -267,6 +267,7 @@ pub struct Service<C: Send + Sync + 'static> {
  client: Arc<C>,
  progress: RwLock<Progress>,
  taking_snapshot: AtomicBool,
+ taking_snapshot_num: AtomicUsize,
  restoring_snapshot: AtomicBool,
 }
 
@@ -288,6 +289,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
  client: params.client,
  progress: RwLock::new(Progress::new()),
  taking_snapshot: AtomicBool::new(false),
+ taking_snapshot_num: AtomicUsize::new(0),
  restoring_snapshot: AtomicBool::new(false),
  };
 
@@ -509,6 +511,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
 
  info!("Taking snapshot at #{}", num);
  {
+ self.taking_snapshot_num.store(num as usize, Ordering::SeqCst);
  scopeguard::defer! {{
  self.taking_snapshot.store(false, Ordering::SeqCst);
  }}
@@ -613,13 +616,15 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
 
  self.restoring_snapshot.store(true, Ordering::SeqCst);
 
+ let block_number = manifest.block_number;
  // Import previous chunks, continue if it fails
  self.import_prev_chunks(&mut res, manifest).ok();
 
  // It could be that the restoration failed or completed in the meanwhile
  let mut restoration_status = self.status.lock();
  if let RestorationStatus::Initializing { .. } = *restoration_status {
  *restoration_status = RestorationStatus::Ongoing {
+ block_number,
  state_chunks: state_chunks as u32,
  block_chunks: block_chunks as u32,
  state_chunks_done: self.state_chunks.load(Ordering::SeqCst) as u32,
@@ -759,7 +764,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
  /// Feed a chunk with the Restoration
  fn feed_chunk_with_restoration(&self, restoration: &mut Option<Restoration>, hash: H256, chunk: &[u8], is_state: bool) -> Result<(), Error> {
  let (result, db) = {
- match self.status() {
+ match self.restoration_status() {
  RestorationStatus::Inactive | RestorationStatus::Failed | RestorationStatus::Finalizing => {
  trace!(target: "snapshot", "Tried to restore chunk {:x} while inactive, failed or finalizing", hash);
  return Ok(());
@@ -852,7 +857,16 @@ impl<C: Send + Sync> SnapshotService for Service<C> {
  self.reader.read().as_ref().and_then(|r| r.chunk(hash).ok())
  }
 
- fn status(&self) -> RestorationStatus {
+ fn creation_status(&self) -> CreationStatus {
+ match self.taking_snapshot.load(Ordering::SeqCst) {
+ false => CreationStatus::Inactive,
+ true => CreationStatus::Ongoing {
+ block_number: self.taking_snapshot_num.load(Ordering::SeqCst) as u32
+ },
+ }
+ }
+
+ fn restoration_status(&self) -> RestorationStatus {
  let mut cur_status = self.status.lock();
 
  match *cur_status {

ethcore/snapshot/src/traits.rs

@@ -22,7 +22,7 @@ use client_traits::{BlockChainClient, BlockInfo, DatabaseRestore, BlockChainRese
 use common_types::{
  ids::BlockId,
  errors::{EthcoreError as Error, SnapshotError},
- snapshot::{ManifestData, ChunkSink, Progress, RestorationStatus},
+ snapshot::{ManifestData, ChunkSink, Progress, CreationStatus, RestorationStatus},
 };
 use engine::Engine;
 use ethereum_types::H256;
@@ -49,7 +49,10 @@ pub trait SnapshotService : Sync + Send {
  fn chunk(&self, hash: H256) -> Option<Bytes>;
 
  /// Ask the snapshot service for the restoration status.
- fn status(&self) -> RestorationStatus;
+ fn restoration_status(&self) -> RestorationStatus;
+
+ /// Ask the snapshot service for the creation status.
+ fn creation_status(&self) -> CreationStatus;
 
  /// Begin snapshot restoration.
  /// If a restoration is in progress, this will reset it and clear all data.

ethcore/src/client/client.rs

@@ -136,6 +136,7 @@ use types::data_format::DataFormat;
 use verification::{self, BlockQueue};
 use verification::queue::kind::BlockLike;
 use vm::{CreateContractAddress, EnvInfo, LastHashes};
+use stats::{prometheus_counter, prometheus_gauge, prometheus, PrometheusMetrics};
 
 const MAX_ANCIENT_BLOCKS_QUEUE_SIZE: usize = 4096;
 // Max number of blocks imported at once.
@@ -2857,6 +2858,66 @@ impl IoChannelQueue {
  }
 }
 
+impl PrometheusMetrics for Client {
+ fn prometheus_metrics(&self, r: &mut prometheus::Registry) {
+
+ // gas, tx & blocks
+ let report = self.report();
+
+ prometheus_counter(r, "import_gas", "Gas processed", report.gas_processed.as_u64() as i64);
+ prometheus_counter(r, "import_blocks", "Blocks imported", report.blocks_imported as i64);
+ prometheus_counter(r, "import_txs", "Transactions applied", report.transactions_applied as i64);
+
+ let state_db = self.state_db.read();
+ prometheus_gauge(r, "statedb_mem_used", "State DB memory used", state_db.mem_used() as i64);
+ prometheus_gauge(r, "statedb_cache_size", "State DB cache size", state_db.cache_size() as i64);
+
+ // blockchain caché
+ let blockchain_cache_info = self.blockchain_cache_info();
+ prometheus_gauge(r, "blockchaincache_block_details", "BlockDetails cache size", blockchain_cache_info.block_details as i64);
+ prometheus_gauge(r, "blockchaincache_block_recipts", "Block receipts size", blockchain_cache_info.block_receipts as i64);
+ prometheus_gauge(r, "blockchaincache_blocks", "Blocks cache size", blockchain_cache_info.blocks as i64);
+ prometheus_gauge(r, "blockchaincache_txaddrs", "Transaction addresses cache size", blockchain_cache_info.transaction_addresses as i64);
+ prometheus_gauge(r, "blockchaincache_size", "Total blockchain cache size", blockchain_cache_info.total() as i64);
+
+ // io
+ let io_stats = self.db.read().key_value().io_stats(kvdb::IoStatsKind::Overall);
+ prometheus_counter(r, "io_transactions", "Number of transactions", io_stats.transactions as i64);
+ prometheus_counter(r, "io_reads", "Number of read operations", io_stats.reads as i64);
+ prometheus_counter(r, "io_writes", "Number of write operations", io_stats.writes as i64);
+ prometheus_counter(r, "io_cache_reads", "Number of reads resulted in a read from cache", io_stats.cache_reads as i64);
+ prometheus_counter(r, "io_bytes_read", "Number of bytes read", io_stats.bytes_read as i64);
+ prometheus_counter(r, "io_cache_read_bytes", "Number of cache bytes read", io_stats.cache_read_bytes as i64);
+ prometheus_counter(r, "io_bytes_written", "Number of bytes written", io_stats.bytes_written as i64);
+
+ // chain info
+ let chain = self.chain_info();
+
+ let gap = chain.ancient_block_number.map(|x| U256::from(x + 1))
+ .and_then(|first| chain.first_block_number.map(|last| (first, U256::from(last))));
+ if let Some((first,last)) = gap {
+ prometheus_gauge(r, "chain_warpsync_gap_first", "Warp sync gap, first block", first.as_u64() as i64);
+ prometheus_gauge(r, "chain_warpsync_gap_last", "Warp sync gap, last block", last.as_u64() as i64);
+ }
+
+ prometheus_gauge(r, "chain_block", "Best block number", chain.best_block_number as i64);
+
+ // prunning info
+ let prunning = self.pruning_info();
+ prometheus_gauge(r, "prunning_earliest_chain", "The first block which everything can be served after", prunning.earliest_chain as i64);
+ prometheus_gauge(r, "prunning_earliest_state", "The first block where state requests may be served", prunning.earliest_state as i64);
+
+ // queue info
+ let queue = self.queue_info();
+ prometheus_gauge(r, "queue_mem_used", "Queue heap memory used in bytes", queue.mem_used as i64);
+ prometheus_gauge(r, "queue_size_total", "The total size of the queues", queue.total_queue_size() as i64);
+ prometheus_gauge(r, "queue_size_unverified", "Number of queued items pending verification", queue.unverified_queue_size as i64);
+ prometheus_gauge(r, "queue_size_verified", "Number of verified queued items pending import", queue.verified_queue_size as i64);
+ prometheus_gauge(r, "queue_size_verifying", "Number of items being verified", queue.verifying_queue_size as i64);
+
+ }
+}
+
 #[cfg(test)]
 mod tests {
  use std::sync::Arc;

ethcore/src/lib.rs

@@ -49,6 +49,7 @@ extern crate rlp;
 extern crate rustc_hex;
 extern crate serde;
 extern crate snapshot;
+extern crate stats;
 extern crate spec;
 extern crate state_db;
 extern crate trace;

ethcore/src/test_helpers/test_client.rs

@@ -81,6 +81,7 @@ use spec::{Spec, self};
 use account_state::state::StateInfo;
 use state_db::StateDB;
 use trace::LocalizedTrace;
+use stats::{PrometheusMetrics, prometheus};
 
 /// Test client.
 pub struct TestBlockChainClient {
@@ -995,3 +996,8 @@ impl client_traits::EngineClient for TestBlockChainClient {
  BlockChainClient::block_header(self, id)
  }
 }
+
+impl PrometheusMetrics for TestBlockChainClient {
+ fn prometheus_metrics(&self, _r: &mut prometheus::Registry) {
+ }
+}

ethcore/sync/Cargo.toml

@@ -26,13 +26,14 @@ light = { package = "ethcore-light", path = "../light" }
 log = "0.4"
 network = { package = "ethcore-network", path = "../../util/network" }
 num-traits = "0.2"
-parity-runtime = "0.1.1"
+parity-runtime = "0.1.2"
 parity-crypto = { version = "0.6.1", features = ["publickey"] }
 parity-util-mem = "0.6.0"
 rand = "0.7.3"
 parking_lot = "0.10.0"
 rlp = "0.4.5"
 snapshot = { path = "../snapshot" }
+stats = { path = "../../util/stats" }
 trace-time = "0.1"
 triehash-ethereum = { version = "0.2", path = "../../util/triehash-ethereum" }