feat: Measure block maker to induction / execution latency (#2498)

Add a couple of histograms measuring ingress message latency from block
making (the earliest reliable timestamp we have) to induction; and to
execution.

These measurements don't cover the latency from agent to block maker,
which could vary considerably with a backlogged ingress pool or with
dropped transport connections.

Author: alin-at-dfinity

rs/execution_environment/src/history.rs

@@ -3,6 +3,7 @@ use ic_error_types::{ErrorCode, RejectCode};
 use ic_interfaces::execution_environment::{
     IngressHistoryError, IngressHistoryReader, IngressHistoryWriter,
 };
+use ic_interfaces::time_source::system_time_now;
 use ic_interfaces_state_manager::{StateManagerError, StateReader};
 use ic_logger::{fatal, ReplicaLogger};
 use ic_metrics::{buckets::decimal_buckets, MetricsRegistry};
@@ -82,6 +83,9 @@ pub struct IngressHistoryWriterImpl {
     // Wrapped in a RwLock for interior mutability, otherwise &self in methods
     // has to be &mut self.
     received_time: RwLock<HashMap<MessageId, TransitionStartTime>>,
+    message_state_transition_received_duration_seconds: Histogram,
+    message_state_transition_processing_duration_seconds: Histogram,
+    message_state_transition_received_to_processing_duration_seconds: Histogram,
     message_state_transition_completed_ic_duration_seconds: Histogram,
     message_state_transition_completed_wall_clock_duration_seconds: Histogram,
     message_state_transition_failed_ic_duration_seconds: HistogramVec,
@@ -102,6 +106,24 @@ impl IngressHistoryWriterImpl {
             config,
             log,
             received_time: RwLock::new(HashMap::new()),
+            message_state_transition_received_duration_seconds: metrics_registry.histogram(
+                "message_state_transition_received_duration_seconds",
+                "Time taken by messages to get from block maker into ingress queue",
+                // 10ms, 20ms, 50ms, ..., 100s, 200s, 500s
+                decimal_buckets(-2,2),
+            ),
+            message_state_transition_processing_duration_seconds: metrics_registry.histogram(
+                "message_state_transition_processing_duration_seconds",
+                "Time taken by messages to get from block maker to execution",
+                // 10ms, 20ms, 50ms, ..., 100s, 200s, 500s
+                decimal_buckets(-2,2),
+            ),
+            message_state_transition_received_to_processing_duration_seconds: metrics_registry.histogram(
+                "message_state_transition_received_to_processing_duration_seconds",
+                "Time spent by messages between induction and execution",
+                // 10ms, 20ms, 50ms, ..., 100s, 200s, 500s
+                decimal_buckets(-2,2),
+            ),
             message_state_transition_completed_ic_duration_seconds: metrics_registry.histogram(
                 "message_state_transition_completed_ic_duration_seconds",
                 "The IC time taken for a message to transition from the Received state to Completed state",
@@ -164,8 +186,60 @@ impl IngressHistoryWriter for IngressHistoryWriterImpl {
                 status
             );
         }
+
         use IngressState::*;
         use IngressStatus::*;
+
+        // Computes the duration elapsed since the time when a message was included into
+        // a block (approximated by block time); and since its induction.
+        let message_latencies_seconds = |message_id| {
+            let received_time = self.received_time.read().unwrap();
+            received_time.get(message_id).map(|timer| {
+                (
+                    system_time_now()
+                        .saturating_duration_since(timer.ic_time)
+                        .as_secs_f64(),
+                    Instant::now()
+                        .saturating_duration_since(timer.system_time)
+                        .as_secs_f64(),
+                )
+            })
+        };
+        // Latency instrumentation.
+        match (&current_status, &status) {
+            // Newly received message: observe its induction latency.
+            (
+                Unknown,
+                Known {
+                    state: Received, ..
+                },
+            ) => {
+                if let Some((seconds_since_block_made, _)) = message_latencies_seconds(&message_id)
+                {
+                    self.message_state_transition_received_duration_seconds
+                        .observe(seconds_since_block_made);
+                }
+            }
+
+            // Message popped from ingress queue: observe its processing latency.
+            (
+                Known {
+                    state: Received, ..
+                },
+                Known { state, .. },
+            ) if state != &Received => {
+                if let Some((seconds_since_block_made, seconds_since_induction)) =
+                    message_latencies_seconds(&message_id)
+                {
+                    self.message_state_transition_processing_duration_seconds
+                        .observe(seconds_since_block_made);
+                    self.message_state_transition_received_to_processing_duration_seconds
+                        .observe(seconds_since_induction);
+                }
+            }
+            _ => {}
+        }
+
         match &status {
             Known {
                 state: Received, ..

rs/interfaces/src/time_source.rs

@@ -58,7 +58,7 @@ impl TimeSource for SysTimeSource {
 /// Return the current system time. Note that the value returned is not
 /// guaranteed to be monotonic.
 #[inline]
-fn system_time_now() -> Time {
+pub fn system_time_now() -> Time {
     UNIX_EPOCH
         + SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)