Expose system health via RPC and REST (#1269)

* Implement health endpoint.

* Expose health API.

substrate/Cargo.lock

@@ -3591,12 +3591,15 @@ dependencies = [
  "parity-codec 2.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "parking_lot 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-hex 2.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.80 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde_derive 1.0.80 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.33 (registry+https://github.com/rust-lang/crates.io-index)",
  "sr-primitives 0.1.0",
  "sr-version 0.1.0",
  "substrate-client 0.1.0",
  "substrate-consensus-common 0.1.0",
  "substrate-executor 0.1.0",
+ "substrate-network 0.1.0",
  "substrate-primitives 0.1.0",
  "substrate-test-client 0.1.0",
  "substrate-transaction-pool 0.1.0",
@@ -3648,7 +3651,6 @@ dependencies = [
  "substrate-keystore 0.1.0",
  "substrate-network 0.1.0",
  "substrate-primitives 0.1.0",
- "substrate-rpc 0.1.0",
  "substrate-rpc-servers 0.1.0",
  "substrate-telemetry 0.3.0",
  "substrate-transaction-pool 0.1.0",

substrate/core/rpc-servers/src/lib.rs

@@ -70,6 +70,7 @@ pub fn start_http(
 ) -> io::Result<http::Server> {
 	http::ServerBuilder::new(io)
 		.threads(4)
+		.health_api(("/health", "system_health"))
 		.rest_api(http::RestApi::Unsecure)
 		.cors(http::DomainsValidation::Disabled)
 		.max_request_body_size(MAX_PAYLOAD)

substrate/core/rpc/Cargo.toml

@@ -11,11 +11,14 @@ jsonrpc-pubsub = { git="https://github.com/paritytech/jsonrpc.git" }
 log = "0.4"
 parking_lot = "0.4"
 parity-codec = "2.1"
+serde = "1.0"
+serde_derive = "1.0"
 serde_json = "1.0"
 substrate-client = { path = "../client" }
 substrate-executor = { path = "../executor" }
-substrate-transaction-pool = { path = "../transaction-pool" }
+substrate-network = { path = "../network" }
 substrate-primitives = { path = "../primitives" }
+substrate-transaction-pool = { path = "../transaction-pool" }
 sr-primitives = { path = "../sr-primitives" }
 sr-version = { path = "../sr-version" }
 tokio = "0.1.7"

substrate/core/rpc/src/lib.rs

@@ -20,13 +20,15 @@
 
 extern crate jsonrpc_core as rpc;
 extern crate jsonrpc_pubsub;
-extern crate parking_lot;
 extern crate parity_codec as codec;
-extern crate substrate_client as client;
-extern crate substrate_transaction_pool as transaction_pool;
-extern crate substrate_primitives as primitives;
+extern crate parking_lot;
+extern crate serde_json;
 extern crate sr_primitives as runtime_primitives;
 extern crate sr_version as runtime_version;
+extern crate substrate_client as client;
+extern crate substrate_network as network;
+extern crate substrate_primitives as primitives;
+extern crate substrate_transaction_pool as transaction_pool;
 extern crate tokio;
 
 #[macro_use]
@@ -35,6 +37,8 @@ extern crate error_chain;
 extern crate jsonrpc_macros;
 #[macro_use]
 extern crate log;
+#[macro_use]
+extern crate serde_derive;
 
 #[cfg(test)]
 #[macro_use]

substrate/core/rpc/src/system/error.rs

@@ -19,9 +19,16 @@
 use rpc;
 
 use errors;
+use system::helpers::Health;
 
 error_chain! {
 	errors {
+		/// Node is not fully functional
+		NotHealthy(h: Health) {
+			description("node is not healthy"),
+			display("Node is not fully functional: {}", h)
+		}
+
 		/// Not implemented yet
 		Unimplemented {
 			description("not yet implemented"),
@@ -30,10 +37,17 @@ error_chain! {
 	}
 }
 
+const ERROR: i64 = 2000;
+
 impl From<Error> for rpc::Error {
 	fn from(e: Error) -> Self {
 		match e {
 			Error(ErrorKind::Unimplemented, _) => errors::unimplemented(),
+			Error(ErrorKind::NotHealthy(h), _) => rpc::Error {
+				code: rpc::ErrorCode::ServerError(ERROR + 1),
+				message: "node is not healthy".into(),
+				data:serde_json::to_value(h).ok(),
+			},
 			e => errors::internal(e),
 		}
 	}

substrate/core/rpc/src/system/helpers.rs

@@ -0,0 +1,54 @@
+// Copyright 2017-2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Substrate system API helpers.
+
+use std::fmt;
+use serde_derive::{Serialize};
+use serde_json::{Value, map::Map};
+
+/// Node properties
+pub type Properties = Map<String, Value>;
+
+/// Running node's static details.
+#[derive(Clone, Debug)]
+pub struct SystemInfo {
+	/// Implementation name.
+	pub impl_name: String,
+	/// Implementation version.
+	pub impl_version: String,
+	/// Chain name.
+	pub chain_name: String,
+	/// A custom set of properties defined in the chain spec.
+	pub properties: Properties,
+}
+
+/// Health struct returned by the RPC
+#[derive(Debug, PartialEq, Serialize)]
+pub struct Health {
+	/// Number of connected peers
+	pub peers: usize,
+	/// Is the node syncing
+	pub is_syncing: bool,
+}
+
+impl fmt::Display for Health {
+	fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
+		write!(fmt, "{} peers ({})", self.peers, if self.is_syncing {
+			"syncing"
+		} else { "idle" })
+	}
+}

substrate/core/rpc/src/system/mod.rs

@@ -18,10 +18,16 @@
 
 pub mod error;
 
+mod helpers;
 #[cfg(test)]
 mod tests;
 
+use std::sync::Arc;
+use network;
+use runtime_primitives::traits;
+
 use self::error::Result;
+pub use self::helpers::{Properties, SystemInfo, Health};
 
 build_rpc_trait! {
 	/// Substrate system RPC API
@@ -40,6 +46,73 @@ build_rpc_trait! {
 
 		/// Get a custom set of properties as a JSON object, defined in the chain spec.
 		#[rpc(name = "system_properties")]
-		fn system_properties(&self) -> Result<serde_json::map::Map<String, serde_json::Value>>;
+		fn system_properties(&self) -> Result<Properties>;
+
+		/// Return health status of the node.
+		///
+		/// Node is considered healthy if it is:
+		/// - connected to some peers (unless running in dev mode)
+		/// - not performing a major sync
+		#[rpc(name = "system_health")]
+		fn system_health(&self) -> Result<Health>;
+	}
+}
+
+/// System API implementation
+pub struct System<B: traits::Block> {
+	info: SystemInfo,
+	sync: Arc<network::SyncProvider<B>>,
+	should_have_peers: bool,
+}
+
+impl<B: traits::Block> System<B> {
+	/// Creates new `System` given the `SystemInfo`.
+	pub fn new(
+		info: SystemInfo,
+		sync: Arc<network::SyncProvider<B>>,
+		should_have_peers: bool,
+	) -> Self {
+		System {
+			info,
+			should_have_peers,
+			sync,
+		}
+	}
+}
+
+impl<B: traits::Block> SystemApi for System<B> {
+	fn system_name(&self) -> Result<String> {
+		Ok(self.info.impl_name.clone())
+	}
+
+	fn system_version(&self) -> Result<String> {
+		Ok(self.info.impl_version.clone())
+	}
+
+	fn system_chain(&self) -> Result<String> {
+		Ok(self.info.chain_name.clone())
+	}
+
+	fn system_properties(&self) -> Result<Properties> {
+		Ok(self.info.properties.clone())
+	}
+
+	fn system_health(&self) -> Result<Health> {
+		let status = self.sync.status();
+
+		let is_syncing = status.sync.is_major_syncing();
+		let peers = status.num_peers;
+
+		let health = Health {
+			peers,
+			is_syncing,
+		};
+
+		let has_no_peers = peers == 0 && self.should_have_peers;
+		if has_no_peers || is_syncing {
+			Err(error::ErrorKind::NotHealthy(health))?
+		} else {
+			Ok(health)
+		}
 	}
 }

substrate/core/rpc/src/system/tests.rs

@@ -15,27 +15,46 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use super::*;
-use super::error::*;
 
-impl SystemApi for () {
-	fn system_name(&self) -> Result<String> {
-		Ok("testclient".into())
-	}
-	fn system_version(&self) -> Result<String> {
-		Ok("0.2.0".into())
-	}
-	fn system_chain(&self) -> Result<String> {
-		Ok("testchain".into())
-	}
-	fn system_properties(&self) -> Result<serde_json::map::Map<String, serde_json::Value>> {
-		Ok(serde_json::map::Map::new())
+use network::{self, SyncState, SyncStatus, ProtocolStatus};
+use test_client::runtime::Block;
+
+#[derive(Default)]
+struct Status {
+	pub peers: usize,
+	pub is_syncing: bool,
+	pub is_dev: bool,
+}
+
+impl network::SyncProvider<Block> for Status {
+	fn status(&self) -> ProtocolStatus<Block> {
+		ProtocolStatus {
+			sync: SyncStatus {
+				state: if self.is_syncing { SyncState::Downloading } else { SyncState::Idle },
+				best_seen_block: None,
+			},
+			num_peers: self.peers,
+			num_active_peers: 0,
+		}
 	}
 }
 
+
+fn api<T: Into<Option<Status>>>(sync: T) -> System<Block> {
+	let status = sync.into().unwrap_or_default();
+	let should_have_peers = !status.is_dev;
+	System::new(SystemInfo {
+		impl_name: "testclient".into(),
+		impl_version: "0.2.0".into(),
+		chain_name: "testchain".into(),
+		properties: Default::default(),
+	}, Arc::new(status), should_have_peers)
+}
+
 #[test]
 fn system_name_works() {
 	assert_eq!(
-		SystemApi::system_name(&()).unwrap(),
+		api(None).system_name().unwrap(),
 		"testclient".to_owned()
 	);
 }
@@ -43,7 +62,7 @@ fn system_name_works() {
 #[test]
 fn system_version_works() {
 	assert_eq!(
-		SystemApi::system_version(&()).unwrap(),
+		api(None).system_version().unwrap(),
 		"0.2.0".to_owned()
 	);
 }
@@ -51,7 +70,7 @@ fn system_version_works() {
 #[test]
 fn system_chain_works() {
 	assert_eq!(
-		SystemApi::system_chain(&()).unwrap(),
+		api(None).system_chain().unwrap(),
 		"testchain".to_owned()
 	);
 }
@@ -59,7 +78,54 @@ fn system_chain_works() {
 #[test]
 fn system_properties_works() {
 	assert_eq!(
-		SystemApi::system_properties(&()).unwrap(),
+		api(None).system_properties().unwrap(),
 		serde_json::map::Map::new()
 	);
 }
+
+#[test]
+fn system_health() {
+	assert_matches!(
+		api(None).system_health().unwrap_err().kind(),
+		error::ErrorKind::NotHealthy(Health {
+			peers: 0,
+			is_syncing: false,
+		})
+	);
+
+	assert_matches!(
+		api(Status {
+			peers: 5,
+			is_syncing: true,
+			is_dev: true,
+		}).system_health().unwrap_err().kind(),
+		error::ErrorKind::NotHealthy(Health {
+			peers: 5,
+			is_syncing: true,
+		})
+	);
+
+	assert_eq!(
+		api(Status {
+			peers: 5,
+			is_syncing: false,
+			is_dev: false,
+		}).system_health().unwrap(),
+		Health {
+			peers: 5,
+			is_syncing: false,
+		}
+	);
+
+	assert_eq!(
+		api(Status {
+			peers: 0,
+			is_syncing: false,
+			is_dev: true,
+		}).system_health().unwrap(),
+		Health {
+			peers: 0,
+			is_syncing: false,
+		}
+	);
+}

substrate/core/service/Cargo.toml

@@ -27,6 +27,5 @@ substrate-client-db = { path = "../../core/client/db" }
 parity-codec = "2.1"
 substrate-executor = { path = "../../core/executor" }
 substrate-transaction-pool = { path = "../../core/transaction-pool" }
-substrate-rpc = { path = "../../core/rpc" }
 substrate-rpc-servers = { path = "../../core/rpc-servers" }
 substrate-telemetry = { path = "../../core/telemetry" }

substrate/core/service/src/components.rs

@@ -19,17 +19,17 @@
 use std::{sync::Arc, net::SocketAddr, marker::PhantomData, ops::Deref, ops::DerefMut};
 use serde::{Serialize, de::DeserializeOwned};
 use tokio::runtime::TaskExecutor;
-use chain_spec::{ChainSpec, Properties};
+use chain_spec::ChainSpec;
 use client_db;
 use client::{self, Client, runtime_api::{Metadata, TaggedTransactionQueue}};
-use {error, Service, RpcConfig, maybe_start_server};
+use {error, Service, maybe_start_server};
 use network::{self, OnDemand, import_queue::ImportQueue};
 use substrate_executor::{NativeExecutor, NativeExecutionDispatch};
 use transaction_pool::txpool::{self, Options as TransactionPoolOptions, Pool as TransactionPool};
 use runtime_primitives::{BuildStorage, traits::{Block as BlockT, Header as HeaderT, ProvideRuntimeApi}, generic::{BlockId, SignedBlock}};
 use config::Configuration;
 use primitives::{Blake2Hasher, H256};
-use rpc;
+use rpc::{self, apis::system::SystemInfo};
 use parking_lot::Mutex;
 
 // Type aliases.
@@ -123,12 +123,11 @@ pub trait StartRPC<C: Components> {
 
 	fn start_rpc(
 		client: Arc<ComponentClient<C>>,
-		chain_name: String,
-		impl_name: &'static str,
-		impl_version: &'static str,
+		network: Arc<network::SyncProvider<ComponentBlock<C>>>,
+		should_have_peers: bool,
+		system_info: SystemInfo,
 		rpc_http: Option<SocketAddr>,
 		rpc_ws: Option<SocketAddr>,
-		properties: Properties,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
 	) -> error::Result<Self::ServersHandle>;
@@ -142,17 +141,14 @@ impl<C: Components> StartRPC<Self> for C where
 
 	fn start_rpc(
 		client: Arc<ComponentClient<C>>,
-		chain_name: String,
-		impl_name: &'static str,
-		impl_version: &'static str,
+		network: Arc<network::SyncProvider<ComponentBlock<C>>>,
+		should_have_peers: bool,
+		rpc_system_info: SystemInfo,
 		rpc_http: Option<SocketAddr>,
 		rpc_ws: Option<SocketAddr>,
-		properties: Properties,
 		task_executor: TaskExecutor,
 		transaction_pool: Arc<TransactionPool<C::TransactionPoolApi>>,
 	) -> error::Result<Self::ServersHandle> {
-		let rpc_config = RpcConfig { properties, chain_name, impl_name, impl_version };
-
 		let handler = || {
 			let client = client.clone();
 			let subscriptions = rpc::apis::Subscriptions::new(task_executor.clone());
@@ -161,11 +157,14 @@ impl<C: Components> StartRPC<Self> for C where
 			let author = rpc::apis::author::Author::new(
 				client.clone(), transaction_pool.clone(), subscriptions
 			);
+			let system = rpc::apis::system::System::new(
+				rpc_system_info.clone(), network.clone(), should_have_peers
+			);
 			rpc::rpc_handler::<ComponentBlock<C>, ComponentExHash<C>, _, _, _, _>(
 				state,
 				chain,
 				author,
-				rpc_config.clone(),
+				system,
 			)
 		};
 

substrate/core/service/src/lib.rs

@@ -34,7 +34,6 @@ extern crate substrate_client as client;
 extern crate substrate_client_db as client_db;
 extern crate parity_codec as codec;
 extern crate substrate_transaction_pool as transaction_pool;
-extern crate substrate_rpc;
 extern crate substrate_rpc_servers as rpc;
 extern crate target_info;
 extern crate tokio;
@@ -189,6 +188,7 @@ impl<Components: components::Components> Service<Components> {
 			protocol_id
 		};
 
+		let has_bootnodes = !network_params.network_config.boot_nodes.is_empty();
 		let network = network::Service::new(
 			network_params,
 			protocol_id,
@@ -240,10 +240,14 @@ impl<Components: components::Components> Service<Components> {
 
 
 		// RPC
+		let system_info = rpc::apis::system::SystemInfo {
+			chain_name: config.chain_spec.name().into(),
+			impl_name: config.impl_name.into(),
+			impl_version: config.impl_version.into(),
+			properties: config.chain_spec.properties(),
+		};
 		let rpc = Components::RPC::start_rpc(
-			client.clone(), config.chain_spec.name().to_string(), config.impl_name,
-			config.impl_version, config.rpc_http, config.rpc_ws, config.chain_spec.properties(),
-			task_executor.clone(), transaction_pool.clone()
+			client.clone(), network.clone(), has_bootnodes, system_info, config.rpc_http, config.rpc_ws, task_executor.clone(), transaction_pool.clone(),
 		)?;
 
 		// Telemetry
@@ -358,32 +362,6 @@ fn maybe_start_server<T, F>(address: Option<SocketAddr>, start: F) -> Result<Opt
 	})
 }
 
-#[derive(Clone)]
-struct RpcConfig {
-	chain_name: String,
-	properties: Properties,
-	impl_name: &'static str,
-	impl_version: &'static str,
-}
-
-impl substrate_rpc::system::SystemApi for RpcConfig {
-	fn system_name(&self) -> substrate_rpc::system::error::Result<String> {
-		Ok(self.impl_name.into())
-	}
-
-	fn system_version(&self) -> substrate_rpc::system::error::Result<String> {
-		Ok(self.impl_version.into())
-	}
-
-	fn system_chain(&self) -> substrate_rpc::system::error::Result<String> {
-		Ok(self.chain_name.clone())
-	}
-
-	fn system_properties(&self) -> substrate_rpc::system::error::Result<Properties> {
-		Ok(self.properties.clone())
-	}
-}
-
 /// Transaction pool adapter.
 pub struct TransactionPoolAdapter<C: Components> {
 	imports_external_transactions: bool,