split collector as a multi-module crate

Author: rochdev

collector/Cargo.lock

@@ -1,5 +1,5 @@
 [package]
-name = "dd-trace-collector"
+name = "ddcollector"
 version = "0.1.0"
 edition = "2021"
 
@@ -13,6 +13,8 @@ strip = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-trait = "0.1.64"
+hashbrown = "0.13.2"
 hyper = { version = "0.14.24", features = ["full"] }
 rmp = "0.8.11"
 tokio = { version = "1.25.0", features = ["full"] }

collector/Cargo.toml

@@ -0,0 +1,9 @@
+use crate::tracing::{Traces};
+use async_trait::async_trait;
+
+pub mod agent;
+
+#[async_trait]
+pub trait Exporter {
+    async fn export(&self, traces: Traces);
+}

collector/src/exporting.rs

@@ -0,0 +1,117 @@
+use crate::tracing::{Trace, Traces};
+use super::Exporter;
+use async_trait::async_trait;
+use rmp::encode;
+use rmp::encode::ByteBuf;
+use hashbrown::HashMap;
+
+pub struct AgentExporter {}
+
+#[async_trait]
+impl Exporter for AgentExporter {
+    async fn export(&self, traces: Traces) {
+        let mut wr = ByteBuf::new();
+        let trace_count = traces.len();
+
+        if trace_count > 0 {
+            // println!("{:#?}", traces);
+
+            self.encode_traces(&mut wr, traces);
+
+            let client = hyper::Client::new();
+            let data: Vec<u8> = wr.as_vec().to_vec();
+            let req = hyper::Request::builder()
+                .method(hyper::Method::PUT)
+                .uri("http://localhost:8126/v0.4/traces")
+                .header("Content-Type", "application/msgpack")
+                .header("X-Datadog-Trace-Count", trace_count.to_string())
+                // .header("Datadog-Meta-Tracer-Version", "")
+                // .header("Datadog-Meta-Lang", "")
+                // .header("Datadog-Meta-Lang-Version", "")
+                // .header("Datadog-Meta-Lang-Interpreter", "")
+                .body(hyper::Body::from(data))
+                .unwrap();
+
+            client.request(req).await.unwrap();
+        }
+    }
+}
+
+impl Default for AgentExporter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AgentExporter {
+    pub fn new() -> Self {
+        Self {}
+    }
+
+    fn encode_traces(&self, wr: &mut ByteBuf, traces: Traces) {
+        encode::write_array_len(wr, traces.len() as u32).unwrap();
+
+        for trace in traces.values() {
+            self.encode_trace(wr, trace);
+        }
+    }
+
+    fn encode_trace(&self, wr: &mut ByteBuf, trace: &Trace) {
+        encode::write_array_len(wr, trace.spans.len() as u32).unwrap();
+
+        for span in trace.spans.values() {
+            match &span.span_type {
+                Some(span_type) => {
+                    encode::write_map_len(wr, 12).unwrap();
+                    encode::write_str(wr, "type").unwrap();
+                    encode::write_str(wr, span_type.as_str()).unwrap();
+                },
+                None => {
+                    encode::write_map_len(wr, 11).unwrap();
+                }
+            }
+
+            encode::write_str(wr, "trace_id").unwrap();
+            encode::write_uint(wr, span.trace_id).unwrap();
+            encode::write_str(wr, "span_id").unwrap();
+            encode::write_uint(wr, span.span_id).unwrap();
+            encode::write_str(wr, "parent_id").unwrap();
+            encode::write_uint(wr, span.parent_id).unwrap();
+            encode::write_str(wr, "name").unwrap();
+            encode::write_str(wr, span.name.as_str()).unwrap();
+            encode::write_str(wr, "resource").unwrap();
+            encode::write_str(wr, span.resource.as_str()).unwrap();
+            encode::write_str(wr, "service").unwrap();
+            encode::write_str(wr, span.service.as_str()).unwrap();
+            encode::write_str(wr, "error").unwrap();
+            encode::write_uint(wr, span.error).unwrap();
+            encode::write_str(wr, "start").unwrap();
+            encode::write_uint(wr, span.start).unwrap();
+            encode::write_str(wr, "duration").unwrap();
+            encode::write_uint(wr, span.duration + 1).unwrap();
+
+            self.encode_meta(wr, &span.meta);
+            self.encode_metrics(wr, &span.metrics);
+        }
+    }
+
+    fn encode_meta(&self, wr: &mut ByteBuf, map: &HashMap<String, String>) {
+        encode::write_str(wr, "meta").unwrap();
+        encode::write_map_len(wr, map.len() as u32).unwrap();
+
+        for (k, v) in map {
+            encode::write_str(wr, k.as_str()).unwrap();
+            encode::write_str(wr, v.as_str()).unwrap();
+        }
+    }
+
+    fn encode_metrics(&self, wr: &mut ByteBuf, map: &HashMap<String, f64>) {
+        encode::write_str(wr, "metrics").unwrap();
+        encode::write_map_len(wr, map.len() as u32).unwrap();
+
+        for (k, v) in map {
+            encode::write_str(wr, k.as_str()).unwrap();
+            encode::write_f64(wr, *v).unwrap();
+        }
+    }
+}

collector/src/exporting/agent.rs

@@ -0,0 +1,4 @@
+pub mod exporting;
+pub mod msgpack;
+pub mod processing;
+pub mod tracing;