Shared reference to conversion rate metric value (#1034)

* shared conversion rate metric value

* clippy

bridges/relays/utils/src/metrics.rs

@@ -21,12 +21,16 @@ pub use substrate_prometheus_endpoint::{
 	register, Counter, CounterVec, Gauge, GaugeVec, Opts, PrometheusError, Registry, F64, U64,
 };
 
+use async_std::sync::{Arc, RwLock};
 use async_trait::async_trait;
 use std::{fmt::Debug, time::Duration};
 
 mod float_json_value;
 mod global;
 
+/// Shared reference to `f64` value that is updated by the metric.
+pub type F64SharedRef = Arc<RwLock<Option<f64>>>;
+
 /// Unparsed address that needs to be used to expose Prometheus metrics.
 #[derive(Debug, Clone)]
 pub struct MetricsAddress {

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

@@ -14,8 +14,9 @@
 // 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::{metric_name, register, Gauge, PrometheusError, Registry, StandaloneMetrics, F64};
+use crate::metrics::{metric_name, register, F64SharedRef, Gauge, PrometheusError, Registry, StandaloneMetrics, F64};
 
+use async_std::sync::{Arc, RwLock};
 use async_trait::async_trait;
 use std::time::Duration;
 
@@ -28,6 +29,7 @@ pub struct FloatJsonValueMetric {
 	url: String,
 	json_path: String,
 	metric: Gauge<F64>,
+	shared_value_ref: F64SharedRef,
 }
 
 impl FloatJsonValueMetric {
@@ -40,13 +42,20 @@ impl FloatJsonValueMetric {
 		name: String,
 		help: String,
 	) -> Result<Self, PrometheusError> {
+		let shared_value_ref = Arc::new(RwLock::new(None));
 		Ok(FloatJsonValueMetric {
 			url,
 			json_path,
 			metric: register(Gauge::new(metric_name(prefix, &name), help)?, registry)?,
+			shared_value_ref,
 		})
 	}
 
+	/// Get shared reference to metric value.
+	pub fn shared_value_ref(&self) -> F64SharedRef {
+		self.shared_value_ref.clone()
+	}
+
 	/// Read value from HTTP service.
 	async fn read_value(&self) -> Result<f64, String> {
 		use isahc::{AsyncReadResponseExt, HttpClient, Request};
@@ -79,7 +88,9 @@ impl StandaloneMetrics for FloatJsonValueMetric {
 	}
 
 	async fn update(&self) {
-		crate::metrics::set_gauge_value(&self.metric, self.read_value().await.map(Some));
+		let value = self.read_value().await;
+		crate::metrics::set_gauge_value(&self.metric, value.clone().map(Some));
+		*self.shared_value_ref.write().await = value.ok();
 	}
 }
 

bridges/relays/utils/src/relay_loop.rs

@@ -105,7 +105,7 @@ impl<SC, TC, LM> Loop<SC, TC, LM> {
 	/// This function represents an outer loop, which in turn calls provided `run_loop` function to do
 	/// actual job. When `run_loop` returns, this outer loop reconnects to failed client (source,
 	/// target or both) and calls `run_loop` again.
-	pub async fn run<R, F>(mut self, loop_name: String, run_loop: R) -> Result<(), String>
+	pub async fn run<R, F>(mut self, loop_name: String, run_loop: R) -> anyhow::Result<()>
 	where
 		R: 'static + Send + Fn(SC, TC, Option<LM>) -> F,
 		F: 'static + Send + Future<Output = Result<(), FailedClient>>,
@@ -151,8 +151,8 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 	pub fn loop_metric<NewLM: Metrics>(
 		self,
 		create_metric: impl FnOnce(&Registry, Option<&str>) -> Result<NewLM, PrometheusError>,
-	) -> Result<LoopMetrics<SC, TC, NewLM>, String> {
-		let loop_metric = create_metric(&self.registry, self.metrics_prefix.as_deref()).map_err(|e| e.to_string())?;
+	) -> anyhow::Result<LoopMetrics<SC, TC, NewLM>> {
+		let loop_metric = create_metric(&self.registry, self.metrics_prefix.as_deref())?;
 
 		Ok(LoopMetrics {
 			relay_loop: self.relay_loop,
@@ -167,13 +167,13 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 	pub fn standalone_metric<M: StandaloneMetrics>(
 		self,
 		create_metric: impl FnOnce(&Registry, Option<&str>) -> Result<M, PrometheusError>,
-	) -> Result<Self, String> {
+	) -> anyhow::Result<Self> {
 		// since standalone metrics are updating themselves, we may just ignore the fact that the same
 		// standalone metric is exposed by several loops && only spawn single metric
 		match create_metric(&self.registry, self.metrics_prefix.as_deref()) {
 			Ok(standalone_metrics) => standalone_metrics.spawn(),
 			Err(PrometheusError::AlreadyReg) => (),
-			Err(e) => return Err(e.to_string()),
+			Err(e) => anyhow::bail!(e),
 		}
 
 		Ok(self)
@@ -191,13 +191,14 @@ impl<SC, TC, LM> LoopMetrics<SC, TC, LM> {
 	/// Expose metrics using address passed at creation.
 	///
 	/// If passed `address` is `None`, metrics are not exposed.
-	pub async fn expose(self) -> Result<Loop<SC, TC, LM>, String> {
+	pub async fn expose(self) -> anyhow::Result<Loop<SC, TC, LM>> {
 		if let Some(address) = self.address {
 			let socket_addr = SocketAddr::new(
 				address.host.parse().map_err(|err| {
-					format!(
+					anyhow::format_err!(
 						"Invalid host {} is used to expose Prometheus metrics: {}",
-						address.host, err,
+						address.host,
+						err,
 					)
 				})?,
 				address.port,