[telemetry] Detect service rechability issues

Summary:
Introducing a counter for number of "rechability" issues for a service
that can detect a service is down or un-responsive

by visualizing `rate(restate.invoker.service_unreachable_errors.total)` coupled with alerts, operator
can know when a service is facing connectivity problems

Author: muhamadazmy

crates/invoker-impl/src/lib.rs

@@ -54,14 +54,14 @@ use tracing::{error, instrument};
 
 use crate::error::SdkInvocationErrorV2;
 use crate::metric_definitions::{
-    INVOKER_ENQUEUE, INVOKER_INVOCATION_TASKS, TASK_OP_COMPLETED, TASK_OP_FAILED, TASK_OP_STARTED,
-    TASK_OP_SUSPENDED,
+    INVOKER_ENQUEUE, INVOKER_INVOCATION_TASKS, INVOKER_SERVICE_UNREACHABLE_ERRORS,
+    TASK_OP_COMPLETED, TASK_OP_FAILED, TASK_OP_STARTED, TASK_OP_SUSPENDED,
 };
 use error::InvokerError;
 pub use input_command::ChannelStatusReader;
 pub use input_command::InvokerHandle;
 use restate_invoker_api::invocation_reader::InvocationReader;
-use restate_service_client::{AssumeRoleCacheMode, ServiceClient};
+use restate_service_client::{AssumeRoleCacheMode, HttpError, ServiceClient, ServiceClientError};
 use restate_types::deployment::PinnedDeployment;
 use restate_types::invocation::{InvocationEpoch, InvocationTarget};
 use restate_types::journal_v2;
@@ -1090,6 +1090,11 @@ where
             .remove_invocation_with_epoch(partition, &invocation_id, invocation_epoch)
         {
             debug_assert_eq!(invocation_epoch, ism.invocation_epoch);
+
+            if self.is_service_down_error(&error) {
+                counter!(INVOKER_SERVICE_UNREACHABLE_ERRORS, "service" => ism.invocation_target.service_name().to_string()).increment(1);
+            }
+
             self.handle_error_event(options, partition, invocation_id, error, ism)
                 .await;
         } else {
@@ -1098,6 +1103,22 @@ where
         }
     }
 
+    fn is_service_down_error(&self, error: &InvokerError) -> bool {
+        if let InvokerError::Client(client_err) = error {
+            match client_err.as_ref() {
+                ServiceClientError::Http(_, HttpError::Connect(_)) => true,
+                ServiceClientError::Lambda(_, error) => {
+                    // service down errors are those which might indicate that the service is down or
+                    // unreachable.
+                    error.is_service_down()
+                }
+                _ => false,
+            }
+        } else {
+            false
+        }
+    }
+
     #[instrument(
         level = "trace",
         skip_all,

crates/invoker-impl/src/metric_definitions.rs

@@ -21,6 +21,8 @@ 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 INVOKER_SERVICE_UNREACHABLE_ERRORS: &str =
+    "restate.invoker.service_unreachable_errors.total";
 
 pub const TASK_OP_STARTED: &str = "started";
 pub const TASK_OP_SUSPENDED: &str = "suspended";
@@ -81,4 +83,10 @@ pub(crate) fn describe_metrics() {
         Unit::Count,
         "Number of inflight invoker tasks"
     );
+
+    describe_counter!(
+        INVOKER_SERVICE_UNREACHABLE_ERRORS,
+        Unit::Count,
+        "Number of service down errors"
+    );
 }

crates/service-client/src/lambda.rs

@@ -277,6 +277,18 @@ pub enum LambdaError {
 }
 
 impl LambdaError {
+    /// Service down errors are those which might indicate that the service is down or
+    /// unreachable.
+    pub fn is_service_down(&self) -> bool {
+        match self {
+            LambdaError::SdkError(err) => matches!(
+                err.as_ref(),
+                SdkError::DispatchFailure(_) | SdkError::TimeoutError(_)
+            ),
+            _ => false,
+        }
+    }
+
     /// Retryable errors are those which can be caused by transient faults and where
     /// retrying can succeed.
     pub fn is_retryable(&self) -> bool {