feat(fortuna): Traced client for better observability

apps/fortuna/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name    = "fortuna"
-version = "6.2.2"
+version = "6.2.3"
 edition = "2021"
 
 [dependencies]

apps/fortuna/src/chain.rs

@@ -1,3 +1,4 @@
 pub(crate) mod eth_gas_oracle;
 pub(crate) mod ethereum;
 pub(crate) mod reader;
+pub(crate) mod traced_client;

apps/fortuna/src/chain/ethereum.rs

@@ -1,5 +1,6 @@
 use {
     crate::{
+        api::ChainId,
         chain::{
             eth_gas_oracle::EthProviderOracle,
             reader::{
@@ -9,6 +10,10 @@ use {
                 EntropyReader,
                 RequestedWithCallbackEvent,
             },
+            traced_client::{
+                RpcMetrics,
+                TracedClient,
+            },
         },
         config::EthereumConfig,
     },
@@ -22,7 +27,6 @@ use {
         abi::RawLog,
         contract::{
             abigen,
-            ContractError,
             EthLogDecode,
         },
         core::types::Address,
@@ -34,6 +38,7 @@ use {
         },
         prelude::{
             BlockId,
+            JsonRpcClient,
             PendingTransaction,
             TransactionRequest,
         },
@@ -67,15 +72,19 @@ abigen!(
     "../../target_chains/ethereum/entropy_sdk/solidity/abis/IEntropy.json"
 );
 
-pub type SignablePythContract = PythRandom<
+pub type SignablePythContractInner<T> = PythRandom<
     LegacyTxMiddleware<
         GasOracleMiddleware<
-            NonceManagerMiddleware<SignerMiddleware<Provider<Http>, LocalWallet>>,
-            EthProviderOracle<Provider<Http>>,
+            NonceManagerMiddleware<SignerMiddleware<Provider<T>, LocalWallet>>,
+            EthProviderOracle<Provider<T>>,
         >,
     >,
 >;
+pub type SignablePythContract = SignablePythContractInner<Http>;
+pub type InstrumentedSignablePythContract = SignablePythContractInner<TracedClient>;
+
 pub type PythContract = PythRandom<Provider<Http>>;
+pub type InstrumentedPythContract = PythRandom<Provider<TracedClient>>;
 
 /// Middleware that converts a transaction into a legacy transaction if use_legacy_tx is true.
 /// We can not use TransformerMiddleware because keeper calls fill_transaction first which bypasses
@@ -157,32 +166,7 @@ impl<M: Middleware> Middleware for LegacyTxMiddleware<M> {
     }
 }
 
-impl SignablePythContract {
-    pub async fn from_config(
-        chain_config: &EthereumConfig,
-        private_key: &str,
-    ) -> Result<SignablePythContract> {
-        let provider = Provider::<Http>::try_from(&chain_config.geth_rpc_addr)?;
-        let chain_id = provider.get_chainid().await?;
-        let gas_oracle = EthProviderOracle::new(provider.clone());
-        let wallet__ = private_key
-            .parse::<LocalWallet>()?
-            .with_chain_id(chain_id.as_u64());
-
-        let address = wallet__.address();
-
-        Ok(PythRandom::new(
-            chain_config.contract_addr,
-            Arc::new(LegacyTxMiddleware::new(
-                chain_config.legacy_tx,
-                GasOracleMiddleware::new(
-                    NonceManagerMiddleware::new(SignerMiddleware::new(provider, wallet__), address),
-                    gas_oracle,
-                ),
-            )),
-        ))
-    }
-
+impl<T: JsonRpcClient + 'static + Clone> SignablePythContractInner<T> {
     /// Submit a request for a random number to the contract.
     ///
     /// This method is a version of the autogenned `request` method that parses the emitted logs
@@ -249,10 +233,54 @@ impl SignablePythContract {
             Err(anyhow!("Request failed").into())
         }
     }
+
+    pub async fn from_config_and_provider(
+        chain_config: &EthereumConfig,
+        private_key: &str,
+        provider: Provider<T>,
+    ) -> Result<SignablePythContractInner<T>> {
+        let chain_id = provider.get_chainid().await?;
+        let gas_oracle = EthProviderOracle::new(provider.clone());
+        let wallet__ = private_key
+            .parse::<LocalWallet>()?
+            .with_chain_id(chain_id.as_u64());
+
+        let address = wallet__.address();
+
+        Ok(PythRandom::new(
+            chain_config.contract_addr,
+            Arc::new(LegacyTxMiddleware::new(
+                chain_config.legacy_tx,
+                GasOracleMiddleware::new(
+                    NonceManagerMiddleware::new(SignerMiddleware::new(provider, wallet__), address),
+                    gas_oracle,
+                ),
+            )),
+        ))
+    }
+}
+
+impl SignablePythContract {
+    pub async fn from_config(chain_config: &EthereumConfig, private_key: &str) -> Result<Self> {
+        let provider = Provider::<Http>::try_from(&chain_config.geth_rpc_addr)?;
+        Self::from_config_and_provider(chain_config, private_key, provider).await
+    }
+}
+
+impl InstrumentedSignablePythContract {
+    pub async fn from_config(
+        chain_config: &EthereumConfig,
+        private_key: &str,
+        chain_id: ChainId,
+        metrics: Arc<RpcMetrics>,
+    ) -> Result<Self> {
+        let provider = TracedClient::new(chain_id, &chain_config.geth_rpc_addr, metrics)?;
+        Self::from_config_and_provider(chain_config, private_key, provider).await
+    }
 }
 
 impl PythContract {
-    pub fn from_config(chain_config: &EthereumConfig) -> Result<PythContract> {
+    pub fn from_config(chain_config: &EthereumConfig) -> Result<Self> {
         let provider = Provider::<Http>::try_from(&chain_config.geth_rpc_addr)?;
 
         Ok(PythRandom::new(
@@ -262,8 +290,23 @@ impl PythContract {
     }
 }
 
+impl InstrumentedPythContract {
+    pub fn from_config(
+        chain_config: &EthereumConfig,
+        chain_id: ChainId,
+        metrics: Arc<RpcMetrics>,
+    ) -> Result<Self> {
+        let provider = TracedClient::new(chain_id, &chain_config.geth_rpc_addr, metrics)?;
+
+        Ok(PythRandom::new(
+            chain_config.contract_addr,
+            Arc::new(provider),
+        ))
+    }
+}
+
 #[async_trait]
-impl EntropyReader for PythContract {
+impl<T: JsonRpcClient + 'static> EntropyReader for PythRandom<Provider<T>> {
     async fn get_request(
         &self,
         provider_address: Address,
@@ -330,7 +373,7 @@ impl EntropyReader for PythContract {
         user_random_number: [u8; 32],
         provider_revelation: [u8; 32],
     ) -> Result<U256> {
-        let result: Result<U256, ContractError<Provider<Http>>> = self
+        let result = self
             .reveal_with_callback(
                 provider,
                 sequence_number,

apps/fortuna/src/chain/traced_client.rs

@@ -0,0 +1,135 @@
+use {
+    crate::api::ChainId,
+    anyhow::Result,
+    axum::async_trait,
+    ethers::{
+        prelude::Http,
+        providers::{
+            HttpClientError,
+            JsonRpcClient,
+            Provider,
+        },
+    },
+    prometheus_client::{
+        encoding::EncodeLabelSet,
+        metrics::{
+            counter::Counter,
+            family::Family,
+            histogram::Histogram,
+        },
+        registry::Registry,
+    },
+    std::sync::Arc,
+    tokio::{
+        sync::RwLock,
+        time::Instant,
+    },
+};
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, EncodeLabelSet)]
+pub struct RpcLabel {
+    chain_id: ChainId,
+    method:   String,
+}
+
+#[derive(Debug)]
+pub struct RpcMetrics {
+    count:        Family<RpcLabel, Counter>,
+    latency:      Family<RpcLabel, Histogram>,
+    errors_count: Family<RpcLabel, Counter>,
+}
+
+impl RpcMetrics {
+    pub async fn new(metrics_registry: Arc<RwLock<Registry>>) -> Self {
+        let count = Family::default();
+        let mut guard = metrics_registry.write().await;
+        let sub_registry = guard.sub_registry_with_prefix("rpc_requests");
+        sub_registry.register(
+            "count",
+            "The number of RPC requests made to the chain with the specified method.",
+            count.clone(),
+        );
+
+        let latency = Family::<RpcLabel, Histogram>::new_with_constructor(|| {
+            Histogram::new(
+                [
+                    0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0,
+                ]
+                .into_iter(),
+            )
+        });
+        sub_registry.register(
+            "latency",
+            "The latency of RPC requests to the chain with the specified method.",
+            latency.clone(),
+        );
+
+        let errors_count = Family::default();
+        sub_registry.register(
+            "errors_count",
+            "The number of RPC requests made to the chain that failed.",
+            errors_count.clone(),
+        );
+
+        Self {
+            count,
+            latency,
+            errors_count,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct TracedClient {
+    inner: Http,
+
+    chain_id: ChainId,
+    metrics:  Arc<RpcMetrics>,
+}
+
+#[async_trait]
+impl JsonRpcClient for TracedClient {
+    type Error = HttpClientError;
+
+    async fn request<
+        T: serde::Serialize + Send + Sync + std::fmt::Debug,
+        R: serde::de::DeserializeOwned + Send,
+    >(
+        &self,
+        method: &str,
+        params: T,
+    ) -> Result<R, HttpClientError> {
+        let start = Instant::now();
+        let label = &RpcLabel {
+            chain_id: self.chain_id.clone(),
+            method:   method.to_string(),
+        };
+        self.metrics.count.get_or_create(label).inc();
+        let res = match self.inner.request(method, params).await {
+            Ok(result) => Ok(result),
+            Err(e) => {
+                self.metrics.errors_count.get_or_create(label).inc();
+                Err(e)
+            }
+        };
+
+        let latency = start.elapsed().as_secs_f64();
+        self.metrics.latency.get_or_create(label).observe(latency);
+        res
+    }
+}
+
+impl TracedClient {
+    pub fn new(
+        chain_id: ChainId,
+        url: &str,
+        metrics: Arc<RpcMetrics>,
+    ) -> Result<Provider<TracedClient>> {
+        let url = url::Url::parse(url)?;
+        Ok(Provider::new(TracedClient {
+            inner: Http::new(url),
+            chain_id,
+            metrics,
+        }))
+    }
+}