feat(sidecar)!: V1 trace builder and send FFI for tracers

datadog-sidecar-ffi/src/lib.rs

@@ -12,8 +12,10 @@ pub mod remote_config_notification;
 #[cfg(target_os = "linux")]
 pub mod signal_flush;
 pub mod span;
+pub mod span_v04;
 
-use crate::span::TracesBytes;
+use crate::span::{TracerPayloadV1Builder, populate_payload_metadata};
+use crate::span_v04::TracesBytes;
 use datadog_sidecar::agent_remote_config::{AgentRemoteConfigWriter, new_reader, reader_from_shm};
 use datadog_sidecar::config;
 use datadog_sidecar::config::LogMethod;
@@ -691,6 +693,7 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
     root_service: ffi::CharSlice,
     root_session_id: ffi::CharSlice,
     parent_session_id: ffi::CharSlice,
+    force_v04_traces: bool,
 ) -> MaybeError {
     unsafe {
         let session_id_str: String = session_id.to_utf8_lossy().into();
@@ -751,6 +754,7 @@ pub unsafe extern "C" fn ddog_sidecar_session_set_config(
                 Some(parent_session_id.to_utf8_lossy().into())
             },
             retry_interval: Duration::from_millis(retry_interval_milliseconds as u64),
+            force_v04_traces,
         };
         #[cfg(unix)]
         let _ = win_remote_config_notification;
@@ -2010,6 +2014,13 @@ pub struct SenderParameters {
     pub url: CharSlice<'static>,
 }
 
+/// Payload-level tracer metadata for the V1 send path not already carried by the sender's
+/// `tracer_headers_tags` (lang, tracer_version, container_id live there and are routed from there).
+#[repr(C)]
+pub struct TracerMetadataV1 {
+    pub runtime_id: CharSlice<'static>,
+}
+
 #[unsafe(no_mangle)]
 #[allow(clippy::missing_safety_doc)]
 pub unsafe extern "C" fn ddog_send_traces_to_sidecar(
@@ -2099,6 +2110,143 @@ pub unsafe extern "C" fn ddog_send_traces_to_sidecar(
     }
 }
 
+/// V1 counterpart of `ddog_send_traces_to_sidecar`: sends the [`crate::span`] builder's native V1
+/// payload to the agent's `/v1.0/traces`. Consumes `builder`. Payload metadata comes at send time
+/// from `parameters.tracer_headers_tags` and `metadata`; lang_interpreter/lang_vendor go as
+/// headers.
+#[unsafe(no_mangle)]
+#[allow(clippy::missing_safety_doc)]
+pub unsafe extern "C" fn ddog_send_traces_to_sidecar_v1(
+    builder: Box<TracerPayloadV1Builder>,
+    parameters: &mut SenderParameters,
+    metadata: &TracerMetadataV1,
+) {
+    let mut payload = builder.into_payload();
+    let size: usize = payload.chunks.iter().map(|c| c.spans.len()).sum();
+
+    // Check connection to the sidecar
+    if parameters.transport.is_closed() {
+        tracing::info!(
+            "Skipping flushing traces of size {} as connection to sidecar failed",
+            size
+        );
+        return;
+    }
+
+    let tags = &parameters.tracer_headers_tags;
+    populate_payload_metadata(
+        &mut payload,
+        &tags.container_id.to_utf8_lossy(),
+        &tags.lang.to_utf8_lossy(),
+        &tags.lang_version.to_utf8_lossy(),
+        &tags.tracer_version.to_utf8_lossy(),
+        &metadata.runtime_id.to_utf8_lossy(),
+    );
+
+    let lang_interpreter = parameters
+        .tracer_headers_tags
+        .lang_interpreter
+        .to_utf8_lossy()
+        .into_owned();
+    let lang_vendor = parameters
+        .tracer_headers_tags
+        .lang_vendor
+        .to_utf8_lossy()
+        .into_owned();
+
+    // Create and map shared memory
+    let shm = check!(
+        ShmHandle::new(parameters.limit),
+        "Failed to create shared memory"
+    );
+
+    let mut mapped_shm = check!(shm.clone().map(), "Failed to map shared memory");
+
+    // Write traces to the shared memory as a native V1 payload
+    let mut shm_slice = mapped_shm.as_slice_mut();
+    let shm_slice_len = shm_slice.len();
+    let written = match msgpack_encoder::v1::write_to_slice_from_v1(&mut shm_slice, &payload) {
+        Ok(()) => shm_slice_len - shm_slice.len(),
+        Err(_) => {
+            tracing::error!("Failed serializing the traces");
+            return;
+        }
+    };
+
+    // Send traces to the sidecar via the shared memory handler
+    let mut size_hint = written;
+    if parameters.n_requests > 0 {
+        size_hint = size_hint.max((parameters.buffer_size / parameters.n_requests + 1) as usize);
+    }
+
+    let send_error = blocking::send_trace_v1_shm(
+        &mut parameters.transport,
+        &parameters.instance_id,
+        shm,
+        size_hint,
+        TracerGenericTags {
+            client_computed_top_level: parameters.tracer_headers_tags.client_computed_top_level,
+            client_computed_stats: parameters.tracer_headers_tags.client_computed_stats,
+            ..Default::default()
+        },
+        lang_interpreter,
+        lang_vendor,
+    );
+
+    // Retry sending traces via bytes if there was an error. `lang_interpreter`/`lang_vendor` were
+    // moved into the shm send above, so re-derive them here (only on this rare path).
+    if send_error.is_err() {
+        match blocking::send_trace_v1_bytes(
+            &mut parameters.transport,
+            &parameters.instance_id,
+            msgpack_encoder::v1::to_vec_with_capacity_from_v1(&payload, written as u32),
+            TracerGenericTags {
+                client_computed_top_level: parameters.tracer_headers_tags.client_computed_top_level,
+                client_computed_stats: parameters.tracer_headers_tags.client_computed_stats,
+                ..Default::default()
+            },
+            parameters
+                .tracer_headers_tags
+                .lang_interpreter
+                .to_utf8_lossy()
+                .into_owned(),
+            parameters
+                .tracer_headers_tags
+                .lang_vendor
+                .to_utf8_lossy()
+                .into_owned(),
+        ) {
+            Ok(_) => {}
+            Err(_) => tracing::debug!(
+                "Failed sending traces via shm to sidecar: {}",
+                unsafe { send_error.err().unwrap_unchecked() }.to_string()
+            ),
+        };
+    }
+
+    tracing::event!(target: "info", tracing::Level::INFO, "Flushing v1 trace of size {} to send-queue for {}", size, parameters.url);
+}
+
+/// Downgrades a native V1 builder to the in-memory v0.4 collection for the in-process `coms.c`
+/// sender (PHP <= 8.2). Consumes `builder`. Returns the collection (not one whole-payload
+/// CharSlice) so `auto_flush` can frame each trace individually; a single CharSlice would drop
+/// extra traces of a multi-trace payload. Empty collection on error; free with
+/// [`crate::span_v04::ddog_free_traces`].
+#[unsafe(no_mangle)]
+pub extern "C" fn ddog_downgrade_v1_builder_to_v04_traces(
+    builder: Box<TracerPayloadV1Builder>,
+) -> Box<TracesBytes> {
+    let payload = builder.into_payload();
+    let v04_bytes = msgpack_encoder::v04::to_vec_from_v1(&payload);
+    match libdd_trace_utils::tracer_payload::decode_to_trace_chunks(
+        libdd_tinybytes::Bytes::from(v04_bytes),
+        libdd_trace_utils::tracer_payload::TraceEncoding::V04,
+    ) {
+        Ok((libdd_trace_utils::tracer_payload::TraceChunks::V04(traces), _)) => Box::new(traces),
+        _ => Box::default(),
+    }
+}
+
 /// Drops the agent info reader.
 #[unsafe(no_mangle)]
 #[allow(clippy::missing_safety_doc)]
@@ -2223,6 +2371,50 @@ mod tests {
         }
     }
 
