Implement retry for http trace

Author: scottgerring

opentelemetry-otlp/Cargo.toml

@@ -83,6 +83,11 @@ tls-webpki-roots = ["tls", "tonic/tls-webpki-roots"]
 
 # http binary
 http-proto = ["prost", "opentelemetry-http", "opentelemetry-proto/gen-tonic-messages", "http", "trace", "metrics"]
+
+# http with retry support.
+# What should we do with this? We need the async_runtime. gRPC exporters already need it.
+http-retry = ["opentelemetry_sdk/experimental_async_runtime", "opentelemetry_sdk/rt-tokio", "tokio"]
+
 http-json = ["serde_json", "prost", "opentelemetry-http", "opentelemetry-proto/gen-tonic-messages", "opentelemetry-proto/with-serde", "http", "trace", "metrics"]
 reqwest-blocking-client = ["reqwest/blocking", "opentelemetry-http/reqwest-blocking"]
 reqwest-client = ["reqwest", "opentelemetry-http/reqwest"]

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

@@ -8,7 +8,138 @@ use opentelemetry_sdk::{
     trace::{SpanData, SpanExporter},
 };
 
+#[cfg(feature = "http-retry")]
+use crate::retry_classification::http::classify_http_error;
+#[cfg(feature = "http-retry")]
+use opentelemetry_sdk::retry::{retry_with_backoff, RetryErrorType, RetryPolicy};
+#[cfg(feature = "http-retry")]
+use opentelemetry_sdk::runtime::Tokio;
+
+#[cfg(feature = "http-retry")]
+/// HTTP-specific error wrapper for retry classification
+#[derive(Debug)]
+struct HttpExportError {
+    status_code: u16,
+    retry_after: Option<String>,
+    message: String,
+}
+
+#[cfg(feature = "http-retry")]
+/// Classify HTTP export errors for retry decisions
+fn classify_http_export_error(error: &HttpExportError) -> RetryErrorType {
+    classify_http_error(error.status_code, error.retry_after.as_deref())
+}
+
 impl SpanExporter for OtlpHttpClient {
+    #[cfg(feature = "http-retry")]
+    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
+        let policy = RetryPolicy {
+            max_retries: 3,
+            initial_delay_ms: 100,
+            max_delay_ms: 1600,
+            jitter_ms: 100,
+        };
+
+        let batch = Arc::new(batch);
+
+        retry_with_backoff(
+            Tokio,
+            policy,
+            classify_http_export_error,
+            "HttpTracesClient.Export",
+            || async {
+                let batch_clone = Arc::clone(&batch);
+
+                // Get client
+                let client = self
+                    .client
+                    .lock()
+                    .map_err(|e| HttpExportError {
+                        status_code: 500,
+                        retry_after: None,
+                        message: format!("Mutex lock failed: {e}"),
+                    })?
+                    .as_ref()
+                    .ok_or_else(|| HttpExportError {
+                        status_code: 500,
+                        retry_after: None,
+                        message: "Exporter already shutdown".to_string(),
+                    })?
+                    .clone();
+
+                // Build request body
+                let (body, content_type, content_encoding) = self
+                    .build_trace_export_body((*batch_clone).clone())
+                    .map_err(|e| HttpExportError {
+                        status_code: 400,
+                        retry_after: None,
+                        message: format!("Failed to build request body: {e}"),
+                    })?;
+
+                // Build HTTP request
+                let mut request_builder = http::Request::builder()
+                    .method(Method::POST)
+                    .uri(&self.collector_endpoint)
+                    .header(CONTENT_TYPE, content_type);
+
+                if let Some(encoding) = content_encoding {
+                    request_builder = request_builder.header("Content-Encoding", encoding);
+                }
+
+                let mut request =
+                    request_builder
+                        .body(body.into())
+                        .map_err(|e| HttpExportError {
+                            status_code: 400,
+                            retry_after: None,
+                            message: format!("Failed to build HTTP request: {e}"),
+                        })?;
+
+                for (k, v) in &self.headers {
+                    request.headers_mut().insert(k.clone(), v.clone());
+                }
+
+                let request_uri = request.uri().to_string();
+                otel_debug!(name: "HttpTracesClient.ExportStarted");
+
+                // Send request
+                let response = client.send_bytes(request).await.map_err(|e| {
+                    HttpExportError {
+                        status_code: 0, // Network error
+                        retry_after: None,
+                        message: format!("Network error: {e:?}"),
+                    }
+                })?;
+
+                let status_code = response.status().as_u16();
+                let retry_after = response
+                    .headers()
+                    .get("retry-after")
+                    .and_then(|v| v.to_str().ok())
+                    .map(|s| s.to_string());
+
+                if !response.status().is_success() {
+                    return Err(HttpExportError {
+                        status_code,
+                        retry_after,
+                        message: format!(
+                            "HTTP export failed. Url: {}, Status: {}, Response: {:?}",
+                            request_uri,
+                            status_code,
+                            response.body()
+                        ),
+                    });
+                }
+
+                otel_debug!(name: "HttpTracesClient.ExportSucceeded");
+                Ok(())
+            },
+        )
+        .await
+        .map_err(|e| OTelSdkError::InternalFailure(e.message))
+    }
+
+    #[cfg(not(feature = "http-retry"))]
     async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
         let client = match self
             .client

