feat(data-pipeline)!: OTLP gRPC trace export

libdd-data-pipeline-ffi/src/trace_exporter.rs

@@ -508,8 +508,8 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_endpoint(
     )
 }
 
-/// Sets the OTLP export protocol. Accepts the OTel-standard values `http/json` (default) or
-/// `http/protobuf`; `grpc` is rejected as not yet supported. The host language resolves the value
+/// Sets the OTLP export protocol. Accepts the OTel-standard values `http/json` (default),
+/// `http/protobuf`, or `grpc`; unknown values are rejected. The host language resolves the value
 /// (e.g. from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`).
 ///
 /// Has no effect unless an OTLP endpoint is also configured via
@@ -529,9 +529,9 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol(
                 Ok(s) => s,
                 Err(e) => return Some(e),
             };
-            // `FromStr` is the single source of truth for string -> OtlpProtocol. It accepts only
-            // the supported HTTP encodings (`http/json`, `http/protobuf`); `grpc` and any unknown
-            // value are rejected with an error, so an unsupported protocol can never be stored.
+            // `FromStr` is the single source of truth for string -> OtlpProtocol. It accepts the
+            // OTel-standard `http/json`, `http/protobuf`, and `grpc`; any unknown value is
+            // rejected with an error, so an unsupported protocol can never be stored.
             match value.parse::<OtlpProtocol>() {
                 Ok(p) => {
                     handle.otlp_protocol = Some(p);
@@ -1477,14 +1477,17 @@ mod tests {
                 Some(OtlpProtocol::HttpProtobuf)
             );
 
-            // "grpc" → InvalidArgument
+            // "grpc" → success, stored
             let mut config = Some(TraceExporterConfig::default());
             let error = ddog_trace_exporter_config_set_otlp_protocol(
                 config.as_mut(),
                 CharSlice::from("grpc"),
             );
-            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
-            ddog_trace_exporter_error_free(error);
+            assert_eq!(error, None);
+            assert_eq!(
+                config.as_ref().unwrap().otlp_protocol,
+                Some(OtlpProtocol::Grpc)
+            );
 
             // Garbage value → InvalidArgument
             let mut config = Some(TraceExporterConfig::default());
@@ -1565,9 +1568,9 @@ mod tests {
     }
 
     #[test]
-    fn set_otlp_protocol_rejects_grpc_and_unknown() {
+    fn set_otlp_protocol_rejects_unknown() {
         let mut cfg = TraceExporterConfig::default();
-        for bad in ["grpc", "nonsense"] {
+        for bad in ["nonsense", "grcp"] {
             let err = unsafe {
                 ddog_trace_exporter_config_set_otlp_protocol(Some(&mut cfg), CharSlice::from(bad))
             };

libdd-data-pipeline/src/otlp/config.rs

@@ -6,19 +6,16 @@
 use http::HeaderMap;
 use std::time::Duration;
 
-/// OTLP trace export protocol — selects the HTTP body encoding and `Content-Type`.
-///
-/// Only the HTTP encodings libdatadog actually supports are representable. A `grpc` value (e.g.
-/// resolved from the OTel-default `OTEL_EXPORTER_OTLP_PROTOCOL`) is rejected by
-/// [`FromStr`](std::str::FromStr) rather than represented here, so an unsupported protocol can
-/// never be constructed and silently mishandled downstream.
+/// OTLP trace export protocol: selects the wire transport and body encoding.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub enum OtlpProtocol {
     /// HTTP with a JSON body (`Content-Type: application/json`). The default.
     #[default]
     HttpJson,
     /// HTTP with a protobuf body (`Content-Type: application/x-protobuf`).
     HttpProtobuf,
+    /// gRPC over HTTP/2.
+    Grpc,
 }
 
 impl std::str::FromStr for OtlpProtocol {
@@ -27,37 +24,37 @@ impl std::str::FromStr for OtlpProtocol {
         match s {
             "http/json" => Ok(OtlpProtocol::HttpJson),
             "http/protobuf" => Ok(OtlpProtocol::HttpProtobuf),
-            // gRPC is a valid OTLP protocol in the OTel spec but is not implemented in
-            // libdatadog. Reject it explicitly so callers get a clean error at the parse
-            // boundary, rather than constructing an unsupported value that has to be guarded
-            // against everywhere downstream.
-            "grpc" => Err("OTLP gRPC export is not supported".to_string()),
+            "grpc" => Ok(OtlpProtocol::Grpc),
             other => Err(format!("unknown OTLP protocol: {other}")),
         }
     }
 }
 
 impl OtlpProtocol {
-    /// The HTTP `Content-Type` for this protocol's body encoding. Crate-internal: the public type
-    /// is only constructed/selected by callers; encoding is the exporter's job.
-    pub(crate) fn content_type(&self) -> http::HeaderValue {
+    /// The HTTP `Content-Type` for this protocol's body encoding, or `None` for [`Self::Grpc`]
+    /// (framed by tonic's codec instead). Crate-internal: the public type is only
+    /// constructed/selected by callers; encoding is the exporter's job.
+    pub(crate) fn content_type(&self) -> Option<http::HeaderValue> {
         match self {
-            OtlpProtocol::HttpJson => libdd_common::header::APPLICATION_JSON,
-            OtlpProtocol::HttpProtobuf => libdd_common::header::APPLICATION_PROTOBUF,
+            OtlpProtocol::HttpJson => Some(libdd_common::header::APPLICATION_JSON),
+            OtlpProtocol::HttpProtobuf => Some(libdd_common::header::APPLICATION_PROTOBUF),
+            OtlpProtocol::Grpc => None,
         }
     }
 
-    /// Encode the prost OTLP request to this protocol's wire format. Crate-internal so the
-    /// third-party `serde_json::Error` does not leak into the public API.
+    /// Encode the prost OTLP request to this protocol's wire format, or `None` for
+    /// [`Self::Grpc`] (encoded by tonic's codec instead). Crate-internal so the third-party
+    /// `serde_json::Error` does not leak into the public API.
     pub(crate) fn encode(
         &self,
         req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest,
-    ) -> Result<Vec<u8>, serde_json::Error> {
+    ) -> Option<Result<Vec<u8>, serde_json::Error>> {
         match self {
-            OtlpProtocol::HttpJson => libdd_trace_utils::otlp_encoder::encode_otlp_json(req),
-            OtlpProtocol::HttpProtobuf => {
-                Ok(libdd_trace_utils::otlp_encoder::encode_otlp_protobuf(req))
-            }
+            OtlpProtocol::HttpJson => Some(libdd_trace_utils::otlp_encoder::encode_otlp_json(req)),
+            OtlpProtocol::HttpProtobuf => Some(Ok(
+                libdd_trace_utils::otlp_encoder::encode_otlp_protobuf(req),
+            )),
+            OtlpProtocol::Grpc => None,
         }
     }
 }
@@ -86,7 +83,8 @@ pub struct OtlpTraceConfig {
 }
 
 /// Per-request OTLP gRPC trace exporter configuration.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
+// Consumed only by the native gRPC export path (`grpc_exporter`), which is compiled out on
+// wasm32, so this is dead code on wasm targets.
 #[allow(dead_code)]
 #[derive(Clone, Debug)]
 pub struct OtlpGrpcTraceConfig {
@@ -112,26 +110,21 @@ mod tests {
             OtlpProtocol::from_str("http/protobuf").unwrap(),
             OtlpProtocol::HttpProtobuf
         );
+        assert_eq!(OtlpProtocol::from_str("grpc").unwrap(), OtlpProtocol::Grpc);
         assert!(OtlpProtocol::from_str("nonsense").is_err());
     }
 
-    #[test]
-    fn grpc_is_rejected_at_parse() {
-        // gRPC is unsupported, so it must not parse into a protocol: an unsupported value can
-        // never be constructed.
-        assert!(OtlpProtocol::from_str("grpc").is_err());
-    }
-
     #[test]
     fn protocol_content_types() {
         assert_eq!(
             OtlpProtocol::HttpJson.content_type(),
-            libdd_common::header::APPLICATION_JSON
+            Some(libdd_common::header::APPLICATION_JSON)
         );
         assert_eq!(
             OtlpProtocol::HttpProtobuf.content_type(),
-            libdd_common::header::APPLICATION_PROTOBUF
+            Some(libdd_common::header::APPLICATION_PROTOBUF)
         );
+        assert_eq!(OtlpProtocol::Grpc.content_type(), None);
     }
 
     #[test]

libdd-data-pipeline/src/otlp/exporter.rs

@@ -17,7 +17,7 @@ use std::time::Duration;
 pub(crate) const OTLP_MAX_RETRIES: u32 = 4;
 /// No retries on shutdown to avoid a long backoff in the shutdown window.
 pub(crate) const OTLP_SHUTDOWN_MAX_RETRIES: u32 = 0;
-const OTLP_RETRY_DELAY_MS: u64 = 100;
+pub(crate) const OTLP_RETRY_DELAY_MS: u64 = 100;
 
 /// POST an OTLP HTTP payload to `endpoint_url` with the given `content_type` (callers pass JSON or
 /// protobuf); `test_token` enables snapshot tests.
@@ -84,13 +84,18 @@ pub async fn send_otlp_traces_http<C: HttpClientCapability + SleepCapability>(
     test_token: Option<&str>,
     body: Vec<u8>,
 ) -> Result<(), TraceExporterError> {
+    let content_type = config.protocol.content_type().ok_or_else(|| {
+        TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(
+            "OTLP gRPC protocol cannot be sent over the HTTP export path".to_string(),
+        ))
+    })?;
     send_otlp_http(
         capabilities,
         &config.endpoint_url,
         &config.headers,
         config.timeout,
         test_token,
-        config.protocol.content_type(),
+        content_type,
         body,
         OTLP_MAX_RETRIES,
     )

libdd-data-pipeline/src/otlp/grpc_exporter.rs

@@ -197,8 +197,6 @@ pub(crate) struct OtlpGrpcTransport {
 }
 
 /// Validate a gRPC endpoint (plaintext `http://` only) and build the transport.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
-#[allow(dead_code)]
 pub(crate) fn build_grpc_transport(
     endpoint_url: &str,
     config: OtlpGrpcTraceConfig,
@@ -280,8 +278,6 @@ type ExportCodec =
     prost_codec::ProstCodecImpl<ExportTraceServiceRequest, ExportTraceServiceResponse>;
 
 /// Send one OTLP trace export request over gRPC. Bounds connect + RPC with a single timeout.
-// Not yet wired to the trace exporter's send loop; exercised by tests only.
-#[allow(dead_code)]
 pub(crate) async fn send_otlp_traces_grpc(
     transport: &OtlpGrpcTransport,
     test_token: Option<&str>,
@@ -370,6 +366,13 @@ fn grpc_status_to_error(status: Status) -> TraceExporterError {
             std::io::ErrorKind::TimedOut,
             status.message(),
         )),
+        // A clean server GOAWAY / stream cancel (e.g. graceful connection close) surfaces as
+        // `Cancelled` with no io source. Treat it as a transient transport failure so it is
+        // retried, matching the HTTP path's behavior on transient network errors.
+        Code::Cancelled => TraceExporterError::Io(std::io::Error::new(
+            std::io::ErrorKind::ConnectionAborted,
+            status.message(),
+        )),
         _ => TraceExporterError::Request(RequestError::new(
             http::StatusCode::INTERNAL_SERVER_ERROR,
             status.message(),
@@ -568,10 +571,10 @@ mod integration_tests {
         let addr = listener.local_addr().unwrap();
         // Complete the H2 handshake and accept the request, then abort the TCP connection with a
         // RST (SO_LINGER=0) instead of a graceful close or a gRPC-status trailer. The client's
-        // in-flight read then fails with ECONNRESET. This is the FIX-1 case: the resulting
-        // `std::io::Error` is wrapped inside a `hyper::Error` below the tonic `Status`, so it is
-        // recovered only by walking the source chain in `grpc_status_to_error` — a graceful FIN
-        // instead surfaces as an h2 "canceled" status that would be misreported as `Request`.
+        // in-flight read then fails with ECONNRESET. The resulting `std::io::Error` is wrapped
+        // inside a `hyper::Error` below the tonic `Status`, so it is recovered only by walking the
+        // source chain in `grpc_status_to_error`. (A graceful FIN instead surfaces as a `Cancelled`
+        // status, which is now also mapped to `Io` and retried.)
         let server = tokio::spawn(async move {
             let (socket, _) = listener.accept().await.unwrap();
             // Force a RST on close rather than a graceful FIN. `set_linger` is deprecated because a
@@ -625,6 +628,11 @@ mod send_tests {
                 Status::deadline_exceeded("slow"),
                 std::io::ErrorKind::TimedOut,
             ),
+            // Graceful GOAWAY/stream cancel is retried like other transient transport failures.
+            (
+                Status::new(Code::Cancelled, "canceled"),
+                std::io::ErrorKind::ConnectionAborted,
+            ),
         ] {
             match grpc_status_to_error(s) {
                 TraceExporterError::Io(e) => assert_eq!(e.kind(), want),

libdd-data-pipeline/src/otlp/mod.rs

@@ -39,3 +39,10 @@ pub use config::{OtlpMetricsConfig, OtlpProtocol, OtlpTraceConfig};
 pub use exporter::send_otlp_traces_http;
 pub use libdd_trace_utils::otlp_encoder::{map_traces_to_otlp, OtlpResourceInfo};
 pub use metrics::OtlpStatsExporter;
+
+// gRPC OTLP export is native-only (tonic/hyper don't build for wasm32); the config type and the
+// transport/send symbols are consumed only by the native trace-exporter send loop and builder.
+#[cfg(not(target_arch = "wasm32"))]
+pub use config::OtlpGrpcTraceConfig;
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) use grpc_exporter::{build_grpc_transport, send_otlp_traces_grpc, OtlpGrpcTransport};