Merge branch 'main' into propogate-trace-errors

Author: lalitb

examples/tracing-http-propagator/src/server.rs

@@ -121,6 +121,10 @@ impl LogProcessor for EnrichWithBaggageLogProcessor {
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
+
+    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+        Ok(())
+    }
 }
 
 /// A custom span processor that enriches spans with baggage attributes. Baggage

opentelemetry-appender-log/examples/logs-basic.rs

@@ -1,12 +1,15 @@
 //! run with `$ cargo run --example logs-basic`
 
-/// This example shows how to use in_memory_exporter for logs. This uses opentelemetry-appender-log crate, which is a
-/// [logging appender](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/glossary.md#log-appender--bridge) that bridges logs from the [log crate](https://docs.rs/log/latest/log/) to OpenTelemetry.
-/// The example setups a LoggerProvider with a in-memory exporter, so emitted logs are stored in memory.
+/// This example shows how to use stdout exporter for logs. This uses
+/// opentelemetry-appender-log crate, which is a [logging
+/// appender](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/glossary.md#log-appender--bridge)
+/// that bridges logs from the [log crate](https://docs.rs/log/latest/log/) to
+/// OpenTelemetry. The example setups a LoggerProvider with a stdout exporter,
+/// so emitted logs are written to stdout.
 ///
 use log::{error, info, warn, Level};
 use opentelemetry_appender_log::OpenTelemetryLogBridge;
-use opentelemetry_sdk::logs::{BatchLogProcessor, SdkLoggerProvider};
+use opentelemetry_sdk::{logs::SdkLoggerProvider, Resource};
 use opentelemetry_stdout::LogExporter;
 
 #[tokio::main]
@@ -15,7 +18,12 @@ async fn main() {
     let exporter = LogExporter::default();
     //Create a LoggerProvider and register the exporter
     let logger_provider = SdkLoggerProvider::builder()
-        .with_log_processor(BatchLogProcessor::builder(exporter).build())
+        .with_resource(
+            Resource::builder()
+                .with_service_name("log-appender-log-example")
+                .build(),
+        )
+        .with_simple_exporter(exporter)
         .build();
 
     // Setup Log Appender for the log crate.

opentelemetry-appender-tracing/benches/log-attributes.rs

@@ -43,6 +43,10 @@ impl LogProcessor for NoopProcessor {
     fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
+
+    fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+        Ok(())
+    }
 }
 
 /// Creates a single benchmark for a specific number of attributes

opentelemetry-appender-tracing/benches/logs.rs

@@ -62,6 +62,10 @@ impl LogProcessor for NoopProcessor {
     ) -> bool {
         self.enabled
     }
+
+    fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+        Ok(())
+    }
 }
 
 struct NoOpLogLayer {

opentelemetry-appender-tracing/src/layer.rs

@@ -934,6 +934,10 @@ mod tests {
         fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
+
+        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
+            Ok(())
+        }
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]

opentelemetry-otlp/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## vNext
 
