Feature: Add request_id in logs (#55)

* first pass at adding request id to logs

* Update rust/lit-node/lit-node/src/main.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* change how we set the request id on the span

* clippy

* Refactor spans behaviour

* Fix fallback ID

* Add Rocket fairings

* Remove redundant tracing

* optimise tests

* Address feedback

* Remove fallback UUID

* Leverage Dashmap

* build lit-os to rebuild cargo.lock

* rebuild lit actions and lit-node

* remove correlation_id from /web/job_status/v2 instrumentation since it's only on this one endpoint

* Revert "remove correlation_id from /web/job_status/v2 instrumentation since it's only on this one endpoint"

This reverts commit 7c4f5f9.

---------

Co-authored-by: Chris Cassano <chris@litprotocol.com>
Co-authored-by: Chris Cassano <1285652+glitch003@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Adam Reif <adam@litprotocol.com>

Author: 4 people

rust/lit-actions/Cargo.lock

@@ -2,18 +2,15 @@ use std::collections::HashMap;
 use std::fmt;
 use std::future::Future;
 
-use opentelemetry::propagation::{Extractor, Injector, TextMapPropagator};
-use opentelemetry_sdk::propagation::TraceContextPropagator;
+use lit_observability::logging::set_request_context;
+use opentelemetry::propagation::Injector;
 use rocket::Request;
 use rocket::request::{FromRequest, Outcome};
 use semver::Version;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 use tokio::task::futures::TaskLocalFuture;
 use tokio::task_local;
-use tracing::{Span, info_span};
-use tracing_opentelemetry::OpenTelemetrySpanExt;
-use uuid::Uuid;
 
 use crate::error::{EC, Error, Result, conversion_err_code, validation_err_code};
 
@@ -27,77 +24,6 @@ task_local! {
     pub static TRACING: Box<dyn Tracer>;
 }
 
-/// The TracingSpan request guard creates a new tracing span for the request. If the request
-/// contains a parent span ID, it will be used as the parent of this new span. Otherwise, a new
-/// root span will be created.
-#[allow(dead_code)]
-#[derive(Clone, Debug)]
-pub struct TracingSpan {
-    span: Span,
-}
-
-impl TracingSpan {
-    pub fn new(span: Span) -> Self {
-        Self { span }
-    }
-
-    pub fn span(&self) -> &Span {
-        &self.span
-    }
-}
-
-#[rocket::async_trait]
-impl<'r> FromRequest<'r> for TracingSpan {
-    type Error = crate::error::Error;
-
-    async fn from_request(
-        req: &'r rocket::Request<'_>,
-    ) -> rocket::request::Outcome<Self, Self::Error> {
-        // Extract the propagated context
-        let propagator = TraceContextPropagator::new();
-        // Initialize some container to hold the header information.
-        let mut carrier = HashMap::new();
-        // Transfer header information from request to carrier.
-        for header in req.headers().iter() {
-            carrier.insert(header.name().to_string(), header.value().to_string());
-        }
-        // Extract the context from the carrier
-        let context = propagator.extract(&HeaderExtractor::from(&carrier));
-
-        // Initialize a new span with the propagated context as the parent
-        let req_method = req.method();
-        let req_path = req.uri().path();
-        let new_span = info_span!(
-            "handle_request",
-            method = req_method.as_str(),
-            path = req_path.to_string(),
-        );
-        new_span.set_parent(context);
-
-        Outcome::Success(TracingSpan { span: new_span })
-    }
-}
-
-pub struct HeaderExtractor<'a> {
-    headers: &'a HashMap<String, String>,
-}
-
-impl<'a> From<&'a HashMap<String, String>> for HeaderExtractor<'a> {
-    fn from(headers: &'a HashMap<String, String>) -> Self {
-        HeaderExtractor { headers }
-    }
-}
-
-impl<'a> Extractor for HeaderExtractor<'a> {
-    fn get(&self, key: &str) -> Option<&'a str> {
-        self.headers.get(key).map(|v| v.as_str())
-    }
-
-    fn keys(&self) -> Vec<&str> {
-        self.headers.keys().map(|v| v.as_str()).collect()
-    }
-}
-
 pub struct HeaderInjector<'a> {
     headers: &'a mut HashMap<String, String>,
 }
