Rename tx_latency to tx_latency_confirmed and add tx_latency_submitted

Author: romac

relayer/src/link/pending.rs

@@ -242,7 +242,7 @@ impl<Chain: ChainHandle> PendingTxs<Chain> {
                         "transactions confirmed",
                     );
 
-                    telemetry!(end_process_batch, &self.chain_id(), pending.tracking_id());
+                    telemetry!(tx_confirmed, &self.chain_id(), pending.tracking_id());
 
                     // Convert the events to RelaySummary and return them.
                     let mut summary = RelaySummary::from_events(events);

relayer/src/link/relay_path.rs

@@ -57,6 +57,7 @@ use crate::link::pending::PendingTxs;
 use crate::link::relay_sender::{AsyncReply, SubmitReply};
 use crate::link::relay_summary::RelaySummary;
 use crate::link::{pending, relay_sender};
+use crate::telemetry;
 use crate::util::queue::Queue;
 
 const MAX_RETRIES: usize = 5;
@@ -682,8 +683,16 @@ impl<ChainA: ChainHandle, ChainB: ChainHandle> RelayPath<ChainA, ChainB> {
         let msgs = odata.assemble_msgs(self)?;
 
         match odata.target {
-            OperationalDataTarget::Source => S::submit(self.src_chain(), msgs),
-            OperationalDataTarget::Destination => S::submit(self.dst_chain(), msgs),
+            OperationalDataTarget::Source => {
+                telemetry!(tx_submitted, &self.src_chain().id(), msgs.tracking_id);
+
+                S::submit(self.src_chain(), msgs)
+            }
+            OperationalDataTarget::Destination => {
+                telemetry!(tx_submitted, &self.dst_chain().id(), msgs.tracking_id);
+
+                S::submit(self.dst_chain(), msgs)
+            }
         }
     }
 

relayer/src/supervisor.rs

@@ -608,7 +608,7 @@ fn process_batch<Chain: ChainHandle>(
 ) -> Result<(), Error> {
     assert_eq!(src_chain.id(), batch.chain_id);
 
-    telemetry!(start_process_batch, batch.tracking_id);
+    telemetry!(received_event_batch, batch.tracking_id);
 
     let collected = collect_events(config, workers, &src_chain, batch);
 

telemetry/src/state.rs

@@ -71,10 +71,15 @@ pub struct TelemetryState {
     /// The balance in each wallet that Hermes is using, per wallet, denom and chain
     wallet_balance: ValueRecorder<u64>,
 
+    /// Indicates the latency for all transactions submitted to a specific chain,
+    /// i.e. the difference between the moment when Hermes received a batch of events
+    /// until the corresponding transaction(s) were submitted. Milliseconds.
+    tx_latency_submitted: ValueRecorder<u64>,
+
     /// Indicates the latency for all transactions submitted to a specific chain,
     /// i.e. the difference between the moment when Hermes received a batch of events
     /// until the corresponding transaction(s) were confirmed. Milliseconds.
-    tx_latency: ValueRecorder<u64>,
+    tx_latency_confirmed: ValueRecorder<u64>,
 
     /// Records the time at which we started processing an event batch.
     /// Used for computing the `tx_latency` metric.
@@ -216,12 +221,27 @@ impl TelemetryState {
         self.wallet_balance.record(amount, labels);
     }
 
-    pub fn start_process_batch(&self, tracking_id: impl ToString) {
+    pub fn received_event_batch(&self, tracking_id: impl ToString) {
         self.in_flight_events
             .insert(tracking_id.to_string(), Instant::now());
     }
 
-    pub fn end_process_batch(&self, chain_id: &ChainId, tracking_id: impl ToString) {
+    pub fn tx_submitted(&self, chain_id: &ChainId, tracking_id: impl ToString) {
+        let tracking_id = tracking_id.to_string();
+
+        if let Some(start) = self.in_flight_events.get(&tracking_id) {
+            let latency = start.elapsed().as_millis() as u64;
+
+            let labels = &[
+                KeyValue::new("chain", chain_id.to_string()),
+                KeyValue::new("tracking_id", tracking_id),
+            ];
+
+            self.tx_latency_submitted.record(latency, labels);
+        }
+    }
+
+    pub fn tx_confirmed(&self, chain_id: &ChainId, tracking_id: impl ToString) {
         let tracking_id = tracking_id.to_string();
 
         if let Some(start) = self.in_flight_events.get(&tracking_id) {
@@ -232,7 +252,7 @@ impl TelemetryState {
                 KeyValue::new("tracking_id", tracking_id),
             ];
 
-            self.tx_latency.record(latency, labels);
+            self.tx_latency_confirmed.record(latency, labels);
         }
     }
 }
@@ -307,8 +327,15 @@ impl Default for TelemetryState {
                 .with_description("The balance in each wallet that Hermes is using, per wallet, denom and chain")
                 .init(),
 
-            tx_latency: meter
-                .u64_value_recorder("tx_latency")
+            tx_latency_submitted: meter
+                .u64_value_recorder("tx_latency_submitted")
+                .with_description("The latency for all transactions submitted to a specific chain, \
+                    i.e. the difference between the moment when Hermes received a batch of events \
+                    and when it submitted the corresponding transaction(s). Milliseconds.")
+                .init(),
+
+            tx_latency_confirmed: meter
+                .u64_value_recorder("tx_latency_confirmed")
                 .with_description("The latency for all transactions submitted to a specific chain, \
                     i.e. the difference between the moment when Hermes received a batch of events \
                     until the corresponding transaction(s) were confirmed. Milliseconds.")