Opentelemetry - set status, moved counter, retry_attempts counter

.github/workflows/rust.yml

@@ -111,7 +111,7 @@ jobs:
 
             - name: Run glide-core telemetry tests
               working-directory: ./glide-core/telemetry
-              run: cargo test --all-features
+              run: cargo test --all-features -- --test-threads=1
 
             - name: Run glide-ffi tests
               working-directory: ./ffi

glide-core/redis-rs/redis/Cargo.toml

@@ -99,6 +99,8 @@ uuid = { version = "1.6.1", optional = true }
 
 telemetrylib = { path = "../../telemetry" }
 
+logger_core = { path = "../../../logger_core" } 
+
 lazy_static = "1"
 
 [features]
@@ -129,7 +131,7 @@ aio = [
     "dispose",
 ]
 geospatial = []
-json = ["serde", "serde/derive", "serde_json"]
+json = ["serde", "serde/derive"]
 cluster = ["crc16", "rand"]
 script = ["sha1_smol"]
 tls-native-tls = ["native-tls"]
@@ -178,6 +180,7 @@ tokio = { version = "1", features = [
 tempfile = "=3.6.0"
 once_cell = "1"
 anyhow = "1"
+serde_json = "1.0.82"
 sscanf = "0.4.1"
 serial_test = "^2"
 versions = "6.3"

glide-core/redis-rs/redis/src/cluster_async/mod.rs

@@ -48,6 +48,9 @@ use pipeline_routing::{
     collect_and_send_pending_requests, map_pipeline_to_nodes, process_and_retry_pipeline_responses,
     route_for_pipeline, PipelineResponses, ResponsePoliciesMap,
 };
+
+use logger_core::log_error;
+
 use std::{
     collections::{HashMap, HashSet},
     fmt, io, mem,
@@ -66,7 +69,7 @@ use tokio::task::JoinHandle;
 
 #[cfg(feature = "tokio-comp")]
 use crate::aio::DisconnectNotifier;
-use telemetrylib::Telemetry;
+use telemetrylib::{GlideOpenTelemetry, Telemetry};
 
 use crate::{
     aio::{get_socket_addrs, ConnectionLike, MultiplexedConnection, Runtime},
@@ -2826,6 +2829,13 @@ where
             match result {
                 Next::Done => {}
                 Next::Retry { request } => {
+                    // Record retries error metric if telemetry is initialized
+                    if let Err(e) = GlideOpenTelemetry::record_retry_attempt() {
+                        log_error(
+                            "OpenTelemetry:retry_error",
+                            format!("Failed to record retry attempt: {}", e),
+                        );
+                    }
                     let future = Self::try_request(request.info.clone(), self.inner.clone());
                     self.in_flight_requests.push(Box::pin(Request {
                         retry_params: retry_params.clone(),
@@ -2836,6 +2846,13 @@ where
                     }));
                 }
                 Next::RetryBusyLoadingError { request, address } => {
+                    // Record retry attempt metric if telemetry is initialized
+                    if let Err(e) = GlideOpenTelemetry::record_retry_attempt() {
+                        log_error(
+                            "OpenTelemetry:retry_error",
+                            format!("Failed to record retry attempt: {}", e),
+                        );
+                    }
                     // TODO - do we also want to try and reconnect to replica if it is loading?
                     let future = Self::handle_loading_error_and_retry(
                         self.inner.clone(),
@@ -2862,6 +2879,13 @@ where
                     let future: Option<
                         RequestState<Pin<Box<dyn Future<Output = OperationResult> + Send>>>,
                     > = if let Some(moved_redirect) = moved_redirect {
+                        // record moved error metric if telemetry is initialized
+                        if let Err(e) = GlideOpenTelemetry::record_moved_error() {
+                            log_error(
+                                "OpenTelemetry:moved_error",
+                                format!("Failed to record moved error: {}", e),
+                            );
+                        }
                         Some(RequestState::UpdateMoved {
                             future: Box::pin(ClusterConnInner::update_upon_moved_error(
                                 self.inner.clone(),

glide-core/redis-rs/redis/src/cluster_async/pipeline_routing.rs

@@ -13,10 +13,12 @@ use crate::{cluster_routing, RedisResult, Value};
 use crate::{cluster_routing::Route, Cmd, ErrorKind, RedisError};
 use cluster_routing::RoutingInfo::{MultiNode, SingleNode};
 use futures::FutureExt;
+use logger_core::log_error;
 use rand::prelude::IteratorRandom;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
+use telemetrylib::GlideOpenTelemetry;
 use tokio::sync::oneshot;
 use tokio::sync::oneshot::error::RecvError;
 
@@ -95,6 +97,8 @@ pub(crate) type ResponsePoliciesMap =
 /// Adds a command to the pipeline map for a specific node address.
 ///
 /// `add_asking` is a boolean flag that determines whether to add an `ASKING` command before the command.
+/// `is_retrying` is a boolean flag that indicates whether this is a retry attempt.
+#[allow(clippy::too_many_arguments)]
 fn add_command_to_node_pipeline_map<C>(
     pipeline_map: &mut NodePipelineMap<C>,
     address: String,
@@ -103,9 +107,19 @@ fn add_command_to_node_pipeline_map<C>(
     index: usize,
     inner_index: Option<usize>,
     add_asking: bool,
+    is_retrying: bool,
 ) where
     C: Clone,
 {
+    if is_retrying {
+        // Record retry attempt metric if telemetry is initialized
+        if let Err(e) = GlideOpenTelemetry::record_retry_attempt() {
+            log_error(
+                "OpenTelemetry:retry_error",
+                format!("Failed to record retry attempt: {}", e),
+            );
+        }
+    }
     if add_asking {
         let asking_cmd = Arc::new(crate::cmd::cmd("ASKING"));
         pipeline_map
@@ -224,6 +238,7 @@ where
                                     index,
                                     Some(inner_index),
                                     false,
+                                    false,
                                 );
                             }
                         }
@@ -279,7 +294,7 @@ where
         .await
         .map_err(|err| (OperationTarget::NotFound, err))?;
 
-    add_command_to_node_pipeline_map(pipeline_map, address, conn, cmd, index, None, false);
+    add_command_to_node_pipeline_map(pipeline_map, address, conn, cmd, index, None, false, false);
     Ok(())
 }
 
@@ -324,6 +339,7 @@ where
                 index,
                 Some(inner_index),
                 false,
+                false,
             );
         } else {
             return Err((
@@ -1042,6 +1058,13 @@ where
 
         // Handle MOVED redirect by updating the topology
         if matches!(retry_method, RetryMethod::MovedRedirect) {
+            // record moved error metric if telemetry is initialized
+            if let Err(e) = GlideOpenTelemetry::record_moved_error() {
+                log_error(
+                    "OpenTelemetry:moved_error",
+                    format!("Failed to record moved error: {}", e),
+                );
+            }
             if let Err(server_error) =
                 pipeline_handle_moved_redirect(core.clone(), &redis_error).await
             {
@@ -1080,6 +1103,7 @@ where
                                 index,
                                 inner_index,
                                 matches!(retry_method, RetryMethod::AskRedirect),
+                                true,
                             );
                             continue;
                         }
@@ -1195,6 +1219,7 @@ where
                     index,
                     inner_index,
                     false,
+                    true,
                 );
             }
             Err(redis_error) => {