Initial logic to include the telemetry in the Supervisor (#868)

informalsystems · May 19, 2021 · 6954493 · 6954493
1 parent 415dead
commit 6954493
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 20 deletions.
diff --git a/relayer/src/supervisor.rs b/relayer/src/supervisor.rs
@@ -25,12 +25,13 @@ use crate::{
     },
     object::{Client, Object, UnidirectionalChannelPath},
     registry::Registry,
+    telemetry::service::TelemetryService,
     util::try_recv_multiple,
     worker::{WorkerMap, WorkerMsg},
 };
 
 mod error;
-use crate::telemetry::service::TelemetryService;
+use crate::telemetry::state::TelemetryState;
 pub use error::Error;
 
 /// The supervisor listens for events on multiple pairs of chains,
@@ -41,6 +42,7 @@ pub struct Supervisor {
     registry: Registry,
     workers: WorkerMap,
     worker_msg_rx: Receiver<WorkerMsg>,
+    telemetry_state: Arc<TelemetryState>,
 }
 
 impl Supervisor {
@@ -50,24 +52,18 @@ impl Supervisor {
         let (worker_msg_tx, worker_msg_rx) = crossbeam_channel::unbounded();
 
         // 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,
+        let telemetry_state = TelemetryState::init();
+        match config.global.telemetry_enabled {
+            true => TelemetryService::run(telemetry_state.clone(), config.global.telemetry_port),
+            false => println!("Telemetry not enabled"),
         };
 
         Ok(Self {
             config,
             registry,
             workers: WorkerMap::new(worker_msg_tx),
             worker_msg_rx,
+            telemetry_state,
         })
     }
 
@@ -112,6 +108,8 @@ impl Supervisor {
                 IbcEvent::SendPacket(ref packet) => {
                     if let Ok(object) = Object::for_send_packet(packet, src_chain) {
                         collected.per_object.entry(object).or_default().push(event);
+                        // Increase counter
+                        self.telemetry_state.tx_counter.add(1);
                     }
                 }
                 IbcEvent::TimeoutPacket(ref packet) => {

diff --git a/relayer/src/telemetry/service.rs b/relayer/src/telemetry/service.rs
@@ -1,16 +1,13 @@
 use prometheus::{Encoder, TextEncoder};
 
 use crate::telemetry::state::TelemetryState;
+use std::sync::Arc;
 
-pub struct TelemetryService {
-    pub(crate) listen_port: u16,
-}
+pub struct TelemetryService {}
 
 impl TelemetryService {
-    pub async fn run(self) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
-        let state = TelemetryState::init();
-
-        rouille::start_server(format!("localhost:{}", self.listen_port), move |request| {
+    pub fn run(state: Arc<TelemetryState>, listen_port: u16) -> () {
+        rouille::start_server(format!("localhost:{}", listen_port), move |request| {
             router!(request,
                 // The prometheus endpoint
                 (GET) (/metrics) => {
@@ -31,6 +28,6 @@ impl TelemetryService {
                     rouille::Response::empty_404()
                 }
             )
-        })
+        });
     }
 }