opentelemetry-otlp/src/lib.rs

@@ -366,7 +366,7 @@ mod metric;
 #[cfg(any(feature = "http-proto", feature = "http-json", feature = "grpc-tonic"))]
 mod span;
 
-#[cfg(feature = "grpc-tonic")]
+#[cfg(any(feature = "grpc-tonic", feature = "http-retry"))]
 pub mod retry_classification;
 
 pub use crate::exporter::Compression;

opentelemetry-otlp/src/retry_classification.rs

@@ -60,14 +60,14 @@ pub mod http {
     fn parse_retry_after(retry_after: &str) -> Option<Duration> {
         // Try parsing as seconds first
         if let Ok(seconds) = retry_after.trim().parse::<u64>() {
-            // Cap at 10 minutes for safety
+            // Cap at 10 minutes. TODO - what's sensible here?
             let capped_seconds = seconds.min(600);
             return Some(Duration::from_secs(capped_seconds));
         }
 
         // Try parsing as HTTP date
         if let Ok(delay_seconds) = parse_http_date_to_delay(retry_after) {
-            // Cap at 10 minutes for safety
+            // Cap at 10 minutes. TODO - what's sensible here?
             let capped_seconds = delay_seconds.min(600);
             return Some(Duration::from_secs(capped_seconds));
         }
@@ -78,7 +78,7 @@ pub mod http {
     /// Parses HTTP date format and returns delay in seconds from now.
     ///
     /// This is a simplified parser for the most common HTTP date format.
-    /// In production, you might want to use a proper HTTP date parsing library.
+    /// TODO - should we use a library here?
     fn parse_http_date_to_delay(date_str: &str) -> Result<u64, ()> {
         // For now, return error - would need proper HTTP date parsing
         // This could be implemented with chrono or similar
@@ -88,6 +88,7 @@ pub mod http {
 }
 
 /// gRPC-specific error classification with RetryInfo support.
+#[cfg(feature = "grpc-tonic")]
 pub mod grpc {
     use super::*;
 
@@ -122,7 +123,7 @@ pub mod grpc {
             tonic::Code::ResourceExhausted => {
                 if let Some(seconds) = retry_info_seconds {
                     // Server signals recovery is possible - use throttled retry
-                    let capped_seconds = seconds.min(600); // Cap at 10 minutes for safety
+                    let capped_seconds = seconds.min(600); // Cap at 10 minutes. TODO - what's sensible here?
                     return RetryErrorType::Throttled(std::time::Duration::from_secs(
                         capped_seconds,
                     ));