Prometheus metrics for RPC calls (#7088)

* WS and HTTP middlewares added

* Prometheus endpoint added

* Counters renamed

* Proper style for inc

* Metrics initialization re-written

* Rework to handler middleware

* Introduce transport prefix for metrics

* String shortened

* Commented code removed, new line inserted

* One more string shortened

* Wasm build fixed

* Wasm build fixed once again

* Rework to shared metrics

* Added collectors label

* Tilde removed from cargo

* Switch to owned metrics in parameters

substrate/Cargo.lock

@@ -7111,6 +7111,7 @@ dependencies = [
 name = "sc-rpc-server"
 version = "2.0.0-rc6"
 dependencies = [
+ "futures 0.1.29",
  "jsonrpc-core",
  "jsonrpc-http-server",
  "jsonrpc-ipc-server",
@@ -7120,6 +7121,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sp-runtime",
+ "substrate-prometheus-endpoint",
 ]
 
 [[package]]

substrate/client/rpc-servers/Cargo.toml

@@ -12,14 +12,16 @@ description = "Substrate RPC servers."
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
+futures = "0.1.6"
 jsonrpc-core = "14.2.0"
 pubsub = { package = "jsonrpc-pubsub", version = "14.2.0" }
 log = "0.4.8"
+prometheus-endpoint = { package = "substrate-prometheus-endpoint", path = "../../utils/prometheus", version = "0.8.0-rc6"}
 serde = "1.0.101"
 serde_json = "1.0.41"
 sp-runtime = { version = "2.0.0-rc6", path = "../../primitives/runtime" }
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
 http = { package = "jsonrpc-http-server", version = "14.2.0" }
-ws = { package = "jsonrpc-ws-server", version = "14.2.0" }
 ipc = { version = "14.2.0", package = "jsonrpc-ipc-server" }
+ws = { package = "jsonrpc-ws-server", version = "14.2.0" }

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

@@ -20,8 +20,10 @@
 
 #![warn(missing_docs)]
 
+mod middleware;
+
 use std::io;
-use jsonrpc_core::IoHandlerExtension;
+use jsonrpc_core::{IoHandlerExtension, MetaIoHandler};
 use log::error;
 use pubsub::PubSubMetadata;
 
@@ -32,15 +34,18 @@ const MAX_PAYLOAD: usize = 15 * 1024 * 1024;
 const WS_MAX_CONNECTIONS: usize = 100;
 
 /// The RPC IoHandler containing all requested APIs.
-pub type RpcHandler<T> = pubsub::PubSubHandler<T>;
+pub type RpcHandler<T> = pubsub::PubSubHandler<T, RpcMiddleware>;
 
 pub use self::inner::*;
+pub use middleware::{RpcMiddleware, RpcMetrics};
 
 /// Construct rpc `IoHandler`
 pub fn rpc_handler<M: PubSubMetadata>(
-	extension: impl IoHandlerExtension<M>
+	extension: impl IoHandlerExtension<M>,
+	rpc_middleware: RpcMiddleware,
 ) -> RpcHandler<M> {
-	let mut io = pubsub::PubSubHandler::default();
+	let io_handler = MetaIoHandler::with_middleware(rpc_middleware);
+	let mut io = pubsub::PubSubHandler::new(io_handler);
 	extension.augment(&mut io);
 
 	// add an endpoint to list all available methods.

substrate/client/rpc-servers/src/middleware.rs

@@ -0,0 +1,87 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2020 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program 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.
+
+// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Middleware for RPC requests.
+
+use jsonrpc_core::{
+	Middleware as RequestMiddleware, Metadata,
+	Request, Response, FutureResponse, FutureOutput
+};
+use prometheus_endpoint::{
+	Registry, CounterVec, PrometheusError,
+	Opts, register, U64
+};
+
+use futures::{future::Either, Future};
+
+/// Metrics for RPC middleware
+#[derive(Debug, Clone)]
+pub struct RpcMetrics {
+	rpc_calls: CounterVec<U64>,
+}
+
+impl RpcMetrics {
+	/// Create an instance of metrics
+	pub fn new(metrics_registry: Option<&Registry>) -> Result<Self, PrometheusError> {
+		metrics_registry.and_then(|r| {
+			Some(RpcMetrics {
+				rpc_calls: register(CounterVec::new(
+					Opts::new(
+						"rpc_calls_total",
+						"Number of rpc calls received",
+					),
+					&["protocol"]
+				).ok()?, r).ok()?,
+			})
+		}).ok_or(PrometheusError::Msg("Cannot register metric".to_string()))
+	}
+}
+
+/// Middleware for RPC calls
+pub struct RpcMiddleware {
+	metrics: Option<RpcMetrics>,
+	transport_label: String,
+}
+
+impl RpcMiddleware {
+	/// Create an instance of middleware with provided metrics
+	/// transport_label is used as a label for Prometheus collector
+	pub fn new(metrics: Option<RpcMetrics>, transport_label: &str) -> Self {
+		RpcMiddleware {
+			metrics,
+			transport_label: String::from(transport_label),
+		}
+	}
+}
+
+impl<M: Metadata> RequestMiddleware<M> for RpcMiddleware {
+	type Future = FutureResponse;
+	type CallFuture = FutureOutput;
+
+	fn on_request<F, X>(&self, request: Request, meta: M, next: F) -> Either<FutureResponse, X>
+	where
+		F: Fn(Request, M) -> X + Send + Sync,
+		X: Future<Item = Option<Response>, Error = ()> + Send + 'static,
+	{
+		if let Some(ref metrics) = self.metrics {
+			metrics.rpc_calls.with_label_values(&[self.transport_label.as_str()]).inc();
+		}
+
+		Either::B(next(request, meta))
+	}
+}

substrate/client/service/src/builder.rs

@@ -545,14 +545,22 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 	);
 
 	// RPC
-	let gen_handler = |deny_unsafe: sc_rpc::DenyUnsafe| gen_handler(
-		deny_unsafe, &config, task_manager.spawn_handle(), client.clone(), transaction_pool.clone(),
-		keystore.clone(), on_demand.clone(), remote_blockchain.clone(), &*rpc_extensions_builder,
+	let gen_handler = |
+		deny_unsafe: sc_rpc::DenyUnsafe,
+		rpc_middleware: sc_rpc_server::RpcMiddleware
+	| gen_handler(
+		deny_unsafe, rpc_middleware, &config, task_manager.spawn_handle(),
+		client.clone(), transaction_pool.clone(), keystore.clone(),
+		on_demand.clone(), remote_blockchain.clone(), &*rpc_extensions_builder,
 		backend.offchain_storage(), system_rpc_tx.clone()
 	);
-	let rpc = start_rpc_servers(&config, gen_handler)?;
+	let rpc_metrics = sc_rpc_server::RpcMetrics::new(config.prometheus_registry()).ok();
+	let rpc = start_rpc_servers(&config, gen_handler, rpc_metrics.as_ref())?;
 	// This is used internally, so don't restrict access to unsafe RPC
-	let rpc_handlers = RpcHandlers(Arc::new(gen_handler(sc_rpc::DenyUnsafe::No).into()));
+	let rpc_handlers = RpcHandlers(Arc::new(gen_handler(
+		sc_rpc::DenyUnsafe::No,
+		sc_rpc_server::RpcMiddleware::new(rpc_metrics.as_ref().cloned(), "inbrowser")
+	).into()));
 
 	// Telemetry
 	let telemetry = config.telemetry_endpoints.clone().and_then(|endpoints| {
@@ -660,6 +668,7 @@ fn build_telemetry<TBl: BlockT>(
 
 fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 	deny_unsafe: sc_rpc::DenyUnsafe,
+	rpc_middleware: sc_rpc_server::RpcMiddleware,
 	config: &Configuration,
 	spawn_handle: SpawnTaskHandle,
 	client: Arc<TCl>,
@@ -670,7 +679,7 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 	rpc_extensions_builder: &(dyn RpcExtensionBuilder<Output = TRpc> + Send),
 	offchain_storage: Option<<TBackend as sc_client_api::backend::Backend<TBl>>::OffchainStorage>,
 	system_rpc_tx: TracingUnboundedSender<sc_rpc::system::Request<TBl>>
-) -> jsonrpc_pubsub::PubSubHandler<sc_rpc::Metadata>
+) -> sc_rpc_server::RpcHandler<sc_rpc::Metadata>
 	where
 		TBl: BlockT,
 		TCl: ProvideRuntimeApi<TBl> + BlockchainEvents<TBl> + HeaderBackend<TBl> +
@@ -735,15 +744,18 @@ fn gen_handler<TBl, TBackend, TExPool, TRpc, TCl>(
 		offchain::OffchainApi::to_delegate(offchain)
 	});
 
-	sc_rpc_server::rpc_handler((
-		state::StateApi::to_delegate(state),
-		state::ChildStateApi::to_delegate(child_state),
-		chain::ChainApi::to_delegate(chain),
-		maybe_offchain_rpc,
-		author::AuthorApi::to_delegate(author),
-		system::SystemApi::to_delegate(system),
-		rpc_extensions_builder.build(deny_unsafe, task_executor),
-	))
+	sc_rpc_server::rpc_handler(
+		(
+			state::StateApi::to_delegate(state),
+			state::ChildStateApi::to_delegate(child_state),
+			chain::ChainApi::to_delegate(chain),
+			maybe_offchain_rpc,
+			author::AuthorApi::to_delegate(author),
+			system::SystemApi::to_delegate(system),
+			rpc_extensions_builder.build(deny_unsafe, task_executor),
+		),
+		rpc_middleware
+	)
 }
 
 /// Parameters to pass into `build_network`.

substrate/client/service/src/lib.rs

@@ -97,7 +97,7 @@ impl<T> MallocSizeOfWasm for T {}
 
 /// RPC handlers that can perform RPC queries.
 #[derive(Clone)]
-pub struct RpcHandlers(Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata>>);
+pub struct RpcHandlers(Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata, sc_rpc_server::RpcMiddleware>>);
 
 impl RpcHandlers {
 	/// Starts an RPC query.
@@ -118,7 +118,8 @@ impl RpcHandlers {
 	}
 
 	/// Provides access to the underlying `MetaIoHandler`
-	pub fn io_handler(&self) -> Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata>> {
+	pub fn io_handler(&self)
+		-> Arc<jsonrpc_core::MetaIoHandler<sc_rpc::Metadata, sc_rpc_server::RpcMiddleware>> {
 		self.0.clone()
 	}
 }
@@ -382,9 +383,13 @@ mod waiting {
 
 /// Starts RPC servers that run in their own thread, and returns an opaque object that keeps them alive.
 #[cfg(not(target_os = "unknown"))]
-fn start_rpc_servers<H: FnMut(sc_rpc::DenyUnsafe) -> sc_rpc_server::RpcHandler<sc_rpc::Metadata>>(
+fn start_rpc_servers<
+	H: FnMut(sc_rpc::DenyUnsafe, sc_rpc_server::RpcMiddleware)
+	-> sc_rpc_server::RpcHandler<sc_rpc::Metadata>
+>(
 	config: &Configuration,
-	mut gen_handler: H
+	mut gen_handler: H,
+	rpc_metrics: Option<&sc_rpc_server::RpcMetrics>
 ) -> Result<Box<dyn std::any::Any + Send + Sync>, error::Error> {
 	fn maybe_start_server<T, F>(address: Option<SocketAddr>, mut start: F) -> Result<Option<T>, io::Error>
 		where F: FnMut(&SocketAddr) -> Result<T, io::Error>,
@@ -414,13 +419,21 @@ fn start_rpc_servers<H: FnMut(sc_rpc::DenyUnsafe) -> sc_rpc_server::RpcHandler<s
 	}
 
 	Ok(Box::new((
-		config.rpc_ipc.as_ref().map(|path| sc_rpc_server::start_ipc(&*path, gen_handler(sc_rpc::DenyUnsafe::No))),
+		config.rpc_ipc.as_ref().map(|path| sc_rpc_server::start_ipc(
+			&*path, gen_handler(
+				sc_rpc::DenyUnsafe::No,
+				sc_rpc_server::RpcMiddleware::new(rpc_metrics.cloned(), "ipc")
+			)
+		)),
 		maybe_start_server(
 			config.rpc_http,
 			|address| sc_rpc_server::start_http(
 				address,
 				config.rpc_cors.as_ref(),
-				gen_handler(deny_unsafe(&address, &config.rpc_methods)),
+				gen_handler(
+					deny_unsafe(&address, &config.rpc_methods),
+					sc_rpc_server::RpcMiddleware::new(rpc_metrics.cloned(), "http")
+				),
 			),
 		)?.map(|s| waiting::HttpServer(Some(s))),
 		maybe_start_server(
@@ -429,7 +442,10 @@ fn start_rpc_servers<H: FnMut(sc_rpc::DenyUnsafe) -> sc_rpc_server::RpcHandler<s
 				address,
 				config.rpc_ws_max_connections,
 				config.rpc_cors.as_ref(),
-				gen_handler(deny_unsafe(&address, &config.rpc_methods)),
+				gen_handler(
+					deny_unsafe(&address, &config.rpc_methods),
+					sc_rpc_server::RpcMiddleware::new(rpc_metrics.cloned(), "ws")
+				),
 			),
 		)?.map(|s| waiting::WsServer(Some(s))),
 	)))
@@ -437,9 +453,13 @@ fn start_rpc_servers<H: FnMut(sc_rpc::DenyUnsafe) -> sc_rpc_server::RpcHandler<s
 
 /// Starts RPC servers that run in their own thread, and returns an opaque object that keeps them alive.
 #[cfg(target_os = "unknown")]
-fn start_rpc_servers<H: FnMut(sc_rpc::DenyUnsafe) -> sc_rpc_server::RpcHandler<sc_rpc::Metadata>>(
+fn start_rpc_servers<
+	H: FnMut(sc_rpc::DenyUnsafe, sc_rpc_server::RpcMiddleware)
+	-> sc_rpc_server::RpcHandler<sc_rpc::Metadata>
+>(
 	_: &Configuration,
-	_: H
+	_: H,
+	_: Option<&sc_rpc_server::RpcMetrics>
 ) -> Result<Box<dyn std::any::Any + Send + Sync>, error::Error> {
 	Ok(Box::new(()))
 }