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 <admin@parity.io>"
@@ -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.7", features = ["jemalloc-global"] }
@@ -75,10 +77,13 @@ 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"
+tokio = { version = "0.2", features = ["blocking"] }
 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.7"
 kvdb-rocksdb = "0.9"
 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;
@@ -265,6 +265,7 @@ pub struct Service<C: Send + Sync + 'static> {
 	client: Arc<C>,
 	progress: RwLock<Progress>,
 	taking_snapshot: AtomicBool,
+	taking_snapshot_at: AtomicUsize,
 	restoring_snapshot: AtomicBool,
 }
 
@@ -286,6 +287,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
 			client: params.client,
 			progress: RwLock::new(Progress::new()),
 			taking_snapshot: AtomicBool::new(false),
+			taking_snapshot_at: AtomicUsize::new(0),
 			restoring_snapshot: AtomicBool::new(false),
 		};
 
@@ -505,6 +507,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
 
 		info!("Taking snapshot at #{}", num);
 		{
+			self.taking_snapshot_at.store(num as usize, Ordering::SeqCst);
 			scopeguard::defer! {{
 				self.taking_snapshot.store(false, Ordering::SeqCst);
 			}}
@@ -609,13 +612,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,
@@ -760,7 +765,7 @@ impl<C> Service<C> where C: SnapshotClient + ChainInfo {
 		chunk: &[u8],
 		is_state: bool
 	) -> Result<(), Error> {
-		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);
 				Ok(())
@@ -832,7 +837,17 @@ 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 {
+		if self.taking_snapshot.load(Ordering::SeqCst) {
+			CreationStatus::Ongoing {
+				block_number: self.taking_snapshot_at.load(Ordering::SeqCst) as u32
+			}
+		} else {
+			CreationStatus::Inactive
+		}
+	}
+
+	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.
@@ -2855,6 +2856,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 cache
+		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 {
@@ -997,3 +998,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

@@ -27,13 +27,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.2", features = ["publickey"] }
 parity-util-mem = "0.7"
 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" }