Pre-create metrics registry before loop is started + administrative metrics (#848)

* administrative metrics

* fmt

* fix compilation

* fix compilation again

* and another one

* remove GenericLoopMetrics

* chttp -> isahc

* remove redundant marker

* not about price metrics

* fmt

bridges/relays/utils/src/lib.rs

@@ -16,7 +16,7 @@
 
 //! Utilities used by different relays.
 
-pub use relay_loop::relay_loop;
+pub use relay_loop::{relay_loop, relay_metrics};
 
 use backoff::{backoff::Backoff, ExponentialBackoff};
 use futures::future::FutureExt;

bridges/relays/utils/src/metrics.rs

@@ -14,23 +14,37 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
+pub use float_json_value::FloatJsonValueMetric;
 pub use global::GlobalMetrics;
-pub use substrate_prometheus_endpoint::{register, Counter, CounterVec, Gauge, GaugeVec, Opts, Registry, F64, U64};
+pub use substrate_prometheus_endpoint::{
+	prometheus::core::{Atomic, Collector},
+	register, Counter, CounterVec, Gauge, GaugeVec, Opts, Registry, F64, U64,
+};
 
 use async_trait::async_trait;
-use std::time::Duration;
+use std::{fmt::Debug, time::Duration};
 
+mod float_json_value;
 mod global;
 
-/// Prometheus endpoint MetricsParams.
+/// Unparsed address that needs to be used to expose Prometheus metrics.
 #[derive(Debug, Clone)]
-pub struct MetricsParams {
+pub struct MetricsAddress {
 	/// Serve HTTP requests at given host.
 	pub host: String,
 	/// Serve HTTP requests at given port.
 	pub port: u16,
 }
 
+/// Prometheus endpoint MetricsParams.
+#[derive(Debug, Clone)]
+pub struct MetricsParams {
+	/// Interface and TCP port to be used when exposing Prometheus metrics.
+	pub address: Option<MetricsAddress>,
+	/// Metrics registry. May be `Some(_)` if several components share the same endpoint.
+	pub registry: Option<Registry>,
+}
+
 /// Metrics API.
 pub trait Metrics: Clone + Send + Sync + 'static {
 	/// Register metrics in the registry.
@@ -61,11 +75,64 @@ pub trait StandaloneMetrics: Metrics {
 	}
 }
 
-impl Default for MetricsParams {
+impl Default for MetricsAddress {
 	fn default() -> Self {
-		MetricsParams {
+		MetricsAddress {
 			host: "127.0.0.1".into(),
 			port: 9616,
 		}
 	}
 }
+
+impl MetricsParams {
+	/// Creates metrics params so that metrics are not exposed.
+	pub fn disabled() -> Self {
+		MetricsParams {
+			address: None,
+			registry: None,
+		}
+	}
+}
+
+impl From<Option<MetricsAddress>> for MetricsParams {
+	fn from(address: Option<MetricsAddress>) -> Self {
+		MetricsParams {
+			address,
+			registry: None,
+		}
+	}
+}
+
+/// Set value of gauge metric.
+///
+/// If value is `Ok(None)` or `Err(_)`, metric would have default value.
+pub fn set_gauge_value<T: Default + Debug, V: Atomic<T = T>, E: Debug>(gauge: &Gauge<V>, value: Result<Option<T>, E>) {
+	gauge.set(match value {
+		Ok(Some(value)) => {
+			log::trace!(
+				target: "bridge-metrics",
+				"Updated value of metric '{:?}': {:?}",
+				gauge.desc().first().map(|d| &d.fq_name),
+				value,
+			);
+			value
+		}
+		Ok(None) => {
+			log::warn!(
+				target: "bridge-metrics",
+				"Failed to update metric '{:?}': value is empty",
+				gauge.desc().first().map(|d| &d.fq_name),
+			);
+			Default::default()
+		}
+		Err(error) => {
+			log::warn!(
+				target: "bridge-metrics",
+				"Failed to update metric '{:?}': {:?}",
+				gauge.desc().first().map(|d| &d.fq_name),
+				error,
+			);
+			Default::default()
+		}
+	})
+}

bridges/relays/utils/src/metrics/float_json_value.rs

@@ -0,0 +1,125 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Parity Bridges Common.
+
+// Parity Bridges Common 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.
+
+// Parity Bridges Common 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 Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
+
+use crate::metrics::{register, Gauge, Metrics, Registry, StandaloneMetrics, F64};
+
+use async_trait::async_trait;
+use std::time::Duration;
+
+/// Value update interval.
+const UPDATE_INTERVAL: Duration = Duration::from_secs(60);
+
+/// Metric that represents float value received from HTTP service as float gauge.
+#[derive(Debug, Clone)]
+pub struct FloatJsonValueMetric {
+	url: String,
+	json_path: String,
+	metric: Gauge<F64>,
+}
+
+impl FloatJsonValueMetric {
+	/// Create new metric instance with given name and help.
+	pub fn new(url: String, json_path: String, name: String, help: String) -> Self {
+		FloatJsonValueMetric {
+			url,
+			json_path,
+			metric: Gauge::new(name, help).expect(
+				"only fails if gauge options are customized;\
+					we use default options;\
+					qed",
+			),
+		}
+	}
+
+	/// Read value from HTTP service.
+	async fn read_value(&self) -> Result<f64, String> {
+		use isahc::{AsyncReadResponseExt, HttpClient, Request};
+
+		fn map_isahc_err(err: impl std::fmt::Display) -> String {
+			format!("Failed to fetch token price from remote server: {}", err)
+		}
+
+		let request = Request::get(&self.url)
+			.header("Accept", "application/json")
+			.body(())
+			.map_err(map_isahc_err)?;
+		let raw_response = HttpClient::new()
+			.map_err(map_isahc_err)?
+			.send_async(request)
+			.await
+			.map_err(map_isahc_err)?
+			.text()
+			.await
+			.map_err(map_isahc_err)?;
+
+		parse_service_response(&self.json_path, &raw_response)
+	}
+}
+
+impl Metrics for FloatJsonValueMetric {
+	fn register(&self, registry: &Registry) -> Result<(), String> {
+		register(self.metric.clone(), registry).map_err(|e| e.to_string())?;
+		Ok(())
+	}
+}
+
+#[async_trait]
+impl StandaloneMetrics for FloatJsonValueMetric {
+	fn update_interval(&self) -> Duration {
+		UPDATE_INTERVAL
+	}
+
+	async fn update(&self) {
+		crate::metrics::set_gauge_value(&self.metric, self.read_value().await.map(Some));
+	}
+}
+
+/// Parse HTTP service response.
+fn parse_service_response(json_path: &str, response: &str) -> Result<f64, String> {
+	let json = serde_json::from_str(response).map_err(|err| {
+		format!(
+			"Failed to parse HTTP service response: {:?}. Response: {:?}",
+			err, response,
+		)
+	})?;
+
+	let mut selector = jsonpath_lib::selector(&json);
+	let maybe_selected_value = selector(json_path).map_err(|err| {
+		format!(
+			"Failed to select value from response: {:?}. Response: {:?}",
+			err, response,
+		)
+	})?;
+	let selected_value = maybe_selected_value
+		.first()
+		.and_then(|v| v.as_f64())
+		.ok_or_else(|| format!("Missing required value from response: {:?}", response,))?;
+
+	Ok(selected_value)
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn parse_service_response_works() {
+		assert_eq!(
+			parse_service_response("$.kusama.usd", r#"{"kusama":{"usd":433.05}}"#).map_err(drop),
+			Ok(433.05),
+		);
+	}
+}

bridges/relays/utils/src/metrics/global.rs

@@ -16,12 +16,11 @@
 
 //! Global system-wide Prometheus metrics exposed by relays.
 
-use crate::metrics::{Metrics, StandaloneMetrics};
+use crate::metrics::{register, Gauge, GaugeVec, Metrics, Opts, Registry, StandaloneMetrics, F64, U64};
 
 use async_std::sync::{Arc, Mutex};
 use async_trait::async_trait;
 use std::time::Duration;
-use substrate_prometheus_endpoint::{register, Gauge, GaugeVec, Opts, Registry, F64, U64};
 use sysinfo::{ProcessExt, RefreshKind, System, SystemExt};
 
 /// Global metrics update interval.
@@ -79,7 +78,7 @@ impl StandaloneMetrics for GlobalMetrics {
 			}
 			_ => {
 				log::warn!(
-					target: "bridge",
+					target: "bridge-metrics",
 					"Failed to refresh process information. Metrics may show obsolete values",
 				);
 			}

bridges/relays/utils/src/relay_loop.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Bridges Common.  If not, see <http://www.gnu.org/licenses/>.
 
-use crate::metrics::{Metrics, MetricsParams, StandaloneMetrics};
+use crate::metrics::{Metrics, MetricsAddress, MetricsParams, StandaloneMetrics};
 use crate::{FailedClient, MaybeConnectionError};
 
 use async_trait::async_trait;
@@ -44,6 +44,24 @@ pub fn relay_loop<SC, TC>(source_client: SC, target_client: TC) -> Loop<SC, TC,
 	}
 }
 
+/// Returns generic relay loop metrics that may be customized and used in one or several relay loops.
+pub fn relay_metrics(prefix: String, address: Option<MetricsAddress>) -> LoopMetrics<(), (), ()> {
+	assert!(!prefix.is_empty(), "Metrics prefix can not be empty");
+
+	LoopMetrics {
+		relay_loop: Loop {
+			reconnect_delay: RECONNECT_DELAY,
+			source_client: (),
+			target_client: (),
+			loop_metric: None,
+		},
+		address,
+		registry: Registry::new_custom(Some(prefix), None)
+			.expect("only fails if prefix is empty; prefix is not empty; qed"),
+		loop_metric: None,
+	}
+}
+
 /// Generic relay loop.
 pub struct Loop<SC, TC, LM> {
 	reconnect_delay: Duration,
@@ -55,6 +73,7 @@ pub struct Loop<SC, TC, LM> {
 /// Relay loop metrics builder.
 pub struct LoopMetrics<SC, TC, LM> {
 	relay_loop: Loop<SC, TC, ()>,
+	address: Option<MetricsAddress>,
 	registry: Registry,
 	loop_metric: Option<LM>,
 }
@@ -69,7 +88,7 @@ impl<SC, TC, LM> Loop<SC, TC, LM> {
 	/// Start building loop metrics using given prefix.
 	///
 	/// Panics if `prefix` is empty.
-	pub fn with_metrics(self, prefix: String) -> LoopMetrics<SC, TC, ()> {
+	pub fn with_metrics(self, prefix: String, params: MetricsParams) -> LoopMetrics<SC, TC, ()> {
 		assert!(!prefix.is_empty(), "Metrics prefix can not be empty");
 
 		LoopMetrics {
@@ -79,8 +98,12 @@ impl<SC, TC, LM> Loop<SC, TC, LM> {
 				target_client: self.target_client,
 				loop_metric: None,
 			},
-			registry: Registry::new_custom(Some(prefix), None)
-				.expect("only fails if prefix is empty; prefix is not empty; qed"),
+			address: params.address,
+			registry: match params.registry {
+				Some(registry) => registry,
+				None => Registry::new_custom(Some(prefix), None)
+					.expect("only fails if prefix is empty; prefix is not empty; qed"),
+			},
 			loop_metric: None,
 		}
 	}
@@ -159,6 +182,7 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 
 		Ok(LoopMetrics {
 			relay_loop: self.relay_loop,
+			address: self.address,
 			registry: self.registry,
 			loop_metric: Some(loop_metric),
 		})
@@ -171,19 +195,27 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 		Ok(self)
 	}
 
-	/// Expose metrics using given params.
+	/// Convert into `MetricsParams` structure so that metrics registry may be extended later.
+	pub fn into_params(self) -> MetricsParams {
+		MetricsParams {
+			address: self.address,
+			registry: Some(self.registry),
+		}
+	}
+
+	/// Expose metrics using address passed at creation.
 	///
-	/// If `params` is `None`, metrics are not exposed.
-	pub async fn expose(self, params: Option<MetricsParams>) -> Result<Loop<SC, TC, LM>, String> {
-		if let Some(params) = params {
+	/// If passed `address` is `None`, metrics are not exposed.
+	pub async fn expose(self) -> Result<Loop<SC, TC, LM>, String> {
+		if let Some(address) = self.address {
 			let socket_addr = SocketAddr::new(
-				params.host.parse().map_err(|err| {
+				address.host.parse().map_err(|err| {
 					format!(
 						"Invalid host {} is used to expose Prometheus metrics: {}",
-						params.host, err,
+						address.host, err,
 					)
 				})?,
-				params.port,
+				address.port,
 			);
 
 			let registry = self.registry;