feat: add independent OTLP span instrumentation for V8 event loop (#674)

Author: nyannyacha

Cargo.lock

@@ -195,9 +195,9 @@ yoke = { version = "0.7.4", features = ["derive"] }
 
 opentelemetry = "0.27.0"
 opentelemetry-http = "0.27.0"
-opentelemetry-otlp = { version = "0.27.0", features = ["logs", "http-proto", "http-json"] }
+opentelemetry-otlp = { version = "0.27.0", features = ["logs", "trace", "http-proto", "http-json"] }
 opentelemetry-semantic-conventions = { version = "0.27.0", features = ["semconv_experimental"] }
-opentelemetry_sdk = "0.27.0"
+opentelemetry_sdk = { version = "0.27.0", features = ["rt-tokio"] }
 
 # upstream resolvers
 deno_npm_cache = "0.2.0"

Cargo.toml

@@ -91,6 +91,7 @@ fn main() -> Result<ExitCode, anyhow::Error> {
           deno::versions::otel_runtime_config(),
           OtelConfig::default(),
         )?;
+        base::runtime::otel::init_if_needed();
 
         let ip = sub_matches.get_one::<String>("ip").cloned().unwrap();
         let ip = IpAddr::from_str(&ip)

cli/src/main.rs

@@ -54,6 +54,9 @@ monch.workspace = true
 notify.workspace = true
 num-traits.workspace = true
 once_cell.workspace = true
+opentelemetry.workspace = true
+opentelemetry-otlp.workspace = true
+opentelemetry_sdk.workspace = true
 pin-project.workspace = true
 reqwest_v011.workspace = true
 rustls-pemfile.workspace = true

crates/base/Cargo.toml

@@ -80,6 +80,7 @@ use deno_facade::Metadata;
 use either::Either;
 use either::Either::Left;
 use either::Either::Right;
+use ext_event_worker::events::EventMetadata;
 use ext_event_worker::events::WorkerEventWithMetadata;
 use ext_runtime::external_memory::CustomAllocator;
 use ext_runtime::MemCheckWaker;
@@ -127,6 +128,7 @@ use crate::worker::Worker;
 mod ops;
 mod unsync;
 
+pub mod otel;
 pub mod permissions;
 
 const DEFAULT_ALLOC_CHECK_INT_MSEC: u64 = 1000;
@@ -1312,6 +1314,15 @@ where
       };
     }
 
+    let otel_attrs = {
+      let op_state = self.js_runtime.op_state();
+      let op_state_ref = op_state.borrow();
+      op_state_ref
+        .try_borrow::<EventMetadata>()
+        .map(otel::WorkerSpanAttrs::from_event_metadata)
+        .unwrap_or_default()
+    };
+
     let inspector = self.inspector();
     let mod_fut_ret = unsafe {
       if let Err(err) = self.init_main_module().await {
@@ -1371,7 +1382,10 @@ where
               op_state,
               &maybe_cpu_usage_metrics_tx,
               &mut accumulated_cpu_time_ns,
-              || locker.js_runtime.mod_evaluate(main_module_id),
+              || {
+                let _span = otel::start_span("v8.mod_evaluate", &otel_attrs);
+                locker.js_runtime.mod_evaluate(main_module_id)
+              },
             ))
           }
           .instrument(span),
@@ -1449,7 +1463,11 @@ where
             locker.js_runtime.op_state(),
             &maybe_cpu_usage_metrics_tx,
             &mut accumulated_cpu_time_ns,
-            || MaybeDenoRuntime::DenoRuntime(*locker).dispatch_load_event(),
+            || {
+              let _span =
+                otel::start_span("v8.dispatch_load_event", &otel_attrs);
+              MaybeDenoRuntime::DenoRuntime(*locker).dispatch_load_event()
+            },
           ) {
             return (Err(err), get_accumulated_cpu_time_ms!());
           }
@@ -1482,7 +1500,10 @@ where
         locker.js_runtime.op_state(),
         &maybe_cpu_usage_metrics_tx,
         &mut accumulated_cpu_time_ns,
-        || MaybeDenoRuntime::DenoRuntime(&mut locker).dispatch_unload_event(),
+        || {
+          let _span = otel::start_span("v8.dispatch_unload_event", &otel_attrs);
+          MaybeDenoRuntime::DenoRuntime(&mut locker).dispatch_unload_event()
+        },
       ) {
         return (Err(err), get_accumulated_cpu_time_ms!());
       }
