feat(data-pipeline): OTLP HTTP/protobuf trace export (#2115)

# What does this PR do?

Adds OTLP HTTP/protobuf as a trace-export encoding alongside HTTP/JSON, selectable via the OTel-standard `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` (`http/json` default, `http/protobuf`). The generated `prost` OTLP types are the single intermediate representation: the mapper builds them directly from native spans, protobuf is `prost::encode_to_vec`, and JSON is a serde serializer over the same types.

# Motivation

libdatadog's OTLP trace export only spoke HTTP/JSON. SDKs that honor `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` need `http/protobuf` to match the OTel default and to talk to collectors that expect protobuf.

# Additional Notes

- Prost is the single IR: no parallel hand-rolled JSON model, no string round-trip. `encode_otlp_protobuf` is `encode_to_vec`; `encode_otlp_json` is a serde serializer (`json_serializer.rs`) emitting OTLP-spec JSON (hex ids, base64 bytes, int64-as-string, lowerCamelCase, proto3 defaults omitted).
- `OtlpProtocol` is `{HttpJson, HttpProtobuf}`; `grpc` is rejected at the parse boundary (`FromStr`) rather than carried as an unsupported variant. `content_type()`/`encode()` live on the type.
- Integrates with the OTLP metrics exporter already on main: the shared low-level sender takes the content-type per request, so traces use the configured protocol while metrics stay JSON.
- Carries through `otel_trace_semantics_enabled` (OTel attribute compatibility, #2091): when set, the prost mapper omits the DD-specific span attributes and promotes the error message to the OTLP `Status`.
- Span-link W3C trace flags are carried through to OTLP `Link.flags`.
- Benchmarks for the encoder hot paths, plus allocation tuning (pre-sized Vecs, allocation-free id/timestamp/int serialization).

# How to test the change?

- `cargo test -p libdd-trace-utils -p libdd-data-pipeline -p libdd-data-pipeline-ffi`, `cargo test --doc`, clippy, fmt, and `cargo ffi-test` pass. A parity test asserts the JSON and protobuf encodings carry the same span from the one prost IR; a protobuf round-trip test asserts the encoding is lossless.
- End-to-end through dd-trace-py (DataDog/dd-trace-py#18609): emitted protobuf-only OTLP traces through a local Agent to the backend. Wire was `application/x-protobuf` (HTTP 200), spans ingested with correct service/resource and a preserved 128-bit trace id.
- Benchmarks: `cargo bench -p libdd-trace-utils --bench main -- otlp/`.

BREAKING CHANGE: removes the previously public `libdd_trace_utils::otlp_encoder::json_types` module (the hand-rolled OTLP JSON model). OTLP encoding now builds prost-generated types as the single IR. libdatadog consumers pin by version, so they pick this up on the next release.


Co-authored-by: brian.marks <brian.marks@datadoghq.com>

Author: bm1549

Cargo.lock

@@ -13,6 +13,7 @@ use libdd_data_pipeline::trace_exporter::{
     TelemetryConfig, TelemetryInstrumentationSessions, TraceExporter as GenericTraceExporter,
     TraceExporterInputFormat, TraceExporterOutputFormat,
 };
+use libdd_data_pipeline::OtlpProtocol;
 
 pub(crate) type TraceExporter = GenericTraceExporter<NativeCapabilities>;
 
@@ -83,6 +84,7 @@ pub struct TraceExporterConfig {
     connection_timeout: Option<u64>,
     shared_runtime: Option<Arc<SharedRuntime>>,
     otlp_endpoint: Option<String>,
+    otlp_protocol: Option<OtlpProtocol>,
 }
 
 #[no_mangle]
@@ -498,12 +500,50 @@ 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
+/// (e.g. from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`).
+///
+/// Has no effect unless an OTLP endpoint is also configured via
+/// `ddog_trace_exporter_config_set_otlp_endpoint`; without one, traces are sent to the
+/// Datadog agent and this protocol selection is ignored.
+///
+/// Returns `None` on success, `ErrorCode::InvalidArgument` for a null config or an unaccepted
+/// value, and `ErrorCode::InvalidInput` for a non-UTF-8 string.
+#[no_mangle]
+pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_protocol(
+    config: Option<&mut TraceExporterConfig>,
+    protocol: CharSlice,
+) -> Option<Box<ExporterError>> {
+    catch_panic!(
+        if let Some(handle) = config {
+            let value = match sanitize_string(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.
+            match value.parse::<OtlpProtocol>() {
+                Ok(p) => {
+                    handle.otlp_protocol = Some(p);
+                    None
+                }
+                Err(_) => gen_error!(ErrorCode::InvalidArgument),
+            }
+        } else {
+            gen_error!(ErrorCode::InvalidArgument)
+        },
+        gen_error!(ErrorCode::Panic)
+    )
+}
+
 /// Create a new TraceExporter instance.
 ///
-/// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces in
-/// OTLP HTTP/JSON to that endpoint instead of the Datadog agent. The same payload (e.g.
-/// MessagePack) is passed to `ddog_trace_exporter_send`; the library decodes and converts to
-/// OTLP when OTLP is enabled.
+/// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces to
+/// that endpoint in OTLP over HTTP — JSON or protobuf per the configured protocol — instead of
+/// to the Datadog agent. The same payload (e.g. MessagePack) is passed to
+/// `ddog_trace_exporter_send`; the library decodes and converts it to OTLP when OTLP is enabled.
 ///
 /// # Arguments
 ///
@@ -565,6 +605,9 @@ pub unsafe extern "C" fn ddog_trace_exporter_new(
 
             if let Some(ref url) = config.otlp_endpoint {
                 builder.set_otlp_endpoint(url);
+                if let Some(protocol) = config.otlp_protocol {
+                    builder.set_otlp_protocol(protocol);
+                }
             }
 
             match builder.build() {
@@ -1283,6 +1326,95 @@ mod tests {
         }
     }
 
+    #[test]
+    fn config_otlp_protocol_test() {
+        unsafe {
+            // Null config → InvalidArgument
+            let error =
+                ddog_trace_exporter_config_set_otlp_protocol(None, CharSlice::from("http/json"));
+            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
+            ddog_trace_exporter_error_free(error);
+
+            // "http/json" → success, stored
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_otlp_protocol(
+                config.as_mut(),
+                CharSlice::from("http/json"),
+            );
+            assert_eq!(error, None);
+            assert_eq!(
+                config.as_ref().unwrap().otlp_protocol,
+                Some(OtlpProtocol::HttpJson)
+            );
+
+            // "http/protobuf" → success, stored
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_otlp_protocol(
+                config.as_mut(),
+                CharSlice::from("http/protobuf"),
+            );
+            assert_eq!(error, None);
+            assert_eq!(
+                config.as_ref().unwrap().otlp_protocol,
+                Some(OtlpProtocol::HttpProtobuf)
+            );
+
+            // "grpc" → InvalidArgument
+            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);
+
+            // Garbage value → InvalidArgument
+            let mut config = Some(TraceExporterConfig::default());
+            let error = ddog_trace_exporter_config_set_otlp_protocol(
+                config.as_mut(),
+                CharSlice::from("nonsense"),
+            );
+            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidArgument);
+            ddog_trace_exporter_error_free(error);
+
+            // Non-UTF-8 input → InvalidInput
+            let mut config = Some(TraceExporterConfig::default());
+            let invalid: [u8; 2] = [0x80u8, 0xFFu8];
+            let error = ddog_trace_exporter_config_set_otlp_protocol(
+                config.as_mut(),
+                CharSlice::from_bytes(&invalid),
+            );
+            assert_eq!(error.as_ref().unwrap().code, ErrorCode::InvalidInput);
+            ddog_trace_exporter_error_free(error);
+        }
+    }
+
+    #[test]
+    fn set_otlp_protocol_stores_parsed_enum() {
+        use libdd_data_pipeline::OtlpProtocol;
+        let mut cfg = TraceExporterConfig::default();
+        let err = unsafe {
+            ddog_trace_exporter_config_set_otlp_protocol(
+                Some(&mut cfg),
+                CharSlice::from("http/protobuf"),
+            )
+        };
+        assert!(err.is_none());
+        assert_eq!(cfg.otlp_protocol, Some(OtlpProtocol::HttpProtobuf));
+    }
+
+    #[test]
+    fn set_otlp_protocol_rejects_grpc_and_unknown() {
+        let mut cfg = TraceExporterConfig::default();
+        for bad in ["grpc", "nonsense"] {
+            let err = unsafe {
+                ddog_trace_exporter_config_set_otlp_protocol(Some(&mut cfg), CharSlice::from(bad))
+            };
+            assert!(err.is_some(), "expected error for {bad}");
+            assert_eq!(cfg.otlp_protocol, None, "{bad} must not be stored");
+        }
+    }
+
     #[cfg(all(feature = "catch_panic", panic = "unwind"))]
     #[test]
     fn catch_panic_test() {

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

@@ -74,6 +74,7 @@ libdd-trace-utils = { path = "../libdd-trace-utils", features = [
     "test-utils",
 ] }
 httpmock = "0.8.0-alpha.1"
+prost = "0.14.1"
 rand = "0.8.5"
 tempfile = "3.3.0"
 tokio = { version = "1.23", features = [

libdd-data-pipeline/Cargo.toml

@@ -14,6 +14,9 @@
 pub mod agent_info;
 mod health_metrics;
 pub(crate) mod otlp;
+// `OtlpProtocol` (documented on the enum itself) is the only public symbol from the otherwise
+// crate-internal `otlp` module; re-exported here so the FFI crate can name it.
+pub use otlp::OtlpProtocol;
 #[cfg(feature = "telemetry")]
 pub(crate) mod telemetry;
 #[cfg(not(target_arch = "wasm32"))]

libdd-data-pipeline/src/lib.rs

@@ -6,18 +6,60 @@
 use http::HeaderMap;
 use std::time::Duration;
 
-/// OTLP trace export protocol. HTTP/JSON is currently supported.
+/// 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.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
-pub(crate) enum OtlpProtocol {
-    /// HTTP with JSON body (Content-Type: application/json). Default for HTTP.
+pub enum OtlpProtocol {
+    /// HTTP with a JSON body (`Content-Type: application/json`). The default.
     #[default]
     HttpJson,
-    /// HTTP with protobuf body. (Not supported yet)
-    #[allow(dead_code)]
+    /// HTTP with a protobuf body (`Content-Type: application/x-protobuf`).
     HttpProtobuf,
-    /// gRPC. (Not supported yet)
-    #[allow(dead_code)]
-    Grpc,
+}
+
+impl std::str::FromStr for OtlpProtocol {
+    type Err = String;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        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()),
+            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 {
+        match self {
+            OtlpProtocol::HttpJson => libdd_common::header::APPLICATION_JSON,
+            OtlpProtocol::HttpProtobuf => libdd_common::header::APPLICATION_PROTOBUF,
+        }
+    }
+
+    /// 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.
+    pub(crate) fn encode(
+        &self,
+        req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest,
+    ) -> 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))
+            }
+        }
+    }
 }
 
 /// Default timeout for OTLP export requests.
@@ -32,15 +74,50 @@ pub struct OtlpTraceConfig {
     pub headers: HeaderMap,
     /// Request timeout.
     pub timeout: Duration,
-    /// Protocol (for future use; currently only HttpJson is supported).
-    #[allow(dead_code)]
-    pub(crate) protocol: OtlpProtocol,
-    /// When `true`, does not add DD-specific per-span attributes to the OTLP payload.
-    //  These attributes are: (`service.name`, `operation.name`, `resource.name`,
-    //  `span.type`, `error.msg`, `error.message`, `span.kind`)
+    /// OTLP export protocol (selects body encoding and content-type).
+    pub protocol: OtlpProtocol,
+    /// When `true`, omit DD-specific per-span attributes (`service.name`, `operation.name`,
+    /// `resource.name`, `span.type`, `error.*`, `span.kind`) from the OTLP payload.
     pub otel_trace_semantics_enabled: bool,
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::str::FromStr;
+    #[test]
+    fn protocol_from_str() {
+        assert_eq!(
+            OtlpProtocol::from_str("http/json").unwrap(),
+            OtlpProtocol::HttpJson
+        );
+        assert_eq!(
+            OtlpProtocol::from_str("http/protobuf").unwrap(),
+            OtlpProtocol::HttpProtobuf
+        );
+        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
+        );
+        assert_eq!(
+            OtlpProtocol::HttpProtobuf.content_type(),
+            libdd_common::header::APPLICATION_PROTOBUF
+        );
+    }
+}
+
 /// Parsed OTLP trace-metrics exporter configuration.
 #[derive(Clone, Debug)]
 pub struct OtlpMetricsConfig {