Put block timestamp lag next to other tracking tasks

Author: m30m

apps/fortuna/src/command/run.rs

@@ -4,40 +4,24 @@ use {
         chain::ethereum::InstrumentedPythContract,
         command::register_provider::CommitmentMetadata,
         config::{Commitment, Config, EthereumConfig, RunOptions},
-        eth_utils::traced_client::{RpcMetrics, TracedClient},
+        eth_utils::traced_client::RpcMetrics,
         keeper::{self, keeper_metrics::KeeperMetrics},
         state::{HashChainState, PebbleHashChain},
     },
     anyhow::{anyhow, Error, Result},
     axum::Router,
-    ethers::{
-        middleware::Middleware,
-        types::{Address, BlockNumber},
-    },
-    prometheus_client::{
-        encoding::EncodeLabelSet,
-        metrics::{family::Family, gauge::Gauge},
-        registry::Registry,
-    },
-    std::{
-        collections::HashMap,
-        net::SocketAddr,
-        sync::Arc,
-        time::{Duration, SystemTime, UNIX_EPOCH},
-    },
+    ethers::types::Address,
+    prometheus_client::{encoding::EncodeLabelSet, registry::Registry},
+    std::{collections::HashMap, net::SocketAddr, sync::Arc},
     tokio::{
         spawn,
         sync::{watch, RwLock},
-        time,
     },
     tower_http::cors::CorsLayer,
     utoipa::OpenApi,
     utoipa_swagger_ui::SwaggerUi,
 };
 
-/// Track metrics in this interval
-const TRACK_INTERVAL: Duration = Duration::from_secs(10);
-
 pub async fn run_api(
     socket_addr: SocketAddr,
     chains: Arc<RwLock<HashMap<String, api::BlockchainState>>>,
@@ -158,16 +142,7 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
         Ok::<(), Error>(())
     });
 
-    // Spawn a thread to track latest block lag. This helps us know if the rpc is up and updated with the latest block.
-    //TODO: only run this after the chain is setup
-    spawn(track_block_timestamp_lag(
-        config,
-        metrics_registry.clone(),
-        rpc_metrics.clone(),
-    ));
-
     run_api(opts.addr, chains.clone(), metrics_registry.clone(), rx_exit).await?;
-    tracing::info!("Shut down server");
     Ok(())
 }
 
@@ -271,74 +246,3 @@ async fn setup_chain_state(
 pub struct ChainLabel {
     pub chain_id: String,
 }
-
-#[tracing::instrument(name = "block_timestamp_lag", skip_all, fields(chain_id = chain_id))]
-pub async fn check_block_timestamp_lag(
-    chain_id: String,
-    chain_config: EthereumConfig,
-    metrics: Family<ChainLabel, Gauge>,
-    rpc_metrics: Arc<RpcMetrics>,
-) {
-    let provider =
-        match TracedClient::new(chain_id.clone(), &chain_config.geth_rpc_addr, rpc_metrics) {
-            Ok(r) => r,
-            Err(e) => {
-                tracing::error!("Failed to create provider for chain id - {:?}", e);
-                return;
-            }
-        };
-
-    const INF_LAG: i64 = 1000000; // value that definitely triggers an alert
-    let lag = match provider.get_block(BlockNumber::Latest).await {
-        Ok(block) => match block {
-            Some(block) => {
-                let block_timestamp = block.timestamp;
-                let server_timestamp = SystemTime::now()
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs();
-                let lag: i64 = (server_timestamp as i64) - (block_timestamp.as_u64() as i64);
-                lag
-            }
-            None => {
-                tracing::error!("Block is None");
-                INF_LAG
-            }
-        },
-        Err(e) => {
-            tracing::error!("Failed to get block - {:?}", e);
-            INF_LAG
-        }
-    };
-    metrics
-        .get_or_create(&ChainLabel {
-            chain_id: chain_id.clone(),
-        })
-        .set(lag);
-}
-
-/// Tracks the difference between the server timestamp and the latest block timestamp for each chain
-pub async fn track_block_timestamp_lag(
-    config: Config,
-    metrics_registry: Arc<RwLock<Registry>>,
-    rpc_metrics: Arc<RpcMetrics>,
-) {
-    let metrics = Family::<ChainLabel, Gauge>::default();
-    metrics_registry.write().await.register(
-        "block_timestamp_lag",
-        "The difference between server timestamp and latest block timestamp",
-        metrics.clone(),
-    );
-    loop {
-        for (chain_id, chain_config) in &config.chains {
-            spawn(check_block_timestamp_lag(
-                chain_id.clone(),
-                chain_config.clone(),
-                metrics.clone(),
-                rpc_metrics.clone(),
-            ));
-        }
-
-        time::sleep(TRACK_INTERVAL).await;
-    }
-}

apps/fortuna/src/keeper.rs

@@ -1,3 +1,4 @@
+use crate::keeper::track::track_block_timestamp_lag;
 use {
     crate::{
         api::{BlockchainState, ChainId},
@@ -215,6 +216,14 @@ pub async fn run_keeper_threads(
                     )
                     .in_current_span(),
                 );
+                spawn(
+                    track_block_timestamp_lag(
+                        chain_id.clone(),
+                        contract.client(),
+                        keeper_metrics.clone(),
+                    )
+                    .in_current_span(),
+                );
 
                 time::sleep(TRACK_INTERVAL).await;
             }

apps/fortuna/src/keeper/keeper_metrics.rs

@@ -42,6 +42,7 @@ pub struct KeeperMetrics {
     pub final_fee_multiplier: Family<AccountLabel, Histogram>,
     pub gas_price_estimate: Family<AccountLabel, Gauge<f64, AtomicU64>>,
     pub accrued_pyth_fees: Family<ChainIdLabel, Gauge<f64, AtomicU64>>,
+    pub block_timestamp_lag: Family<ChainIdLabel, Gauge>,
 }
 
 impl Default for KeeperMetrics {
@@ -83,6 +84,7 @@ impl Default for KeeperMetrics {
             }),
             gas_price_estimate: Family::default(),
             accrued_pyth_fees: Family::default(),
+            block_timestamp_lag: Family::default(),
         }
     }
 }
