feat(ext/cache): inject W3C traceparent/tracestate into LSC HTTP requests

Extract the active trace context on the JS side using the telemetry
propagators (ContextManager.active() + PROPAGATORS.inject()) and pass
the traceparent and tracestate values as optional string fields through
the cache op calls (op_cache_put, op_cache_match, op_cache_delete).

On the Rust side, CacheShard::get_object, put_object, and
put_object_empty now accept an optional trace_headers parameter and
inject the corresponding HTTP headers into outgoing LSC requests,
enabling distributed trace propagation through the cache layer.

Author: wperron

ext/cache/01_cache.js

@@ -11,13 +11,34 @@ import {
 const {
   ArrayPrototypePush,
   ObjectPrototypeIsPrototypeOf,
+  SafeArrayIterator,
   StringPrototypeSplit,
   StringPrototypeTrim,
   Symbol,
   SymbolFor,
   TypeError,
 } = primordials;
 
+import {
+  ContextManager,
+  PROPAGATORS,
+  TRACING_ENABLED,
+} from "ext:deno_telemetry/telemetry.ts";
+
+function getTraceHeaders() {
+  if (!TRACING_ENABLED) return { __proto__: null };
+  const headers = { __proto__: null };
+  const context = ContextManager.active();
+  for (const propagator of new SafeArrayIterator(PROPAGATORS)) {
+    propagator.inject(context, headers, {
+      set(carrier, key, value) {
+        carrier[key] = value;
+      },
+    });
+  }
+  return headers;
+}
+
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 import {
   Request,
@@ -152,6 +173,7 @@ class Cache {
 
     // Step 9-11.
     // Step 12-19: TODO(@satyarohith): do the insertion in background.
+    const putTraceHeaders = getTraceHeaders();
     await op_cache_put(
       {
         cacheId: this[_id],
@@ -162,6 +184,8 @@ class Cache {
         responseStatus: innerResponse.status,
         responseStatusText: innerResponse.statusMessage,
         responseRid: rid,
+        traceparent: putTraceHeaders["traceparent"],
+        tracestate: putTraceHeaders["tracestate"],
       },
     );
   }
@@ -208,9 +232,12 @@ class Cache {
     ) {
       r = new Request(request);
     }
+    const deleteTraceHeaders = getTraceHeaders();
     return await op_cache_delete({
       cacheId: this[_id],
       requestUrl: r.url,
+      traceparent: deleteTraceHeaders["traceparent"],
+      tracestate: deleteTraceHeaders["tracestate"],
     });
   }
 
@@ -252,12 +279,15 @@ class Cache {
       const url = new URL(r.url);
       url.hash = "";
       const innerRequest = toInnerRequest(r);
+      const matchTraceHeaders = getTraceHeaders();
       const matchResult = await op_cache_match(
         {
           cacheId: this[_id],
           // deno-lint-ignore prefer-primordials
           requestUrl: url.toString(),
           requestHeaders: innerRequest.headerList,
+          traceparent: matchTraceHeaders["traceparent"],
+          tracestate: matchTraceHeaders["tracestate"],
         },
       );
       if (matchResult) {

ext/cache/lib.rs

@@ -98,7 +98,7 @@ pub enum CacheError {
 pub struct CreateCache(pub Arc<dyn Fn() -> Result<CacheImpl, CacheError>>);
 
 deno_core::extension!(deno_cache,
-  deps = [ deno_webidl, deno_web, deno_fetch ],
+  deps = [ deno_webidl, deno_web, deno_fetch, deno_telemetry ],
   ops = [
     op_cache_storage_open,
     op_cache_storage_has,
@@ -128,6 +128,8 @@ pub struct CachePutRequest {
   pub response_status: u16,
   pub response_status_text: String,
   pub response_rid: Option<ResourceId>,
+  pub traceparent: Option<String>,
+  pub tracestate: Option<String>,
 }
 
 #[derive(Deserialize, Serialize, Debug)]
@@ -136,6 +138,8 @@ pub struct CacheMatchRequest {
   pub cache_id: i64,
   pub request_url: String,
   pub request_headers: Vec<(ByteString, ByteString)>,
+  pub traceparent: Option<String>,
+  pub tracestate: Option<String>,
 }
 
 #[derive(Debug, Deserialize, Serialize)]
@@ -156,6 +160,8 @@ pub struct CacheMatchResponseMeta {
 pub struct CacheDeleteRequest {
   pub cache_id: i64,
   pub request_url: String,
+  pub traceparent: Option<String>,
+  pub tracestate: Option<String>,
 }
 
 #[async_trait(?Send)]

ext/cache/lsc_shard.rs

@@ -14,13 +14,14 @@ use hyper::HeaderMap;
 use hyper::StatusCode;
 use hyper::body::Incoming;
 use hyper::header::AUTHORIZATION;
+use hyper::header::HeaderValue;
 use hyper_util::client::legacy::Client;
 use hyper_util::client::legacy::connect::HttpConnector;
 use hyper_util::rt::tokio::TokioExecutor;
 
 use crate::CacheError;
 
-type ClientBody =
+pub type ClientBody =
   Either<UnsyncBoxBody<Bytes, CacheError>, UnsyncBoxBody<Bytes, Infallible>>;
 
 pub struct CacheShard {
@@ -44,16 +45,28 @@ impl CacheShard {
   pub async fn get_object(
     &self,
     object_key: &str,
+    trace_headers: Option<(&str, Option<&str>)>,
   ) -> Result<Option<Response<Incoming>>, CacheError> {
     let body = Either::Right(UnsyncBoxBody::new(Empty::new()));
-    let req = Request::builder()
+    let mut req = Request::builder()
       .method(Method::GET)
       .uri(format!("{}/objects/{}", self.endpoint, object_key))
       .header(&AUTHORIZATION, format!("Bearer {}", self.token))
       .header("x-ryw", "1")
       .body(body)
       .unwrap();
 
+    if let Some((traceparent, tracestate)) = trace_headers {
+      req
+        .headers_mut()
+        .insert("traceparent", HeaderValue::from_str(traceparent)?);
+      if let Some(tracestate) = tracestate {
+        req
+          .headers_mut()
+          .insert("tracestate", HeaderValue::from_str(tracestate)?);
+      }
+    }
+
     let res = self.client.request(req).await?;
 
     if res.status().is_success() {
@@ -72,6 +85,7 @@ impl CacheShard {
     &self,
     object_key: &str,
     headers: HeaderMap,
+    trace_headers: Option<(&str, Option<&str>)>,
   ) -> Result<(), CacheError> {
     let body = Either::Right(UnsyncBoxBody::new(Empty::new()));
     let mut builder = Request::builder()
@@ -83,7 +97,17 @@ impl CacheShard {
       builder = builder.header(key, val)
     }
 
-    let req = builder.body(body).unwrap();
+    let mut req = builder.body(body).unwrap();
+    if let Some((traceparent, tracestate)) = trace_headers {
+      req
+        .headers_mut()
+        .insert("traceparent", HeaderValue::from_str(traceparent)?);
+      if let Some(tracestate) = tracestate {
+        req
+          .headers_mut()
+          .insert("tracestate", HeaderValue::from_str(tracestate)?);
+      }
+    }
 
     let res = self.client.request(req).await?;
 
@@ -107,6 +131,7 @@ impl CacheShard {
     object_key: &str,
     headers: HeaderMap,
     body: UnsyncBoxBody<Bytes, CacheError>,
+    trace_headers: Option<(&str, Option<&str>)>,
   ) -> Result<(), CacheError> {
     let mut builder = Request::builder()
       .method(Method::PUT)
@@ -117,7 +142,17 @@ impl CacheShard {
       builder = builder.header(key, val)
     }
 
-    let req = builder.body(Either::Left(body)).unwrap();
+    let mut req = builder.body(Either::Left(body)).unwrap();
+    if let Some((traceparent, tracestate)) = trace_headers {
+      req
+        .headers_mut()
+        .insert("traceparent", HeaderValue::from_str(traceparent)?);
+      if let Some(tracestate) = tracestate {
+        req
+          .headers_mut()
+          .insert("tracestate", HeaderValue::from_str(tracestate)?);
+      }
+    }
 
     let res = self.client.request(req).await?;
 

ext/cache/lscache.rs

@@ -144,7 +144,13 @@ impl LscBackend {
       body_rx.into_stream().map_ok(http_body::Frame::data),
     );
     let body = UnsyncBoxBody::new(body);
-    shard.put_object(&object_key, headers, body).await?;
+    let trace_headers = request_response
+      .traceparent
+      .as_deref()
+      .map(|tp| (tp, request_response.tracestate.as_deref()));
+    shard
+      .put_object(&object_key, headers, body, trace_headers)
+      .await?;
     Ok(())
   }
 
@@ -171,7 +177,11 @@ impl LscBackend {
       cache_name.as_bytes(),
       request.request_url.as_bytes(),
     );
-    let Some(res) = shard.get_object(&object_key).await? else {
+    let trace_headers = request
+      .traceparent
+      .as_deref()
+      .map(|tp| (tp, request.tracestate.as_deref()));
+    let Some(res) = shard.get_object(&object_key, trace_headers).await? else {
       return Ok(None);
     };
 
@@ -285,7 +295,13 @@ impl LscBackend {
           .as_bytes(),
       )?,
     );
-    shard.put_object_empty(&object_key, headers).await?;
+    let trace_headers = request
+      .traceparent
+      .as_deref()
+      .map(|tp| (tp, request.tracestate.as_deref()));
+    shard
+      .put_object_empty(&object_key, headers, trace_headers)
+      .await?;
     Ok(true)
   }
 }

ext/cache/sqlite.rs

@@ -316,6 +316,8 @@ impl SqliteBackedCache {
               .delete(CacheDeleteRequest {
                 cache_id: request.cache_id,
                 request_url: request.request_url,
+                traceparent: None,
+                tracestate: None,
               })
               .await;
             return Ok(None);

runtime/worker.rs

@@ -377,8 +377,8 @@ impl MainWorker {
           let token = elems[1];
           use deno_cache::CacheShard;
 
-          let shard =
-            Rc::new(CacheShard::new(endpoint.to_string(), token.to_string()));
+          let shard = CacheShard::new(endpoint.to_string(), token.to_string());
+          let shard = Rc::new(shard);
           let create_cache_fn = move || {
             let x = deno_cache::LscBackend::default();
             x.set_shard(shard.clone());