[telemetry] Improve invoker observability

- Add time taken to receive the first response (headers) from the user service
- Add Time taken to replay the journal

Author: muhamadazmy

crates/invoker-impl/src/invocation_task/service_protocol_runner.rs

@@ -15,6 +15,7 @@ use crate::invocation_task::{
     ResponseChunk, ResponseStreamState, TerminalLoopState, X_RESTATE_SERVER,
     invocation_id_to_header_value, service_protocol_version_to_header_value,
 };
+use crate::metric_definitions::{INVOKER_JOURNAL_REPLAY_TIME, INVOKER_SERVICE_RESPONSE_TIME};
 use bytes::Bytes;
 use futures::future::FusedFuture;
 use futures::{FutureExt, Stream, StreamExt};
@@ -44,7 +45,7 @@ use restate_types::schema::deployment::{
 use restate_types::service_protocol::ServiceProtocolVersion;
 use std::collections::HashSet;
 use std::future::poll_fn;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, trace, warn};
@@ -70,6 +71,7 @@ pub struct ServiceProtocolRunner<'a, IR, EE, DMR> {
 
     // task state
     next_journal_index: EntryIndex,
+    http_stream_start_time: Instant,
 }
 
 impl<'a, IR, EE, DMR> ServiceProtocolRunner<'a, IR, EE, DMR>
@@ -93,6 +95,8 @@ where
             encoder,
             decoder,
             next_journal_index: 0,
+            // initial value will be overridden to the actual start time of the http stream
+            http_stream_start_time: Instant::now(),
         }
     }
 
@@ -161,6 +165,8 @@ where
             .await
         );
 
+        // update the start time for the service response time metric
+        self.http_stream_start_time = Instant::now();
         // Initialize the response stream state
         let mut http_stream_rx =
             ResponseStreamState::initialize(&self.invocation_task.client, request);
@@ -171,6 +177,9 @@ where
                 .await
         );
 
+        metrics::histogram!(INVOKER_JOURNAL_REPLAY_TIME)
+            .record(self.http_stream_start_time.elapsed());
+
         // Check all the entries have been replayed
         debug_assert_eq!(self.next_journal_index, journal_size);
 
@@ -460,6 +469,9 @@ where
         &mut self,
         mut parts: http::response::Parts,
     ) -> Result<(), InvokerError> {
+        metrics::histogram!(INVOKER_SERVICE_RESPONSE_TIME)
+            .record(self.http_stream_start_time.elapsed());
+
         // if service is running behind a gateway, the service can be down
         // but we still get a response code from the gateway itself. In that
         // case we still need to return the proper error

crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs

@@ -11,7 +11,7 @@
 use std::collections::HashSet;
 use std::future::poll_fn;
 use std::ops::Deref;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use bytes::Bytes;
 use bytestring::ByteString;
@@ -63,6 +63,7 @@ use crate::invocation_task::{
     ResponseChunk, ResponseStreamState, TerminalLoopState, X_RESTATE_SERVER,
     invocation_id_to_header_value, service_protocol_version_to_header_value,
 };
+use crate::metric_definitions::{INVOKER_JOURNAL_REPLAY_TIME, INVOKER_SERVICE_RESPONSE_TIME};
 
 ///  Provides the value of the invocation id
 const INVOCATION_ID_HEADER_NAME: HeaderName = HeaderName::from_static("x-restate-invocation-id");
@@ -85,6 +86,7 @@ pub struct ServiceProtocolRunner<'a, IR, EE, Schemas> {
 
     // task state
     command_index: CommandIndex,
+    http_stream_start_time: Instant,
 }
 
 impl<'a, IR, EE, Schemas> ServiceProtocolRunner<'a, IR, EE, Schemas>
@@ -108,6 +110,8 @@ where
             encoder,
             decoder,
             command_index: 0,
+            // initial value will be overridden to the actual start time of the http stream
+            http_stream_start_time: Instant::now(),
         }
     }
 
@@ -176,6 +180,9 @@ where
             .await
         );
 
+        // update the start time for the service response time metric
+        self.http_stream_start_time = Instant::now();
+
         // Initialize the response stream state
         let mut http_stream_rx =
             ResponseStreamState::initialize(&self.invocation_task.client, request);
@@ -186,6 +193,9 @@ where
                 .await
         );
 
+        metrics::histogram!(INVOKER_JOURNAL_REPLAY_TIME)
+            .record(self.http_stream_start_time.elapsed());
+
         // If we have the invoker_rx and the protocol type is bidi stream,
         // then we can use the bidi_stream loop reading the invoker_rx and the http_stream_rx
         if protocol_type == ProtocolType::BidiStream {
@@ -308,6 +318,7 @@ where
         let got_headers_future = poll_fn(|cx| http_stream_rx.poll_only_headers(cx)).fuse();
         tokio::pin!(got_headers_future);
 
+        // TODO: add metrics for delays for when headers are received (first response from service)
         loop {
             tokio::select! {
                 got_headers_res = got_headers_future.as_mut(), if !got_headers_future.is_terminated() => {
@@ -518,6 +529,9 @@ where
         &mut self,
         mut parts: http::response::Parts,
     ) -> Result<(), InvokerError> {
+        metrics::histogram!(INVOKER_SERVICE_RESPONSE_TIME)
+            .record(self.http_stream_start_time.elapsed());
+
         // if service is running behind a gateway, the service can be down
         // but we still get a response code from the gateway itself. In that
         // case we still need to return the proper error

crates/invoker-impl/src/metric_definitions.rs

@@ -18,7 +18,9 @@ pub const INVOKER_INVOCATION_TASKS: &str = "restate.invoker.invocation_tasks.tot
 pub const INVOKER_AVAILABLE_SLOTS: &str = "restate.invoker.available_slots";
 pub const INVOKER_CONCURRENCY_LIMIT: &str = "restate.invoker.concurrency_limit";
 pub const INVOKER_TASK_DURATION: &str = "restate.invoker.task_duration.seconds";
+pub const INVOKER_SERVICE_RESPONSE_TIME: &str = "restate.invoker.service_response_time.seconds";
 pub const INVOKER_TASKS_IN_FLIGHT: &str = "restate.invoker.inflight_tasks";
+pub const INVOKER_JOURNAL_REPLAY_TIME: &str = "restate.invoker.journal_replay_time.seconds";
 
 pub const TASK_OP_STARTED: &str = "started";
 pub const TASK_OP_SUSPENDED: &str = "suspended";
@@ -62,6 +64,18 @@ pub(crate) fn describe_metrics() {
         "Time taken to complete an invoker task"
     );
 
+    describe_histogram!(
+        INVOKER_SERVICE_RESPONSE_TIME,
+        Unit::Seconds,
+        "Time taken to receive the first response (headers) from the user service"
+    );
+
+    describe_histogram!(
+        INVOKER_JOURNAL_REPLAY_TIME,
+        Unit::Seconds,
+        "Time taken to replay the journal"
+    );
+
     describe_gauge!(
         INVOKER_TASKS_IN_FLIGHT,
         Unit::Count,