[PLATFORM-1568]: Add support for log error tracking in datadog (#91)

* Use normalized_metadata instead of manually normalizing metadata

* Linters

* Fix tests

* Serialize metadata as is

* Make sure that events contain metadata

* Remove duplicate environment key

* Update changelog

* Update log message

.github/workflows/ci.yml

@@ -36,6 +36,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@cargo-make
+      - uses: taiki-e/install-action@nextest
       - run: cargo make test
 
   alls-green:

CHANGELOG.md

@@ -5,6 +5,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- All log metadata is now stored in the top level object instead of under a `metadata` key.
+
 ---
 
 ## [0.8.1] - 2024-01-05

Makefile.toml

@@ -15,13 +15,15 @@ command = "cargo"
 args = ["fmt", "--", "--check"]
 
 [tasks.test]
+install_crate = "cargo-nextest"
 description = "Run tests."
 command = "cargo"
-args = ["test", "--all-features", "${@}"]
+args = ["nextest", "run", "--all-features", "${@}"]
 
 [tasks.clippy]
 command = "cargo"
-args = ["clippy", "--all-features", "--all-targets", "--", "-D", "warnings"]
+args = ["clippy", "--all-features", "--all-targets", "${@}"]
+env.RUSTFLAGS = "-Dwarnings"
 
 [tasks.docs]
 description = "Build docs as they are rendered on docs.rs"

docker-compose.yml

@@ -27,7 +27,7 @@ services:
   jaeger:
     image: jaegertracing/all-in-one:1.35
     ports:
-      - "16685:16685"
+      - "16686:16686"
       - "55681:55681"
     environment:
       COLLECTOR_OTLP_ENABLED: "true"

examples/ping.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use actix_web::{middleware::Logger, web, App, HttpResponse, HttpServer};
-use prima_bridge::{prelude::*, Bridge, Request};
+use prima_bridge::prelude::*;
 use prima_tracing::{builder, configure_subscriber, init_subscriber, Country, Environment};
 use tracing_actix_web::TracingLogger;
 

src/json/formatter.rs

@@ -3,7 +3,8 @@ use std::io::Write;
 
 use serde::ser::{SerializeMap, Serializer};
 use serde::Serialize;
-use tracing::{Event, Metadata, Subscriber};
+use tracing::{Event, Subscriber};
+use tracing_log::NormalizeEvent;
 use tracing_subscriber::{
     fmt::MakeWriter,
     layer::Context,
@@ -118,10 +119,25 @@ impl EventFormatter for DefaultEventFormatter {
     where
         S: Subscriber + for<'a> tracing_subscriber::registry::LookupSpan<'a>,
     {
-        let metadata = event.metadata();
+        let normalized_metadata = event.normalized_metadata();
+
+        let metadata = normalized_metadata
+            .as_ref()
+            .unwrap_or_else(|| event.metadata());
+
         let mut buffer = Vec::new();
         let mut serializer = serde_json::Serializer::new(&mut buffer);
         let mut map_serializer = serializer.serialize_map(None)?;
+        let mut visitor = PrimaJsonVisitor::default();
+        event.record(&mut visitor);
+
+        for (key, value) in visitor
+            .fields()
+            .iter()
+            .filter(|(&key, _)| key != "message" && !key.starts_with("log."))
+        {
+            map_serializer.serialize_entry(key, value)?;
+        }
 
         map_serializer.serialize_entry("timestamp", &chrono::Utc::now())?;
         map_serializer.serialize_entry(
@@ -132,20 +148,13 @@ impl EventFormatter for DefaultEventFormatter {
         map_serializer.serialize_entry("environment", info.environment())?;
         map_serializer.serialize_entry("type", info.app_name())?;
 
-        let mut visitor = PrimaJsonVisitor::default();
-        event.record(&mut visitor);
-
         map_serializer.serialize_entry("message", &visitor.fields().get("message"))?;
 
-        map_serializer.serialize_entry(
-            "metadata",
-            &MetadataSerializer {
-                ctx: &ctx,
-                metadata,
-                visitor: &visitor,
-                environment: info.environment(),
-            },
-        )?;
+        if let Some(current_span) = ctx.current_span().id().and_then(|id| ctx.span(id)) {
+            map_serializer.serialize_entry("current_span", &SpanSerializer(&current_span))?;
+        }
+
+        map_serializer.serialize_entry("spans", &SpanListSerializer(&ctx))?;
 
         // Adds support for correlating logs and traces on datadog
         // In order for Datadog to be able to correlate the logs with the traces we need to insert `dd.trace_id` and `dd.span_id` at root level
@@ -188,73 +197,6 @@ impl EventFormatter for DefaultEventFormatter {
     }
 }
 
-pub struct MetadataSerializer<'a, S>
-where
-    S: Subscriber + tracing_subscriber::registry::LookupSpan<'a>,
-{
-    ctx: &'a Context<'a, S>,
-    metadata: &'a Metadata<'a>,
-    visitor: &'a PrimaJsonVisitor<'a>,
-    environment: &'a str,
-}
-
-impl<'a, Sub> Serialize for MetadataSerializer<'a, Sub>
-where
-    Sub: Subscriber + for<'lookup> LookupSpan<'lookup>,
-{
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        let mut map_serializer = serializer.serialize_map(None)?;
-
-        map_serializer.serialize_entry("environment", self.environment)?;
-        map_serializer.serialize_entry(
-            "target",
-            self.visitor
-                .get("log.target")
-                .unwrap_or_else(|| self.metadata.target()),
-        )?;
-        map_serializer.serialize_entry(
-            "file",
-            self.metadata
-                .file()
-                .or_else(|| self.visitor.get("log.file"))
-                .unwrap_or("-"),
-        )?;
-        map_serializer.serialize_entry(
-            "line",
-            &self
-                .metadata
-                .line()
-                .or_else(|| self.visitor.get("log.line"))
-                .unwrap_or(0),
-        )?;
-
-        for (key, value) in self
-            .visitor
-            .fields()
-            .iter()
-            .filter(|(&key, _)| key != "message" && !key.starts_with("log."))
-        {
-            map_serializer.serialize_entry(key, value)?;
-        }
-
-        if let Some(current_span) = self
-            .ctx
-            .current_span()
-            .id()
-            .and_then(|id| self.ctx.span(id))
-        {
-            map_serializer.serialize_entry("current_span", &SpanSerializer(&current_span))?;
-        }
-
-        map_serializer.serialize_entry("spans", &SpanListSerializer(self.ctx))?;
-
-        map_serializer.end()
-    }
-}
-
 struct SpanSerializer<'a, 'b, Span>(&'b SpanRef<'a, Span>)
 where
     Span: for<'lookup> LookupSpan<'lookup>;

tests/e2e/mod.rs

@@ -1,16 +1,12 @@
 use opentelemetry_jaeger::testing::jaeger_api_v2::Span;
 use opentelemetry_jaeger::testing::jaeger_client::JaegerTestClient;
 use prima_tracing::{builder, configure_subscriber, init_subscriber, Country, Environment};
-use std::time::SystemTime;
 
 async fn get_spans(f: impl FnOnce()) -> Option<Vec<Span>> {
     std::env::set_var("RUST_LOG", "info");
 
     // Unique id for this test run
-    let seed = SystemTime::now()
-        .duration_since(SystemTime::UNIX_EPOCH)
-        .unwrap()
-        .as_millis();
+    let seed = std::fs::read_to_string("/proc/sys/kernel/random/uuid").unwrap();
     let service_name = format!("e2e-test-{seed}");
 
     let collector_url = "http://jaeger:55681";
@@ -41,8 +37,8 @@ async fn get_spans(f: impl FnOnce()) -> Option<Vec<Span>> {
 
 #[cfg(feature = "traces")]
 #[tokio::test(flavor = "multi_thread")]
-async fn it_sends_traces_to_jaeger() {
-    let log_message = "hello it_sends_traces_to_jaeger";
+async fn traces_are_sent_to_datadog() {
+    let log_message = "hello traces_are_sent_to_datadog";
 
     let spans = get_spans(|| {
         let span = tracing::info_span!("my span");
@@ -59,3 +55,26 @@ async fn it_sends_traces_to_jaeger() {
     let msg = spans[0].logs[0].fields[0].v_str.as_str();
     assert_eq!(log_message, msg);
 }
+
+#[cfg(feature = "traces")]
+#[tokio::test(flavor = "multi_thread")]
+async fn events_contain_metadata() {
+    let spans = get_spans(|| {
+        let span = tracing::info_span!("my span");
+        span.in_scope(|| {
+            tracing::info!(hello = "meta!", "meta?");
+        });
+    })
+    .await
+    .expect("Failed to fetch traces from jaeger");
+
+    assert_eq!(
+        "meta!",
+        &spans[0].logs[0]
+            .fields
+            .iter()
+            .find(|f| f.key == "hello")
+            .unwrap()
+            .v_str
+    );
+}