add retry to traces

Author: scottgerring

opentelemetry-otlp/src/exporter/tonic/logs.rs

@@ -92,7 +92,7 @@ impl LogExporter for TonicLogsClient {
                         None => return Err(OTelSdkError::AlreadyShutdown),
                     };
 
-                    let resource_logs = group_logs_by_resource_and_scope(&*batch, resource);
+                    let resource_logs = group_logs_by_resource_and_scope(&batch, resource);
 
                     otel_debug!(name: "TonicLogsClient.ExportStarted");
 

opentelemetry-otlp/src/exporter/tonic/trace.rs

@@ -1,4 +1,5 @@
 use core::fmt;
+use std::sync::Arc;
 use tokio::sync::Mutex;
 
 use opentelemetry::otel_debug;
@@ -15,6 +16,9 @@ use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Chann
 
 use super::BoxInterceptor;
 
+use opentelemetry_sdk::retry::{retry_with_exponential_backoff, RetryPolicy};
+use opentelemetry_sdk::runtime::Tokio;
+
 pub(crate) struct TonicTracesClient {
     inner: Option<ClientInner>,
     #[allow(dead_code)]
@@ -60,43 +64,63 @@ impl TonicTracesClient {
 
 impl SpanExporter for TonicTracesClient {
     async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
-        let (mut client, metadata, extensions) = match &self.inner {
-            Some(inner) => {
-                let (m, e, _) = inner
-                    .interceptor
-                    .lock()
-                    .await // tokio::sync::Mutex doesn't return a poisoned error, so we can safely use the interceptor here
-                    .call(Request::new(()))
-                    .map_err(|e| OTelSdkError::InternalFailure(format!("error: {e:?}")))?
-                    .into_parts();
-                (inner.client.clone(), m, e)
-            }
-            None => return Err(OTelSdkError::AlreadyShutdown),
+        let policy = RetryPolicy {
+            max_retries: 3,
+            initial_delay_ms: 100,
+            max_delay_ms: 1600,
+            jitter_ms: 100,
         };
 
-        let resource_spans = group_spans_by_resource_and_scope(batch, &self.resource);
+        let batch = Arc::new(batch);
 
-        otel_debug!(name: "TonicTracesClient.ExportStarted");
+        retry_with_exponential_backoff(Tokio, policy, "TonicTracesClient.Export", {
+            let batch = Arc::clone(&batch);
+            let inner = &self.inner;
+            let resource = &self.resource;
+            move || {
+                let batch = Arc::clone(&batch);
+                Box::pin(async move {
+                    let (mut client, metadata, extensions) = match inner {
+                        Some(inner) => {
+                            let (m, e, _) = inner
+                                .interceptor
+                                .lock()
+                                .await // tokio::sync::Mutex doesn't return a poisoned error, so we can safely use the interceptor here
+                                .call(Request::new(()))
+                                .map_err(|e| OTelSdkError::InternalFailure(format!("error: {e:?}")))?
+                                .into_parts();
+                            (inner.client.clone(), m, e)
+                        }
+                        None => return Err(OTelSdkError::AlreadyShutdown),
+                    };
 
-        let result = client
-            .export(Request::from_parts(
-                metadata,
-                extensions,
-                ExportTraceServiceRequest { resource_spans },
-            ))
-            .await;
+                    let resource_spans = group_spans_by_resource_and_scope((*batch).clone(), resource);
 
-        match result {
-            Ok(_) => {
-                otel_debug!(name: "TonicTracesClient.ExportSucceeded");
-                Ok(())
-            }
-            Err(e) => {
-                let error = e.to_string();
-                otel_debug!(name: "TonicTracesClient.ExportFailed", error = &error);
-                Err(OTelSdkError::InternalFailure(error))
+                    otel_debug!(name: "TonicTracesClient.ExportStarted");
+
+                    let result = client
+                        .export(Request::from_parts(
+                            metadata,
+                            extensions,
+                            ExportTraceServiceRequest { resource_spans },
+                        ))
+                        .await;
+
+                    match result {
+                        Ok(_) => {
+                            otel_debug!(name: "TonicTracesClient.ExportSucceeded");
+                            Ok(())
+                        }
+                        Err(e) => {
+                            let error = format!("export error: {e:?}");
+                            otel_debug!(name: "TonicTracesClient.ExportFailed", error = &error);
+                            Err(OTelSdkError::InternalFailure(error))
+                        }
+                    }
+                })
             }
-        }
+        })
+        .await
     }
 
     fn shutdown(&mut self) -> OTelSdkResult {

opentelemetry-sdk/src/retry.rs

@@ -34,7 +34,7 @@ pub struct RetryPolicy {
 /// A runtime stub for when experimental_async_runtime is not enabled.
 /// This allows retry policy to be configured but no actual retries occur.
 #[cfg(not(feature = "experimental_async_runtime"))]
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct NoOpRuntime;
 
 #[cfg(not(feature = "experimental_async_runtime"))]