+- Add partial success response handling for OTLP exporters (traces, metrics, logs) per OTLP spec. Exporters now log warnings when the server returns partial success responses with rejected items and error messages. [#865](https://github.com/open-telemetry/opentelemetry-rust/issues/865)
 - Refactor `internal-logs` feature in `opentelemetry-otlp` to reduce unnecessary dependencies[3191](https://github.com/open-telemetry/opentelemetry-rust/pull/3192)
 
 ## 0.31.0

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

@@ -1,16 +1,23 @@
 use super::OtlpHttpClient;
+use crate::Protocol;
+use opentelemetry::{otel_debug, otel_warn};
 use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
+use prost::Message;
 use std::time;
 
 impl LogExporter for OtlpHttpClient {
     async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
-        self.export_http_with_retry(
-            batch,
-            OtlpHttpClient::build_logs_export_body,
-            "HttpLogsClient.Export",
-        )
-        .await
+        let response_body = self
+            .export_http_with_retry(
+                batch,
+                OtlpHttpClient::build_logs_export_body,
+                "HttpLogsClient.Export",
+            )
+            .await?;
+
+        handle_partial_success(&response_body, self.protocol);
+        Ok(())
     }
 
     fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
@@ -29,3 +36,68 @@ impl LogExporter for OtlpHttpClient {
         self.resource = resource.into();
     }
 }
+
+/// Handles partial success returned by OTLP endpoints. We log the rejected log records,
+/// as well as the error message returned.
+fn handle_partial_success(response_body: &[u8], protocol: Protocol) {
+    use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceResponse;
+
+    let response: ExportLogsServiceResponse = match protocol {
+        #[cfg(feature = "http-json")]
+        Protocol::HttpJson => match serde_json::from_slice(response_body) {
+            Ok(r) => r,
+            Err(e) => {
+                otel_debug!(name: "HttpLogsClient.ResponseParseError", error = e.to_string());
+                return;
+            }
+        },
+        _ => match Message::decode(response_body) {
+            Ok(r) => r,
+            Err(e) => {
+                otel_debug!(name: "HttpLogsClient.ResponseParseError", error = e.to_string());
+                return;
+            }
+        },
+    };
+
+    if let Some(partial_success) = response.partial_success {
+        if partial_success.rejected_log_records > 0 || !partial_success.error_message.is_empty() {
+            otel_warn!(
+                name: "HttpLogsClient.PartialSuccess",
+                rejected_log_records = partial_success.rejected_log_records,
+                error_message = partial_success.error_message.as_str(),
+            );
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_handle_invalid_protobuf() {
+        // Corrupted/invalid protobuf data
+        let invalid = vec![0xFF, 0xFF, 0xFF, 0xFF];
+
+        // Should not panic - logs debug and returns early
+        handle_partial_success(&invalid, Protocol::HttpBinary);
+    }
+
+    #[test]
+    fn test_handle_empty_response() {
+        let empty = vec![];
+
+        // Should not panic
+        handle_partial_success(&empty, Protocol::HttpBinary);
+    }
+
+    #[cfg(feature = "http-json")]
+    #[test]
+    fn test_handle_invalid_json() {
+        let invalid_json = b"{not valid json}";
+
+        // Should not panic - logs debug and returns
+        handle_partial_success(invalid_json, Protocol::HttpJson);
+    }
+}

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

@@ -1,6 +1,9 @@
 use crate::metric::MetricsClient;
+use crate::Protocol;
+use opentelemetry::{otel_debug, otel_warn};
 use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
+use prost::Message;
 
 use super::OtlpHttpClient;
 
@@ -12,8 +15,12 @@ impl MetricsClient for OtlpHttpClient {
                 .ok_or_else(|| "Failed to serialize metrics".to_string())
         };
 
-        self.export_http_with_retry(metrics, build_body_wrapper, "HttpMetricsClient.Export")
-            .await
+        let response_body = self
+            .export_http_with_retry(metrics, build_body_wrapper, "HttpMetricsClient.Export")
+            .await?;
+
+        handle_partial_success(&response_body, self.protocol);
+        Ok(())
     }
 
     fn shutdown(&self) -> OTelSdkResult {
@@ -25,3 +32,68 @@ impl MetricsClient for OtlpHttpClient {
         Ok(())
     }
 }
+
+/// Handles partial success returned by OTLP endpoints. We log the rejected data points,
+/// as well as the error message returned.
+fn handle_partial_success(response_body: &[u8], protocol: Protocol) {
+    use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceResponse;
+
+    let response: ExportMetricsServiceResponse = match protocol {
+        #[cfg(feature = "http-json")]
+        Protocol::HttpJson => match serde_json::from_slice(response_body) {
+            Ok(r) => r,
+            Err(e) => {
+                otel_debug!(name: "HttpMetricsClient.ResponseParseError", error = e.to_string());
+                return;
+            }
+        },
+        _ => match Message::decode(response_body) {
+            Ok(r) => r,
+            Err(e) => {
+                otel_debug!(name: "HttpMetricsClient.ResponseParseError", error = e.to_string());
+                return;
+            }
+        },
+    };
+
+    if let Some(partial_success) = response.partial_success {
+        if partial_success.rejected_data_points > 0 || !partial_success.error_message.is_empty() {
+            otel_warn!(
+                name: "HttpMetricsClient.PartialSuccess",
+                rejected_data_points = partial_success.rejected_data_points,
+                error_message = partial_success.error_message.as_str(),
+            );
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_handle_invalid_protobuf() {
+        // Corrupted/invalid protobuf data
+        let invalid = vec![0xFF, 0xFF, 0xFF, 0xFF];
+
+        // Should not panic - logs debug and returns early
+        handle_partial_success(&invalid, Protocol::HttpBinary);
+    }
+
+    #[test]
+    fn test_handle_empty_response() {
+        let empty = vec![];
+
+        // Should not panic
+        handle_partial_success(&empty, Protocol::HttpBinary);
+    }
+
+    #[cfg(feature = "http-json")]
+    #[test]
+    fn test_handle_invalid_json() {
+        let invalid_json = b"{not valid json}";
+
+        // Should not panic - logs debug and returns
+        handle_partial_success(invalid_json, Protocol::HttpJson);
+    }
+}

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

@@ -5,7 +5,7 @@ use super::{
 use crate::{ExportConfig, Protocol, OTEL_EXPORTER_OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS};
 use http::{HeaderName, HeaderValue, Uri};
 use opentelemetry::otel_debug;
-use opentelemetry_http::HttpClient;
+use opentelemetry_http::{Bytes, HttpClient};
 use opentelemetry_proto::transform::common::tonic::ResourceAttributesWithSchema;
 #[cfg(feature = "logs")]
 use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
@@ -381,7 +381,7 @@ impl OtlpHttpClient {
         data: T,
         build_body_fn: F,
         operation_name: &'static str,
-    ) -> opentelemetry_sdk::error::OTelSdkResult
+    ) -> Result<Bytes, opentelemetry_sdk::error::OTelSdkError>
     where
         F: Fn(&Self, T) -> Result<(Vec<u8>, &'static str, Option<&'static str>), String>,
     {
@@ -408,7 +408,7 @@ impl OtlpHttpClient {
             #[cfg(not(feature = "reqwest-blocking-client"))]
             let runtime = opentelemetry_sdk::runtime::Tokio;
 
-            retry_with_backoff(
+            let response_body = retry_with_backoff(
                 runtime,
                 self.retry_policy.clone(),
                 classify_http_export_error,
@@ -424,7 +424,9 @@ impl OtlpHttpClient {
                 },
             )
             .await
-            .map_err(|e| opentelemetry_sdk::error::OTelSdkError::InternalFailure(e.message))
+            .map_err(|e| opentelemetry_sdk::error::OTelSdkError::InternalFailure(e.message))?;
+
+            Ok(response_body)
         }
 
         #[cfg(not(feature = "experimental-http-retry"))]
@@ -438,9 +440,12 @@ impl OtlpHttpClient {
                 endpoint: self.collector_endpoint.to_string(),
             };
 
-            self.export_http_once(&retry_data, content_type, content_encoding, operation_name)
+            let response_body = self
+                .export_http_once(&retry_data, content_type, content_encoding, operation_name)
                 .await
-                .map_err(|e| opentelemetry_sdk::error::OTelSdkError::InternalFailure(e.message))
+                .map_err(|e| opentelemetry_sdk::error::OTelSdkError::InternalFailure(e.message))?;
+
+            Ok(response_body)
         }
     }
 
@@ -451,7 +456,7 @@ impl OtlpHttpClient {
         content_type: &'static str,
         content_encoding: Option<&'static str>,
         _operation_name: &'static str,
-    ) -> Result<(), HttpExportError> {
+    ) -> Result<Bytes, HttpExportError> {
         // Get client
         let client = self
             .client
@@ -510,7 +515,9 @@ impl OtlpHttpClient {
         }
 
         otel_debug!(name: "HttpClient.ExportSucceeded");
-        Ok(())
+
+        // Return the response, consuming the body to save a copy
+        Ok(response.into_body())
     }
 
     /// Compress data using gzip or zstd if the user has requested it and the relevant feature