feat: support dd integration

agent/crates/public/src/sender.rs

@@ -56,6 +56,7 @@ pub enum SendMessageType {
     ApplicationLog = 17,
     SyslogDetail = 18,
     SkyWalking = 19,
+    Datadog = 20,
 }
 
 impl fmt::Display for SendMessageType {
@@ -80,6 +81,7 @@ impl fmt::Display for SendMessageType {
             Self::ApplicationLog => write!(f, "application_log"),
             Self::SyslogDetail => write!(f, "syslog_detail"),
             Self::SkyWalking => write!(f, "skywalking"),
+            Self::Datadog => write!(f, "datadog"),
         }
     }
 }

agent/plugins/integration_skywalking/src/lib.rs

@@ -24,7 +24,7 @@ use public::{
 use std::net::SocketAddr;
 
 #[derive(Debug, PartialEq)]
-pub struct SkyWalkingExtra(pub flow_log::SkyWalkingExtra);
+pub struct SkyWalkingExtra(pub flow_log::ThirdPartyTrace);
 
 impl Sendable for SkyWalkingExtra {
     fn encode(self, _: &mut Vec<u8>) -> Result<usize, EncodeError> {

agent/src/config/handler.rs

@@ -1158,6 +1158,7 @@ pub enum TraceType {
     NewRpcTraceContext,
     XTingyun(String),
     Customize(String),
+    Ddog,
 }
 
 // The value here must be lower case
@@ -1169,6 +1170,7 @@ const TRACE_TYPE_SW6: &str = "sw6";
 const TRACE_TYPE_SW8: &str = "sw8";
 const TRACE_TYPE_TRACE_PARENT: &str = "traceparent";
 const TRACE_TYPE_X_TINGYUN: &str = "x-tingyun";
+const TRACE_TYPE_DDOG: &str = "x-datadog-trace-id";
 
 impl From<&str> for TraceType {
     // The parameter supports the following two formats:
@@ -1194,6 +1196,7 @@ impl From<&str> for TraceType {
             TRACE_TYPE_TRACE_PARENT => TraceType::TraceParent,
             SOFA_NEW_RPC_TRACE_CTX_KEY => TraceType::NewRpcTraceContext,
             TRACE_TYPE_X_TINGYUN => TraceType::XTingyun(sub_tag),
+            TRACE_TYPE_DDOG => TraceType::Ddog,
             _ if tag.len() > 0 => TraceType::Customize(tag),
             _ => TraceType::Disabled,
         }
@@ -1215,6 +1218,7 @@ impl TraceType {
             }
             TraceType::XTingyun(_) => context.eq_ignore_ascii_case(TRACE_TYPE_X_TINGYUN),
             TraceType::Customize(tag) => context.eq_ignore_ascii_case(&tag),
+            TraceType::Ddog => context.eq_ignore_ascii_case(TRACE_TYPE_DDOG),
             _ => false,
         }
     }
@@ -1231,6 +1235,7 @@ impl TraceType {
             TraceType::NewRpcTraceContext => SOFA_NEW_RPC_TRACE_CTX_KEY,
             TraceType::XTingyun(_) => TRACE_TYPE_X_TINGYUN,
             TraceType::Customize(tag) => &tag,
+            TraceType::Ddog => TRACE_TYPE_DDOG,
             _ => "",
         }
     }
@@ -1318,6 +1323,25 @@ impl TraceType {
         }
     }
 
+    // Datadog HTTP trace format:
+    // X-Datadog-Trace-Id: 5634230643731801278 (u64)
+    fn decode_ddog_id(value: &str, id_type: u8) -> Option<Cow<'_, str>> {
+        if id_type == Self::TRACE_ID {
+            let trace_id: u64 = value.parse().unwrap();
+            let mut trace_id_bytes = [0u8; 16];
+            // only use low 8 bits
+            trace_id_bytes[8..].copy_from_slice(&trace_id.to_be_bytes());
+            Some(Cow::Owned(hex::encode(trace_id_bytes)))
+        } else if id_type == Self::SPAN_ID {
+            let span_id: u64 = value.parse().unwrap();
+            let mut span_id_bytes = [0u8; 8];
+            span_id_bytes[..].copy_from_slice(&span_id.to_be_bytes());
+            Some(Cow::Owned(hex::encode(span_id_bytes)))
+        } else {
+            unreachable!()
+        }
+    }
+
     fn decode_tingyun<'a, 'b>(value: &'a str, sub_tag: &'b str) -> Option<Cow<'a, str>> {
         cloud_platform::tingyun::decode_trace_id(value, sub_tag)
     }
