Initial commit

Author: davidsteiner

.gitignore

@@ -0,0 +1,3 @@
+/target
+/Cargo.lock
+.idea/

Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "ddtrace"
+version = "0.0.1"
+edition = "2021"
+
+[features]
+axum = ["dep:axum", "dep:axum-tracing-opentelemetry"]
+
+[dependencies]
+axum = { version = "^0.6.10", optional = true }
+axum-tracing-opentelemetry = { version = "^0.10.0", optional = true }
+chrono = "^0.4.24"
+opentelemetry = { version = "^0.18.0", features = ["rt-tokio"] }
+opentelemetry-datadog = "^0.6.0"
+opentelemetry-otlp = { version = "^0.11.0" }
+serde = { version = "^1.0.156", features = ["derive"] }
+serde_json = "^1.0.95"
+tracing = "^0.1.37"
+tracing-opentelemetry = "^0.18.0"
+tracing-serde = "^0.1.3"
+tracing-subscriber = { version = "^0.3.16", features = ["env-filter", "json"] }

README.md

@@ -0,0 +1,59 @@
+This repository is home to a Rust crate `ddtrace` with various Datadog
+utilities for tracing and logging in Rust.
+
+# Background
+
+Datadog has official support for Python, which includes various SDKs and
+other utilities (such as the Python `ddtrace` library)
+for tracing and logging in Python applications.
+
+They don't have similar support for Rust. However, they do support the
+[OpenTelemetry](https://opentelemetry.io/) format for both logs and traces.
+This crate contains the necessary glue to bridge the gap between OpenTelemetry
+and Datadog.
+
+# Features
+## Tracing
+For traces, the official Datadog agent
+[can ingest OTel trace data](https://docs.datadoghq.com/opentelemetry/)
+with the correct environment variable settings. The traces can be sent 
+via either HTTP or gRPC. More information on this can be found here:
+https://docs.datadoghq.com/opentelemetry/otlp_ingest_in_the_agent/?tab=docker
+
+OpenTelemetry has an official Rust crate with extensions for major 
+formats/providers. This includes a datadog exporter. We have found
+this exporter to be less reliable than the standard OTel exporter
+sending data to the OTel endpoint of the Datadog agent, though.
+
+This library provides utilities to set up the Rust `tracing` crate
+for sending data to the agent in the correct way.
+
+## Logging
+Datadog can ingest OpenTelemetry logs with two caveats - 
+it expects the `dd.trace_id` and `dd.span_id` attributes
+to be set, and it expects a slightly different format for
+trace ID.
+
+This crate contains a JSON formatter layer that also correctly
+transform the trace ID to the Datadog native format.
+
+## Propagation
+The Python library takes care of propagation of the trace context automatically.
+Unfortunately, we need to do this manually in Rust. There are many protocols and
+corresponding libraries we could support, but the main one is HTTP requests.
+
+In Rust, `reqwest` is the most commonly used HTTP client crate. We provide a 
+`reqwest` middleware that injects the necessary headers using the Datadog native
+propagation standard (common alternatives would be Jaeger and B3, more on this:
+https://opentelemetry.io/docs/reference/specification/context/api-propagators/#propagators-distribution).
+
+## Axum Support
+The trace context propagated from other services needs to be extracted and injected
+into the propagator. For `axum`, our choice of HTTP API framework, a third-party crate
+exists that supports this: https://github.com/davidB/axum-tracing-opentelemetry.
+
+We re-expose this library for convenience.
+
+# Full Example
+
+TODO

src/formatter.rs

@@ -0,0 +1,109 @@
+use std::io;
+
+use chrono::Utc;
+use opentelemetry::trace::{SpanId, TraceContextExt, TraceId};
+use serde::ser::{SerializeMap, Serializer as _};
+use tracing::{Event, Subscriber};
+use tracing_opentelemetry::OtelData;
+use tracing_serde::fields::AsMap;
+use tracing_serde::AsSerde;
+use tracing_subscriber::fmt::format::Writer;
+use tracing_subscriber::fmt::{FmtContext, FormatEvent, FormatFields};
+use tracing_subscriber::registry::{LookupSpan, SpanRef};
+
+pub struct TraceInfo {
+    pub trace_id: String,
+    pub span_id: String,
+}
+
+fn convert_trace_id(id: TraceId) -> String {
+    let bytes = &id.to_bytes()[std::mem::size_of::<u64>()..std::mem::size_of::<u128>()];
+    let dd_id = u64::from_be_bytes(bytes.try_into().unwrap());
+
+    dd_id.to_string()
+}
+
+fn convert_span_id(id: SpanId) -> String {
+    let dd_id = u64::from_be_bytes(id.to_bytes().try_into().unwrap());
+
+    dd_id.to_string()
+}
+
+fn lookup_trace_info<S>(span_ref: &SpanRef<S>) -> Option<TraceInfo>
+    where
+        S: Subscriber + for<'a> LookupSpan<'a>,
+{
+    span_ref.extensions().get::<OtelData>().map(|o| TraceInfo {
+        trace_id: convert_trace_id(o.parent_cx.span().span_context().trace_id()),
+        span_id: convert_span_id(o.builder.span_id.unwrap_or(SpanId::INVALID)),
+    })
+}
+
+// mostly stolen from here: https://github.com/tokio-rs/tracing/issues/1531
+pub struct TraceIdFormat;
+
+impl<S, N> FormatEvent<S, N> for TraceIdFormat
+    where
+        S: Subscriber + for<'lookup> LookupSpan<'lookup>,
+        N: for<'writer> FormatFields<'writer> + 'static,
+{
+    fn format_event(
+        &self,
+        ctx: &FmtContext<'_, S, N>,
+        mut writer: Writer<'_>,
+        event: &Event<'_>,
+    ) -> std::fmt::Result
+        where
+            S: Subscriber + for<'a> LookupSpan<'a>,
+    {
+        let meta = event.metadata();
+
+        let mut visit = || {
+            let mut serializer = serde_json::Serializer::new(WriteAdaptor::new(&mut writer));
+            let mut serializer = serializer.serialize_map(None)?;
+            serializer.serialize_entry("timestamp", &Utc::now().to_rfc3339())?;
+            serializer.serialize_entry("level", &meta.level().as_serde())?;
+            serializer.serialize_entry("fields", &event.field_map())?;
+            serializer.serialize_entry("target", meta.target())?;
+
+            if let Some(ref span_ref) = ctx.lookup_current() {
+                if let Some(trace_info) = lookup_trace_info(span_ref) {
+                    serializer.serialize_entry("dd.span_id", &trace_info.span_id)?;
+                    serializer.serialize_entry("dd.trace_id", &trace_info.trace_id)?;
+                }
+            }
+
+            serializer.end()
+        };
+
+        visit().map_err(|_| std::fmt::Error)?;
+        writeln!(writer)
+    }
+}
+
+pub struct WriteAdaptor<'a> {
+    fmt_write: &'a mut dyn std::fmt::Write,
+}
+
+impl<'a> WriteAdaptor<'a> {
+    pub fn new(fmt_write: &'a mut dyn std::fmt::Write) -> Self {
+        Self { fmt_write }
+    }
+}
+
+impl<'a> io::Write for WriteAdaptor<'a> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        let s =
+            std::str::from_utf8(buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
+
+        self.fmt_write
+            .write_str(s)
+            .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
+
+        Ok(s.as_bytes().len())
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}

src/lib.rs

@@ -0,0 +1,11 @@
+mod formatter;
+mod propagator;
+mod tracer;
+
+pub use formatter::TraceIdFormat;
+pub use propagator::set_global_propagator;
+pub use tracer::{build_tracer, build_layer};
+pub use opentelemetry::trace::{TraceError, TraceResult};
+
+#[cfg(feature = "axum")]
+pub use axum_tracing_opentelemetry::opentelemetry_tracing_layer;

src/propagator.rs

@@ -0,0 +1,4 @@
+pub fn set_global_propagator() {
+    let propagator = opentelemetry_datadog::DatadogPropagator::new();
+    opentelemetry::global::set_text_map_propagator(propagator);
+}

src/tracer.rs

@@ -0,0 +1,38 @@
+use std::time::Duration;
+use opentelemetry::KeyValue;
+use opentelemetry::trace::{TraceResult};
+use opentelemetry::sdk::{Resource, trace};
+use opentelemetry::sdk::trace::{RandomIdGenerator, Sampler, Tracer};
+use opentelemetry_otlp::WithExportConfig;
+use tracing::Subscriber;
+use tracing_opentelemetry::{OpenTelemetryLayer, PreSampledTracer};
+use tracing_subscriber::registry::LookupSpan;
+
+pub fn build_tracer(service_name: &str) -> TraceResult<Tracer> {
+    let exporter = opentelemetry_otlp::new_exporter()
+        .tonic()
+        .with_timeout(Duration::from_secs(3));
+
+    opentelemetry_otlp::new_pipeline()
+        .tracing()
+        .with_trace_config(
+            trace::config()
+                .with_sampler(Sampler::AlwaysOn)
+                .with_resource(Resource::new(vec![KeyValue::new(
+                    "service.name",
+                    service_name.to_string(),
+                )]))
+                .with_id_generator(RandomIdGenerator::default()),
+        )
+        .with_exporter(exporter)
+        .install_batch(opentelemetry::runtime::Tokio)
+}
+
+pub fn build_layer<S>(service_name: &str) -> TraceResult<OpenTelemetryLayer<S, Tracer>>
+where
+    Tracer: opentelemetry::trace::Tracer + PreSampledTracer + 'static,
+    S: Subscriber + for<'span> LookupSpan<'span>,
+{
+    let tracer = build_tracer(service_name)?;
+    Ok(tracing_opentelemetry::layer().with_tracer(tracer))
+}