@@ -212,6 +214,12 @@ impl KeeperMetrics {
             keeper_metrics.accrued_pyth_fees.clone(),
         );
 
+        writable_registry.register(
+            "block_timestamp_lag",
+            "The difference between server timestamp and latest block timestamp",
+            keeper_metrics.block_timestamp_lag.clone(),
+        );
+
         // *Important*: When adding a new metric:
         // 1. Register it above using `writable_registry.register(...)`
         // 2. Add a get_or_create call in the add_chain function below to initialize it for each chain/provider pair
@@ -220,9 +228,11 @@ impl KeeperMetrics {
     }
 
     pub fn add_chain(&self, chain_id: String, provider_address: Address) {
-        let _ = self.accrued_pyth_fees.get_or_create(&ChainIdLabel {
+        let chain_id_label = ChainIdLabel {
             chain_id: chain_id.clone(),
-        });
+        };
+        let _ = self.accrued_pyth_fees.get_or_create(&chain_id_label);
+        let _ = self.block_timestamp_lag.get_or_create(&chain_id_label);
 
         let account_label = AccountLabel {
             chain_id,

apps/fortuna/src/keeper/track.rs

@@ -5,8 +5,11 @@ use {
         eth_utils::traced_client::TracedClient,
     },
     ethers::middleware::Middleware,
-    ethers::{providers::Provider, types::Address},
-    std::sync::Arc,
+    ethers::{prelude::BlockNumber, providers::Provider, types::Address},
+    std::{
+        sync::Arc,
+        time::{SystemTime, UNIX_EPOCH},
+    },
     tracing,
 };
 
@@ -41,6 +44,43 @@ pub async fn track_balance(
         .set(balance);
 }
 
+/// Tracks the difference between the server timestamp and the latest block timestamp for each chain
+#[tracing::instrument(skip_all)]
+pub async fn track_block_timestamp_lag(
+    chain_id: String,
+    provider: Arc<Provider<TracedClient>>,
+    metrics: Arc<KeeperMetrics>,
+) {
+    const INF_LAG: i64 = 1000000; // value that definitely triggers an alert
+    let lag = match provider.get_block(BlockNumber::Latest).await {
+        Ok(block) => match block {
+            Some(block) => {
+                let block_timestamp = block.timestamp;
+                let server_timestamp = SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs();
+                let lag: i64 = (server_timestamp as i64) - (block_timestamp.as_u64() as i64);
+                lag
+            }
+            None => {
+                tracing::error!("Block is None");
+                INF_LAG
+            }
+        },
+        Err(e) => {
+            tracing::error!("Failed to get block - {:?}", e);
+            INF_LAG
+        }
+    };
+    metrics
+        .block_timestamp_lag
+        .get_or_create(&ChainIdLabel {
+            chain_id: chain_id.clone(),
+        })
+        .set(lag);
+}
+
 /// tracks the collected fees and the hashchain data of the given provider address on the given chain
 /// if there is a error the function will just return
 #[tracing::instrument(skip_all)]