@@ -167,10 +93,13 @@ impl<'r> FromRequest<'r> for Tracing {
     type Error = crate::error::Error;
 
     async fn from_request(req: &'r Request<'_>) -> Outcome<Self, Self::Error> {
-        let correlation_id =
-            extract_correlation_id(req).unwrap_or_else(|| format!("LD-{}", Uuid::new_v4()));
+        let (request_id, correlation_id) = extract_request_and_correlation_ids(req);
+
+        // Set request context for log injection; no fallback IDs.
+        set_request_context(request_id, correlation_id.clone());
 
-        let mut tracing = Self::new(correlation_id);
+        // For the Tracing struct, use empty string if no correlation_id was provided.
+        let mut tracing = Self::new(correlation_id.unwrap_or_default());
         apply_req_tracing_fields(req, &mut tracing);
 
         Outcome::Success(tracing)
@@ -227,7 +156,16 @@ impl<'r> FromRequest<'r> for TracingRequired {
     type Error = crate::error::Error;
 
     async fn from_request(req: &'r Request<'_>) -> Outcome<Self, Self::Error> {
-        if let Some(correlation_id) = extract_correlation_id(req) {
+        let (request_id, correlation_id) = extract_request_and_correlation_ids(req);
+
+        // TracingRequired requires at least one header
+        if let Some(correlation_id) = correlation_id {
+            // Preserve distinct values when both headers are present
+            let request_id = request_id.unwrap_or_else(|| correlation_id.clone());
+
+            // Set request context (span extensions + OTel attributes) for consistency
+            set_request_context(Some(request_id), Some(correlation_id.clone()));
+
             let mut tracing = Self::new(correlation_id);
             apply_req_tracing_fields(req, &mut tracing);
 
@@ -257,12 +195,23 @@ where
     TRACING.scope(Box::new(tracing), f)
 }
 
-pub(crate) fn extract_correlation_id(req: &Request<'_>) -> Option<String> {
-    req.headers()
-        .get(HEADER_KEY_X_CORRELATION_ID)
-        .next()
-        .or_else(|| req.headers().get(HEADER_KEY_X_REQUEST_ID).next())
-        .map(|val| val.to_string())
+/// Extracts both request_id and correlation_id from headers, preserving distinct values.
+/// Returns (request_id, correlation_id) tuple.
+/// - request_id: X-Request-Id header, falls back to X-Correlation-Id
+/// - correlation_id: X-Correlation-Id header, falls back to X-Request-Id
+pub(crate) fn extract_request_and_correlation_ids(
+    req: &Request<'_>,
+) -> (Option<String>, Option<String>) {
+    let x_request_id = req.headers().get(HEADER_KEY_X_REQUEST_ID).next().map(|v| v.to_string());
+    let x_correlation_id =
+        req.headers().get(HEADER_KEY_X_CORRELATION_ID).next().map(|v| v.to_string());
+
+    // request_id: prefer X-Request-Id, fall back to X-Correlation-Id
+    let request_id = x_request_id.clone().or_else(|| x_correlation_id.clone());
+    // correlation_id: prefer X-Correlation-Id, fall back to X-Request-Id
+    let correlation_id = x_correlation_id.or(x_request_id);
+
+    (request_id, correlation_id)
 }
 
 pub(crate) fn apply_req_tracing_fields(req: &Request<'_>, tracing: &mut (impl Tracer + 'static)) {

rust/lit-core/Cargo.lock

@@ -10,7 +10,6 @@ use hyper::body::Bytes;
 use hyper::http::HeaderValue;
 use hyper::{HeaderMap, Method, Request as HyperRequest, Response as HyperResponse};
 use tracing::debug;
-use uuid::Uuid;
 
 use crate::context::{HEADER_KEY_X_CORRELATION_ID, HEADER_KEY_X_REQUEST_ID, Tracing, with_context};
 
@@ -129,13 +128,10 @@ impl Default for Router {
     }
 }
 
-// Get Tracing from request headers.
+// Get Tracing from request headers; no fallback ID generation.
 fn get_tracing_from_request_header(headers: HeaderMap<HeaderValue>) -> Tracing {
-    if let Some(correlation_id) = extract_correlation_id(headers) {
-        Tracing::new(correlation_id)
-    } else {
-        Tracing::new(Uuid::new_v4().simple().to_string())
-    }
+    let correlation_id = extract_correlation_id(headers).unwrap_or_default();
+    Tracing::new(correlation_id)
 }
 
 fn extract_correlation_id(headers: HeaderMap<HeaderValue>) -> Option<String> {

rust/lit-core/lit-api-core/src/context/mod.rs

@@ -14,12 +14,12 @@ proxy-collector = []
 channels = ["dep:flume"]
 
 [dependencies]
+dashmap = "6"
 derive_more.workspace = true
 flume = { version = "0.11", optional = true }
 hyper-util.workspace = true
 nu-ansi-term = { version = "0.50.1" }
 opentelemetry.workspace = true
-opentelemetry-appender-tracing = { version = "0.5.0", default-features = false }
 opentelemetry-otlp = { workspace = true, features = ["logs"] }
 opentelemetry-semantic-conventions.workspace = true
 opentelemetry_sdk = { workspace = true, features = ["logs"] }

rust/lit-core/lit-api-core/src/server/hyper/handler/router.rs

@@ -3,11 +3,10 @@ use std::str::FromStr;
 pub use config::LitObservabilityConfig;
 use error::unexpected_err;
 use lit_core::config::LitConfig;
-use logging::init_logger_provider;
+use logging::{ContextAwareOtelLogLayer, CustomEventFormatter, init_logger_provider};
 use metrics::init_metrics_provider;
 use net::init_tonic_exporter_builder;
 use opentelemetry::trace::TracerProvider;
-use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
 
 use opentelemetry_sdk::logs::LoggerProvider;
 use opentelemetry_sdk::metrics::SdkMeterProvider;
@@ -80,8 +79,7 @@ pub async fn create_providers(
     };
     let logger_provider = init_logger_provider(tonic_exporter_builder, resource.clone())?;
 
-    // Create a new OpenTelemetryTracingBridge using the above LoggerProvider.
-    let tracing_bridge_layer = OpenTelemetryTracingBridge::new(&logger_provider);
+    let context_aware_log_layer = ContextAwareOtelLogLayer::new(&logger_provider);
 
     // Add a tracing filter to filter events from crates used by opentelemetry-otlp.
     // The filter levels are set as follows:
@@ -101,10 +99,12 @@ pub async fn create_providers(
         .add_directive("h2=error".parse().unwrap())
         .add_directive("reqwest=error".parse().unwrap());
 
+    let custom_formatter = CustomEventFormatter::default();
+
     let sub = tracing_subscriber::registry()
         .with(level_filter)
-        .with(fmt::layer())
-        .with(tracing_bridge_layer)
+        .with(fmt::layer().event_format(custom_formatter))
+        .with(context_aware_log_layer)
         .with(MetricsLayer::new(meter_provider.clone()))
         .with(OpenTelemetryLayer::new(tracer));