Initial telemetry support implementation (#868)

informalsystems · May 13, 2021 · 26f2728 · 26f2728
1 parent 8910d88
commit 26f2728
Show file tree

Hide file tree

Showing 8 changed files with 193 additions and 0 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/relayer/Cargo.toml b/relayer/Cargo.toml
@@ -72,3 +72,19 @@ ibc = { version = "0.3.0", path = "../modules", features = ["mocks"] }
 
 # Needed for generating (synthetic) light blocks.
 tendermint-testgen = { version = "=0.19.0" }
+
+# Dependencies needed for telemetry support
+[dependencies.opentelemetry]
+version = "0.14.0"
+
+[dependencies.opentelemetry-prometheus]
+version = "0.7.0"
+
+[dependencies.hyper]
+version = "0.14.7"
+
+[dependencies.lazy_static]
+version = "1.4.0"
+
+[dependencies.prometheus]
+version = "0.12.0"
diff --git a/relayer/src/config.rs b/relayer/src/config.rs
@@ -87,13 +87,19 @@ pub struct GlobalConfig {
     /// All valid log levels, as defined in tracing:
     /// https://docs.rs/tracing-core/0.1.17/tracing_core/struct.Level.html
     pub log_level: String,
+
+    pub telemetry_enabled: bool,
+
+    pub telemetry_port: u16,
 }
 
 impl Default for GlobalConfig {
     fn default() -> Self {
         Self {
             strategy: Strategy::default(),
             log_level: "info".to_string(),
+            telemetry_enabled: true,
+            telemetry_port: 3000,
         }
     }
 }

diff --git a/relayer/src/lib.rs b/relayer/src/lib.rs
@@ -13,6 +13,9 @@
 //!
 //! [Hermes]: https://docs.rs/ibc-relayer-cli/0.2.0/
 
+#[macro_use]
+extern crate lazy_static;
+
 pub mod chain;
 pub mod channel;
 pub mod config;
@@ -28,6 +31,7 @@ pub mod object;
 pub mod registry;
 pub mod relay;
 pub mod supervisor;
+pub mod telemetry;
 pub mod transfer;
 pub mod upgrade_chain;
 pub mod util;

diff --git a/relayer/src/supervisor.rs b/relayer/src/supervisor.rs
@@ -29,6 +29,7 @@ use crate::{
 };
 
 mod error;
+use crate::telemetry::service::TelemetryService;
 pub use error::Error;
 
 /// The supervisor listens for events on multiple pairs of chains,
@@ -38,17 +39,33 @@ pub struct Supervisor {
     config: Config,
     registry: Registry,
     workers: HashMap<Object, WorkerHandle>,
+    telemetry: Option<TelemetryService>,
 }
 
 impl Supervisor {
     /// Spawns a [`Supervisor`] which will listen for events on all the chains in the [`Config`].
     pub fn spawn(config: Config) -> Result<Self, BoxError> {
         let registry = Registry::new(config.clone());
 
+        // Start the telemetry service
+        let telemetry = match config.global.telemetry_enabled {
+            true => {
+                println!(
+                    "TELEMETRY ENABLED ON PORT: {:?}",
+                    config.global.telemetry_port
+                );
+                Some(TelemetryService {
+                    listen_port: config.global.telemetry_port,
+                })
+            }
+            false => None,
+        };
+
         Ok(Self {
             config,
             registry,
             workers: HashMap::new(),
+            telemetry,
         })
     }
 

diff --git a/relayer/src/telemetry.rs b/relayer/src/telemetry.rs
@@ -0,0 +1,2 @@
+pub mod relayer_state;
+pub mod service;
diff --git a/relayer/src/telemetry/relayer_state.rs b/relayer/src/telemetry/relayer_state.rs
@@ -0,0 +1,7 @@
+use opentelemetry::metrics::BoundCounter;
+use opentelemetry_prometheus::PrometheusExporter;
+
+pub struct RelayerState {
+    pub exporter: PrometheusExporter,
+    pub tx_counter: BoundCounter<'static, u64>,
+}
diff --git a/relayer/src/telemetry/service.rs b/relayer/src/telemetry/service.rs
@@ -0,0 +1,75 @@
+use hyper::{
+    header::CONTENT_TYPE,
+    service::{make_service_fn, service_fn},
+    Body, Request, Response, Server,
+};
+
+use crate::telemetry::relayer_state::RelayerState;
+use opentelemetry::{global, KeyValue};
+use prometheus::{Encoder, TextEncoder};
+use std::convert::Infallible;
+use std::sync::Arc;
+
+lazy_static! {
+    static ref HANDLER_ALL: [KeyValue; 1] = [KeyValue::new("handler", "all")];
+}
+
+pub struct TelemetryService {
+    pub(crate) listen_port: u16,
+}
+
+async fn serve_req(
+    _req: Request<Body>,
+    state: Arc<RelayerState>,
+) -> Result<Response<Body>, hyper::Error> {
+    let mut buffer = vec![];
+    let encoder = TextEncoder::new();
+    let metric_families = state.exporter.registry().gather();
+    encoder.encode(&metric_families, &mut buffer).unwrap();
+
+    state.tx_counter.add(1);
+
+    let response = Response::builder()
+        .status(200)
+        .header(CONTENT_TYPE, encoder.format_type())
+        .body(Body::from(buffer))
+        .unwrap();
+
+    Ok(response)
+}
+
+impl TelemetryService {
+    pub async fn run(self) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+        let exporter = opentelemetry_prometheus::exporter().init();
+
+        let meter = global::meter("hermes/relayer");
+        let state = Arc::new(RelayerState {
+            exporter,
+            tx_counter: meter
+                .u64_counter("hermes.tx_count")
+                .with_description("Total number of transactions processed via the relayer.")
+                .init()
+                .bind(HANDLER_ALL.as_ref()),
+        });
+
+        // For every connection, we must make a `Service` to handle all
+        // incoming HTTP requests on said connection.
+        let make_svc = make_service_fn(move |_conn| {
+            let state = state.clone();
+            // This is the `Service` that will handle the connection.
+            // `service_fn` is a helper to convert a function that
+            // returns a Response into a `Service`.
+            async move { Ok::<_, Infallible>(service_fn(move |req| serve_req(req, state.clone()))) }
+        });
+
+        let addr = ([127, 0, 0, 1], self.listen_port).into();
+
+        let server = Server::bind(&addr).serve(make_svc);
+
+        println!("Telemetry service listening on http://{}", addr);
+
+        server.await?;
+
+        Ok(self)
+    }
+}