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

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

@@ -9,6 +9,7 @@ use libdd_common_ffi::{
     CharSlice,
     {slice::AsBytes, slice::ByteSlice},
 };
+use libdd_data_pipeline::otlp::OtlpProtocol;
 use libdd_data_pipeline::trace_exporter::{
     TelemetryConfig, TelemetryInstrumentationSessions, TraceExporter as GenericTraceExporter,
     TraceExporterInputFormat, TraceExporterOutputFormat,
@@ -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,51 @@ 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. The OTLP trace
+            // exporter is HTTP-only, so we additionally reject `Grpc` here (it parses, but is
+            // unsupported) rather than storing a value the exporter would refuse at send time.
+            // The `_` arm also covers any future non_exhaustive variant.
+            match value.parse::<OtlpProtocol>() {
+                Ok(p @ (OtlpProtocol::HttpJson | OtlpProtocol::HttpProtobuf)) => {
+                    handle.otlp_protocol = Some(p);
+                    None
+                }
+                _ => 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 +606,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 +1327,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::otlp::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/Cargo.toml

@@ -73,6 +73,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/src/lib.rs

@@ -13,7 +13,7 @@
 
 pub mod agent_info;
 mod health_metrics;
-pub(crate) mod otlp;
+pub mod otlp;
 #[cfg(feature = "telemetry")]
 pub(crate) mod telemetry;
 #[cfg(not(target_arch = "wasm32"))]

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

@@ -6,20 +6,77 @@
 use http::HeaderMap;
 use std::time::Duration;
 
-/// OTLP trace export protocol. HTTP/JSON is currently supported.
+/// OTLP trace export protocol.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
-pub(crate) enum OtlpProtocol {
+#[non_exhaustive]
+pub enum OtlpProtocol {
     /// HTTP with JSON body (Content-Type: application/json). Default for HTTP.
     #[default]
     HttpJson,
-    /// HTTP with protobuf body. (Not supported yet)
-    #[allow(dead_code)]
+    /// HTTP with protobuf body (Content-Type: application/x-protobuf).
     HttpProtobuf,
-    /// gRPC. (Not supported yet)
-    #[allow(dead_code)]
+    /// gRPC. Parsed by `FromStr` so callers get a clean error, but rejected at export time
+    /// (unsupported).
     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" => Ok(OtlpProtocol::Grpc),
+            other => Err(format!("unknown OTLP protocol: {other}")),
+        }
+    }
+}
+
+/// The wire encoding actually used to send OTLP traces over HTTP. Internal, closed set: the
+/// only encodings the exporter supports. The user-facing [`OtlpProtocol`] (which also carries the
+/// unsupported `Grpc`) converts into this at the send boundary via `TryFrom`.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OtlpWireProtocol {
+    Json,
+    Protobuf,
+}
+
+impl std::convert::TryFrom<OtlpProtocol> for OtlpWireProtocol {
+    type Error = OtlpProtocol;
+    /// Maps the user-facing protocol to a supported wire encoding. `Grpc` is unsupported and
+    /// returns `Err(Grpc)` so the caller surfaces a clean error instead of silently downgrading.
+    fn try_from(p: OtlpProtocol) -> Result<Self, Self::Error> {
+        match p {
+            OtlpProtocol::HttpJson => Ok(OtlpWireProtocol::Json),
+            OtlpProtocol::HttpProtobuf => Ok(OtlpWireProtocol::Protobuf),
+            other => Err(other),
+        }
+    }
+}
+
+impl OtlpWireProtocol {
+    /// The HTTP `Content-Type` for this encoding.
+    pub fn content_type(&self) -> http::HeaderValue {
+        match self {
+            OtlpWireProtocol::Json => libdd_common::header::APPLICATION_JSON,
+            OtlpWireProtocol::Protobuf => libdd_common::header::APPLICATION_PROTOBUF,
+        }
+    }
+
+    /// Encode the prost OTLP request to this wire format.
+    pub fn encode(
+        &self,
+        req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest,
+    ) -> Result<Vec<u8>, serde_json::Error> {
+        match self {
+            OtlpWireProtocol::Json => libdd_trace_utils::otlp_encoder::encode_otlp_json(req),
+            OtlpWireProtocol::Protobuf => {
+                Ok(libdd_trace_utils::otlp_encoder::encode_otlp_protobuf(req))
+            }
+        }
+    }
+}
+
 /// Default timeout for OTLP export requests.
 pub const DEFAULT_OTLP_TIMEOUT: Duration = Duration::from_secs(10);
 
@@ -32,7 +89,51 @@ 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,
+    /// OTLP export protocol (selects body encoding and content-type).
+    pub protocol: OtlpProtocol,
+}
+
+#[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_eq!(OtlpProtocol::from_str("grpc").unwrap(), OtlpProtocol::Grpc);
+        assert!(OtlpProtocol::from_str("nonsense").is_err());
+    }
+
+    #[test]
+    fn wire_protocol_from_user_protocol() {
+        use std::convert::TryFrom;
+        assert_eq!(
+            OtlpWireProtocol::try_from(OtlpProtocol::HttpJson).unwrap(),
+            OtlpWireProtocol::Json
+        );
+        assert_eq!(
+            OtlpWireProtocol::try_from(OtlpProtocol::HttpProtobuf).unwrap(),
+            OtlpWireProtocol::Protobuf
+        );
+        assert!(OtlpWireProtocol::try_from(OtlpProtocol::Grpc).is_err());
+    }
+
+    #[test]
+    fn wire_protocol_content_types() {
+        assert_eq!(
+            OtlpWireProtocol::Json.content_type(),
+            libdd_common::header::APPLICATION_JSON
+        );
+        assert_eq!(
+            OtlpWireProtocol::Protobuf.content_type(),
+            libdd_common::header::APPLICATION_PROTOBUF
+        );
+    }
 }