Relayer balance metrics

rust/agents/relayer/src/relayer.rs

@@ -9,11 +9,15 @@ use derive_more::AsRef;
 use eyre::Result;
 use hyperlane_base::{
     db::{HyperlaneRocksDB, DB},
-    run_all, BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
-    SequencedDataContractSync, WatermarkContractSync,
+    metrics::{AgentMetrics, Metrics},
+    run_all,
+    settings::ChainConf,
+    BaseAgent, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore, SequencedDataContractSync,
+    WatermarkContractSync,
 };
 use hyperlane_core::{
-    HyperlaneDomain, HyperlaneMessage, InterchainGasPayment, MerkleTreeInsertion, U256,
+    metrics::agent::METRICS_SCRAPE_INTERVAL, HyperlaneDomain, HyperlaneMessage,
+    InterchainGasPayment, MerkleTreeInsertion, U256,
 };
 use tokio::{
     sync::{
@@ -49,7 +53,7 @@ struct ContextKey {
 #[derive(AsRef)]
 pub struct Relayer {
     origin_chains: HashSet<HyperlaneDomain>,
-    destination_chains: HashSet<HyperlaneDomain>,
+    destination_chains: HashMap<HyperlaneDomain, ChainConf>,
     #[as_ref]
     core: HyperlaneAgentCore,
     message_syncs: HashMap<HyperlaneDomain, Arc<SequencedDataContractSync<HyperlaneMessage>>>,
@@ -67,6 +71,8 @@ pub struct Relayer {
     transaction_gas_limit: Option<U256>,
     skip_transaction_gas_limit_for: HashSet<u32>,
     allow_local_checkpoint_syncers: bool,
+    core_metrics: Arc<CoreMetrics>,
+    agent_metrics: Metrics,
 }
 
 impl Debug for Relayer {
@@ -92,11 +98,15 @@ impl BaseAgent for Relayer {
 
     type Settings = RelayerSettings;
 
-    async fn from_settings(settings: Self::Settings, metrics: Arc<CoreMetrics>) -> Result<Self>
+    async fn from_settings(
+        settings: Self::Settings,
+        core_metrics: Arc<CoreMetrics>,
+        agent_metrics: Metrics,
+    ) -> Result<Self>
     where
         Self: Sized,
     {
-        let core = settings.build_hyperlane_core(metrics.clone());
+        let core = settings.build_hyperlane_core(core_metrics.clone());
         let db = DB::from_path(&settings.db)?;
         let dbs = settings
             .origin_chains
@@ -105,18 +115,18 @@ impl BaseAgent for Relayer {
             .collect::<HashMap<_, _>>();
 
         let mailboxes = settings
-            .build_mailboxes(settings.destination_chains.iter(), &metrics)
+            .build_mailboxes(settings.destination_chains.iter(), &core_metrics)
             .await?;
         let validator_announces = settings
-            .build_validator_announces(settings.origin_chains.iter(), &metrics)
+            .build_validator_announces(settings.origin_chains.iter(), &core_metrics)
             .await?;
 
-        let contract_sync_metrics = Arc::new(ContractSyncMetrics::new(&metrics));
+        let contract_sync_metrics = Arc::new(ContractSyncMetrics::new(&core_metrics));
 
         let message_syncs = settings
             .build_message_indexers(
                 settings.origin_chains.iter(),
-                &metrics,
+                &core_metrics,
                 &contract_sync_metrics,
                 dbs.iter()
                     .map(|(d, db)| (d.clone(), Arc::new(db.clone()) as _))
@@ -126,7 +136,7 @@ impl BaseAgent for Relayer {
         let interchain_gas_payment_syncs = settings
             .build_interchain_gas_payment_indexers(
                 settings.origin_chains.iter(),
-                &metrics,
+                &core_metrics,
                 &contract_sync_metrics,
                 dbs.iter()
                     .map(|(d, db)| (d.clone(), Arc::new(db.clone()) as _))
@@ -136,7 +146,7 @@ impl BaseAgent for Relayer {
         let merkle_tree_hook_syncs = settings
             .build_merkle_tree_hook_indexers(
                 settings.origin_chains.iter(),
-                &metrics,
+                &core_metrics,
                 &contract_sync_metrics,
                 dbs.iter()
                     .map(|(d, db)| (d.clone(), Arc::new(db.clone()) as _))
@@ -188,9 +198,10 @@ impl BaseAgent for Relayer {
             .collect();
 
         let mut msg_ctxs = HashMap::new();
+        let mut destination_chains = HashMap::new();
         for destination in &settings.destination_chains {
             let destination_chain_setup = core.settings.chain_setup(destination).unwrap().clone();
-
+            destination_chains.insert(destination.clone(), destination_chain_setup.clone());
             let transaction_gas_limit: Option<U256> =
                 if skip_transaction_gas_limit_for.contains(&destination.id()) {
                     None
@@ -221,7 +232,7 @@ impl BaseAgent for Relayer {
                         metadata_builder,
                         origin_gas_payment_enforcer: gas_payment_enforcers[origin].clone(),
                         transaction_gas_limit,
-                        metrics: MessageSubmissionMetrics::new(&metrics, origin, destination),
+                        metrics: MessageSubmissionMetrics::new(&core_metrics, origin, destination),
                     }),
                 );
             }
@@ -230,7 +241,7 @@ impl BaseAgent for Relayer {
         Ok(Self {
             dbs,
             origin_chains: settings.origin_chains,
-            destination_chains: settings.destination_chains,
+            destination_chains,
             msg_ctxs,
             core,
             message_syncs,
@@ -242,6 +253,8 @@ impl BaseAgent for Relayer {
             transaction_gas_limit,
             skip_transaction_gas_limit_for,
             allow_local_checkpoint_syncers: settings.allow_local_checkpoint_syncers,
+            core_metrics,
+            agent_metrics,
         })
     }
 
@@ -251,12 +264,35 @@ impl BaseAgent for Relayer {
 
         // send channels by destination chain
         let mut send_channels = HashMap::with_capacity(self.destination_chains.len());
-        for destination in &self.destination_chains {
+        for (dest_domain, dest_conf) in &self.destination_chains {
             let (send_channel, receive_channel) =
                 mpsc::unbounded_channel::<Box<DynPendingOperation>>();
-            send_channels.insert(destination.id(), send_channel);
+            send_channels.insert(dest_domain.id(), send_channel);
+
+            tasks.push(self.run_destination_submitter(dest_domain, receive_channel));
 
-            tasks.push(self.run_destination_submitter(destination, receive_channel));
+            let agent_metrics_conf = dest_conf
+                .agent_metrics_conf(Self::AGENT_NAME.to_string())
+                .await
+                .unwrap();
+            let agent_metrics_fetcher = dest_conf
+                .build_agent_metrics_fetcher(&self.core_metrics)
+                .await
+                .unwrap();
+            let agent_metrics = AgentMetrics::new(
+                self.agent_metrics.clone(),
+                agent_metrics_conf,
+                agent_metrics_fetcher,
+            );
+
+            let fetcher_task = tokio::spawn(async move {
+                agent_metrics
+                    .start_updating_on_interval(METRICS_SCRAPE_INTERVAL)
+                    .await;
+                Ok(())
+            })
+            .instrument(info_span!("AgentMetricsFetcher"));
+            tasks.push(fetcher_task);
         }
 
         for origin in &self.origin_chains {
@@ -330,11 +366,11 @@ impl Relayer {
         let metrics = MessageProcessorMetrics::new(
             &self.core.metrics,
             origin,
-            self.destination_chains.iter(),
+            self.destination_chains.keys(),
         );
         let destination_ctxs = self
             .destination_chains
-            .iter()
+            .keys()
             .filter(|&destination| destination != origin)
             .map(|destination| {
                 (

rust/agents/scraper/migration/bin/common.rs

@@ -1,9 +1,11 @@
-use std::env;
+use std::{env, time::Duration};
 
 use migration::sea_orm::{Database, DatabaseConnection};
 pub use migration::{DbErr, Migrator, MigratorTrait as _};
+use sea_orm::ConnectOptions;
 
 const LOCAL_DATABASE_URL: &str = "postgresql://postgres:47221c18c610@localhost:5432/postgres";
+const CONNECT_TIMEOUT: u64 = 20;
 
 pub fn url() -> String {
     env::var("DATABASE_URL").unwrap_or_else(|_| LOCAL_DATABASE_URL.into())
@@ -16,6 +18,8 @@ pub async fn init() -> Result<DatabaseConnection, DbErr> {
         .init();
 
     let url = url();
+    let mut options: ConnectOptions = url.clone().into();
+    options.connect_timeout(Duration::from_secs(CONNECT_TIMEOUT));
     println!("Connecting to {url}");
-    Database::connect(url).await
+    Database::connect(options).await
 }

rust/agents/scraper/src/agent.rs

@@ -3,8 +3,8 @@ use std::{collections::HashMap, sync::Arc};
 use async_trait::async_trait;
 use derive_more::AsRef;
 use hyperlane_base::{
-    run_all, settings::IndexSettings, BaseAgent, ContractSyncMetrics, CoreMetrics,
-    HyperlaneAgentCore,
+    metrics::Metrics as AgentMetrics, run_all, settings::IndexSettings, BaseAgent,
+    ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
 };
 use hyperlane_core::HyperlaneDomain;
 use tokio::task::JoinHandle;
@@ -38,6 +38,7 @@ impl BaseAgent for Scraper {
     async fn from_settings(
         settings: Self::Settings,
         metrics: Arc<CoreMetrics>,
+        _agent_metrics: AgentMetrics,
     ) -> eyre::Result<Self>
     where
         Self: Sized,

rust/agents/validator/src/validator.rs

@@ -10,6 +10,7 @@ use tracing::{error, info, info_span, instrument::Instrumented, warn, Instrument
 
 use hyperlane_base::{
     db::{HyperlaneRocksDB, DB},
+    metrics::Metrics as AgentMetrics,
     run_all, BaseAgent, CheckpointSyncer, ContractSyncMetrics, CoreMetrics, HyperlaneAgentCore,
     SequencedDataContractSync,
 };
@@ -51,7 +52,11 @@ impl BaseAgent for Validator {
 
     type Settings = ValidatorSettings;
 
-    async fn from_settings(settings: Self::Settings, metrics: Arc<CoreMetrics>) -> Result<Self>
+    async fn from_settings(
+        settings: Self::Settings,
+        metrics: Arc<CoreMetrics>,
+        _agent_metrics: AgentMetrics,
+    ) -> Result<Self>
     where
         Self: Sized,
     {

rust/chains/hyperlane-cosmos/src/aggregation_ism.rs

@@ -2,7 +2,7 @@ use std::str::FromStr;
 
 use crate::{
     address::CosmosAddress,
-    grpc::{WasmGrpcProvider, WasmProvider},
+    grpc::WasmProvider,
     payloads::aggregate_ism::{ModulesAndThresholdRequest, ModulesAndThresholdResponse},
     ConnectionConf, CosmosProvider, Signer,
 };
@@ -18,7 +18,7 @@ use tracing::instrument;
 pub struct CosmosAggregationIsm {
     domain: HyperlaneDomain,
     address: H256,
-    provider: Box<WasmGrpcProvider>,
+    provider: Box<CosmosProvider>,
 }
 
 impl CosmosAggregationIsm {
@@ -28,7 +28,12 @@ impl CosmosAggregationIsm {
         locator: ContractLocator,
         signer: Option<Signer>,
     ) -> ChainResult<Self> {
-        let provider = WasmGrpcProvider::new(conf.clone(), locator.clone(), signer)?;
+        let provider = CosmosProvider::new(
+            locator.domain.clone(),
+            conf.clone(),
+            Some(locator.clone()),
+            signer,
+        )?;
 
         Ok(Self {
             domain: locator.domain.clone(),
@@ -50,7 +55,7 @@ impl HyperlaneChain for CosmosAggregationIsm {
     }
 
     fn provider(&self) -> Box<dyn HyperlaneProvider> {
-        Box::new(CosmosProvider::new(self.domain.clone()))
+        self.provider.clone()
     }
 }
 
@@ -63,7 +68,7 @@ impl AggregationIsm for CosmosAggregationIsm {
     ) -> ChainResult<(Vec<H256>, u8)> {
         let payload = ModulesAndThresholdRequest::new(message);
 
-        let data = self.provider.wasm_query(payload, None).await?;
+        let data = self.provider.grpc().wasm_query(payload, None).await?;
         let response: ModulesAndThresholdResponse = serde_json::from_slice(&data)?;
 
         let modules: ChainResult<Vec<H256>> = response

rust/chains/hyperlane-cosmos/src/interchain_gas.rs

@@ -10,7 +10,6 @@ use once_cell::sync::Lazy;
 use std::ops::RangeInclusive;
 
 use crate::{
-    grpc::WasmGrpcProvider,
     rpc::{CosmosWasmIndexer, ParsedEvent, WasmIndexer},
     signers::Signer,
     utils::{CONTRACT_ADDRESS_ATTRIBUTE_KEY, CONTRACT_ADDRESS_ATTRIBUTE_KEY_BASE64},
@@ -22,6 +21,7 @@ use crate::{
 pub struct CosmosInterchainGasPaymaster {
     domain: HyperlaneDomain,
     address: H256,
+    provider: CosmosProvider,
 }
 
 impl HyperlaneContract for CosmosInterchainGasPaymaster {
@@ -36,7 +36,7 @@ impl HyperlaneChain for CosmosInterchainGasPaymaster {
     }
 
     fn provider(&self) -> Box<dyn HyperlaneProvider> {
-        Box::new(CosmosProvider::new(self.domain.clone()))
+        Box::new(self.provider.clone())
     }
 }
 
@@ -49,11 +49,17 @@ impl CosmosInterchainGasPaymaster {
         locator: ContractLocator,
         signer: Option<Signer>,
     ) -> ChainResult<Self> {
-        let provider = WasmGrpcProvider::new(conf.clone(), locator.clone(), signer)?;
+        let provider = CosmosProvider::new(
+            locator.domain.clone(),
+            conf.clone(),
+            Some(locator.clone()),
+            signer,
+        )?;
 
         Ok(Self {
             domain: locator.domain.clone(),
             address: locator.address,
+            provider,
         })
     }
 }