Add telemetry and Prometheus endpoint

Cargo.toml

@@ -4,7 +4,8 @@ members = [
     "modules",
     "relayer",
     "relayer-cli",
-    "proto"
+    "telemetry",
+    "proto",
 ]
 
 exclude = [

config.toml

@@ -5,6 +5,10 @@
 strategy = 'packets'
 log_level = 'info'
 
+[telemetry]
+enabled = true
+port = 3001
+
 [[chains]]
 id = 'ibc-0'
 rpc_addr = 'http://127.0.0.1:26657'

modules/src/events.rs

@@ -66,11 +66,11 @@ pub enum IbcEvent {
 }
 
 /// For use in debug messages
-pub struct VecIbcEvents(pub Vec<IbcEvent>);
-impl fmt::Display for VecIbcEvents {
+pub struct PrettyEvents<'a>(pub &'a [IbcEvent]);
+impl<'a> fmt::Display for PrettyEvents<'a> {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         writeln!(f, "events:")?;
-        for v in &self.0 {
+        for v in self.0 {
             writeln!(f, "\t{}", v)?;
         }
         Ok(())

relayer-cli/Cargo.toml

@@ -17,12 +17,15 @@ description = """
 name = "hermes"
 
 [features]
+default   = ["telemetry"]
 profiling = ["ibc-relayer/profiling"]
+telemetry = ["ibc-relayer/telemetry", "ibc-telemetry"]
 
 [dependencies]
-ibc         = { version = "0.3.2", path = "../modules" }
-ibc-relayer = { version = "0.3.2", path = "../relayer" }
-ibc-proto   = { version = "0.8.0", path = "../proto" }
+ibc           = { version = "0.3.2", path = "../modules" }
+ibc-relayer   = { version = "0.3.2", path = "../relayer" }
+ibc-proto     = { version = "0.8.0", path = "../proto" }
+ibc-telemetry = { version = "0.1.0", path = "../telemetry", optional = true }
 
 anomaly = "0.2.0"
 gumdrop = { version = "0.7", features = ["default_expr"] }

relayer-cli/src/commands/start.rs

@@ -1,5 +1,6 @@
 use abscissa_core::{Command, Options, Runnable};
 
+use ibc_relayer::config::Config;
 use ibc_relayer::supervisor::Supervisor;
 
 use crate::conclude::Output;
@@ -11,10 +12,35 @@ pub struct StartCmd {}
 impl Runnable for StartCmd {
     fn run(&self) {
         let config = app_config();
-        let supervisor = Supervisor::spawn(config.clone()).expect("failed to spawn supervisor");
+
+        let supervisor = spawn_supervisor(config.clone());
         match supervisor.run() {
             Ok(()) => Output::success_msg("done").exit(),
             Err(e) => Output::error(e).exit(),
         }
     }
 }
+
+#[cfg(feature = "telemetry")]
+fn spawn_supervisor(config: Config) -> Supervisor {
+    let state = ibc_telemetry::new_state();
+
+    if config.telemetry.enabled {
+        ibc_telemetry::spawn(config.telemetry.port, state.clone());
+    }
+
+    Supervisor::spawn(config, state)
+}
+
+#[cfg(not(feature = "telemetry"))]
+fn spawn_supervisor(config: Config) -> Supervisor {
+    if config.telemetry.enabled {
+        warn!(
+            "telemetry enabled in the config but Hermes was built without telemetry support, \
+             build Hermes with --features=telemetry to enable telemetry support."
+        );
+    }
+
+    let telemetry = ibc_relayer::telemetry::TelemetryDisabled;
+    Supervisor::spawn(config, telemetry)
+}

relayer/Cargo.toml

@@ -14,10 +14,12 @@ description = """
 
 [features]
 profiling = []
+telemetry = ["ibc-telemetry"]
 
 [dependencies]
-ibc       = { version = "0.3.2", path = "../modules" }
-ibc-proto = { version = "0.8.0", path = "../proto" }
+ibc           = { version = "0.3.2", path = "../modules" }
+ibc-proto     = { version = "0.8.0", path = "../proto" }
+ibc-telemetry = { version = "0.1.0", path = "../telemetry", optional = true }
 
 subtle-encoding = "0.5"
 anomaly = "0.2.0"

relayer/src/chain/handle.rs

@@ -1,39 +1,45 @@
-use std::fmt::Debug;
-use std::sync::Arc;
+use std::{
+    fmt::{self, Debug},
+    sync::Arc,
+};
 
 use crossbeam_channel as channel;
 use dyn_clone::DynClone;
 use serde::{Serialize, Serializer};
 
-use ibc::ics02_client::client_consensus::{AnyConsensusState, AnyConsensusStateWithHeight};
-use ibc::ics02_client::client_state::{AnyClientState, IdentifiedAnyClientState};
-use ibc::ics02_client::events::UpdateClient;
-use ibc::ics02_client::misbehaviour::AnyMisbehaviour;
-use ibc::ics04_channel::channel::IdentifiedChannelEnd;
-use ibc::query::QueryTxRequest;
 use ibc::{
     events::IbcEvent,
-    ics02_client::header::AnyHeader,
+    ics02_client::{
+        client_consensus::{AnyConsensusState, AnyConsensusStateWithHeight},
+        client_state::{AnyClientState, IdentifiedAnyClientState},
+        events::UpdateClient,
+        header::AnyHeader,
+        misbehaviour::AnyMisbehaviour,
+    },
     ics03_connection::{connection::ConnectionEnd, version::Version},
     ics04_channel::{
-        channel::ChannelEnd,
+        channel::{ChannelEnd, IdentifiedChannelEnd},
         packet::{PacketMsgType, Sequence},
     },
     ics23_commitment::commitment::CommitmentPrefix,
     ics24_host::identifier::{ChainId, ChannelId, ClientId, ConnectionId, PortId},
     proofs::Proofs,
+    query::QueryTxRequest,
     signer::Signer,
     Height,
 };
-use ibc_proto::ibc::core::channel::v1::{
-    PacketState, QueryChannelClientStateRequest, QueryChannelsRequest,
-    QueryConnectionChannelsRequest, QueryNextSequenceReceiveRequest,
-    QueryPacketAcknowledgementsRequest, QueryPacketCommitmentsRequest, QueryUnreceivedAcksRequest,
-    QueryUnreceivedPacketsRequest,
+
+use ibc_proto::ibc::core::{
+    channel::v1::{
+        PacketState, QueryChannelClientStateRequest, QueryChannelsRequest,
+        QueryConnectionChannelsRequest, QueryNextSequenceReceiveRequest,
+        QueryPacketAcknowledgementsRequest, QueryPacketCommitmentsRequest,
+        QueryUnreceivedAcksRequest, QueryUnreceivedPacketsRequest,
+    },
+    client::v1::{QueryClientStatesRequest, QueryConsensusStatesRequest},
+    commitment::v1::MerkleProof,
+    connection::v1::QueryClientConnectionsRequest,
 };
-use ibc_proto::ibc::core::client::v1::{QueryClientStatesRequest, QueryConsensusStatesRequest};
-use ibc_proto::ibc::core::commitment::v1::MerkleProof;
-use ibc_proto::ibc::core::connection::v1::QueryClientConnectionsRequest;
 
 pub use prod::ProdChainHandle;
 
@@ -63,6 +69,15 @@ impl ChainHandlePair {
     }
 }
 
+impl Debug for ChainHandlePair {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ChainHandlePair")
+            .field("a", &self.a.id())
+            .field("b", &self.b.id())
+            .finish()
+    }
+}
+
 pub type Subscription = channel::Receiver<Arc<MonitorResult<EventBatch>>>;
 
 pub type ReplyTo<T> = channel::Sender<Result<T, Error>>;

relayer/src/config.rs

@@ -39,6 +39,8 @@ pub mod default {
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]
 pub struct Config {
     pub global: GlobalConfig,
+    #[serde(default)]
+    pub telemetry: TelemetryConfig,
     #[serde(default = "Vec::new", skip_serializing_if = "Vec::is_empty")]
     pub chains: Vec<ChainConfig>,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -102,6 +104,21 @@ impl Default for GlobalConfig {
     }
 }
 
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct TelemetryConfig {
+    pub enabled: bool,
+    pub port: u16,
+}
+
+impl Default for TelemetryConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            port: 3001,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct ChainConfig {
     pub id: ChainId,

relayer/src/lib.rs

@@ -30,6 +30,7 @@ pub mod macros;
 pub mod object;
 pub mod registry;
 pub mod supervisor;
+pub mod telemetry;
 pub mod transfer;
 pub mod upgrade_chain;
 pub mod util;