From f0e93c2dceb5ab426de44849c2854d984d2939dd Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Fri, 24 Jul 2026 16:15:21 -0400 Subject: [PATCH 1/6] feat(data-pipeline)!: OTLP gRPC trace export Wires OTLP gRPC trace export into TraceExporter on top of the fork-safe gRPC transport, selectable via OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=grpc. OtlpProtocol gains a Grpc variant; the exporter dispatches through a new OtlpExportMode (Http | Grpc) to either the existing HTTP path or the gRPC transport (send_otlp_traces_grpc), with bounded exponential retry on transient IO matching the HTTP path. The OTLP resource info is built once at construction and shared by both paths. gRPC is native-only; wasm32 rejects it at build time. Includes a public-API end-to-end gRPC export test. BREAKING CHANGE: adds the Grpc variant to the exhaustive public OtlpProtocol enum, so exhaustive matches on it must add an arm. libdatadog consumers pin by version and pick this up on the next release. Co-Authored-By: Claude Opus 4.8 (1M context) --- libdd-data-pipeline-ffi/src/trace_exporter.rs | 28 +- libdd-data-pipeline/src/otlp/config.rs | 56 ++-- libdd-data-pipeline/src/otlp/exporter.rs | 9 +- libdd-data-pipeline/src/otlp/grpc_exporter.rs | 220 ++++++++++++---- libdd-data-pipeline/src/otlp/mod.rs | 10 +- .../src/trace_exporter/builder.rs | 227 ++++++++++++---- libdd-data-pipeline/src/trace_exporter/mod.rs | 112 ++++++-- .../tests/test_trace_exporter_otlp_grpc.rs | 248 ++++++++++++++++++ 8 files changed, 731 insertions(+), 179 deletions(-) create mode 100644 libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs index 6016c32dcb..3daf2d624b 100644 --- a/libdd-data-pipeline-ffi/src/trace_exporter.rs +++ b/libdd-data-pipeline-ffi/src/trace_exporter.rs @@ -495,11 +495,12 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_shared_runtime( ) } -/// Enables OTLP HTTP/JSON export and sets the endpoint URL. +/// Enables OTLP trace export and sets the endpoint URL. /// -/// When set, traces are sent to this URL in OTLP HTTP/JSON format instead of the Datadog -/// agent. The host language is responsible for resolving the endpoint from its configuration -/// (e.g. `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this function. +/// When set, traces are sent to this URL using the protocol selected by +/// `ddog_trace_exporter_config_set_otlp_protocol` instead of the Datadog agent. The host language +/// is responsible for resolving the endpoint from its configuration (e.g. +/// `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this function. #[no_mangle] pub unsafe extern "C" fn ddog_trace_exporter_config_set_otlp_endpoint( config: Option<&mut TraceExporterConfig>, @@ -519,8 +520,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 @@ -540,9 +541,6 @@ 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. match value.parse::() { Ok(p) => { handle.otlp_protocol = Some(p); @@ -1626,14 +1624,16 @@ mod tests { 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); + assert_eq!(error, None); + assert_eq!( + config.as_ref().unwrap().otlp_protocol, + Some(OtlpProtocol::Grpc) + ); // Garbage value → InvalidArgument let mut config = Some(TraceExporterConfig::default()); @@ -1714,9 +1714,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)) }; diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index 39dd54a38e..cf72f0add4 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -6,12 +6,7 @@ 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. @@ -19,6 +14,8 @@ pub enum OtlpProtocol { 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,34 @@ 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`]. + pub(crate) fn content_type(&self) -> Option { 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`]. pub(crate) fn encode( &self, req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest, - ) -> Result, serde_json::Error> { + ) -> Option, 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,8 +80,7 @@ pub struct OtlpTraceConfig { } /// Per-request OTLP gRPC trace exporter configuration. -// Not yet wired to the trace exporter's send loop; exercised by tests only. -#[allow(dead_code)] +#[cfg(not(target_arch = "wasm32"))] #[derive(Clone, Debug)] pub struct OtlpGrpcTraceConfig { /// Custom key-value pairs forwarded as gRPC request metadata. @@ -112,26 +105,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); } } diff --git a/libdd-data-pipeline/src/otlp/exporter.rs b/libdd-data-pipeline/src/otlp/exporter.rs index 1297a7a25e..103cc0f427 100644 --- a/libdd-data-pipeline/src/otlp/exporter.rs +++ b/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. @@ -93,13 +93,18 @@ pub async fn send_otlp_traces_http( test_token: Option<&str>, body: Vec, ) -> 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, ) diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs index ea96524204..a949871e3e 100644 --- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs +++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs @@ -10,11 +10,11 @@ use crate::otlp::config::OtlpGrpcTraceConfig; use crate::trace_exporter::error::{BuilderErrorKind, RequestError, TraceExporterError}; use bytes::Bytes; -use http_body_util::{BodyExt, Collected}; +use http_body_util::{BodyExt, Collected, Limited}; use hyper::client::conn::http2; use hyper_util::rt::{TokioExecutor, TokioIo}; use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::{ - ExportTraceServiceRequest, ExportTraceServiceResponse, + ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse, }; use std::error::Error as StdError; use std::future::Future; @@ -29,6 +29,7 @@ use tonic::{Code, Request, Status}; use tracing::warn; type BoxError = Box; +const MAX_GRPC_RESPONSE_SIZE: usize = 4 * 1024 * 1024; /// tonic 0.14 moved `ProstCodec` to the separate `tonic-prost` crate; we hand-roll a minimal /// codec here to avoid that extra dependency and keep tonic at `default-features = false`. @@ -172,7 +173,9 @@ impl GrpcService for H2Service { let request = async move { let resp = sender.send_request(req).await?; let (parts, incoming) = resp.into_parts(); - let collected = incoming.collect().await?; + let collected = Limited::new(incoming, MAX_GRPC_RESPONSE_SIZE) + .collect() + .await?; Ok::<_, BoxError>(http::Response::from_parts(parts, collected)) }; tokio::pin!(conn); @@ -210,8 +213,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, @@ -293,8 +294,6 @@ type ExportCodec = prost_codec::ProstCodecImpl; /// 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>, @@ -317,18 +316,23 @@ pub(crate) async fn send_otlp_traces_grpc( client.ready().await.map_err(|e| { TraceExporterError::Io(std::io::Error::other(format!("gRPC not ready: {e}"))) })?; - client + let response = client .unary(req, path, codec) .await - .map(|_resp| ()) - .map_err(grpc_status_to_error) + .map_err(grpc_status_to_error)?; + if let Some(details) = partial_success_details(response.get_ref()) { + warn!( + rejected_spans = details.rejected_spans, + error_message = %details.error_message, + "OTLP gRPC export was only partially accepted" + ); + } + Ok(()) }) .await .map_err(|_| TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)))? } -// Insert the pre-validated custom headers, the optional test-session token, and (when enabled) the -// client-computed-stats marker into the request metadata. fn attach_metadata( req: &mut Request, headers: &[(AsciiMetadataKey, AsciiMetadataValue)], @@ -357,6 +361,15 @@ fn attach_metadata( } } +fn partial_success_details( + response: &ExportTraceServiceResponse, +) -> Option<&ExportTracePartialSuccess> { + response + .partial_success + .as_ref() + .filter(|details| details.rejected_spans != 0 || !details.error_message.is_empty()) +} + fn grpc_status_to_error(status: Status) -> TraceExporterError { // Transport/IO failures from our bare `H2Service` fall through to `Code::Unknown` with the // original `std::io::Error` somewhere in the source chain (directly for a connect failure, or @@ -383,10 +396,29 @@ fn grpc_status_to_error(status: Status) -> TraceExporterError { std::io::ErrorKind::TimedOut, status.message(), )), - _ => TraceExporterError::Request(RequestError::new( - http::StatusCode::INTERNAL_SERVER_ERROR, - status.message(), - )), + Code::Cancelled | Code::Aborted | Code::OutOfRange | Code::DataLoss => { + TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionAborted, + status.message(), + )) + } + code => { + let http_status = match code { + Code::InvalidArgument => http::StatusCode::BAD_REQUEST, + Code::Unauthenticated => http::StatusCode::UNAUTHORIZED, + Code::PermissionDenied => http::StatusCode::FORBIDDEN, + Code::NotFound => http::StatusCode::NOT_FOUND, + Code::AlreadyExists => http::StatusCode::CONFLICT, + Code::ResourceExhausted => http::StatusCode::TOO_MANY_REQUESTS, + Code::FailedPrecondition => http::StatusCode::PRECONDITION_FAILED, + Code::Unimplemented => http::StatusCode::NOT_IMPLEMENTED, + _ => http::StatusCode::INTERNAL_SERVER_ERROR, + }; + TraceExporterError::Request(RequestError::new( + http_status, + &format!("gRPC {code:?}: {}", status.message()), + )) + } } } @@ -432,13 +464,11 @@ mod build_tests { let config = OtlpGrpcTraceConfig { headers: vec![ ("good-key".to_string(), "ok".to_string()), - // Invalid metadata key (contains a space): skipped, not fatal to the build. ("bad key".to_string(), "v".to_string()), ], timeout: Duration::from_secs(5), otel_trace_semantics_enabled: false, }; - // A malformed header must not fail the build; it is dropped and the valid one retained. let t = build_grpc_transport("http://localhost:4317", config).unwrap(); assert_eq!(t.metadata_headers.len(), 1); assert_eq!(t.metadata_headers[0].0.as_str(), "good-key"); @@ -467,7 +497,6 @@ mod integration_tests { #[cfg_attr(miri, ignore)] #[tokio::test] async fn connection_refused_maps_to_io() { - // Port 1 has no listener → connect fails. let transport = build_grpc_transport("http://127.0.0.1:1", cfg()).unwrap(); let err = send_otlp_traces_grpc( &transport, @@ -480,13 +509,70 @@ mod integration_tests { assert!(matches!(err, TraceExporterError::Io(_)), "got: {err:?}"); } - // Minimal in-process gRPC server: accepts one unary Export call, decodes the request, and - // replies OK with an empty ExportTraceServiceResponse plus a `grpc-status` trailer. + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn response_body_is_limited() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut conn = server::handshake(socket).await.unwrap(); + let (_req, mut respond) = conn.accept().await.unwrap().unwrap(); + let sender = tokio::spawn(async move { + let response = http::Response::builder().status(200).body(()).unwrap(); + let mut stream = respond.send_response(response, false).unwrap(); + let data = Bytes::from(vec![0; MAX_GRPC_RESPONSE_SIZE + 1]); + let mut offset = 0; + while offset < data.len() { + stream.reserve_capacity((data.len() - offset).min(16 * 1024)); + let Some(Ok(capacity)) = + std::future::poll_fn(|cx| stream.poll_capacity(cx)).await + else { + break; + }; + let end = (offset + capacity.min(data.len() - offset)).min(data.len()); + if stream + .send_data(data.slice(offset..end), end == data.len()) + .is_err() + { + break; + } + offset = end; + } + }); + while conn.accept().await.is_some() {} + sender.await.unwrap(); + }); + + let mut service = H2Service { + authority: Arc::from(addr.to_string()), + }; + let request = http::Request::builder() + .uri("http://localhost/") + .body(TonicBody::empty()) + .unwrap(); + let error = service.call(request).await.unwrap_err(); + server.await.unwrap(); + + let mut source: Option<&(dyn StdError + 'static)> = Some(error.as_ref()); + let mut length_limited = false; + while let Some(error) = source { + if error + .downcast_ref::() + .is_some() + { + length_limited = true; + break; + } + source = error.source(); + } + assert!(length_limited, "got: {error:?}"); + } + async fn run_one_shot_grpc_server(listener: TcpListener) -> ExportTraceServiceRequest { let (socket, _) = listener.accept().await.unwrap(); let mut conn = server::handshake(socket).await.unwrap(); let (req, mut respond) = conn.accept().await.unwrap().unwrap(); - // Drain the gRPC-framed request body: 1-byte compression flag + 4-byte length + message. let mut body = req.into_body(); let mut buf = Vec::new(); while let Some(chunk) = body.data().await { @@ -496,7 +582,6 @@ mod integration_tests { } let decoded = ExportTraceServiceRequest::decode(&buf[5..]).unwrap(); - // Respond: headers, one empty framed message, then grpc-status trailer. let resp = http::Response::builder() .status(200) .header("content-type", "application/grpc") @@ -506,15 +591,12 @@ mod integration_tests { let msg = ExportTraceServiceResponse::default(); let mut framed = vec![0u8; 5]; msg.encode(&mut framed).unwrap(); - let len = (framed.len() - 5) as u32; + let len = u32::try_from(framed.len() - 5).expect("response exceeds gRPC frame length"); framed[1..5].copy_from_slice(&len.to_be_bytes()); send.send_data(Bytes::from(framed), false).unwrap(); let mut trailers = http::HeaderMap::new(); trailers.insert("grpc-status", "0".parse().unwrap()); send.send_trailers(trailers).unwrap(); - // Keep the connection future alive (driving any remaining H2 protocol traffic, e.g. the - // client's final stream-closing frames) until the client drops its side and this - // `accept()` resolves to `None`, so the connection task doesn't outlive the test. let _ = tokio::time::timeout(Duration::from_secs(2), conn.accept()).await; decoded } @@ -543,8 +625,6 @@ mod integration_tests { async fn timeout_maps_to_io_timedout() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); - // Accept the connection but never drive the H2 handshake or send a response, so the send - // blocks until the timeout fires rather than completing or failing fast. let server = tokio::spawn(async move { let (socket, _) = listener.accept().await.unwrap(); tokio::time::sleep(Duration::from_secs(30)).await; @@ -579,28 +659,17 @@ mod integration_tests { async fn post_connect_transport_failure_maps_to_io() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); 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`. 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 - // non-zero linger blocks the thread on drop; a zero duration instead drops buffered - // data and sends the RST immediately without blocking, which is exactly what we want. #[allow(deprecated)] socket.set_linger(Some(Duration::ZERO)).unwrap(); let mut conn = server::handshake(socket).await.unwrap(); let (req, _respond) = conn.accept().await.unwrap().unwrap(); - // Drain the request body so the client is parked awaiting the response when we reset. let mut body = req.into_body(); while let Some(chunk) = body.data().await { let chunk = chunk.unwrap(); body.flow_control().release_capacity(chunk.len()).unwrap(); } - // Drop the connection (and its socket) -> RST. drop(conn); }); @@ -638,6 +707,22 @@ mod send_tests { Status::deadline_exceeded("slow"), std::io::ErrorKind::TimedOut, ), + ( + Status::new(Code::Cancelled, "canceled"), + std::io::ErrorKind::ConnectionAborted, + ), + ( + Status::new(Code::Aborted, "aborted"), + std::io::ErrorKind::ConnectionAborted, + ), + ( + Status::new(Code::OutOfRange, "out of range"), + std::io::ErrorKind::ConnectionAborted, + ), + ( + Status::new(Code::DataLoss, "data loss"), + std::io::ErrorKind::ConnectionAborted, + ), ] { match grpc_status_to_error(s) { TraceExporterError::Io(e) => assert_eq!(e.kind(), want), @@ -648,7 +733,6 @@ mod send_tests { #[test] fn status_unknown_with_io_source_maps_to_io() { - // Exercises the `Code::Unknown` + io-source recovery path in `grpc_status_to_error`. let io_err = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "refused"); let status = Status::from_error(Box::new(io_err)); assert_eq!(status.code(), Code::Unknown); @@ -670,10 +754,56 @@ mod send_tests { #[test] fn status_application_errors_map_to_request() { - assert!(matches!( - grpc_status_to_error(Status::new(Code::Internal, "boom")), - TraceExporterError::Request(_) - )); + for (code, http_status) in [ + (Code::InvalidArgument, http::StatusCode::BAD_REQUEST), + (Code::Unauthenticated, http::StatusCode::UNAUTHORIZED), + (Code::PermissionDenied, http::StatusCode::FORBIDDEN), + (Code::NotFound, http::StatusCode::NOT_FOUND), + (Code::AlreadyExists, http::StatusCode::CONFLICT), + (Code::ResourceExhausted, http::StatusCode::TOO_MANY_REQUESTS), + ( + Code::FailedPrecondition, + http::StatusCode::PRECONDITION_FAILED, + ), + (Code::Unimplemented, http::StatusCode::NOT_IMPLEMENTED), + (Code::Internal, http::StatusCode::INTERNAL_SERVER_ERROR), + ] { + match grpc_status_to_error(Status::new(code, "failure")) { + TraceExporterError::Request(error) => { + assert_eq!(error.status(), http_status); + assert_eq!(error.msg(), format!("gRPC {code:?}: failure")); + } + other => panic!("expected Request, got {other:?}"), + } + } + } + + #[test] + fn partial_success_details_ignores_empty_response() { + let empty = ExportTraceServiceResponse::default(); + assert!(partial_success_details(&empty).is_none()); + + let present_but_empty = ExportTraceServiceResponse { + partial_success: Some( + libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTracePartialSuccess::default(), + ), + }; + assert!(partial_success_details(&present_but_empty).is_none()); + } + + #[test] + fn partial_success_details_returns_rejections_and_warnings() { + let response = ExportTraceServiceResponse { + partial_success: Some( + libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTracePartialSuccess { + rejected_spans: 3, + error_message: "too many spans".to_string(), + }, + ), + }; + let details = partial_success_details(&response).unwrap(); + assert_eq!(details.rejected_spans, 3); + assert_eq!(details.error_message, "too many spans"); } #[test] diff --git a/libdd-data-pipeline/src/otlp/mod.rs b/libdd-data-pipeline/src/otlp/mod.rs index c124fa662e..738d14fe8a 100644 --- a/libdd-data-pipeline/src/otlp/mod.rs +++ b/libdd-data-pipeline/src/otlp/mod.rs @@ -5,8 +5,8 @@ //! //! When an OTLP endpoint is configured via //! [`crate::trace_exporter::TraceExporterBuilder::set_otlp_endpoint`], the trace exporter sends -//! traces in OTLP HTTP format to that endpoint instead of the Datadog agent; the wire encoding -//! (JSON or protobuf) is selected via [`OtlpProtocol`]. The host language is responsible for +//! traces in OTLP format to that endpoint instead of the Datadog agent; the transport and wire +//! encoding are selected via [`OtlpProtocol`]. The host language is responsible for //! resolving the endpoint from its own configuration (e.g. //! `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`). //! @@ -31,7 +31,6 @@ pub mod config; pub mod exporter; pub mod metrics; -// gRPC OTLP export depends on tonic/hyper, which do not build for wasm32. #[cfg(not(target_arch = "wasm32"))] pub mod grpc_exporter; @@ -39,3 +38,8 @@ 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; + +#[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}; diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 9eab89862e..2dc45b8565 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -4,6 +4,8 @@ use crate::agent_info::AgentInfoFetcher; use crate::agentless::config::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT}; use crate::otlp::config::{OtlpProtocol, DEFAULT_OTLP_TIMEOUT}; +#[cfg(not(target_arch = "wasm32"))] +use crate::otlp::{build_grpc_transport, OtlpGrpcTraceConfig}; use crate::otlp::{OtlpMetricsConfig, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] use crate::telemetry::TelemetryClientBuilder; @@ -16,8 +18,9 @@ use crate::trace_exporter::TelemetryConfig; use crate::trace_exporter::TelemetryInstrumentationSessions; use crate::trace_exporter::TraceExporterWorkers; use crate::trace_exporter::{ - add_path, StatsComputationStatus, TraceExporter, TraceExporterError, TraceExporterInputFormat, - TraceExporterOutputFormat, TraceSerializer, TracerMetadata, INFO_ENDPOINT, + add_path, OtlpExportMode, StatsComputationStatus, TraceExporter, TraceExporterError, + TraceExporterInputFormat, TraceExporterOutputFormat, TraceSerializer, TracerMetadata, + INFO_ENDPOINT, }; use arc_swap::ArcSwap; #[cfg(feature = "telemetry")] @@ -453,11 +456,12 @@ impl TraceExporterBuilder { self } - /// Enables OTLP HTTP/JSON export and sets the endpoint URL. + /// Enables OTLP trace export and sets the endpoint URL. /// - /// When set, traces are sent to this endpoint in OTLP HTTP/JSON format instead of the - /// Datadog agent. The host language is responsible for resolving the endpoint from its - /// configuration (e.g. `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this method. + /// When set, traces are sent to this endpoint using the protocol selected by + /// [`Self::set_otlp_protocol`] instead of the Datadog agent. The host language is responsible + /// for resolving the endpoint from its configuration (e.g. + /// `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`) before calling this method. /// /// OTLP trace export is mutually exclusive with agentless trace export /// ([`Self::set_agentless_endpoint`]); configuring both causes @@ -472,16 +476,19 @@ impl TraceExporterBuilder { self } - /// Selects the OTLP export protocol: [`OtlpProtocol::HttpJson`] (default) or - /// [`OtlpProtocol::HttpProtobuf`]. The host language resolves this from - /// `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` / `OTEL_EXPORTER_OTLP_PROTOCOL`; a `grpc` value is - /// unsupported and is rejected when parsed into [`OtlpProtocol`], so it never reaches here. + /// Selects the OTLP export protocol: [`OtlpProtocol::HttpJson`] (default), + /// [`OtlpProtocol::HttpProtobuf`], or [`OtlpProtocol::Grpc`]. The host language resolves this + /// from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL` / `OTEL_EXPORTER_OTLP_PROTOCOL`; all three OTel + /// protocol strings (`http/json`, `http/protobuf`, `grpc`) parse into [`OtlpProtocol`]. gRPC + /// export requires a plaintext `http://` endpoint: an `https://` gRPC endpoint is rejected at + /// [`build`](Self::build) time, and gRPC is not supported on wasm32 targets (also rejected at + /// build time). pub fn set_otlp_protocol(&mut self, protocol: OtlpProtocol) -> &mut Self { self.otlp_protocol = protocol; self } - /// Sets additional HTTP headers to include in OTLP trace export requests. + /// Sets additional headers or gRPC metadata to include in OTLP trace export requests. /// /// Headers should be provided as key-value pairs. The host language is responsible for /// resolving headers from its configuration (e.g. `OTEL_EXPORTER_OTLP_TRACES_HEADERS`) @@ -656,6 +663,31 @@ impl TraceExporterBuilder { self.validate_export_targets()?; + let otlp_timeout = self + .connection_timeout + .map(Duration::from_millis) + .unwrap_or(DEFAULT_OTLP_TIMEOUT); + #[cfg(not(target_arch = "wasm32"))] + let grpc_transport = match self.otlp_endpoint.as_deref() { + Some(url) if self.otlp_protocol == OtlpProtocol::Grpc => Some(build_grpc_transport( + url, + OtlpGrpcTraceConfig { + headers: self.otlp_headers.clone(), + timeout: otlp_timeout, + otel_trace_semantics_enabled: self.otel_trace_semantics_enabled, + }, + )?), + _ => None, + }; + #[cfg(target_arch = "wasm32")] + if self.otlp_endpoint.is_some() && self.otlp_protocol == OtlpProtocol::Grpc { + return Err(TraceExporterError::Builder( + BuilderErrorKind::InvalidConfiguration( + "OTLP gRPC export is not supported on wasm32 targets".to_string(), + ), + )); + } + let shared_runtime = match self.shared_runtime { Some(rt) => rt, None => Arc::new(R::new().map_err(|e| { @@ -795,22 +827,31 @@ impl TraceExporterBuilder { _ => None, }; - let otlp_timeout = self - .connection_timeout - .map(Duration::from_millis) - .unwrap_or(DEFAULT_OTLP_TIMEOUT); + let otlp_headers = self.otlp_headers; + let instrumentation_scope_name = self.instrumentation_scope_name; + let instrumentation_scope_version = self.instrumentation_scope_version; - // `self.otlp_protocol` is always an HTTP encoding here: gRPC is rejected at the parse - // boundary (`OtlpProtocol::from_str`) and so can never be constructed. - let otlp_config = otlp_endpoint.map(|url| OtlpTraceConfig { - endpoint_url: url, - headers: build_otlp_header_map(self.otlp_headers), - timeout: otlp_timeout, - protocol: self.otlp_protocol, - instrumentation_scope_name: self.instrumentation_scope_name, - instrumentation_scope_version: self.instrumentation_scope_version, - otel_trace_semantics_enabled: self.otel_trace_semantics_enabled, - }); + let otlp = match otlp_endpoint { + #[cfg(not(target_arch = "wasm32"))] + Some(_) if self.otlp_protocol == OtlpProtocol::Grpc => { + let transport = grpc_transport.ok_or_else(|| { + TraceExporterError::Builder(BuilderErrorKind::InvalidConfiguration( + "OTLP gRPC transport was not initialized".to_string(), + )) + })?; + Some(OtlpExportMode::Grpc(transport)) + } + Some(url) => Some(OtlpExportMode::Http(OtlpTraceConfig { + endpoint_url: url, + headers: build_otlp_header_map(otlp_headers), + timeout: otlp_timeout, + protocol: self.otlp_protocol, + instrumentation_scope_name: instrumentation_scope_name.clone(), + instrumentation_scope_version: instrumentation_scope_version.clone(), + otel_trace_semantics_enabled: self.otel_trace_semantics_enabled, + })), + None => None, + }; let otlp_metrics_config = self.otlp_metrics_endpoint.map(|url| OtlpMetricsConfig { endpoint_url: url, @@ -824,6 +865,17 @@ impl TraceExporterBuilder { .runtime_id .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); + let base_otlp_resource = |rid: &str| { + let mut r = OtlpResourceInfo::default(); + r.service = self.service.clone(); + r.env = self.env.clone(); + r.app_version = self.app_version.clone(); + r.language = self.language.clone(); + r.tracer_version = self.tracer_version.clone(); + r.runtime_id = rid.to_string(); + r + }; + // OTLP metrics + stats bucket size: start the concentrator unconditionally (bypass the // agent gate) so `check_agent_info` cannot later disable stats. let mut otlp_stats_enabled = false; @@ -847,13 +899,7 @@ impl TraceExporterBuilder { #[cfg(feature = "stats-obfuscation")] None, ))); - let mut resource = OtlpResourceInfo::default(); - resource.service = self.service.clone(); - resource.env = self.env.clone(); - resource.app_version = self.app_version.clone(); - resource.language = self.language.clone(); - resource.tracer_version = self.tracer_version.clone(); - resource.runtime_id = runtime_id.clone(); + let mut resource = base_otlp_resource(&runtime_id); resource.hostname = self.hostname.clone(); resource.process_tags = self.process_tags.clone(); resource.tracer_tags = self.tracer_tags.clone(); @@ -1002,6 +1048,25 @@ impl TraceExporterBuilder { }; } + let otlp_resource_info = if let Some(mode) = otlp.as_ref() { + let mut r = base_otlp_resource(&runtime_id); + r.client_computed_stats = self.client_computed_stats || otlp_stats_enabled; + match mode { + OtlpExportMode::Http(config) => { + r.instrumentation_scope_name = config.instrumentation_scope_name.clone(); + r.instrumentation_scope_version = config.instrumentation_scope_version.clone(); + } + #[cfg(not(target_arch = "wasm32"))] + OtlpExportMode::Grpc(_) => { + r.instrumentation_scope_name = instrumentation_scope_name; + r.instrumentation_scope_version = instrumentation_scope_version; + } + } + r + } else { + OtlpResourceInfo::default() + }; + let log_output = self .output_to_log .then(|| self.log_max_line_size.unwrap_or(DEFAULT_LOG_MAX_LINE_SIZE)); @@ -1067,7 +1132,8 @@ impl TraceExporterBuilder { agent_payload_response_version: self .agent_rates_payload_version_enabled .then(AgentResponsePayloadVersion::new), - otlp_config, + otlp, + otlp_resource_info, agentless_config, trace_filterer: ArcSwap::from_pointee(TraceFilterer::with_empty_conf()), otlp_stats_enabled, @@ -1077,23 +1143,6 @@ impl TraceExporterBuilder { } /// Reject configurations that combine mutually exclusive trace export targets. - /// - /// Trace export uses exactly one of three transports: - /// - the Datadog Agent (via [`Self::set_url`], the default when no transport is set), - /// - an OTLP HTTP/JSON endpoint (via [`Self::set_otlp_endpoint`]), or - /// - the agentless intake (via [`Self::set_agentless_endpoint`]). - /// - /// Exclusion rules enforced here: - /// - OTLP and agentless cannot both be configured. - /// - Agentless cannot be combined with a caller-supplied agent URL. - /// - Log output cannot be combined with OTLP or agentless trace export. - /// - [`Self::set_agentless_timeout`] requires [`Self::set_agentless_endpoint`]. - /// - [`Self::set_agentless_stats_endpoint`] requires agentless trace export - /// ([`Self::set_agentless_endpoint`]) and is incompatible with OTLP stats - /// ([`Self::set_otlp_metrics_endpoint`]). - /// - /// OTLP and an agent URL may coexist: the agent URL is still useful for auxiliary - /// agent endpoints (info, stats) even when trace payloads are routed to OTLP. fn validate_export_targets(&self) -> Result<(), TraceExporterError> { let otlp_set = self.otlp_endpoint.is_some(); let agentless_set = self.agentless_endpoint.is_some(); @@ -1230,7 +1279,9 @@ impl TraceExporterBuilder { #[cfg(test)] mod tests { use super::*; + use crate::agent_info; use crate::trace_exporter::error::BuilderErrorKind; + use httpmock::prelude::*; use libdd_capabilities_impl::NativeCapabilities; use libdd_shared_runtime::ForkSafeRuntime; @@ -1307,7 +1358,13 @@ mod tests { assert_eq!(exporter.metadata.language_interpreter_vendor, "node"); assert_eq!(exporter.metadata.git_commit_sha, "797e9ea"); assert!(exporter.metadata.client_computed_stats); - let otlp_config = exporter.otlp_config.as_ref().unwrap(); + #[cfg(not(target_arch = "wasm32"))] + let otlp_config = match exporter.otlp.as_ref().unwrap() { + OtlpExportMode::Http(c) => c, + OtlpExportMode::Grpc(_) => panic!("expected HTTP OTLP mode"), + }; + #[cfg(target_arch = "wasm32")] + let OtlpExportMode::Http(otlp_config) = exporter.otlp.as_ref().unwrap(); assert_eq!(otlp_config.instrumentation_scope_name, "dd-trace-js"); assert_eq!(otlp_config.instrumentation_scope_version, "7.0.0-pre"); assert!(!exporter.restart_after_fork); @@ -1529,6 +1586,72 @@ mod tests { assert!(builder.build::().is_ok()); } + #[cfg_attr(miri, ignore)] + #[test] + fn build_with_grpc_protocol_and_endpoint_succeeds() { + let mut builder = TraceExporterBuilder::default(); + builder + .set_otlp_endpoint("http://localhost:4317") + .set_otlp_protocol(OtlpProtocol::Grpc) + .set_otlp_instrumentation_scope("dd-trace-js", "7.0.0-pre"); + let exporter = builder.build::().unwrap(); + assert!(matches!(exporter.otlp, Some(OtlpExportMode::Grpc(_)))); + assert_eq!( + exporter.otlp_resource_info.instrumentation_scope_name, + "dd-trace-js" + ); + assert_eq!( + exporter.otlp_resource_info.instrumentation_scope_version, + "7.0.0-pre" + ); + } + + #[cfg_attr(miri, ignore)] + #[test] + fn invalid_grpc_endpoint_does_not_start_info_worker() { + agent_info::clear_cache_for_test(); + let server = MockServer::start(); + let info = server.mock(|when, then| { + when.method(GET).path("/info"); + then.status(200).body("{}"); + }); + let shared_runtime = Arc::new(ForkSafeRuntime::new().unwrap()); + let mut builder = TraceExporterBuilder::default(); + builder + .set_shared_runtime(shared_runtime.clone()) + .set_url(&server.base_url()) + .set_otlp_endpoint("https://localhost:4317") + .set_otlp_protocol(OtlpProtocol::Grpc); + + assert!(builder.build::().is_err()); + for _ in 0..50 { + if info.calls() > 0 { + break; + } + std::thread::sleep(Duration::from_millis(10)); + } + assert_eq!(info.calls(), 0); + } + + #[cfg_attr(miri, ignore)] + #[test] + fn build_with_grpc_protocol_no_endpoint_uses_agent_path() { + let mut builder = TraceExporterBuilder::default(); + builder.set_otlp_protocol(OtlpProtocol::Grpc); + let exporter = builder.build::().unwrap(); + assert!(exporter.otlp.is_none()); + } + + #[cfg_attr(miri, ignore)] + #[test] + fn build_with_grpc_https_endpoint_rejected() { + let mut builder = TraceExporterBuilder::default(); + builder + .set_otlp_endpoint("https://localhost:4317") + .set_otlp_protocol(OtlpProtocol::Grpc); + assert!(builder.build::().is_err()); + } + #[cfg_attr(miri, ignore)] #[test] fn test_build_with_v1_starts_inactive() { diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 83c8b84048..d94dce746c 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -20,6 +20,11 @@ use self::trace_serializer::TraceSerializer; use crate::agent_info::ResponseObserver; use crate::agentless::exporter::send_agentless_traces; use crate::agentless::AgentlessTraceConfig; +#[cfg(not(target_arch = "wasm32"))] +use crate::otlp::{ + exporter::{OTLP_MAX_RETRIES, OTLP_RETRY_DELAY_MS}, + send_otlp_traces_grpc, OtlpGrpcTransport, +}; use crate::otlp::{map_traces_to_otlp, send_otlp_traces_http, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] use crate::telemetry::{SendPayloadTelemetry, TelemetryClient}; @@ -145,6 +150,14 @@ fn add_path(url: &Uri, path: &str) -> Uri { pub use libdd_trace_utils::tracer_metadata::TracerMetadata; +/// The transport used for OTLP trace export. +#[derive(Debug)] +pub(crate) enum OtlpExportMode { + Http(OtlpTraceConfig), + #[cfg(not(target_arch = "wasm32"))] + Grpc(OtlpGrpcTransport), +} + /// Handles for the background workers owned by a [`TraceExporter`]. #[derive(Debug)] pub(crate) struct TraceExporterWorkers { @@ -225,8 +238,10 @@ pub struct TraceExporter< capabilities: C, workers: TraceExporterWorkers, agent_payload_response_version: Option, - /// When set, traces are exported via OTLP HTTP/JSON instead of the Datadog agent. - otlp_config: Option, + /// When set, traces are exported via OTLP instead of the Datadog agent. + otlp: Option, + /// OTLP Resource attributes derived from tracer metadata. + otlp_resource_info: OtlpResourceInfo, /// When set, APM trace spans are exported directly to the Datadog HTTP intake (agentless) /// instead of via the Datadog Agent agentless_config: Option, @@ -637,31 +652,25 @@ impl< traces: Vec>>, config: &OtlpTraceConfig, ) -> Result { - let resource_info = { - let mut r = OtlpResourceInfo::default(); - r.service = self.metadata.service.clone(); - r.env = self.metadata.env.clone(); - r.app_version = self.metadata.app_version.clone(); - r.language = self.metadata.language.clone(); - r.tracer_version = self.metadata.tracer_version.clone(); - r.runtime_id = self.metadata.runtime_id.clone(); - r.client_computed_stats = - self.metadata.client_computed_stats || self.otlp_stats_enabled; - r.instrumentation_scope_name = config.instrumentation_scope_name.clone(); - r.instrumentation_scope_version = config.instrumentation_scope_version.clone(); - r - }; - // Single prost OTLP IR; the configured protocol encodes the same request to its wire - // format (JSON or protobuf). OTel-semantics gating (omit DD-specific attrs) happens in - // the mapper. - let request = - map_traces_to_otlp(traces, &resource_info, config.otel_trace_semantics_enabled); - let body = config.protocol.encode(&request).map_err(|e| { - error!("OTLP serialization error: {e}"); - TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!( - "failed to encode OTLP request: {e}" - ))) - })?; + let request = map_traces_to_otlp( + traces, + &self.otlp_resource_info, + config.otel_trace_semantics_enabled, + ); + let body = config + .protocol + .encode(&request) + .ok_or_else(|| { + TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState( + "OTLP gRPC protocol cannot be encoded on the HTTP export path".to_string(), + )) + })? + .map_err(|e| { + error!("OTLP serialization error: {e}"); + TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!( + "failed to encode OTLP request: {e}" + ))) + })?; // Also set the header: resource attributes survive Collector hops, headers don't. let effective_config; let config_to_use = if self.metadata.client_computed_stats || self.otlp_stats_enabled { @@ -687,6 +696,45 @@ impl< Ok(AgentResponse::Unchanged) } + /// Sends trace chunks via OTLP gRPC. + #[cfg(not(target_arch = "wasm32"))] + async fn send_otlp_grpc_inner( + &self, + traces: Vec>>, + transport: &OtlpGrpcTransport, + ) -> Result { + let request = map_traces_to_otlp( + traces, + &self.otlp_resource_info, + transport.config.otel_trace_semantics_enabled, + ); + let test_token = self.endpoint.test_token.as_deref(); + let mut attempt: u32 = 1; + loop { + match send_otlp_traces_grpc( + transport, + test_token, + self.metadata.client_computed_stats || self.otlp_stats_enabled, + request.clone(), + ) + .await + { + Ok(()) => return Ok(AgentResponse::Unchanged), + Err(TraceExporterError::Io(e)) => { + if attempt > OTLP_MAX_RETRIES { + return Err(TraceExporterError::Io(e)); + } + let delay_ms = OTLP_RETRY_DELAY_MS * 2u64.pow(attempt - 1); + self.capabilities + .sleep(Duration::from_millis(delay_ms)) + .await; + attempt += 1; + } + Err(e) => return Err(e), + } + } + } + /// Send traces payload to agent with retry and telemetry reporting async fn send_traces_with_telemetry( &self, @@ -788,12 +836,18 @@ impl< // OTLP path: send sampled traces via OTLP when an OTLP endpoint is configured. // Unlike the agent path, there is no downstream agent to drop unsampled traces, // so drop_chunks is always called here regardless of whether stats are enabled. - if let Some(ref config) = self.otlp_config { + if let Some(otlp) = &self.otlp { libdd_trace_utils::span::trace_utils::drop_chunks(&mut traces); if traces.is_empty() { return Ok(AgentResponse::Unchanged); } - return self.send_otlp_traces_inner(traces, config).await; + return match otlp { + OtlpExportMode::Http(config) => self.send_otlp_traces_inner(traces, config).await, + #[cfg(not(target_arch = "wasm32"))] + OtlpExportMode::Grpc(transport) => { + self.send_otlp_grpc_inner(traces, transport).await + } + }; } // Snapshot the effective format once so the serializer and the URL agree even if diff --git a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs new file mode 100644 index 0000000000..057d0950b9 --- /dev/null +++ b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs @@ -0,0 +1,248 @@ +// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 +#[cfg(all(test, not(target_arch = "wasm32")))] +mod grpc_export_tests { + use bytes::Bytes; + use h2::server; + use libdd_capabilities_impl::NativeCapabilities; + use libdd_data_pipeline::{trace_exporter::TraceExporterBuilder, OtlpProtocol}; + use libdd_shared_runtime::{ForkSafeRuntime, SharedRuntime}; + use libdd_trace_protobuf::opentelemetry::proto::{ + collector::trace::v1::{ExportTraceServiceRequest, ExportTraceServiceResponse}, + common::v1::any_value::Value, + }; + use libdd_trace_utils::test_utils::create_test_json_span; + use prost::Message; + use serde_json::json; + use std::sync::{mpsc, Arc}; + use std::time::Duration; + use tokio::net::TcpListener; + use tokio::sync::oneshot; + use tokio::task::JoinSet; + + struct ReceivedExport { + path: String, + client_computed_stats: Option, + request: ExportTraceServiceRequest, + } + + async fn run_grpc_test_server( + listener: TcpListener, + req_tx: mpsc::Sender, + mut shutdown: oneshot::Receiver<()>, + ) { + let mut connections = JoinSet::new(); + loop { + tokio::select! { + _ = &mut shutdown => break, + accepted = listener.accept() => { + let Ok((socket, _)) = accepted else { return }; + let connection_req_tx = req_tx.clone(); + connections.spawn(async move { + let Ok(mut connection) = server::handshake(socket).await else { + return; + }; + let mut handlers = JoinSet::new(); + while let Some(result) = connection.accept().await { + if let Ok((request, respond)) = result { + handlers.spawn(handle_export_stream( + request, + respond, + connection_req_tx.clone(), + )); + } + } + while let Some(result) = handlers.join_next().await { + result.expect("gRPC request handler failed"); + } + }); + } + } + } + connections.abort_all(); + while let Some(result) = connections.join_next().await { + if let Err(error) = result { + assert!(error.is_cancelled(), "gRPC connection task failed: {error}"); + } + } + } + + async fn handle_export_stream( + request: http::Request, + mut respond: h2::server::SendResponse, + req_tx: mpsc::Sender, + ) { + let path = request.uri().path().to_string(); + let client_computed_stats = request + .headers() + .get("datadog-client-computed-stats") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + let mut body = request.into_body(); + let mut frame_data: Vec = Vec::new(); + while let Some(chunk) = body.data().await { + let Ok(chunk) = chunk else { return }; + let len = chunk.len(); + frame_data.extend_from_slice(&chunk); + body.flow_control().release_capacity(len).ok(); + } + + let decoded = if frame_data.len() > 5 { + ExportTraceServiceRequest::decode(&frame_data[5..]).ok() + } else { + None + }; + if let Some(req) = decoded { + let _ = req_tx.send(ReceivedExport { + path, + client_computed_stats, + request: req, + }); + } + + let response_proto = ExportTraceServiceResponse::default(); + let proto_bytes = response_proto.encode_to_vec(); + let mut frame = Vec::with_capacity(5 + proto_bytes.len()); + frame.push(0u8); + frame.extend_from_slice( + &u32::try_from(proto_bytes.len()) + .expect("protobuf response exceeds the gRPC frame length") + .to_be_bytes(), + ); + frame.extend_from_slice(&proto_bytes); + + let response = http::Response::builder() + .status(200) + .header("content-type", "application/grpc") + .body(()) + .unwrap(); + let Ok(mut send_stream) = respond.send_response(response, false) else { + return; + }; + let _ = send_stream.send_data(Bytes::from(frame), false); + + let mut trailers = http::HeaderMap::new(); + trailers.insert("grpc-status", "0".parse().unwrap()); + let _ = send_stream.send_trailers(trailers); + } + + fn resource_attribute<'a>( + request: &'a ExportTraceServiceRequest, + key: &str, + ) -> Option<&'a str> { + request + .resource_spans + .first() + .and_then(|rs| rs.resource.as_ref()) + .and_then(|r| { + r.attributes.iter().find_map(|kv| { + if kv.key == key { + kv.value + .as_ref() + .and_then(|v| v.value.as_ref()) + .and_then(|v| match v { + Value::StringValue(s) => Some(s.as_str()), + _ => None, + }) + } else { + None + } + }) + }) + } + + fn run_grpc_export_end_to_end_and_survives_shared_runtime_restart<'scope, 'env: 'scope>( + scope: &'scope std::thread::Scope<'scope, 'env>, + ) { + let (port_tx, port_rx) = mpsc::channel::(); + let (req_tx, req_rx) = mpsc::channel::(); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + + let server = scope.spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + rt.block_on(async move { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + port_tx.send(listener.local_addr().unwrap().port()).unwrap(); + run_grpc_test_server(listener, req_tx, shutdown_rx).await; + }); + }); + + let port = port_rx + .recv_timeout(Duration::from_secs(10)) + .expect("server did not bind within 10s"); + let endpoint = format!("http://127.0.0.1:{port}/otel/"); + let expected_path = "/otel/opentelemetry.proto.collector.trace.v1.TraceService/Export"; + + let shared_runtime = Arc::new(ForkSafeRuntime::new().expect("build shared runtime")); + + let mut builder = TraceExporterBuilder::default(); + builder + .set_shared_runtime(shared_runtime.clone()) + .set_otlp_endpoint(&endpoint) + .set_otlp_protocol(OtlpProtocol::Grpc) + .set_connection_timeout(Some(30_000)) + .set_language("test-lang") + .set_tracer_version("1.0") + .set_env("grpc-test-env") + .set_service("grpc-test-svc") + .set_client_computed_stats(); + + let exporter = builder + .build::() + .expect("build exporter"); + + let mut span = create_test_json_span(1234, 12342, 12341, 1, false); + span["service"] = json!("grpc-test-svc"); + span["name"] = json!("grpc_span"); + let data = rmp_serde::to_vec_named(&vec![vec![span]]).unwrap(); + + exporter.send(data.as_ref()).expect("initial send ok"); + let initial = req_rx + .recv_timeout(Duration::from_secs(10)) + .expect("server did not receive the initial request"); + assert_eq!(initial.path, expected_path); + assert!( + !initial.request.resource_spans.is_empty(), + "expected at least one ResourceSpans" + ); + assert_eq!( + resource_attribute(&initial.request, "service.name"), + Some("grpc-test-svc"), + "service.name attribute not found or wrong value" + ); + assert_eq!(initial.client_computed_stats.as_deref(), Some("yes")); + assert_eq!( + resource_attribute(&initial.request, "_dd.stats_computed"), + Some("true") + ); + + shared_runtime.before_fork(); + shared_runtime + .after_fork_parent() + .expect("restart shared runtime"); + + exporter.send(data.as_ref()).expect("post-restart send ok"); + let after_restart = req_rx + .recv_timeout(Duration::from_secs(10)) + .expect("server did not receive the post-restart request"); + assert_eq!(after_restart.path, expected_path); + assert_eq!( + resource_attribute(&after_restart.request, "service.name"), + Some("grpc-test-svc"), + "service.name attribute not found or wrong value after runtime restart" + ); + shutdown_tx.send(()).expect("server stopped early"); + server.join().expect("server thread failed"); + } + + #[cfg_attr(miri, ignore)] + #[test] + fn grpc_export_end_to_end_and_survives_shared_runtime_restart() { + std::thread::scope(|scope| { + run_grpc_export_end_to_end_and_survives_shared_runtime_restart(scope) + }); + } +} From 061c9074fcb64b9c6c5fceca9d24b9aa060dfe1c Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Tue, 1 Sep 2026 17:39:18 -0400 Subject: [PATCH 2/6] fix(data-pipeline): harden OTLP gRPC retries and dialing Default portless HTTP endpoints to port 80 while rejecting ambiguous authorities. Decode google.rpc.RetryInfo locally to avoid another tonic dependency, retry recoverable resource exhaustion with capped exponential delays, and share protobuf requests across attempts. --- .github/CODEOWNERS | 2 +- libdd-data-pipeline/src/otlp/grpc_exporter.rs | 355 +++++++++++++++--- libdd-data-pipeline/src/otlp/mod.rs | 4 +- libdd-data-pipeline/src/trace_exporter/mod.rs | 55 ++- 4 files changed, 353 insertions(+), 63 deletions(-) diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS index 9327cd1d02..fb0b8481ac 100644 --- a/.github/CODEOWNERS +++ b/.github/CODEOWNERS @@ -110,8 +110,8 @@ tools/cc_utils/ @DataDog/libdatadog-php tools/sidecar_mockgen/ @DataDog/libdatadog-php libdd-data-pipeline/src/otlp/ @DataDog/apm-sdk-capabilities-rust libdd-data-pipeline/tests/test_trace_exporter_otlp_export.rs @DataDog/apm-sdk-capabilities-rust +libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs @DataDog/apm-sdk-capabilities-rust libdd-trace-utils/src/otlp_encoder/ @DataDog/apm-sdk-capabilities-rust datadog-sidecar/src/service/ffe_exposures_flusher.rs @DataDog/libdatadog-php @DataDog/libdatadog-apm @DataDog/feature-flagging-and-experimentation-sdk datadog-sidecar/src/service/ffe_metrics_flusher.rs @DataDog/libdatadog-php @DataDog/libdatadog-apm @DataDog/feature-flagging-and-experimentation-sdk .github/workflows/nix.yml @DataDog/nix-guild @DataDog/apm-common-components-core - diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs index a949871e3e..c4bba4aa5b 100644 --- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs +++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs @@ -16,11 +16,13 @@ use hyper_util::rt::{TokioExecutor, TokioIo}; use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::{ ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse, }; +use prost::Message as _; use std::error::Error as StdError; use std::future::Future; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; +use std::time::Duration; use tokio::net::TcpStream; use tonic::body::Body as TonicBody; use tonic::client::{Grpc, GrpcService}; @@ -30,12 +32,51 @@ use tracing::warn; type BoxError = Box; const MAX_GRPC_RESPONSE_SIZE: usize = 4 * 1024 * 1024; +const RETRY_INFO_TYPE_URL: &str = "type.googleapis.com/google.rpc.RetryInfo"; + +#[derive(Debug)] +pub(crate) enum GrpcExportError { + Retryable { + error: TraceExporterError, + retry_after: Option, + }, + NonRetryable(TraceExporterError), +} + +#[derive(Clone, PartialEq, prost::Message)] +struct RpcStatusDetails { + #[prost(message, repeated, tag = "3")] + details: Vec, +} + +#[derive(Clone, PartialEq, prost::Message)] +struct RpcStatusDetail { + #[prost(string, tag = "1")] + type_url: String, + #[prost(bytes = "vec", tag = "2")] + value: Vec, +} + +#[derive(Clone, Copy, PartialEq, prost::Message)] +struct RetryInfo { + #[prost(message, optional, tag = "1")] + retry_delay: Option, +} + +#[derive(Clone, Copy, PartialEq, prost::Message)] +struct ProtoDuration { + #[prost(int64, tag = "1")] + seconds: i64, + #[prost(int32, tag = "2")] + nanos: i32, +} /// tonic 0.14 moved `ProstCodec` to the separate `tonic-prost` crate; we hand-roll a minimal /// codec here to avoid that extra dependency and keep tonic at `default-features = false`. pub(crate) mod prost_codec { use prost::Message as ProstMessage; use std::marker::PhantomData; + use std::sync::Arc; use tonic::codec::{Codec, DecodeBuf, Decoder, EncodeBuf, Encoder}; use tonic::Status; @@ -54,15 +95,16 @@ pub(crate) mod prost_codec { // Shared with the `Encoder` impl below. tonic's `EncodeBuf`/`DecodeBuf` constructors are // private to the crate, so tests exercise this generic-over-`BufMut`/`Buf` core directly // instead of going through the `Encoder`/`Decoder` traits (see `codec_tests`). - fn encode_into(item: T, dst: &mut impl bytes::BufMut) -> Result<(), Status> { - item.encode(dst) + fn encode_into(item: Arc, dst: &mut impl bytes::BufMut) -> Result<(), Status> { + item.as_ref() + .encode(dst) .map_err(|e| Status::internal(format!("Failed to encode protobuf message: {e}"))) } } - impl Encoder for ProstEncoder { - type Item = T; + impl Encoder for ProstEncoder { + type Item = Arc; type Error = Status; - fn encode(&mut self, item: T, dst: &mut EncodeBuf<'_>) -> Result<(), Status> { + fn encode(&mut self, item: Arc, dst: &mut EncodeBuf<'_>) -> Result<(), Status> { Self::encode_into(item, dst) } } @@ -90,10 +132,10 @@ pub(crate) mod prost_codec { impl Codec for ProstCodecImpl where - Enc: ProstMessage + Default + Send + 'static, + Enc: ProstMessage + Default + Send + Sync + 'static, Dec: ProstMessage + Default + Send + 'static, { - type Encode = Enc; + type Encode = Arc; type Decode = Dec; type Encoder = ProstEncoder; type Decoder = ProstDecoder; @@ -113,6 +155,7 @@ pub(crate) mod prost_codec { ExportTraceServiceRequest, ExportTraceServiceResponse, }; use libdd_trace_protobuf::opentelemetry::proto::trace::v1::ResourceSpans; + use std::sync::Arc; // Round-trips through the `BufMut`/`Buf`-generic core (`encode_into`/`decode_from`) that // the `Encoder`/`Decoder` impls delegate to, over a plain `BytesMut`; see @@ -127,7 +170,9 @@ pub(crate) mod prost_codec { }], }; let mut buf = BytesMut::new(); - ProstEncoder::encode_into(msg.clone(), &mut buf).unwrap(); + let shared_msg = Arc::new(msg.clone()); + ProstEncoder::encode_into(shared_msg.clone(), &mut buf).unwrap(); + assert_eq!(Arc::strong_count(&shared_msg), 1); assert!(!buf.is_empty()); let out = ProstDecoder::decode_from(&mut buf).unwrap(); @@ -245,8 +290,23 @@ pub(crate) fn build_grpc_transport( "gRPC endpoint must include an authority".to_string(), )) })?; + let authority_text = authority.as_str(); + if authority_text.contains('@') { + return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri( + "gRPC endpoint authority must not include userinfo".to_string(), + ))); + } + let dial_authority = match authority.port_u16() { + Some(_) => authority_text.to_string(), + None if authority_text == authority.host() => format!("{}:80", authority.host()), + None => { + return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri( + "gRPC endpoint authority contains an invalid port".to_string(), + ))); + } + }; let service = H2Service { - authority: Arc::from(authority.as_str()), + authority: Arc::from(dial_authority), }; // Origin = scheme+authority + normalized path prefix; tonic appends the RPC method path. @@ -298,8 +358,8 @@ pub(crate) async fn send_otlp_traces_grpc( transport: &OtlpGrpcTransport, test_token: Option<&str>, client_computed_stats: bool, - request: ExportTraceServiceRequest, -) -> Result<(), TraceExporterError> { + request: Arc, +) -> Result<(), GrpcExportError> { let mut req = Request::new(request); attach_metadata( &mut req, @@ -313,9 +373,15 @@ pub(crate) async fn send_otlp_traces_grpc( tokio::time::timeout(transport.config.timeout, async { let mut client = Grpc::with_origin(transport.service.clone(), transport.origin.clone()); - client.ready().await.map_err(|e| { - TraceExporterError::Io(std::io::Error::other(format!("gRPC not ready: {e}"))) - })?; + client + .ready() + .await + .map_err(|e| GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::other(format!( + "gRPC not ready: {e}" + ))), + retry_after: None, + })?; let response = client .unary(req, path, codec) .await @@ -330,11 +396,14 @@ pub(crate) async fn send_otlp_traces_grpc( Ok(()) }) .await - .map_err(|_| TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)))? + .map_err(|_| GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)), + retry_after: None, + })? } -fn attach_metadata( - req: &mut Request, +fn attach_metadata( + req: &mut Request, headers: &[(AsciiMetadataKey, AsciiMetadataValue)], test_token: Option<&str>, client_computed_stats: bool, @@ -370,7 +439,9 @@ fn partial_success_details( .filter(|details| details.rejected_spans != 0 || !details.error_message.is_empty()) } -fn grpc_status_to_error(status: Status) -> TraceExporterError { +fn grpc_status_to_error(status: Status) -> GrpcExportError { + let retry_after = retry_info_delay(&status); + // Transport/IO failures from our bare `H2Service` fall through to `Code::Unknown` with the // original `std::io::Error` somewhere in the source chain (directly for a connect failure, or // wrapped in a `hyper::Error` for a post-connect handshake/read/write failure). Walk the chain @@ -379,28 +450,40 @@ fn grpc_status_to_error(status: Status) -> TraceExporterError { let mut cause: Option<&(dyn std::error::Error + 'static)> = status.source(); while let Some(err) = cause { if let Some(io_err) = err.downcast_ref::() { - return TraceExporterError::Io(std::io::Error::new( - io_err.kind(), - io_err.to_string(), - )); + return GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + io_err.kind(), + io_err.to_string(), + )), + retry_after, + }; } cause = err.source(); } } match status.code() { - Code::Unavailable => TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionRefused, - status.message(), - )), - Code::DeadlineExceeded => TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - status.message(), - )), - Code::Cancelled | Code::Aborted | Code::OutOfRange | Code::DataLoss => { - TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionAborted, + Code::Unavailable => GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, status.message(), - )) + )), + retry_after, + }, + Code::DeadlineExceeded => GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::TimedOut, + status.message(), + )), + retry_after, + }, + Code::Cancelled | Code::Aborted | Code::OutOfRange | Code::DataLoss => { + GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionAborted, + status.message(), + )), + retry_after, + } } code => { let http_status = match code { @@ -414,14 +497,38 @@ fn grpc_status_to_error(status: Status) -> TraceExporterError { Code::Unimplemented => http::StatusCode::NOT_IMPLEMENTED, _ => http::StatusCode::INTERNAL_SERVER_ERROR, }; - TraceExporterError::Request(RequestError::new( + let error = TraceExporterError::Request(RequestError::new( http_status, &format!("gRPC {code:?}: {}", status.message()), - )) + )); + if code == Code::ResourceExhausted && retry_after.is_some() { + GrpcExportError::Retryable { error, retry_after } + } else { + GrpcExportError::NonRetryable(error) + } } } } +fn retry_info_delay(status: &Status) -> Option { + let rich_status = RpcStatusDetails::decode(status.details()).ok()?; + rich_status.details.into_iter().find_map(|detail| { + if detail.type_url != RETRY_INFO_TYPE_URL { + return None; + } + let retry_info = RetryInfo::decode(detail.value.as_slice()).ok()?; + let Some(delay) = retry_info.retry_delay else { + return Some(Duration::ZERO); + }; + let seconds = u64::try_from(delay.seconds).ok()?; + let nanos = u32::try_from(delay.nanos).ok()?; + if nanos >= 1_000_000_000 { + return None; + } + Some(Duration::new(seconds, nanos)) + }) +} + #[cfg(test)] mod build_tests { use super::*; @@ -453,6 +560,48 @@ mod build_tests { assert!(build_grpc_transport("http://localhost:4317", cfg()).is_ok()); } + #[test] + fn supplies_http_default_port_for_dialing() { + let transport = build_grpc_transport("http://collector", cfg()).unwrap(); + + assert_eq!(transport.service.authority.as_ref(), "collector:80"); + assert_eq!( + transport + .origin + .authority() + .map(|authority| authority.as_str()), + Some("collector") + ); + + let ipv6_transport = build_grpc_transport("http://[::1]", cfg()).unwrap(); + assert_eq!(ipv6_transport.service.authority.as_ref(), "[::1]:80"); + } + + #[test] + fn preserves_explicit_port_for_dialing() { + let transport = build_grpc_transport("http://collector:4317", cfg()).unwrap(); + assert_eq!(transport.service.authority.as_ref(), "collector:4317"); + + let ipv6_transport = build_grpc_transport("http://[::1]:4317", cfg()).unwrap(); + assert_eq!(ipv6_transport.service.authority.as_ref(), "[::1]:4317"); + } + + #[test] + fn rejects_invalid_dial_authorities() { + for endpoint in [ + "http://collector:not-a-port", + "http://collector:99999", + "http://collector:", + "http://user@collector", + "http://user@collector:4317", + ] { + assert!( + build_grpc_transport(endpoint, cfg()).is_err(), + "accepted invalid endpoint {endpoint}" + ); + } + } + #[test] fn normalizes_origin_path_prefix() { let t = build_grpc_transport("http://localhost:4317/otel/", cfg()).unwrap(); @@ -482,7 +631,6 @@ mod integration_tests { use bytes::Bytes; use h2::server; use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest; - use prost::Message as _; use std::time::Duration; use tokio::net::TcpListener; @@ -502,11 +650,20 @@ mod integration_tests { &transport, None, false, - ExportTraceServiceRequest::default(), + Arc::new(ExportTraceServiceRequest::default()), ) .await .unwrap_err(); - assert!(matches!(err, TraceExporterError::Io(_)), "got: {err:?}"); + assert!( + matches!( + err, + GrpcExportError::Retryable { + error: TraceExporterError::Io(_), + retry_after: None + } + ), + "got: {err:?}" + ); } #[cfg_attr(miri, ignore)] @@ -612,7 +769,7 @@ mod integration_tests { let mut request = ExportTraceServiceRequest::default(); request.resource_spans.push(Default::default()); - send_otlp_traces_grpc(&transport, None, false, request.clone()) + send_otlp_traces_grpc(&transport, None, false, Arc::new(request.clone())) .await .expect("send should succeed"); @@ -641,16 +798,19 @@ mod integration_tests { &transport, None, false, - ExportTraceServiceRequest::default(), + Arc::new(ExportTraceServiceRequest::default()), ) .await .unwrap_err(); server.abort(); match err { - TraceExporterError::Io(e) => { + GrpcExportError::Retryable { + error: TraceExporterError::Io(e), + retry_after: None, + } => { assert_eq!(e.kind(), std::io::ErrorKind::TimedOut, "got: {e:?}") } - other => panic!("expected Io(TimedOut), got {other:?}"), + other => panic!("expected retryable Io(TimedOut), got {other:?}"), } } @@ -678,13 +838,19 @@ mod integration_tests { &transport, None, false, - ExportTraceServiceRequest::default(), + Arc::new(ExportTraceServiceRequest::default()), ) .await .unwrap_err(); server.await.unwrap(); assert!( - matches!(err, TraceExporterError::Io(_)), + matches!( + err, + GrpcExportError::Retryable { + error: TraceExporterError::Io(_), + retry_after: None + } + ), "expected Io (post-connect transport failure), got: {err:?}" ); } @@ -694,8 +860,89 @@ mod integration_tests { mod send_tests { use super::*; use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest; + use std::time::Duration; use tonic::{Code, Request, Status}; + #[derive(Clone, PartialEq, prost::Message)] + struct TestRpcStatus { + #[prost(message, repeated, tag = "3")] + details: Vec, + } + + #[derive(Clone, PartialEq, prost::Message)] + struct TestAny { + #[prost(string, tag = "1")] + type_url: String, + #[prost(bytes = "vec", tag = "2")] + value: Vec, + } + + #[derive(Clone, Copy, PartialEq, prost::Message)] + struct TestRetryInfo { + #[prost(message, optional, tag = "1")] + retry_delay: Option, + } + + #[derive(Clone, Copy, PartialEq, prost::Message)] + struct TestDuration { + #[prost(int64, tag = "1")] + seconds: i64, + #[prost(int32, tag = "2")] + nanos: i32, + } + + fn status_with_retry_info(code: Code, retry_delay: Duration) -> Status { + let retry_info = TestRetryInfo { + retry_delay: Some(TestDuration { + seconds: i64::try_from(retry_delay.as_secs()).unwrap(), + nanos: i32::try_from(retry_delay.subsec_nanos()).unwrap(), + }), + }; + let rich_status = TestRpcStatus { + details: vec![TestAny { + type_url: RETRY_INFO_TYPE_URL.to_string(), + value: retry_info.encode_to_vec(), + }], + }; + Status::with_details( + code, + "retry later", + Bytes::from(rich_status.encode_to_vec()), + ) + } + + #[test] + fn resource_exhausted_with_retry_info_is_retryable() { + let retry_after = Duration::new(3, 250_000_000); + + match grpc_status_to_error(status_with_retry_info(Code::ResourceExhausted, retry_after)) { + GrpcExportError::Retryable { + error: TraceExporterError::Request(error), + retry_after: actual_retry_after, + } => { + assert_eq!(error.status(), http::StatusCode::TOO_MANY_REQUESTS); + assert_eq!(actual_retry_after, Some(retry_after)); + } + other => panic!("expected retryable throttling error, got {other:?}"), + } + } + + #[test] + fn unavailable_honors_retry_info_delay() { + let retry_after = Duration::from_secs(7); + + match grpc_status_to_error(status_with_retry_info(Code::Unavailable, retry_after)) { + GrpcExportError::Retryable { + error: TraceExporterError::Io(error), + retry_after: actual_retry_after, + } => { + assert_eq!(error.kind(), std::io::ErrorKind::ConnectionRefused); + assert_eq!(actual_retry_after, Some(retry_after)); + } + other => panic!("expected throttled unavailable error, got {other:?}"), + } + } + #[test] fn status_transient_maps_to_io_kind() { for (s, want) in [ @@ -725,8 +972,11 @@ mod send_tests { ), ] { match grpc_status_to_error(s) { - TraceExporterError::Io(e) => assert_eq!(e.kind(), want), - other => panic!("expected Io, got {other:?}"), + GrpcExportError::Retryable { + error: TraceExporterError::Io(e), + retry_after: None, + } => assert_eq!(e.kind(), want), + other => panic!("expected retryable Io, got {other:?}"), } } } @@ -737,10 +987,13 @@ mod send_tests { let status = Status::from_error(Box::new(io_err)); assert_eq!(status.code(), Code::Unknown); match grpc_status_to_error(status) { - TraceExporterError::Io(e) => { + GrpcExportError::Retryable { + error: TraceExporterError::Io(e), + retry_after: None, + } => { assert_eq!(e.kind(), std::io::ErrorKind::ConnectionRefused) } - other => panic!("expected Io, got {other:?}"), + other => panic!("expected retryable Io, got {other:?}"), } } @@ -748,7 +1001,7 @@ mod send_tests { fn status_unknown_without_io_source_maps_to_request() { assert!(matches!( grpc_status_to_error(Status::new(Code::Unknown, "mystery")), - TraceExporterError::Request(_) + GrpcExportError::NonRetryable(TraceExporterError::Request(_)) )); } @@ -769,7 +1022,7 @@ mod send_tests { (Code::Internal, http::StatusCode::INTERNAL_SERVER_ERROR), ] { match grpc_status_to_error(Status::new(code, "failure")) { - TraceExporterError::Request(error) => { + GrpcExportError::NonRetryable(TraceExporterError::Request(error)) => { assert_eq!(error.status(), http_status); assert_eq!(error.msg(), format!("gRPC {code:?}: failure")); } diff --git a/libdd-data-pipeline/src/otlp/mod.rs b/libdd-data-pipeline/src/otlp/mod.rs index 738d14fe8a..8cfaa85e41 100644 --- a/libdd-data-pipeline/src/otlp/mod.rs +++ b/libdd-data-pipeline/src/otlp/mod.rs @@ -42,4 +42,6 @@ pub use metrics::OtlpStatsExporter; #[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}; +pub(crate) use grpc_exporter::{ + build_grpc_transport, send_otlp_traces_grpc, GrpcExportError, OtlpGrpcTransport, +}; diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index d94dce746c..0387cc6a49 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -23,7 +23,7 @@ use crate::agentless::AgentlessTraceConfig; #[cfg(not(target_arch = "wasm32"))] use crate::otlp::{ exporter::{OTLP_MAX_RETRIES, OTLP_RETRY_DELAY_MS}, - send_otlp_traces_grpc, OtlpGrpcTransport, + send_otlp_traces_grpc, GrpcExportError, OtlpGrpcTransport, }; use crate::otlp::{map_traces_to_otlp, send_otlp_traces_http, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] @@ -73,6 +73,20 @@ const INFO_ENDPOINT: &str = "/info"; const V04_TRACES_ENDPOINT: &str = "/v0.4/traces"; const V05_TRACES_ENDPOINT: &str = "/v0.5/traces"; const V1_TRACES_ENDPOINT: &str = "/v1.0/traces"; +#[cfg(not(target_arch = "wasm32"))] +const OTLP_GRPC_MAX_RETRY_DELAY: Duration = Duration::from_secs(30); + +#[cfg(not(target_arch = "wasm32"))] +fn grpc_retry_delay(attempt: u32, retry_after: Option) -> Duration { + let initial_delay = retry_after.unwrap_or_else(|| Duration::from_millis(OTLP_RETRY_DELAY_MS)); + let multiplier = 2u32 + .checked_pow(attempt.saturating_sub(1)) + .unwrap_or(u32::MAX); + initial_delay + .checked_mul(multiplier) + .unwrap_or(Duration::MAX) + .min(OTLP_GRPC_MAX_RETRY_DELAY) +} /// Values for optional telemetry HTTP session headers (`dd-session-id`, root/parent). #[derive(Debug, Default, Clone)] @@ -703,11 +717,11 @@ impl< traces: Vec>>, transport: &OtlpGrpcTransport, ) -> Result { - let request = map_traces_to_otlp( + let request = Arc::new(map_traces_to_otlp( traces, &self.otlp_resource_info, transport.config.otel_trace_semantics_enabled, - ); + )); let test_token = self.endpoint.test_token.as_deref(); let mut attempt: u32 = 1; loop { @@ -720,17 +734,15 @@ impl< .await { Ok(()) => return Ok(AgentResponse::Unchanged), - Err(TraceExporterError::Io(e)) => { + Err(GrpcExportError::Retryable { error, retry_after }) => { if attempt > OTLP_MAX_RETRIES { - return Err(TraceExporterError::Io(e)); + return Err(error); } - let delay_ms = OTLP_RETRY_DELAY_MS * 2u64.pow(attempt - 1); - self.capabilities - .sleep(Duration::from_millis(delay_ms)) - .await; + let delay = grpc_retry_delay(attempt, retry_after); + self.capabilities.sleep(delay).await; attempt += 1; } - Err(e) => return Err(e), + Err(GrpcExportError::NonRetryable(error)) => return Err(error), } } } @@ -1142,6 +1154,29 @@ mod tests { use libdd_trace_utils::span::v04::SpanBytes; use std::net; + #[test] + fn grpc_retry_delay_applies_backoff_and_cap() { + let retry_after = Duration::new(2, 250_000_000); + + assert_eq!(grpc_retry_delay(1, Some(retry_after)), retry_after); + assert_eq!( + grpc_retry_delay(2, Some(retry_after)), + Duration::new(4, 500_000_000) + ); + assert_eq!( + grpc_retry_delay(3, None), + Duration::from_millis(OTLP_RETRY_DELAY_MS * 4) + ); + assert_eq!( + grpc_retry_delay(3, Some(Duration::from_secs(20))), + Duration::from_secs(30) + ); + assert_eq!( + grpc_retry_delay(u32::MAX, Some(Duration::MAX)), + Duration::from_secs(30) + ); + } + #[test] fn test_from_tracer_tags_to_tracer_header_tags() { let tracer_tags = TracerMetadata { From e675f6431f57c459b91860bdbd76ad835006b83f Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Wed, 2 Sep 2026 16:50:04 -0400 Subject: [PATCH 3/6] fix(data-pipeline): harden OTLP gRPC retries --- libdd-data-pipeline/Cargo.toml | 2 +- libdd-data-pipeline/src/otlp/grpc_exporter.rs | 150 +++++++++++++++--- libdd-data-pipeline/src/trace_exporter/mod.rs | 42 ++--- 3 files changed, 150 insertions(+), 44 deletions(-) diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index 69f12b8b14..e295811613 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -57,6 +57,7 @@ libdd-capabilities-impl = { version = "4.0.0", path = "../libdd-capabilities-imp # (hyper/tokio/socket2) does not build for wasm32, so the whole gRPC path is gated off. tonic = { version = "0.14", default-features = false } prost = "0.14.1" +h2 = "0.4" hyper = { workspace = true, features = ["client", "http2"] } hyper-util = { workspace = true, features = ["tokio"] } @@ -97,7 +98,6 @@ tokio = { version = "1.23", features = [ duplicate = "2.0.1" [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] -h2 = "0.4" zstd = { version = "0.13", default-features = false } [target.'cfg(target_arch = "wasm32")'.dev-dependencies] diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs index c4bba4aa5b..e9daf6a3f6 100644 --- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs +++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs @@ -95,9 +95,8 @@ pub(crate) mod prost_codec { // Shared with the `Encoder` impl below. tonic's `EncodeBuf`/`DecodeBuf` constructors are // private to the crate, so tests exercise this generic-over-`BufMut`/`Buf` core directly // instead of going through the `Encoder`/`Decoder` traits (see `codec_tests`). - fn encode_into(item: Arc, dst: &mut impl bytes::BufMut) -> Result<(), Status> { - item.as_ref() - .encode(dst) + fn encode_into(item: &T, dst: &mut impl bytes::BufMut) -> Result<(), Status> { + item.encode(dst) .map_err(|e| Status::internal(format!("Failed to encode protobuf message: {e}"))) } } @@ -105,7 +104,7 @@ pub(crate) mod prost_codec { type Item = Arc; type Error = Status; fn encode(&mut self, item: Arc, dst: &mut EncodeBuf<'_>) -> Result<(), Status> { - Self::encode_into(item, dst) + Self::encode_into(item.as_ref(), dst) } } @@ -155,8 +154,6 @@ pub(crate) mod prost_codec { ExportTraceServiceRequest, ExportTraceServiceResponse, }; use libdd_trace_protobuf::opentelemetry::proto::trace::v1::ResourceSpans; - use std::sync::Arc; - // Round-trips through the `BufMut`/`Buf`-generic core (`encode_into`/`decode_from`) that // the `Encoder`/`Decoder` impls delegate to, over a plain `BytesMut`; see // `ProstEncoder::encode_into` for why the `Codec` traits can't be driven directly. @@ -170,9 +167,7 @@ pub(crate) mod prost_codec { }], }; let mut buf = BytesMut::new(); - let shared_msg = Arc::new(msg.clone()); - ProstEncoder::encode_into(shared_msg.clone(), &mut buf).unwrap(); - assert_eq!(Arc::strong_count(&shared_msg), 1); + ProstEncoder::encode_into(&msg, &mut buf).unwrap(); assert!(!buf.is_empty()); let out = ProstDecoder::decode_from(&mut buf).unwrap(); @@ -230,17 +225,10 @@ impl GrpcService for H2Service { biased; response = &mut request => response?, connection = &mut conn => { - connection?; - return Err(std::io::Error::new( - std::io::ErrorKind::UnexpectedEof, - "gRPC connection closed before the response completed", - ).into()); + let _ = connection; + request.await? } }; - - // The completed request has dropped its sender. Polling the connection to completion - // lets hyper shut it down gracefully; the exporter's outer timeout bounds this wait. - conn.await?; Ok(response) }) } @@ -250,7 +238,8 @@ impl GrpcService for H2Service { /// service. Holds no live connection and thus no background task (nothing to rebuild across fork). #[derive(Clone, Debug)] pub(crate) struct OtlpGrpcTransport { - pub(crate) config: OtlpGrpcTraceConfig, + pub(crate) timeout: Duration, + pub(crate) otel_trace_semantics_enabled: bool, origin: http::Uri, service: H2Service, /// Custom headers parsed to gRPC metadata once at build time. @@ -262,6 +251,11 @@ pub(crate) fn build_grpc_transport( endpoint_url: &str, config: OtlpGrpcTraceConfig, ) -> Result { + let OtlpGrpcTraceConfig { + headers, + timeout, + otel_trace_semantics_enabled, + } = config; let uri = endpoint_url.parse::()?; let scheme = uri.scheme().ok_or_else(|| { @@ -290,6 +284,11 @@ pub(crate) fn build_grpc_transport( "gRPC endpoint must include an authority".to_string(), )) })?; + if authority.host().is_empty() { + return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri( + "gRPC endpoint authority must include a host".to_string(), + ))); + } let authority_text = authority.as_str(); if authority_text.contains('@') { return Err(TraceExporterError::Builder(BuilderErrorKind::InvalidUri( @@ -322,9 +321,8 @@ pub(crate) fn build_grpc_transport( // Parse custom headers to gRPC metadata once here rather than on every send. Invalid entries // are skipped with a single warning (logging only the key: a value may carry a secret). - let metadata_headers = config - .headers - .iter() + let metadata_headers = headers + .into_iter() .filter_map(|(k, v)| { match ( k.parse::(), @@ -340,7 +338,8 @@ pub(crate) fn build_grpc_transport( .collect(); Ok(OtlpGrpcTransport { - config, + timeout, + otel_trace_semantics_enabled, origin, service, metadata_headers, @@ -371,7 +370,7 @@ pub(crate) async fn send_otlp_traces_grpc( let path = http::uri::PathAndQuery::from_static(GRPC_EXPORT_PATH); let codec = ExportCodec::default(); - tokio::time::timeout(transport.config.timeout, async { + tokio::time::timeout(transport.timeout, async { let mut client = Grpc::with_origin(transport.service.clone(), transport.origin.clone()); client .ready() @@ -458,6 +457,24 @@ fn grpc_status_to_error(status: Status) -> GrpcExportError { retry_after, }; } + let hyper_retryable = err + .downcast_ref::() + .is_some_and(|error| error.is_canceled() || error.is_closed()); + let h2_retryable = err.downcast_ref::().is_some_and(|error| { + matches!( + error.reason(), + Some(h2::Reason::REFUSED_STREAM | h2::Reason::CANCEL) + ) + }); + if hyper_retryable || h2_retryable { + return GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + err.to_string(), + )), + retry_after, + }; + } cause = err.source(); } } @@ -592,6 +609,7 @@ mod build_tests { "http://collector:not-a-port", "http://collector:99999", "http://collector:", + "http://:4317", "http://user@collector", "http://user@collector:4317", ] { @@ -777,6 +795,90 @@ mod integration_tests { assert_eq!(decoded.resource_spans.len(), 1); } + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn successful_response_wins_over_connection_teardown() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut conn = server::handshake(socket).await.unwrap(); + let (req, mut respond) = conn.accept().await.unwrap().unwrap(); + let mut body = req.into_body(); + while let Some(chunk) = body.data().await { + let chunk = chunk.unwrap(); + body.flow_control().release_capacity(chunk.len()).unwrap(); + } + + let response = http::Response::builder() + .status(200) + .header("content-type", "application/grpc") + .body(()) + .unwrap(); + let mut send = respond.send_response(response, false).unwrap(); + send.send_data(Bytes::from_static(&[0, 0, 0, 0, 0]), false) + .unwrap(); + let mut trailers = http::HeaderMap::new(); + trailers.insert("grpc-status", "0".parse().unwrap()); + send.send_trailers(trailers).unwrap(); + let flush_deadline = tokio::time::sleep(Duration::from_millis(25)); + tokio::pin!(flush_deadline); + tokio::select! { + _ = &mut flush_deadline => {} + _ = conn.accept() => {} + } + conn.abrupt_shutdown(h2::Reason::INTERNAL_ERROR); + while conn.accept().await.is_some() {} + }); + + let transport = build_grpc_transport(&format!("http://{addr}"), cfg()).unwrap(); + send_otlp_traces_grpc( + &transport, + None, + false, + Arc::new(ExportTraceServiceRequest::default()), + ) + .await + .expect("completed response should remain successful"); + server.await.unwrap(); + } + + #[cfg_attr(miri, ignore)] + #[tokio::test] + async fn refused_stream_is_retryable() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut conn = server::handshake(socket).await.unwrap(); + let (_req, mut respond) = conn.accept().await.unwrap().unwrap(); + respond.send_reset(h2::Reason::REFUSED_STREAM); + while conn.accept().await.is_some() {} + }); + + let transport = build_grpc_transport(&format!("http://{addr}"), cfg()).unwrap(); + let error = send_otlp_traces_grpc( + &transport, + None, + false, + Arc::new(ExportTraceServiceRequest::default()), + ) + .await + .unwrap_err(); + server.await.unwrap(); + + assert!( + matches!( + error, + GrpcExportError::Retryable { + error: TraceExporterError::Io(_), + retry_after: None + } + ), + "expected retryable transport error, got: {error:?}" + ); + } + #[cfg_attr(miri, ignore)] #[tokio::test] async fn timeout_maps_to_io_timedout() { diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index 0387cc6a49..dc41ccc287 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -77,15 +77,18 @@ const V1_TRACES_ENDPOINT: &str = "/v1.0/traces"; const OTLP_GRPC_MAX_RETRY_DELAY: Duration = Duration::from_secs(30); #[cfg(not(target_arch = "wasm32"))] -fn grpc_retry_delay(attempt: u32, retry_after: Option) -> Duration { - let initial_delay = retry_after.unwrap_or_else(|| Duration::from_millis(OTLP_RETRY_DELAY_MS)); - let multiplier = 2u32 - .checked_pow(attempt.saturating_sub(1)) - .unwrap_or(u32::MAX); - initial_delay - .checked_mul(multiplier) - .unwrap_or(Duration::MAX) - .min(OTLP_GRPC_MAX_RETRY_DELAY) +fn grpc_retry_delay(attempt: u32, retry_after: Option) -> Option { + let initial_delay = match retry_after { + Some(delay) if delay > OTLP_GRPC_MAX_RETRY_DELAY => return None, + Some(delay) if !delay.is_zero() => delay, + _ => Duration::from_millis(OTLP_RETRY_DELAY_MS), + }; + let multiplier = 2u32.saturating_pow(attempt.saturating_sub(1)); + Some( + initial_delay + .saturating_mul(multiplier) + .min(OTLP_GRPC_MAX_RETRY_DELAY), + ) } /// Values for optional telemetry HTTP session headers (`dd-session-id`, root/parent). @@ -164,7 +167,6 @@ fn add_path(url: &Uri, path: &str) -> Uri { pub use libdd_trace_utils::tracer_metadata::TracerMetadata; -/// The transport used for OTLP trace export. #[derive(Debug)] pub(crate) enum OtlpExportMode { Http(OtlpTraceConfig), @@ -710,7 +712,6 @@ impl< Ok(AgentResponse::Unchanged) } - /// Sends trace chunks via OTLP gRPC. #[cfg(not(target_arch = "wasm32"))] async fn send_otlp_grpc_inner( &self, @@ -720,7 +721,7 @@ impl< let request = Arc::new(map_traces_to_otlp( traces, &self.otlp_resource_info, - transport.config.otel_trace_semantics_enabled, + transport.otel_trace_semantics_enabled, )); let test_token = self.endpoint.test_token.as_deref(); let mut attempt: u32 = 1; @@ -738,7 +739,9 @@ impl< if attempt > OTLP_MAX_RETRIES { return Err(error); } - let delay = grpc_retry_delay(attempt, retry_after); + let Some(delay) = grpc_retry_delay(attempt, retry_after) else { + return Err(error); + }; self.capabilities.sleep(delay).await; attempt += 1; } @@ -1158,23 +1161,24 @@ mod tests { fn grpc_retry_delay_applies_backoff_and_cap() { let retry_after = Duration::new(2, 250_000_000); - assert_eq!(grpc_retry_delay(1, Some(retry_after)), retry_after); + assert_eq!(grpc_retry_delay(1, Some(retry_after)), Some(retry_after)); assert_eq!( grpc_retry_delay(2, Some(retry_after)), - Duration::new(4, 500_000_000) + Some(Duration::new(4, 500_000_000)) ); assert_eq!( grpc_retry_delay(3, None), - Duration::from_millis(OTLP_RETRY_DELAY_MS * 4) + Some(Duration::from_millis(OTLP_RETRY_DELAY_MS * 4)) ); assert_eq!( grpc_retry_delay(3, Some(Duration::from_secs(20))), - Duration::from_secs(30) + Some(Duration::from_secs(30)) ); assert_eq!( - grpc_retry_delay(u32::MAX, Some(Duration::MAX)), - Duration::from_secs(30) + grpc_retry_delay(1, Some(Duration::ZERO)), + Some(Duration::from_millis(OTLP_RETRY_DELAY_MS)) ); + assert_eq!(grpc_retry_delay(1, Some(Duration::from_secs(31))), None); } #[test] From 89aff0b7fe135a796543a94a9e4801e7b778c36e Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Wed, 2 Sep 2026 21:05:57 -0400 Subject: [PATCH 4/6] fix(data-pipeline): align OTLP gRPC transport behavior --- libdd-data-pipeline-ffi/src/trace_exporter.rs | 7 +- libdd-data-pipeline/Cargo.toml | 2 +- libdd-data-pipeline/src/otlp/config.rs | 3 - libdd-data-pipeline/src/otlp/grpc_exporter.rs | 135 ++++++++---------- libdd-data-pipeline/src/otlp/mod.rs | 26 ---- libdd-data-pipeline/src/trace_exporter/mod.rs | 42 ++++-- .../tests/test_trace_exporter_otlp_grpc.rs | 84 ++++++++--- libdd-trace-utils/src/send_with_retry/mod.rs | 6 +- 8 files changed, 164 insertions(+), 141 deletions(-) diff --git a/libdd-data-pipeline-ffi/src/trace_exporter.rs b/libdd-data-pipeline-ffi/src/trace_exporter.rs index 3daf2d624b..3525ffd034 100644 --- a/libdd-data-pipeline-ffi/src/trace_exporter.rs +++ b/libdd-data-pipeline-ffi/src/trace_exporter.rs @@ -523,6 +523,7 @@ 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), /// `http/protobuf`, or `grpc`; unknown values are rejected. The host language resolves the value /// (e.g. from `OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`). +/// The `grpc` protocol currently supports plaintext `http://` endpoints only. /// /// 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 @@ -752,9 +753,9 @@ pub unsafe extern "C" fn ddog_trace_exporter_config_set_output_to_log( /// Create a new TraceExporter instance. /// -/// 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 +/// When an OTLP endpoint is configured via `TraceExporterConfig`, the exporter sends traces using +/// the configured `http/json`, `http/protobuf`, or `grpc` protocol instead of 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 diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index e295811613..58845dad7b 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -60,6 +60,7 @@ prost = "0.14.1" h2 = "0.4" hyper = { workspace = true, features = ["client", "http2"] } hyper-util = { workspace = true, features = ["tokio"] } +rand = "0.8.5" [target.'cfg(target_arch = "wasm32")'.dependencies] getrandom = { version = "0.2", features = ["js"] } @@ -88,7 +89,6 @@ libdd-trace-utils = { path = "../libdd-trace-utils", features = [ ] } httpmock = "0.8.0-alpha.1" prost = "0.14.1" -rand = "0.8.5" tempfile.workspace = true tokio = { version = "1.23", features = [ "rt", diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index cf72f0add4..a0ae1de342 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -31,7 +31,6 @@ impl std::str::FromStr for OtlpProtocol { } impl OtlpProtocol { - /// The HTTP `Content-Type` for this protocol's body encoding, or `None` for [`Self::Grpc`]. pub(crate) fn content_type(&self) -> Option { match self { OtlpProtocol::HttpJson => Some(libdd_common::header::APPLICATION_JSON), @@ -40,8 +39,6 @@ impl OtlpProtocol { } } - /// Encode the prost OTLP request to this protocol's wire format, or `None` for - /// [`Self::Grpc`]. pub(crate) fn encode( &self, req: &libdd_trace_utils::otlp_encoder::ProtoExportTraceServiceRequest, diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs index e9daf6a3f6..16a5c03628 100644 --- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs +++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs @@ -16,6 +16,7 @@ use hyper_util::rt::{TokioExecutor, TokioIo}; use libdd_trace_protobuf::opentelemetry::proto::collector::trace::v1::{ ExportTracePartialSuccess, ExportTraceServiceRequest, ExportTraceServiceResponse, }; +use libdd_trace_utils::send_with_retry::TRACE_EXPORTER_USER_AGENT; use prost::Message as _; use std::error::Error as StdError; use std::future::Future; @@ -407,6 +408,18 @@ fn attach_metadata( test_token: Option<&str>, client_computed_stats: bool, ) { + req.metadata_mut().insert( + AsciiMetadataKey::from_static("user-agent"), + AsciiMetadataValue::from_static(TRACE_EXPORTER_USER_AGENT), + ); + for (key, value) in libdd_common::entity_id::get_entity_headers() { + if let (Ok(key), Ok(value)) = ( + key.parse::(), + value.parse::(), + ) { + req.metadata_mut().insert(key, value); + } + } for (key, val) in headers { req.metadata_mut().insert(key.clone(), val.clone()); } @@ -478,52 +491,32 @@ fn grpc_status_to_error(status: Status) -> GrpcExportError { cause = err.source(); } } - match status.code() { - Code::Unavailable => GrpcExportError::Retryable { - error: TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionRefused, - status.message(), - )), - retry_after, - }, - Code::DeadlineExceeded => GrpcExportError::Retryable { - error: TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::TimedOut, - status.message(), - )), - retry_after, - }, - Code::Cancelled | Code::Aborted | Code::OutOfRange | Code::DataLoss => { - GrpcExportError::Retryable { - error: TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionAborted, - status.message(), - )), - retry_after, - } - } - code => { - let http_status = match code { - Code::InvalidArgument => http::StatusCode::BAD_REQUEST, - Code::Unauthenticated => http::StatusCode::UNAUTHORIZED, - Code::PermissionDenied => http::StatusCode::FORBIDDEN, - Code::NotFound => http::StatusCode::NOT_FOUND, - Code::AlreadyExists => http::StatusCode::CONFLICT, - Code::ResourceExhausted => http::StatusCode::TOO_MANY_REQUESTS, - Code::FailedPrecondition => http::StatusCode::PRECONDITION_FAILED, - Code::Unimplemented => http::StatusCode::NOT_IMPLEMENTED, - _ => http::StatusCode::INTERNAL_SERVER_ERROR, - }; - let error = TraceExporterError::Request(RequestError::new( - http_status, - &format!("gRPC {code:?}: {}", status.message()), - )); - if code == Code::ResourceExhausted && retry_after.is_some() { - GrpcExportError::Retryable { error, retry_after } - } else { - GrpcExportError::NonRetryable(error) - } - } + let code = status.code(); + let (http_status, retryable) = match code { + Code::Cancelled => (http::StatusCode::REQUEST_TIMEOUT, true), + Code::InvalidArgument => (http::StatusCode::BAD_REQUEST, false), + Code::OutOfRange => (http::StatusCode::BAD_REQUEST, true), + Code::DeadlineExceeded => (http::StatusCode::GATEWAY_TIMEOUT, true), + Code::NotFound => (http::StatusCode::NOT_FOUND, false), + Code::AlreadyExists => (http::StatusCode::CONFLICT, false), + Code::Aborted => (http::StatusCode::CONFLICT, true), + Code::PermissionDenied => (http::StatusCode::FORBIDDEN, false), + Code::ResourceExhausted => (http::StatusCode::TOO_MANY_REQUESTS, retry_after.is_some()), + Code::FailedPrecondition => (http::StatusCode::PRECONDITION_FAILED, false), + Code::Unauthenticated => (http::StatusCode::UNAUTHORIZED, false), + Code::Unavailable => (http::StatusCode::SERVICE_UNAVAILABLE, true), + Code::Unimplemented => (http::StatusCode::NOT_IMPLEMENTED, false), + Code::DataLoss => (http::StatusCode::INTERNAL_SERVER_ERROR, true), + _ => (http::StatusCode::INTERNAL_SERVER_ERROR, false), + }; + let error = TraceExporterError::Request(RequestError::new( + http_status, + &format!("gRPC {code:?}: {}", status.message()), + )); + if retryable { + GrpcExportError::Retryable { error, retry_after } + } else { + GrpcExportError::NonRetryable(error) } } @@ -1035,10 +1028,11 @@ mod send_tests { match grpc_status_to_error(status_with_retry_info(Code::Unavailable, retry_after)) { GrpcExportError::Retryable { - error: TraceExporterError::Io(error), + error: TraceExporterError::Request(error), retry_after: actual_retry_after, } => { - assert_eq!(error.kind(), std::io::ErrorKind::ConnectionRefused); + assert_eq!(error.status(), http::StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(error.msg(), "gRPC Unavailable: retry later"); assert_eq!(actual_retry_after, Some(retry_after)); } other => panic!("expected throttled unavailable error, got {other:?}"), @@ -1046,39 +1040,24 @@ mod send_tests { } #[test] - fn status_transient_maps_to_io_kind() { - for (s, want) in [ - ( - Status::unavailable("down"), - std::io::ErrorKind::ConnectionRefused, - ), - ( - Status::deadline_exceeded("slow"), - std::io::ErrorKind::TimedOut, - ), - ( - Status::new(Code::Cancelled, "canceled"), - std::io::ErrorKind::ConnectionAborted, - ), - ( - Status::new(Code::Aborted, "aborted"), - std::io::ErrorKind::ConnectionAborted, - ), - ( - Status::new(Code::OutOfRange, "out of range"), - std::io::ErrorKind::ConnectionAborted, - ), - ( - Status::new(Code::DataLoss, "data loss"), - std::io::ErrorKind::ConnectionAborted, - ), + fn status_transient_remains_retryable_request_error() { + for (code, http_status) in [ + (Code::Unavailable, http::StatusCode::SERVICE_UNAVAILABLE), + (Code::DeadlineExceeded, http::StatusCode::GATEWAY_TIMEOUT), + (Code::Cancelled, http::StatusCode::REQUEST_TIMEOUT), + (Code::Aborted, http::StatusCode::CONFLICT), + (Code::OutOfRange, http::StatusCode::BAD_REQUEST), + (Code::DataLoss, http::StatusCode::INTERNAL_SERVER_ERROR), ] { - match grpc_status_to_error(s) { + match grpc_status_to_error(Status::new(code, "transient")) { GrpcExportError::Retryable { - error: TraceExporterError::Io(e), + error: TraceExporterError::Request(error), retry_after: None, - } => assert_eq!(e.kind(), want), - other => panic!("expected retryable Io, got {other:?}"), + } => { + assert_eq!(error.status(), http_status); + assert_eq!(error.msg(), format!("gRPC {code:?}: transient")); + } + other => panic!("expected retryable Request, got {other:?}"), } } } diff --git a/libdd-data-pipeline/src/otlp/mod.rs b/libdd-data-pipeline/src/otlp/mod.rs index 0937cbad3d..ab82f314af 100644 --- a/libdd-data-pipeline/src/otlp/mod.rs +++ b/libdd-data-pipeline/src/otlp/mod.rs @@ -1,32 +1,6 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! OTLP trace and trace-metrics export for libdatadog. -//! -//! When an OTLP endpoint is configured via -//! [`crate::trace_exporter::TraceExporterBuilder::set_otlp_endpoint`], the trace exporter sends -//! traces in OTLP format to that endpoint instead of the Datadog agent; the transport and wire -//! encoding are selected via [`OtlpProtocol`]. The host language is responsible for -//! resolving the endpoint from its own configuration (e.g. -//! `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`). -//! -//! With [`set_otlp_metrics_endpoint`](crate::trace_exporter::TraceExporterBuilder::set_otlp_metrics_endpoint), -//! client-computed span stats ship as a `traces.span.sdk.metrics.duration` OTLP histogram -//! instead of going to the agent `/v0.6/stats` endpoint. -//! -//! ## Sampling -//! -//! The exporter enforces the sampling decision already made by the tracer: unsampled chunks are -//! dropped via `drop_chunks` before export. It does not apply its own sampling policy. The tracer -//! (e.g. dd-trace-py) is responsible for inheriting the sampling decision from the distributed -//! trace context; when no decision is present, the tracer typically uses 100% (always on). -//! -//! ## Partial flush -//! -//! For the POC, partial flush is disabled. The tracer should only invoke the exporter when all -//! spans from a local trace are closed (i.e. send complete trace chunks). This crate does not -//! buffer or flush partially—it exports whatever trace chunks it receives. - pub mod config; pub mod exporter; pub mod metrics; diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index a64e62ee93..9b2cf4d6f6 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -78,9 +78,15 @@ const V05_TRACES_ENDPOINT: &str = "/v0.5/traces"; const V1_TRACES_ENDPOINT: &str = "/v1.0/traces"; #[cfg(not(target_arch = "wasm32"))] const OTLP_GRPC_MAX_RETRY_DELAY: Duration = Duration::from_secs(30); +#[cfg(not(target_arch = "wasm32"))] +const OTLP_GRPC_MAX_JITTER: Duration = Duration::from_millis(100); #[cfg(not(target_arch = "wasm32"))] -fn grpc_retry_delay(attempt: u32, retry_after: Option) -> Option { +fn grpc_retry_delay( + attempt: u32, + retry_after: Option, + jitter: Duration, +) -> Option { let initial_delay = match retry_after { Some(delay) if delay > OTLP_GRPC_MAX_RETRY_DELAY => return None, Some(delay) if !delay.is_zero() => delay, @@ -90,10 +96,17 @@ fn grpc_retry_delay(attempt: u32, retry_after: Option) -> Option Duration { + let max_millis = u64::try_from(OTLP_GRPC_MAX_JITTER.as_millis()).unwrap_or(u64::MAX); + Duration::from_millis(rand::random::() % max_millis + 1) +} + #[derive(Clone, Copy)] struct PayloadCounts { chunks: usize, @@ -840,7 +853,8 @@ impl< if attempt > OTLP_MAX_RETRIES { break Err(error); } - let Some(delay) = grpc_retry_delay(attempt, retry_after) else { + let Some(delay) = grpc_retry_delay(attempt, retry_after, grpc_retry_jitter()) + else { break Err(error); }; self.capabilities.sleep(delay).await; @@ -1262,24 +1276,36 @@ mod tests { fn grpc_retry_delay_applies_backoff_and_cap() { let retry_after = Duration::new(2, 250_000_000); - assert_eq!(grpc_retry_delay(1, Some(retry_after)), Some(retry_after)); assert_eq!( - grpc_retry_delay(2, Some(retry_after)), + grpc_retry_delay(1, Some(retry_after), Duration::ZERO), + Some(retry_after) + ); + assert_eq!( + grpc_retry_delay(2, Some(retry_after), Duration::ZERO), Some(Duration::new(4, 500_000_000)) ); assert_eq!( - grpc_retry_delay(3, None), + grpc_retry_delay(3, None, Duration::ZERO), Some(Duration::from_millis(OTLP_RETRY_DELAY_MS * 4)) ); assert_eq!( - grpc_retry_delay(3, Some(Duration::from_secs(20))), + grpc_retry_delay(3, Some(Duration::from_secs(20)), Duration::ZERO), Some(Duration::from_secs(30)) ); assert_eq!( - grpc_retry_delay(1, Some(Duration::ZERO)), + grpc_retry_delay(1, Some(Duration::ZERO), Duration::ZERO), Some(Duration::from_millis(OTLP_RETRY_DELAY_MS)) ); - assert_eq!(grpc_retry_delay(1, Some(Duration::from_secs(31))), None); + assert_eq!( + grpc_retry_delay(1, Some(Duration::from_secs(31)), Duration::ZERO), + None + ); + } + + #[test] + fn grpc_retry_delay_adds_bounded_jitter() { + let delay = grpc_retry_delay(1, None, Duration::from_millis(50)); + assert_eq!(delay, Some(Duration::from_millis(150))); } #[test] diff --git a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs index 27a124c02e..6086d9d8f8 100644 --- a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs +++ b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs @@ -15,8 +15,13 @@ mod grpc_export_tests { }; use libdd_trace_utils::test_utils::create_test_json_span; use prost::Message; + #[cfg(feature = "telemetry")] + use regex::Regex; use serde_json::json; - use std::sync::{mpsc, Arc}; + use std::sync::{ + atomic::{AtomicUsize, Ordering}, + mpsc, Arc, + }; use std::time::Duration; use tokio::net::TcpListener; use tokio::sync::oneshot; @@ -24,6 +29,7 @@ mod grpc_export_tests { struct ReceivedExport { path: String, + user_agent: Option, client_computed_stats: Option, request: ExportTraceServiceRequest, } @@ -31,6 +37,7 @@ mod grpc_export_tests { async fn run_grpc_test_server( listener: TcpListener, req_tx: mpsc::Sender, + failures_remaining: Arc, mut shutdown: oneshot::Receiver<()>, ) { let mut connections = JoinSet::new(); @@ -40,6 +47,7 @@ mod grpc_export_tests { accepted = listener.accept() => { let Ok((socket, _)) = accepted else { return }; let connection_req_tx = req_tx.clone(); + let connection_failures_remaining = failures_remaining.clone(); connections.spawn(async move { let Ok(mut connection) = server::handshake(socket).await else { return; @@ -51,6 +59,7 @@ mod grpc_export_tests { request, respond, connection_req_tx.clone(), + connection_failures_remaining.clone(), )); } } @@ -73,8 +82,14 @@ mod grpc_export_tests { request: http::Request, mut respond: h2::server::SendResponse, req_tx: mpsc::Sender, + failures_remaining: Arc, ) { let path = request.uri().path().to_string(); + let user_agent = request + .headers() + .get("user-agent") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); let client_computed_stats = request .headers() .get("datadog-client-computed-stats") @@ -97,21 +112,17 @@ mod grpc_export_tests { if let Some(req) = decoded { let _ = req_tx.send(ReceivedExport { path, + user_agent, client_computed_stats, request: req, }); } - let response_proto = ExportTraceServiceResponse::default(); - let proto_bytes = response_proto.encode_to_vec(); - let mut frame = Vec::with_capacity(5 + proto_bytes.len()); - frame.push(0u8); - frame.extend_from_slice( - &u32::try_from(proto_bytes.len()) - .expect("protobuf response exceeds the gRPC frame length") - .to_be_bytes(), - ); - frame.extend_from_slice(&proto_bytes); + let should_fail = failures_remaining + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |remaining| { + remaining.checked_sub(1) + }) + .is_ok(); let response = http::Response::builder() .status(200) @@ -121,10 +132,25 @@ mod grpc_export_tests { let Ok(mut send_stream) = respond.send_response(response, false) else { return; }; - let _ = send_stream.send_data(Bytes::from(frame), false); + if !should_fail { + let response_proto = ExportTraceServiceResponse::default(); + let proto_bytes = response_proto.encode_to_vec(); + let mut frame = Vec::with_capacity(5 + proto_bytes.len()); + frame.push(0u8); + frame.extend_from_slice( + &u32::try_from(proto_bytes.len()) + .expect("protobuf response exceeds the gRPC frame length") + .to_be_bytes(), + ); + frame.extend_from_slice(&proto_bytes); + let _ = send_stream.send_data(Bytes::from(frame), false); + } let mut trailers = http::HeaderMap::new(); - trailers.insert("grpc-status", "0".parse().unwrap()); + trailers.insert( + "grpc-status", + if should_fail { "14" } else { "0" }.parse().unwrap(), + ); let _ = send_stream.send_trailers(trailers); } @@ -168,7 +194,8 @@ mod grpc_export_tests { rt.block_on(async move { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); port_tx.send(listener.local_addr().unwrap().port()).unwrap(); - run_grpc_test_server(listener, req_tx, shutdown_rx).await; + run_grpc_test_server(listener, req_tx, Arc::new(AtomicUsize::new(0)), shutdown_rx) + .await; }); }); @@ -206,6 +233,10 @@ mod grpc_export_tests { .recv_timeout(Duration::from_secs(10)) .expect("server did not receive the initial request"); assert_eq!(initial.path, expected_path); + assert_eq!( + initial.user_agent.as_deref(), + Some(libdd_trace_utils::send_with_retry::TRACE_EXPORTER_USER_AGENT) + ); assert!( !initial.request.resource_spans.is_empty(), "expected at least one ResourceSpans" @@ -251,12 +282,17 @@ mod grpc_export_tests { #[cfg(feature = "telemetry")] #[cfg_attr(miri, ignore)] #[test] - fn grpc_export_emits_native_trace_telemetry() { + fn grpc_retry_emits_native_trace_telemetry() { let telemetry_server = httpmock::MockServer::start(); + let requests_metric = + Regex::new(r#""metric":"trace_api.requests","points":\[\[\d+,2\.0\]\]"#).unwrap(); let metrics_endpoint = telemetry_server.mock(|when, then| { when.method(httpmock::Method::POST) .path("/telemetry/proxy/api/v2/apmtelemetry") - .body_includes("\"metric\":\"trace_api.requests\"") + .is_true(move |request| { + String::from_utf8(request.body_vec()) + .is_ok_and(|body| requests_metric.is_match(&body)) + }) .body_includes("\"metric\":\"spans_enqueued_for_serialization\"") .body_includes("\"metric\":\"trace_chunks_sent\""); then.status(200) @@ -276,7 +312,13 @@ mod grpc_export_tests { rt.block_on(async move { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); port_tx.send(listener.local_addr().unwrap().port()).unwrap(); - run_grpc_test_server(listener, req_tx, shutdown_rx).await; + run_grpc_test_server( + listener, + req_tx, + Arc::new(AtomicUsize::new(1)), + shutdown_rx, + ) + .await; }); }); @@ -306,9 +348,11 @@ mod grpc_export_tests { let span = create_test_json_span(1234, 12342, 12341, 1, false); let data = rmp_serde::to_vec_named(&vec![vec![span]]).unwrap(); exporter.send(data.as_ref()).expect("send traces"); - req_rx - .recv_timeout(Duration::from_secs(10)) - .expect("server did not receive request"); + for _ in 0..2 { + req_rx + .recv_timeout(Duration::from_secs(10)) + .expect("server did not receive request"); + } let deadline = std::time::Instant::now() + Duration::from_secs(10); while metrics_endpoint.calls() == 0 && std::time::Instant::now() < deadline { diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index 4cb3b98f1d..b14ea8af7f 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -22,6 +22,9 @@ pub type Attempts = u32; pub type SendWithRetryResult = Result<(http::Response, Attempts), SendWithRetryError>; +/// User-agent sent by the trace exporter. +pub const TRACE_EXPORTER_USER_AGENT: &str = concat!("Tracer/", env!("CARGO_PKG_VERSION")); + /// All errors contain the number of attempts after which the final error was returned #[derive(Debug)] pub enum SendWithRetryError { @@ -156,8 +159,7 @@ pub async fn send_with_retry_and_size let mut builder = http::Request::builder() .method(http::Method::POST) .uri(target.url.clone()); - builder = - target.set_standard_headers(builder, concat!("Tracer/", env!("CARGO_PKG_VERSION"))); + builder = target.set_standard_headers(builder, TRACE_EXPORTER_USER_AGENT); for (key, value) in headers { builder = builder.header(key, value); } From 90514300981af80e0fee8931bc0ada899093b38e Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Wed, 2 Sep 2026 21:30:10 -0400 Subject: [PATCH 5/6] test(data-pipeline): avoid deprecated atomic API --- .../tests/test_trace_exporter_otlp_grpc.rs | 20 ++++++++++++++----- 1 file changed, 15 insertions(+), 5 deletions(-) diff --git a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs index 6086d9d8f8..1151f0233c 100644 --- a/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs +++ b/libdd-data-pipeline/tests/test_trace_exporter_otlp_grpc.rs @@ -118,11 +118,21 @@ mod grpc_export_tests { }); } - let should_fail = failures_remaining - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |remaining| { - remaining.checked_sub(1) - }) - .is_ok(); + let mut remaining = failures_remaining.load(Ordering::Relaxed); + let should_fail = loop { + let Some(next) = remaining.checked_sub(1) else { + break false; + }; + match failures_remaining.compare_exchange_weak( + remaining, + next, + Ordering::Relaxed, + Ordering::Relaxed, + ) { + Ok(_) => break true, + Err(current) => remaining = current, + } + }; let response = http::Response::builder() .status(200) From a6ae5e4719e77f36c8849b984986d4093dc4428d Mon Sep 17 00:00:00 2001 From: Brian Marks Date: Thu, 3 Sep 2026 00:44:21 -0400 Subject: [PATCH 6/6] fix(data-pipeline): preserve gRPC transport errors --- libdd-data-pipeline/src/otlp/grpc_exporter.rs | 82 +++++++++++-------- 1 file changed, 47 insertions(+), 35 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/grpc_exporter.rs b/libdd-data-pipeline/src/otlp/grpc_exporter.rs index 16a5c03628..41f77e47f7 100644 --- a/libdd-data-pipeline/src/otlp/grpc_exporter.rs +++ b/libdd-data-pipeline/src/otlp/grpc_exporter.rs @@ -454,42 +454,37 @@ fn partial_success_details( fn grpc_status_to_error(status: Status) -> GrpcExportError { let retry_after = retry_info_delay(&status); - // Transport/IO failures from our bare `H2Service` fall through to `Code::Unknown` with the - // original `std::io::Error` somewhere in the source chain (directly for a connect failure, or - // wrapped in a `hyper::Error` for a post-connect handshake/read/write failure). Walk the chain - // and recover it so these map to `Io` rather than an application-level `Request` error. - if status.code() == Code::Unknown { - let mut cause: Option<&(dyn std::error::Error + 'static)> = status.source(); - while let Some(err) = cause { - if let Some(io_err) = err.downcast_ref::() { - return GrpcExportError::Retryable { - error: TraceExporterError::Io(std::io::Error::new( - io_err.kind(), - io_err.to_string(), - )), - retry_after, - }; - } - let hyper_retryable = err - .downcast_ref::() - .is_some_and(|error| error.is_canceled() || error.is_closed()); - let h2_retryable = err.downcast_ref::().is_some_and(|error| { - matches!( - error.reason(), - Some(h2::Reason::REFUSED_STREAM | h2::Reason::CANCEL) - ) - }); - if hyper_retryable || h2_retryable { - return GrpcExportError::Retryable { - error: TraceExporterError::Io(std::io::Error::new( - std::io::ErrorKind::ConnectionReset, - err.to_string(), - )), - retry_after, - }; - } - cause = err.source(); + // Recover local transport failures before classifying source-less remote gRPC statuses. + let mut cause: Option<&(dyn std::error::Error + 'static)> = status.source(); + while let Some(err) = cause { + if let Some(io_err) = err.downcast_ref::() { + return GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + io_err.kind(), + io_err.to_string(), + )), + retry_after, + }; + } + let hyper_retryable = err + .downcast_ref::() + .is_some_and(|error| error.is_canceled() || error.is_closed()); + let h2_retryable = err.downcast_ref::().is_some_and(|error| { + matches!( + error.reason(), + Some(h2::Reason::REFUSED_STREAM | h2::Reason::CANCEL) + ) + }); + if hyper_retryable || h2_retryable { + return GrpcExportError::Retryable { + error: TraceExporterError::Io(std::io::Error::new( + std::io::ErrorKind::ConnectionReset, + err.to_string(), + )), + retry_after, + }; } + cause = err.source(); } let code = status.code(); let (http_status, retryable) = match code { @@ -1078,6 +1073,23 @@ mod send_tests { } } + #[test] + fn status_unavailable_with_io_source_maps_to_io() { + let io_err = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "reset"); + let mut status = Status::unavailable("transport failure"); + status.set_source(Arc::new(io_err)); + assert_eq!(status.code(), Code::Unavailable); + match grpc_status_to_error(status) { + GrpcExportError::Retryable { + error: TraceExporterError::Io(e), + retry_after: None, + } => { + assert_eq!(e.kind(), std::io::ErrorKind::ConnectionReset) + } + other => panic!("expected retryable Io, got {other:?}"), + } + } + #[test] fn status_unknown_without_io_source_maps_to_request() { assert!(matches!(