@@ -1339,6 +1363,7 @@ impl TraceType {
                 decode_new_rpc_trace_context_with_type(value.as_bytes(), id_type)
             }
             TraceType::XTingyun(sub_tag) => Self::decode_tingyun(value, sub_tag),
+            TraceType::Ddog => Self::decode_ddog_id(value, id_type),
         }
     }
 

agent/src/integration_collector.rs

@@ -70,6 +70,7 @@ use public::{
     l7_protocol::L7Protocol,
     proto::{
         agent::Exception,
+        flow_log,
         integration::opentelemetry::proto::{
             common::v1::{
                 any_value::Value::{IntValue, StringValue},
@@ -226,6 +227,19 @@ async fn aggregate_with_catch_exception(
     })
 }
 
+#[derive(Debug, PartialEq)]
+pub struct Datadog(flow_log::ThirdPartyTrace);
+
+impl Sendable for Datadog {
+    fn encode(self, buf: &mut Vec<u8>) -> Result<usize, prost::EncodeError> {
+        self.0.encode(buf).map(|_| self.0.encoded_len())
+    }
+
+    fn message_type(&self) -> SendMessageType {
+        SendMessageType::Datadog
+    }
+}
+
 // for log capture from vector
 #[derive(Debug, PartialEq)]
 pub struct ApplicationLog(Vec<u8>);
@@ -599,6 +613,7 @@ async fn handler(
     profile_sender: DebugSender<Profile>,
     application_log_sender: DebugSender<ApplicationLog>,
     skywalking_sender: DebugSender<SkyWalkingExtra>,
+    datadog_sender: DebugSender<Datadog>,
     exception_handler: ExceptionHandler,
     compressed: bool,
     profile_compressed: bool,
@@ -862,6 +877,37 @@ async fn handler(
             )
             .await)
         }
+        // TODO: confirm regex/fuzzy path
+        // path: /api/v0.2,v0.3,v0.4,v0.5,v0.7,
+        // TODO: repeated code, consider make it common
+        (&Method::POST, "/v0.4/traces") => {
+            if external_trace_integration_disabled {
+                return Ok(Response::builder().body(Body::empty()).unwrap());
+            }
+            let (part, body) = req.into_parts();
+            let whole_body = match aggregate_with_catch_exception(body, &exception_handler).await {
+                Ok(b) => b,
+                Err(e) => {
+                    return Ok(e);
+                }
+            };
+
+            let mut third_party_data = flow_log::ThirdPartyTrace::default();
+            parse_dd_headers(&part.headers, &mut third_party_data);
+            let datadog = decode_metric(whole_body, &part.headers)?;
+            third_party_data.data = datadog;
+            third_party_data.uri = part.uri.path().to_string();
+            third_party_data.peer_ip = match peer_addr.ip() {
+                IpAddr::V4(ip4) => ip4.octets().to_vec(),
+                IpAddr::V6(ip6) => ip6.octets().to_vec(),
+            };
+
+            if let Err(e) = datadog_sender.send(Datadog(third_party_data)) {
+                warn!("datadog_sender failed to send data, because {:?}", e);
+            }
+
+            Ok(Response::builder().body(Body::empty()).unwrap())
+        }
         // Return the 404 Not Found for other routes.
         _ => Ok(Response::builder()
             .status(StatusCode::NOT_FOUND)
@@ -904,6 +950,23 @@ fn parse_profile_query(query: &str, profile: &mut metric::Profile) {
     };
 }
 
+fn parse_dd_headers(headers: &HeaderMap, third_party_data: &mut flow_log::ThirdPartyTrace) {
+    for key in vec![
+        "Datadog-Meta-Lang",           // headers.lang
+        "Datadog-Meta-Lang-Version",   // headers.lang_version
+        "Datadog-Meta-Tracer-Version", // headers.tracer_version
+        "Datadog-Container-Id",        // headers.container_id
+        "Content-Type",                // for decode format validate
+    ] {
+        if let Some(value) = headers.get(key) {
+            third_party_data.extend_keys.push(key.to_string());
+            third_party_data
+                .extend_values
+                .push(value.to_str().unwrap().to_string());
+        }
+    }
+}
+
 #[derive(Default)]
 struct CompressedMetric {
     compressed: AtomicU64,   // unit (bytes)
@@ -963,6 +1026,7 @@ pub struct MetricServer {
     profile_sender: DebugSender<Profile>,
     application_log_sender: DebugSender<ApplicationLog>,
     skywalking_sender: DebugSender<SkyWalkingExtra>,
+    datadog_sender: DebugSender<Datadog>,
     port: Arc<AtomicU16>,
     exception_handler: ExceptionHandler,
     server_shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
@@ -991,6 +1055,7 @@ impl MetricServer {
         profile_sender: DebugSender<Profile>,
         application_log_sender: DebugSender<ApplicationLog>,
         skywalking_sender: DebugSender<SkyWalkingExtra>,
+        datadog_sender: DebugSender<Datadog>,
         port: u16,
         exception_handler: ExceptionHandler,
         compressed: bool,
@@ -1020,6 +1085,7 @@ impl MetricServer {
                 profile_sender,
                 application_log_sender,
                 skywalking_sender,
+                datadog_sender,
                 port: Arc::new(AtomicU16::new(port)),
                 exception_handler,
                 server_shutdown_tx: Default::default(),
@@ -1070,6 +1136,7 @@ impl MetricServer {
         let profile_sender = self.profile_sender.clone();
         let application_log_sender = self.application_log_sender.clone();
         let skywalking_sender = self.skywalking_sender.clone();
+        let datadog_sender = self.datadog_sender.clone();
         let port = self.port.clone();
         let monitor_port = Arc::new(AtomicU16::new(port.load(Ordering::Acquire)));
         let (mon_tx, mon_rx) = oneshot::channel();
@@ -1143,6 +1210,7 @@ impl MetricServer {
                     let profile_sender = profile_sender.clone();
                     let application_log_sender = application_log_sender.clone();
                     let skywalking_sender = skywalking_sender.clone();
+                    let datadog_sender = datadog_sender.clone();
                     let exception_handler_inner = exception_handler.clone();
                     let counter = counter.clone();
                     let compressed = compressed.clone();
@@ -1161,6 +1229,7 @@ impl MetricServer {
                         let profile_sender = profile_sender.clone();
                         let application_log_sender = application_log_sender.clone();
                         let skywalking_sender = skywalking_sender.clone();
+                        let datadog_sender = datadog_sender.clone();
                         let exception_handler = exception_handler_inner.clone();
                         let peer_addr = conn.remote_addr();
                         let counter = counter.clone();
@@ -1185,6 +1254,7 @@ impl MetricServer {
                                     profile_sender.clone(),
                                     application_log_sender.clone(),
                                     skywalking_sender.clone(),
+                                    datadog_sender.clone(),
                                     exception_handler.clone(),
                                     compressed.load(Ordering::Relaxed),
                                     profile_compressed.load(Ordering::Relaxed),

agent/src/trident.rs

@@ -72,8 +72,8 @@ use crate::{
     },
     handler::{NpbBuilder, PacketHandlerBuilder},
     integration_collector::{
-        ApplicationLog, BoxedPrometheusExtra, MetricServer, OpenTelemetry, OpenTelemetryCompressed,
-        Profile, TelegrafMetric,
+        ApplicationLog, BoxedPrometheusExtra, Datadog, MetricServer, OpenTelemetry,
+        OpenTelemetryCompressed, Profile, TelegrafMetric,
     },
     metric::document::BoxedDocument,
     monitor::Monitor,
@@ -1568,6 +1568,7 @@ pub struct AgentComponents {
     pub proc_event_uniform_sender: UniformSenderThread<BoxedProcEvents>,
     pub application_log_uniform_sender: UniformSenderThread<ApplicationLog>,
     pub skywalking_uniform_sender: UniformSenderThread<SkyWalkingExtra>,
+    pub datadog_uniform_sender: UniformSenderThread<Datadog>,
     pub exception_handler: ExceptionHandler,
     pub proto_log_sender: DebugSender<BoxAppProtoLogsData>,
     pub pcap_batch_sender: DebugSender<BoxedPcapBatch>,
@@ -2467,6 +2468,32 @@ impl AgentComponents {
             None,
         );
 
+        let datadog_queue_name = "1-datadog-to-sender";
+        let (datadog_sender, datadog_receiver, counter) = queue::bounded_with_debug(
+            user_config
+                .processors
+                .flow_log
+                .tunning
+                .flow_aggregator_queue_size,
+            datadog_queue_name,
+            &queue_debugger,
+        );
+        stats_collector.register_countable(
+            &QueueStats {
+                module: datadog_queue_name,
+                ..Default::default()
+            },
+            Countable::Owned(Box::new(counter)),
+        );
+        let datadog_uniform_sender = UniformSenderThread::new(
+            datadog_queue_name,
+            Arc::new(datadog_receiver),
+            config_handler.sender(),
+            stats_collector.clone(),
+            exception_handler.clone(),
+            None,
+        );
+
         let ebpf_dispatcher_id = dispatcher_components.len();
         #[cfg(any(target_os = "linux", target_os = "android"))]
         let mut ebpf_dispatcher_component = None;
@@ -2710,6 +2737,7 @@ impl AgentComponents {
             profile_sender,
             application_log_sender,
             skywalking_sender,
+            datadog_sender,
             candidate_config.metric_server.port,
             exception_handler.clone(),
             candidate_config.metric_server.compressed,
@@ -2795,6 +2823,7 @@ impl AgentComponents {
             proc_event_uniform_sender,
             application_log_uniform_sender,
             skywalking_uniform_sender,
+            datadog_uniform_sender,
             capture_mode: candidate_config.capture_mode,
             packet_sequence_uniform_output, // Enterprise Edition Feature: packet-sequence
             packet_sequence_uniform_sender, // Enterprise Edition Feature: packet-sequence
@@ -2884,6 +2913,7 @@ impl AgentComponents {
             self.proc_event_uniform_sender.start();
             self.application_log_uniform_sender.start();
             self.skywalking_uniform_sender.start();
+            self.datadog_uniform_sender.start();
             if self.config.metric_server.enabled {
                 self.metrics_server_component.start();
             }
@@ -2959,6 +2989,9 @@ impl AgentComponents {
         if let Some(h) = self.skywalking_uniform_sender.notify_stop() {
             join_handles.push(h);
         }
+        if let Some(h) = self.datadog_uniform_sender.notify_stop() {
+            join_handles.push(h);
+        }
         // Enterprise Edition Feature: packet-sequence
         if let Some(h) = self.packet_sequence_uniform_sender.notify_stop() {
             join_handles.push(h);

message/flow_log.proto

@@ -296,8 +296,10 @@ message MqttTopic {
     int32 qos = 2; // -1 mean not exist qos
 }
 
-message SkyWalkingExtra {
+message ThirdPartyTrace {
     bytes data = 1;
     bytes peer_ip = 2;
     string uri = 3;
+    repeated string extend_keys = 4;
+    repeated string extend_values = 5;
 }

server/go.mod

@@ -14,6 +14,7 @@ replace (
 	github.com/deepflowio/deepflow/server/controller/http/service/configuration => ./controller/http/service/configuration
 	github.com/deepflowio/deepflow/server/controller/monitor/license => ./controller/monitor/license
 	github.com/deepflowio/deepflow/server/ingester/config/configdefaults => ./ingester/config/configdefaults
+	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/dd_import => ./ingester/flow_log/log_data/dd_import
 	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/sw_import => ./ingester/flow_log/log_data/sw_import
 	github.com/deepflowio/deepflow/server/libs/logger/blocker => ./libs/logger/blocker
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap => ./querier/app/distributed_tracing/service/tracemap
@@ -106,6 +107,7 @@ require (
 	github.com/bytedance/sonic v1.11.8
 	github.com/deepflowio/deepflow/server/controller/http/appender v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/controller/http/service/agentlicense v0.0.0-00010101000000-000000000000
+	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/dd_import v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/ingester/flow_log/log_data/sw_import v0.0.0-00010101000000-000000000000
 	github.com/deepflowio/deepflow/server/libs/logger/blocker v0.0.0-20240822020041-cdaf0f82ce6f
 	github.com/deepflowio/deepflow/server/querier/app/distributed_tracing/service/tracemap v0.0.0-00010101000000-000000000000