+    #[test]
+    fn downgrade_v1_builder_to_v04_traces_preserves_all_traces() {
+        use crate::span::ddog_v1_new_builder;
+        use crate::span_v04::{ddog_free_traces, ddog_serialize_trace_into_charslice};
+        use libdd_tinybytes::BytesString;
+
+        // Two chunks (traces) with a shared service; the whole-payload CharSlice form drops the
+        // second under the coms framing, the TracesBytes form must keep both.
+        let mut builder = ddog_v1_new_builder();
+        let c0 = builder.push_chunk(0, 1);
+        // Safety: `c0`/`c1` are live chunk nodes, `s0`/`s1` their spans (Box-per-node handles).
+        unsafe {
+            let sp = (*(*c0).push_span()).span_mut();
+            sp.service = BytesString::from_slice(b"svc-shared").unwrap();
+            sp.name = BytesString::from_slice(b"op-one").unwrap();
+            sp.span_id = 1;
+        }
+        let c1 = builder.push_chunk(0, 2);
+        unsafe {
+            let sp = (*(*c1).push_span()).span_mut();
+            sp.service = BytesString::from_slice(b"svc-shared").unwrap();
+            sp.name = BytesString::from_slice(b"op-two").unwrap();
+            sp.span_id = 2;
+        }
+
+        let mut traces = ddog_downgrade_v1_builder_to_v04_traces(builder);
+        assert_eq!(traces.len(), 2, "both traces must survive the downgrade");
+        assert_eq!(traces[0].len(), 1);
+        assert_eq!(traces[1].len(), 1);
+        assert_eq!(traces[0][0].service.as_str(), "svc-shared");
+        assert_eq!(traces[0][0].name.as_str(), "op-one");
+        assert_eq!(traces[1][0].name.as_str(), "op-two");
+
+        // Each trace serializes to a valid, non-empty msgpack array-of-1 for the background sender.
+        for i in 0..traces.len() {
+            let slice = ddog_serialize_trace_into_charslice(&mut traces[i]);
+            assert!(!slice.is_empty());
+            // Safety: `slice` is the owned allocation just returned above.
+            unsafe { crate::span::ddog_free_charslice(slice) };
+        }
+
+        ddog_free_traces(traces);
+    }
+
     #[test]
     fn otlp_metrics_endpoint_inherits_agent_test_token_when_missing() {
         let agent_endpoint = Endpoint {