feat(data-pipeline-ffi): add functions to manipulate span from C (#994)

Co-authored-by: bwoebi <[email protected]>

Author: Leiyks

Cargo.lock

@@ -31,4 +31,4 @@ datadog-trace-utils = { path = "../datadog-trace-utils" }
 data-pipeline = { path = "../data-pipeline" }
 ddcommon-ffi = { path = "../ddcommon-ffi", default-features = false }
 tinybytes = { path = "../tinybytes" }
-tracing = { version = "0.1", default-features = false }
+tracing = { version = "0.1", default-features = false }

data-pipeline-ffi/Cargo.toml

@@ -28,6 +28,7 @@ use crate::{
 use arc_swap::{ArcSwap, ArcSwapOption};
 use bytes::Bytes;
 use datadog_trace_utils::msgpack_decoder::{self, decode::error::DecodeError};
+use datadog_trace_utils::msgpack_encoder;
 use datadog_trace_utils::send_with_retry::{
     send_with_retry, RetryStrategy, SendWithRetryError, SendWithRetryResult,
 };
@@ -1057,9 +1058,7 @@ impl TraceExporter {
         payload: &tracer_payload::TraceChunks<T>,
     ) -> Result<Vec<u8>, TraceExporterError> {
         match payload {
-            tracer_payload::TraceChunks::V04(p) => {
-                rmp_serde::to_vec_named(p).map_err(TraceExporterError::Serialization)
-            }
+            tracer_payload::TraceChunks::V04(p) => Ok(msgpack_encoder::v04::to_vec(p)),
             tracer_payload::TraceChunks::V05(p) => {
                 rmp_serde::to_vec(p).map_err(TraceExporterError::Serialization)
             }
@@ -1544,7 +1543,7 @@ mod tests {
             ..Default::default()
         }];
 
-        let data = rmp_serde::to_vec_named(&vec![trace_chunk]).unwrap();
+        let data = msgpack_encoder::v04::to_vec(&[trace_chunk]);
 
         // Wait for the info fetcher to get the config
         while agent_info::get_agent_info().is_none() {
@@ -1645,7 +1644,7 @@ mod tests {
             ..Default::default()
         }];
 
-        let data = rmp_serde::to_vec_named(&vec![trace_chunk]).unwrap();
+        let data = msgpack_encoder::v04::to_vec(&[trace_chunk]);
 
         // Wait for agent_info to be present so that sending a trace will trigger the stats worker
         // to start
@@ -1751,7 +1750,7 @@ mod tests {
                 ..Default::default()
             }],
         ];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
 
         let _result = exporter
             .send(data.as_ref(), 2)
@@ -1851,7 +1850,7 @@ mod tests {
             name: BytesString::from_slice(b"test").unwrap(),
             ..Default::default()
         }]];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
         let result = exporter.send(data.as_ref(), 1);
 
         assert!(result.is_err());
@@ -1958,7 +1957,7 @@ mod tests {
             name: BytesString::from_slice(b"test").unwrap(),
             ..Default::default()
         }]];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
         let result = exporter.send(data.as_ref(), 1);
 
         assert!(result.is_err());
@@ -2068,7 +2067,7 @@ mod tests {
             name: BytesString::from_slice(b"test").unwrap(),
             ..Default::default()
         }]];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
         let result = exporter.send(data.as_ref(), 1).unwrap();
 
         assert_eq!(
@@ -2110,7 +2109,7 @@ mod tests {
             name: BytesString::from_slice(b"test").unwrap(),
             ..Default::default()
         }]];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
         let code = match exporter.send(data.as_ref(), 1).unwrap_err() {
             TraceExporterError::Request(e) => Some(e.status()),
             _ => None,
@@ -2145,7 +2144,7 @@ mod tests {
             name: BytesString::from_slice(b"test").unwrap(),
             ..Default::default()
         }]];
-        let data = rmp_serde::to_vec_named(&traces).expect("failed to serialize static trace");
+        let data = msgpack_encoder::v04::to_vec(&traces);
         let err = exporter.send(data.as_ref(), 1);
 
         assert!(err.is_err());
@@ -2513,7 +2512,7 @@ mod tests {
             ..Default::default()
         }];
 