@@ -1506,6 +1527,16 @@ where
     let is_user_worker = self.conf.is_user_worker();
     let global_waker = self.waker.clone();
 
+    // Collect worker identity for OTLP spans.
+    let otel_attrs = {
+      let op_state = self.js_runtime.op_state();
+      let op_state_ref = op_state.borrow();
+      op_state_ref
+        .try_borrow::<EventMetadata>()
+        .map(otel::WorkerSpanAttrs::from_event_metadata)
+        .unwrap_or_default()
+    };
+
     let mut termination_request_fut = self
       .termination_request_token
       .clone()
@@ -1573,6 +1604,7 @@ where
           Cow::Borrowed(waker)
         };
 
+        let _poll_span = otel::start_span("v8.poll_event_loop", &otel_attrs);
         js_runtime.poll_event_loop(
           &mut std::task::Context::from_waker(waker.as_ref()),
           PollEventLoopOptions {

crates/base/src/runtime/mod.rs

@@ -0,0 +1,99 @@
+use std::sync::OnceLock;
+use std::time::SystemTime;
+
+use ext_event_worker::events::EventMetadata;
+use opentelemetry::global;
+use opentelemetry::trace::Span as _;
+use opentelemetry::trace::SpanKind;
+use opentelemetry::trace::TraceContextExt;
+use opentelemetry::trace::Tracer;
+use opentelemetry::trace::TracerProvider as _;
+use opentelemetry::Context;
+use opentelemetry::KeyValue;
+use opentelemetry_otlp::WithExportConfig;
+use opentelemetry_sdk::trace::BatchSpanProcessor;
+use opentelemetry_sdk::trace::TracerProvider;
+
+static PROVIDER: OnceLock<Option<TracerProvider>> = OnceLock::new();
+
+pub fn init_if_needed() {
+  PROVIDER.get_or_init(|| {
+    if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err() {
+      return None;
+    }
+
+    let exporter = opentelemetry_otlp::SpanExporter::builder()
+      .with_http()
+      .with_export_config(opentelemetry_otlp::ExportConfig::default())
+      .build()
+      .expect("failed to build OTLP span exporter");
+
+    let processor =
+      BatchSpanProcessor::builder(exporter, opentelemetry_sdk::runtime::Tokio)
+        .build();
+
+    let provider = TracerProvider::builder()
+      .with_span_processor(processor)
+      .build();
+
+    global::set_tracer_provider(provider.clone());
+    Some(provider)
+  });
+}
+
+/// Worker identity and custom otel attributes collected once at `run()` entry.
+#[derive(Clone, Default)]
+pub struct WorkerSpanAttrs(Vec<KeyValue>);
+
+impl WorkerSpanAttrs {
+  pub fn from_event_metadata(meta: &EventMetadata) -> Self {
+    let mut attrs = Vec::new();
+
+    if let Some(sp) = &meta.service_path {
+      attrs.push(KeyValue::new("worker.service_path", sp.clone()));
+    }
+    if let Some(id) = meta.execution_id {
+      attrs.push(KeyValue::new("worker.key", id.to_string()));
+    }
+    if let Some(extra) = &meta.otel_attributes {
+      for (k, v) in extra {
+        attrs.push(KeyValue::new(k.clone(), v.clone()));
+      }
+    }
+
+    Self(attrs)
+  }
+}
+
+/// Start a named span with worker attributes. Returns `None` if OTLP is not
+/// configured.
+pub fn start_span(
+  name: &'static str,
+  attrs: &WorkerSpanAttrs,
+) -> Option<OtelSpanGuard> {
+  let provider = PROVIDER.get()?.as_ref()?;
+  let tracer = provider.tracer("edge-runtime");
+  let mut span = tracer
+    .span_builder(name)
+    .with_kind(SpanKind::Internal)
+    .with_start_time(SystemTime::now())
+    .start(&tracer);
+
+  for kv in &attrs.0 {
+    span.set_attribute(kv.clone());
+  }
+
+  let cx = Context::current_with_span(span);
+  Some(OtelSpanGuard { _cx: cx })
+}
+
+/// Ends the span on drop.
+pub struct OtelSpanGuard {
+  _cx: Context,
+}
+
+impl Drop for OtelSpanGuard {
+  fn drop(&mut self) {
+    self._cx.span().end();
+  }
+}