stamp: align with supabase otel requirement

Author: nyannyacha

Cargo.toml

@@ -55,7 +55,7 @@ deno_fs = { version = "0.92.0", features = ["sync_fs"] }
 deno_http = "0.180.0"
 deno_io = "0.92.0"
 deno_net = "0.174.0"
-deno_telemetry = "0.12.0"
+deno_telemetry = "0.17.0"
 deno_tls = "0.169.0"
 deno_url = "0.182.0"
 deno_web = "0.213.0"

cli/src/flags.rs

@@ -11,6 +11,7 @@ use clap::ArgAction;
 use clap::ArgGroup;
 use clap::Command;
 use clap::ValueEnum;
+use deno::deno_telemetry;
 use deno_facade::Checksum;
 
 #[derive(ValueEnum, Default, Clone, Copy)]
@@ -28,6 +29,30 @@ impl From<EszipV2ChecksumKind> for Option<Checksum> {
   }
 }
 
+#[derive(ValueEnum, Clone, Copy)]
+pub(super) enum OtelKind {
+  Main,
+  Event,
+}
+
+#[derive(ValueEnum, Default, Clone, Copy)]
+pub(super) enum OtelConsoleConfig {
+  #[default]
+  Ignore,
+  Capture,
+  Replace,
+}
+
+impl From<OtelConsoleConfig> for deno_telemetry::OtelConsoleConfig {
+  fn from(value: OtelConsoleConfig) -> Self {
+    match value {
+      OtelConsoleConfig::Ignore => Self::Ignore,
+      OtelConsoleConfig::Capture => Self::Capture,
+      OtelConsoleConfig::Replace => Self::Replace,
+    }
+  }
+}
+
 pub(super) fn get_cli() -> Command {
   Command::new(env!("CARGO_BIN_NAME"))
     .about(env!("CARGO_PKG_DESCRIPTION"))
@@ -278,6 +303,21 @@ fn get_start_command() -> Command {
         .value_parser(value_parser!(u8).range(..=99))
         .default_value("90"),
     )
+    .arg(
+      arg!(--"enable-otel")
+        .help("Enable Otel in the main and event workers")
+        .value_delimiter(',')
+        .value_parser(value_parser!(OtelKind))
+        .num_args(0..=1)
+        .default_missing_value("main,event")
+        .action(ArgAction::Append),
+    )
+    .arg(
+      arg!(--"otel-console" <MODE>)
+        // .env("OTEL_DENO_CONSOLE")
+        .help("Configure otel console auto instrumentation")
+        .value_parser(value_parser!(OtelConsoleConfig)),
+    )
 }
 
 fn get_bundle_command() -> Command {

cli/src/main.rs

@@ -10,6 +10,7 @@ use std::sync::Arc;
 use anyhow::bail;
 use anyhow::Context;
 use anyhow::Error;
+use base::server;
 use base::server::Builder;
 use base::server::ServerFlags;
 use base::server::Tls;
@@ -30,6 +31,8 @@ use deno_facade::Metadata;
 use env::resolve_deno_runtime_env;
 use flags::get_cli;
 use flags::EszipV2ChecksumKind;
+use flags::OtelConsoleConfig;
+use flags::OtelKind;
 use log::warn;
 
 mod env;
@@ -83,7 +86,7 @@ fn main() -> Result<ExitCode, anyhow::Error> {
       Some(("start", sub_matches)) => {
         deno_telemetry::init(
           deno::versions::otel_runtime_config(),
-          &OtelConfig::default(),
+          OtelConfig::default(),
         )?;
 
         let ip = sub_matches.get_one::<String>("ip").cloned().unwrap();
@@ -128,6 +131,17 @@ fn main() -> Result<ExitCode, anyhow::Error> {
           .cloned()
           .unwrap();
 
+        let enable_otel = sub_matches
+          .get_many::<OtelKind>("enable-otel")
+          .unwrap_or_default()
+          .cloned()
+          .collect::<Vec<_>>();
+
+        let otel_console = sub_matches
+          .get_one::<OtelConsoleConfig>("otel-console")
+          .cloned()
+          .map(Into::into);
+
         let event_service_manager_path =
           sub_matches.get_one::<String>("event-worker").cloned();
         let maybe_main_entrypoint =
@@ -210,6 +224,21 @@ fn main() -> Result<ExitCode, anyhow::Error> {
           .unwrap();
 
         let flags = ServerFlags {
+          otel: if !enable_otel.is_empty() {
+            if enable_otel.len() > 1 {
+              Some(server::OtelKind::Both)
+            } else {
+              match enable_otel.first() {
+                Some(OtelKind::Main) => Some(server::OtelKind::Main),
+                Some(OtelKind::Event) => Some(server::OtelKind::Event),
+                None => None,
+              }
+            }
+          } else {
+            None
+          },
+          otel_console,
+
           no_module_cache,
           allow_main_inspector,
           tcp_nodelay,

crates/base/src/runtime/mod.rs

@@ -23,7 +23,7 @@ use base_rt::get_current_cpu_time_ns;
 use base_rt::BlockingScopeCPUUsage;
 use base_rt::DenoRuntimeDropToken;
 use base_rt::DropToken;
-use base_rt::RuntimeOtelAttribute;
+use base_rt::RuntimeOtelExtraAttributes;
 use base_rt::RuntimeState;
 use base_rt::RuntimeWaker;
 use cooked_waker::IntoWaker;
@@ -55,6 +55,7 @@ use deno_cache::SqliteBackedCache;
 use deno_core::error::AnyError;
 use deno_core::error::JsError;
 use deno_core::serde_json;
+use deno_core::serde_json::Value;
 use deno_core::url::Url;
 use deno_core::v8;
 use deno_core::v8::GCCallbackFlags;
@@ -1105,6 +1106,7 @@ where
       s3_fs,
       beforeunload_cpu_threshold,
       beforeunload_mem_threshold,
+      context,
       ..
     } = match bootstrap_ret {
       Ok(Ok(v)) => v,
@@ -1116,6 +1118,7 @@ where
       }
     };
 
+    let context = context.unwrap_or_default();
     let span = Span::current();
     let post_task_ret = unsafe {
       spawn_blocking_non_send(|| {
@@ -1143,14 +1146,31 @@ where
             op_state.put::<HashMap<usize, CancellationToken>>(HashMap::new());
           }
 
+          let mut otel_attributes = HashMap::new();
+
+          otel_attributes.insert(
+            "edge_runtime.worker.kind".into(),
+            conf.to_worker_kind().to_string().into(),
+          );
+
           if conf.is_user_worker() {
             let conf = conf.as_user_worker().unwrap();
             let key = conf.key.map_or("".to_string(), |k| k.to_string());
 
             // set execution id for user workers
             env_vars.insert("SB_EXECUTION_ID".to_string(), key.clone());
 
-            op_state.put(RuntimeOtelAttribute(key.into()));
+            if let Some(Value::Object(attributes)) = context.get("otel") {
+              for (k, v) in attributes {
+                otel_attributes.insert(
+                  k.to_string().into(),
+                  match v {
+                    Value::String(str) => str.to_string().into(),
+                    others => others.to_string().into(),
+                  },
+                );
+              }
+            }
 
             if let Some(events_msg_tx) = conf.events_msg_tx.clone() {
               op_state.put::<mpsc::UnboundedSender<WorkerEventWithMetadata>>(
@@ -1161,14 +1181,11 @@ where
                 execution_id: conf.key,
               });
             }
-          } else {
-            op_state.put(RuntimeOtelAttribute(
-              conf.to_worker_kind().to_string().into(),
-            ));
           }
 
           op_state.put(ext_env::EnvVars(env_vars));
           op_state.put(DenoRuntimeDropToken(DropToken(drop_token.clone())));
+          op_state.put(RuntimeOtelExtraAttributes(otel_attributes));
         }
 
         if is_user_worker {

crates/base/src/server.rs

@@ -1,3 +1,4 @@
+use std::collections::HashSet;
 use std::future::pending;
 use std::future::Future;
 use std::net::SocketAddr;
@@ -13,6 +14,7 @@ use anyhow::Context;
 use anyhow::Error;
 use deno::deno_telemetry::OtelConfig;
 use deno::deno_telemetry::OtelConsoleConfig;
+use deno::deno_telemetry::OtelPropagators;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use either::Either;
@@ -269,8 +271,18 @@ pub struct WorkerEntrypoints {
   pub events: Option<String>,
 }
 
+#[derive(Debug, Clone, Copy)]
+pub enum OtelKind {
+  Main,
+  Event,
+  Both,
+}
+
 #[derive(Debug, Default, Clone, Copy)]
 pub struct ServerFlags {
+  pub otel: Option<OtelKind>,
+  pub otel_console: Option<OtelConsoleConfig>,
+
   pub no_module_cache: bool,
   pub allow_main_inspector: bool,
   pub tcp_nodelay: bool,
@@ -401,13 +413,19 @@ impl Server {
         .set_server_flags(Some(Left(flags.clone())))
         .set_termination_token(Some(termination_tokens.event.clone().unwrap()));
 
-      builder
-        .set_entrypoint(maybe_event_entrypoint.as_deref())
-        .set_otel_config(Some(OtelConfig {
+      builder.set_entrypoint(maybe_event_entrypoint.as_deref());
+
+      if let Some(OtelKind::Event | OtelKind::Both) = flags.otel {
+        builder.set_otel_config(Some(OtelConfig {
           tracing_enabled: true,
-          console: OtelConsoleConfig::Capture,
+          console: flags.otel_console.unwrap_or_default(),
+          propagators: HashSet::from([
+            OtelPropagators::TraceContext,
+            OtelPropagators::Baggage,
+          ]),
           ..Default::default()
         }));
+      }
 
       Some(builder.build().await?)
     } else {
@@ -450,12 +468,19 @@ impl Server {
         .set_shared_metric_source(Some(shared_metric_src.clone()))
         .set_event_worker_metric_source(
           event_worker_surface.as_ref().map(|it| it.metric.clone()),
-        )
-        .set_otel_config(Some(OtelConfig {
+        );
+
+      if let Some(OtelKind::Main | OtelKind::Both) = flags.otel {
+        builder.set_otel_config(Some(OtelConfig {
           tracing_enabled: true,
-          console: OtelConsoleConfig::Capture,
+          console: flags.otel_console.unwrap_or_default(),
+          propagators: HashSet::from([
+            OtelPropagators::TraceContext,
+            OtelPropagators::Baggage,
+          ]),
           ..Default::default()
         }));
+      }
 
       builder.build().await?
     };

crates/base/tests/integration_tests.rs

@@ -3827,7 +3827,7 @@ async fn test_should_be_able_to_trigger_early_drop_with_mem() {
   let resp = tb
     .request(|b| {
       b.uri("/early-drop-mem")
-        .header("x-memory-limit-mb", HeaderValue::from_static("22"))
+        .header("x-memory-limit-mb", HeaderValue::from_static("30"))
         .body(Body::empty())
         .context("can't make request")
     })

crates/base_rt/src/lib.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use std::num::NonZeroUsize;
 use std::sync::atomic::AtomicI64;
 use std::sync::atomic::Ordering;
@@ -200,4 +201,6 @@ impl BlockingScopeCPUUsageMetricExt for &mut OpState {
 pub struct RuntimeWaker(pub Arc<AtomicWaker>);
 
 #[derive(Debug, Clone)]
-pub struct RuntimeOtelAttribute(pub opentelemetry::Value);
+pub struct RuntimeOtelExtraAttributes(
+  pub HashMap<opentelemetry::Key, opentelemetry::Value>,
+);

examples/main/index.ts

@@ -3,6 +3,12 @@ import { STATUS_CODE } from "https://deno.land/std/http/status.ts";
 
 import { handleRegistryRequest } from "./registry/mod.ts";
 import { join } from "jsr:@std/path@^1.0";
+import { context, propagation } from "npm:@opentelemetry/api";
+import { W3CBaggagePropagator } from "npm:@opentelemetry/core@1";
+
+// @ts-ignore See https://github.com/denoland/deno/issues/28082
+globalThis[Symbol.for("opentelemetry.js.api.1")].propagation =
+  new W3CBaggagePropagator();
 
 console.log("main function started");
 console.log(Deno.version);
@@ -42,6 +48,17 @@ addEventListener("unhandledrejection", (ev) => {
 // }, 30 * 1000);
 
 Deno.serve(async (req: Request) => {
+  const ctx = propagation.extract(context.active(), req.headers, {
+    get(carrier, key) {
+      return carrier.get(key) ?? void 0;
+    },
+    keys(carrier) {
+      return [...carrier.keys()];
+    },
+  });
+  const baggage = propagation.getBaggage(ctx);
+  const requestId = baggage?.getEntry("sb-request-id")?.value ?? null;
+
   const headers = new Headers({
     "Content-Type": "application/json",
   });
@@ -149,7 +166,7 @@ Deno.serve(async (req: Request) => {
 
   // console.error(`serving the request with ${servicePath}`);
 
-  const createWorker = async () => {
+  const createWorker = async (otelAttributes?: { [_: string]: string }) => {
     const memoryLimitMb = 150;
     const workerTimeoutMs = 5 * 60 * 1000;
     const noModuleCache = false;
@@ -184,10 +201,11 @@ Deno.serve(async (req: Request) => {
       staticPatterns,
       context: {
         useReadSyncFileAPI: true,
+        otel: otelAttributes,
       },
       otelConfig: {
         tracing_enabled: true,
-        console: "Replace",
+        propagators: ["TraceContext", "Baggage"],
       },
       // maybeEszip,
       // maybeEntrypoint,
@@ -200,7 +218,14 @@ Deno.serve(async (req: Request) => {
       // If a worker for the given service path already exists,
       // it will be reused by default.
       // Update forceCreate option in createWorker to force create a new worker for each request.
-      const worker = await createWorker();
+      const worker = await createWorker(
+        requestId
+          ? {
+            "sb_request_id": requestId,
+          }
+          : void 0,
+      );
+
       const controller = new AbortController();
 
       const signal = controller.signal;