-        let data = rmp_serde::to_vec_named(&vec![trace_chunk]).unwrap();
+        let data = msgpack_encoder::v04::to_vec(&[trace_chunk]);
 
         // Wait for the info fetcher to get the config
         while mock_info.hits() == 0 {

data-pipeline/src/trace_exporter/mod.rs

@@ -20,9 +20,13 @@ ddcommon-ffi = { path = "../ddcommon-ffi", default-features = false }
 ddtelemetry-ffi = { path = "../ddtelemetry-ffi", default-features = false }
 datadog-remote-config = { path = "../datadog-remote-config" }
 datadog-live-debugger = { path = "../datadog-live-debugger" }
+dogstatsd-client = { path = "../dogstatsd-client" }
+tinybytes = { path = "../tinybytes", features = ["bytes_string"] }
 paste = "1"
 libc = "0.2"
-dogstatsd-client = { path = "../dogstatsd-client" }
+tracing = { version = "0.1", default-features = false }
+rmp-serde = "1.1.1"
+
 
 [target.'cfg(windows)'.dependencies]
 datadog-crashtracker-ffi = { path = "../datadog-crashtracker-ffi", features = ["collector", "collector_windows"] }

datadog-sidecar-ffi/Cargo.toml

@@ -5,6 +5,12 @@ language = "C"
 tab_width = 2
 header = """// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
+
+typedef struct ddog_SpanBytes ddog_SpanBytes;
+typedef struct ddog_SpanLinkBytes ddog_SpanLinkBytes;
+typedef struct ddog_SpanEventBytes ddog_SpanEventBytes;
+typedef struct ddog_AttributeAnyValueBytes ddog_AttributeAnyValueBytes;
+typedef struct ddog_AttributeArrayValueBytes ddog_AttributeArrayValueBytes;
 """
 include_guard = "DDOG_SIDECAR_H"
 style = "both"

datadog-sidecar-ffi/cbindgen.toml

@@ -7,6 +7,9 @@
 #![cfg_attr(not(test), deny(clippy::todo))]
 #![cfg_attr(not(test), deny(clippy::unimplemented))]
 
+pub mod span;
+
+use crate::span::TracesBytes;
 #[cfg(windows)]
 use datadog_crashtracker_ffi::Metadata;
 use datadog_ipc::platform::{
@@ -29,14 +32,14 @@ use datadog_sidecar::service::{
 };
 use datadog_sidecar::service::{get_telemetry_action_sender, InternalTelemetryActions};
 use datadog_sidecar::shm_remote_config::{path_for_remote_config, RemoteConfigReader};
+use datadog_trace_utils::msgpack_encoder;
 use ddcommon::tag::Tag;
 use ddcommon::Endpoint;
 use ddcommon_ffi as ffi;
 use ddcommon_ffi::{CharSlice, MaybeError};
-use ddtelemetry::worker::LogIdentifier;
 use ddtelemetry::{
     data::{self, Dependency, Integration},
-    worker::{LifecycleAction, TelemetryActions},
+    worker::{LifecycleAction, LogIdentifier, TelemetryActions},
 };
 use ddtelemetry_ffi::try_c;
 use dogstatsd_client::DogStatsDActionOwned;
@@ -1111,6 +1114,111 @@ pub unsafe extern "C" fn ddog_get_agent_info_env<'a>(
         .unwrap_or(ffi::CharSlice::empty())
 }
 
+#[macro_export]
+macro_rules! check {
+    ($failable:expr, $msg:expr) => {
+        match $failable {
+            Ok(o) => o,
+            Err(e) => {
+                tracing::error!("{}: {}", $msg, e);
+                return;
+            }
+        }
+    };
+}
+
+#[repr(C)]
+#[derive()]
+pub struct SenderParameters {
+    pub tracer_headers_tags: TracerHeaderTags<'static>,
+    pub transport: Box<SidecarTransport>,
+    pub instance_id: Box<InstanceId>,
+    pub limit: usize,
+    pub n_requests: i64,
+    pub buffer_size: i64,
+    pub url: CharSlice<'static>,
+}
+
+#[no_mangle]
+#[allow(clippy::missing_safety_doc)]
+pub unsafe extern "C" fn ddog_send_traces_to_sidecar(
+    traces: &mut TracesBytes,
+    parameters: &mut SenderParameters,
+) {
+    let size: usize = traces.iter().map(|trace| trace.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;
+    }
+
+    // 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
+    let mut shm_slice = mapped_shm.as_slice_mut();
+    let shm_slice_len = shm_slice.len();
+    let written = match msgpack_encoder::v04::write_to_slice(&mut shm_slice, traces) {
+        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_v04_shm(
+        &mut parameters.transport,
+        &parameters.instance_id,
+        shm,
+        size_hint,
+        check!(
+            (&parameters.tracer_headers_tags).try_into(),
+            "Failed to convert tracer headers tags"
+        ),
+    );
+
+    // Retry sending traces via bytes if there was an error
+    if send_error.is_err() {
+        match blocking::send_trace_v04_bytes(
+            &mut parameters.transport,
+            &parameters.instance_id,
+            msgpack_encoder::v04::to_vec_with_capacity(traces, written as u32),
+            check!(
+                (&parameters.tracer_headers_tags).try_into(),
+                "Failed to convert tracer headers tags"
+            ),
+        ) {
+            Ok(_) => {}
+            Err(_) => tracing::debug!(
+                "Failed sending traces via shm to sidecar: {}",
+                send_error.err().unwrap_unchecked().to_string()
+            ),
+        };
+    }
+
+    tracing::event!(target: "info", tracing::Level::INFO, "Flushing trace of size {} to send-queue for {}", size, parameters.url);
+    // tracing::info!(
+    //     "Flushing traces of size {} to send-queue for {}",
+    //     size,
+    //     parameters.url
+    // );
+}
+
 /// Drops the agent info reader.
 #[no_mangle]
 #[allow(clippy::missing_safety_doc)]