feat(runtime): send console output to OpenTelemetry collector (#3)

Cargo.lock

@@ -1620,7 +1620,7 @@ dependencies = [
  "http 1.1.0",
  "log",
  "num-bigint",
- "prost",
+ "prost 0.11.9",
  "prost-build",
  "rand",
  "rusqlite",
@@ -1835,6 +1835,7 @@ dependencies = [
 name = "deno_runtime"
 version = "0.169.0"
 dependencies = [
+ "anyhow",
  "deno_ast",
  "deno_broadcast_channel",
  "deno_cache",
@@ -1872,13 +1873,19 @@ dependencies = [
  "hyper-util",
  "libc",
  "log",
+ "maplit",
  "netif",
  "nix 0.26.2",
  "notify",
  "ntapi",
  "once_cell",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry-semantic-conventions",
+ "opentelemetry_sdk",
  "percent-encoding",
  "regex",
+ "reqwest",
  "rustyline",
  "serde",
  "signal-hook",
@@ -2061,7 +2068,7 @@ dependencies = [
  "chrono",
  "futures",
  "num-bigint",
- "prost",
+ "prost 0.11.9",
  "serde",
  "uuid",
 ]
@@ -2081,7 +2088,7 @@ dependencies = [
  "futures",
  "http 1.1.0",
  "log",
- "prost",
+ "prost 0.11.9",
  "rand",
  "serde",
  "serde_json",
@@ -4498,6 +4505,88 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
+[[package]]
+name = "opentelemetry"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4c365a63eec4f55b7efeceb724f1336f26a9cf3427b70e59e2cd2a5b947fba96"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "once_cell",
+ "pin-project-lite",
+ "thiserror",
+]
+
+[[package]]
+name = "opentelemetry-http"
+version = "0.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ad31e9de44ee3538fb9d64fe3376c1362f406162434609e79aea2a41a0af78ab"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "http 1.1.0",
+ "opentelemetry",
+ "reqwest",
+]
+
+[[package]]
+name = "opentelemetry-otlp"
+version = "0.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b925a602ffb916fb7421276b86756027b37ee708f9dce2dbdcc51739f07e727"
+dependencies = [
+ "async-trait",
+ "futures-core",
+ "http 1.1.0",
+ "opentelemetry",
+ "opentelemetry-http",
+ "opentelemetry-proto",
+ "opentelemetry_sdk",
+ "prost 0.13.1",
+ "reqwest",
+ "thiserror",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "30ee9f20bff9c984511a02f082dc8ede839e4a9bf15cc2487c8d6fea5ad850d9"
+dependencies = [
+ "opentelemetry",
+ "opentelemetry_sdk",
+ "prost 0.13.1",
+ "tonic",
+]
+
+[[package]]
+name = "opentelemetry-semantic-conventions"
+version = "0.16.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1cefe0543875379e47eb5f1e68ff83f45cc41366a92dfd0d073d513bf68e9a05"
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.24.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "692eac490ec80f24a17828d49b40b60f5aeaccdfe6a503f939713afd22bc28df"
+dependencies = [
+ "async-trait",
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "glob",
+ "once_cell",
+ "opentelemetry",
+ "percent-encoding",
+ "rand",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "option-ext"
 version = "0.2.0"
@@ -5041,7 +5130,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b82eaa1d779e9a4bc1c3217db8ffbeabaae1dca241bf70183242128d48681cd"
 dependencies = [
  "bytes",
- "prost-derive",
+ "prost-derive 0.11.9",
+]
+
+[[package]]
+name = "prost"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13db3d3fde688c61e2446b4d843bc27a7e8af269a69440c0308021dc92333cc"
+dependencies = [
+ "bytes",
+ "prost-derive 0.13.1",
 ]
 
 [[package]]
@@ -5058,7 +5157,7 @@ dependencies = [
  "multimap",
  "petgraph",
  "prettyplease 0.1.25",
- "prost",
+ "prost 0.11.9",
  "prost-types",
  "regex",
  "syn 1.0.109",
@@ -5079,13 +5178,26 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "prost-derive"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18bec9b0adc4eba778b33684b7ba3e7137789434769ee3ce3930463ef904cfca"
+dependencies = [
+ "anyhow",
+ "itertools",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.58",
+]
+
 [[package]]
 name = "prost-types"
 version = "0.11.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "213622a1460818959ac1181aaeb2dc9c7f63df720db7d788b3e24eacd1983e13"
 dependencies = [
- "prost",
+ "prost 0.11.9",
 ]
 
 [[package]]
@@ -5370,6 +5482,7 @@ dependencies = [
  "async-compression",
  "base64 0.22.1",
  "bytes",
+ "futures-channel",
  "futures-core",
  "futures-util",
  "h2 0.4.4",
@@ -6843,7 +6956,7 @@ dependencies = [
  "os_pipe",
  "parking_lot 0.12.3",
  "pretty_assertions",
- "prost",
+ "prost 0.11.9",
  "prost-build",
  "regex",
  "reqwest",
@@ -7090,6 +7203,27 @@ dependencies = [
  "winnow",
 ]
 
+[[package]]
+name = "tonic"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "38659f4a91aba8598d27821589f5db7dddd94601e7a01b1e485a50e5484c7401"
+dependencies = [
+ "async-trait",
+ "base64 0.22.1",
+ "bytes",
+ "http 1.1.0",
+ "http-body 1.0.0",
+ "http-body-util",
+ "percent-encoding",
+ "pin-project",
+ "prost 0.13.1",
+ "tokio-stream",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
 [[package]]
 name = "tower"
 version = "0.4.13"

cli/args/flags.rs

@@ -589,6 +589,7 @@ pub struct Flags {
   pub permissions: PermissionFlags,
   pub allow_scripts: PackagesAllowedScripts,
   pub eszip: bool,
+  pub otel: bool,
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, Default, Serialize, Deserialize)]
@@ -3330,6 +3331,7 @@ fn runtime_args(
     .arg(enable_testing_features_arg())
     .arg(strace_ops_arg())
     .arg(eszip_arg())
+    .arg(otel_arg())
 }
 
 fn inspect_args(app: Command) -> Command {
@@ -3456,6 +3458,14 @@ fn eszip_arg() -> Arg {
     .help("Run eszip")
 }
 
+fn otel_arg() -> Arg {
+  Arg::new("otel-internal-do-not-use")
+    .long("otel-internal-do-not-use")
+    .action(ArgAction::SetTrue)
+    .help("Enable OpenTelemetry")
+    .hide(true)
+}
+
 /// Used for subcommands that operate on executable scripts only.
 /// `deno fmt` has its own `--ext` arg because its possible values differ.
 /// If --ext is not provided and the script doesn't have a file extension,
@@ -4740,6 +4750,7 @@ fn runtime_args_parse(
   env_file_arg_parse(flags, matches);
   strace_ops_parse(flags, matches);
   eszip_arg_parse(flags, matches);
+  otel_arg_parse(flags, matches);
 }
 
 fn inspect_arg_parse(flags: &mut Flags, matches: &mut ArgMatches) {
@@ -4831,6 +4842,12 @@ fn eszip_arg_parse(flags: &mut Flags, matches: &mut ArgMatches) {
   }
 }
 
+fn otel_arg_parse(flags: &mut Flags, matches: &mut ArgMatches) {
+  if matches.get_flag("otel-internal-do-not-use") {
+    flags.otel = true;
+  }
+}
+
 fn ext_arg_parse(flags: &mut Flags, matches: &mut ArgMatches) {
   flags.ext = matches.remove_one::<String>("ext");
 }

cli/args/mod.rs

@@ -29,6 +29,7 @@ use deno_runtime::deno_fs::DenoConfigFsAdapter;
 use deno_runtime::deno_fs::RealFs;
 use deno_runtime::deno_permissions::PermissionsContainer;
 use deno_runtime::deno_tls::RootCertStoreProvider;
+use deno_runtime::ops::otel::OtelConfig;
 use deno_semver::npm::NpmPackageReqReference;
 use import_map::resolve_import_map_value_from_specifier;
 
@@ -67,6 +68,7 @@ use once_cell::sync::Lazy;
 use once_cell::sync::OnceCell;
 use serde::Deserialize;
 use serde::Serialize;
+use std::borrow::Cow;
 use std::collections::HashMap;
 use std::env;
 use std::io::BufReader;
@@ -1103,6 +1105,13 @@ impl CliOptions {
     }
   }
 
+  pub fn otel_config(&self) -> Option<OtelConfig> {
+    self.flags.otel.then(|| OtelConfig {
+      default_service_name: Cow::Borrowed("deno"),
+      default_service_version: Cow::Borrowed(crate::version::deno()),
+    })
+  }
+
   pub fn env_file_name(&self) -> Option<&String> {
     self.flags.env_file.as_ref()
   }

cli/factory.rs

@@ -833,6 +833,7 @@ impl CliFactory {
       } else {
         None
       },
+      self.options.otel_config(),
     ))
   }
 

cli/standalone/binary.rs

@@ -30,6 +30,7 @@ use deno_core::serde_json;
 use deno_core::url::Url;
 use deno_npm::NpmSystemInfo;
 use deno_runtime::deno_node::PackageJson;
+use deno_runtime::ops::otel::OtelConfig;
 use deno_semver::npm::NpmVersionReqParseError;
 use deno_semver::package::PackageReq;
 use deno_semver::VersionReqSpecifierParseError;
@@ -103,6 +104,7 @@ pub struct Metadata {
   pub node_modules: Option<NodeModules>,
   pub disable_deprecated_api_warning: bool,
   pub unstable_config: UnstableConfig,
+  pub otel_config: Option<OtelConfig>, // None means disabled.
 }
 
 pub fn load_npm_vfs(root_dir_path: PathBuf) -> Result<FileBackedVfs, AnyError> {
@@ -645,6 +647,7 @@ impl<'a> DenoCompileBinaryWriter<'a> {
         sloppy_imports: cli_options.unstable_sloppy_imports(),
         features: cli_options.unstable_features(),
       },
+      otel_config: cli_options.otel_config(),
     };
 
     write_binary_bytes(

cli/standalone/mod.rs

@@ -740,6 +740,7 @@ pub async fn run(
     false,
     // Code cache is not supported for standalone binary yet.
     None,
+    metadata.otel_config,
   );
 
   // Initialize v8 once from the main thread.

cli/tools/run/mod.rs

@@ -1,5 +1,6 @@
 // Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
 
+use std::borrow::Cow;
 use std::io::Read;
 
 use deno_config::workspace::PackageJsonDepResolution;
@@ -11,6 +12,7 @@ use deno_core::futures::StreamExt;
 use deno_core::unsync::spawn;
 use deno_runtime::deno_permissions::Permissions;
 use deno_runtime::deno_permissions::PermissionsContainer;
+use deno_runtime::ops::otel::OtelConfig;
 use deno_runtime::WorkerExecutionMode;
 use eszip::EszipV2;
 use tokio_util::compat::TokioAsyncReadCompatExt;
@@ -289,6 +291,10 @@ pub async fn run_eszip(
         package_jsons: Default::default(),
         pkg_json_resolution: PackageJsonDepResolution::Disabled,
       },
+      otel_config: flags.otel.then(|| OtelConfig {
+        default_service_name: Cow::Borrowed("deno"),
+        default_service_version: Cow::Borrowed(crate::version::deno()),
+      }),
     },
     run_flags.script.as_bytes(),
     "run-eszip",

cli/worker.rs

@@ -31,6 +31,7 @@ use deno_runtime::deno_tls::RootCertStoreProvider;
 use deno_runtime::deno_web::BlobStore;
 use deno_runtime::fmt_errors::format_js_error;
 use deno_runtime::inspector_server::InspectorServer;
+use deno_runtime::ops::otel::OtelConfig;
 use deno_runtime::ops::worker_host::CreateWebWorkerCb;
 use deno_runtime::web_worker::WebWorker;
 use deno_runtime::web_worker::WebWorkerOptions;
@@ -143,6 +144,7 @@ struct SharedWorkerState {
   code_cache: Option<Arc<dyn code_cache::CodeCache>>,
   serve_port: Option<u16>,
   serve_host: Option<String>,
+  otel_config: Option<OtelConfig>, // `None` means OpenTelemetry is disabled.
 }
 
 impl SharedWorkerState {
@@ -417,6 +419,7 @@ impl CliMainWorkerFactory {
     disable_deprecated_api_warning: bool,
     verbose_deprecated_api_warning: bool,
     code_cache: Option<Arc<dyn code_cache::CodeCache>>,
+    otel_config: Option<OtelConfig>,
   ) -> Self {
     Self {
       shared: Arc::new(SharedWorkerState {
@@ -443,6 +446,7 @@ impl CliMainWorkerFactory {
         disable_deprecated_api_warning,
         verbose_deprecated_api_warning,
         code_cache,
+        otel_config,
       }),
     }
   }
@@ -585,6 +589,7 @@ impl CliMainWorkerFactory {
         mode,
         serve_port: shared.serve_port,
         serve_host: shared.serve_host.clone(),
+        otel_config: shared.otel_config.clone(),
       },
       extensions: custom_extensions,
       startup_snapshot: crate::js::deno_isolate_init(),
@@ -789,6 +794,7 @@ fn create_web_worker_callback(
         mode,
         serve_port: shared.serve_port,
         serve_host: shared.serve_host.clone(),
+        otel_config: shared.otel_config.clone(),
       },
       extensions: vec![],
       startup_snapshot: crate::js::deno_isolate_init(),

runtime/Cargo.toml

@@ -68,6 +68,7 @@ serde.workspace = true
 winapi.workspace = true
 
 [dependencies]
+anyhow.workspace = true
 deno_ast.workspace = true
 deno_broadcast_channel.workspace = true
 deno_cache.workspace = true
@@ -105,11 +106,17 @@ hyper-util.workspace = true
 hyper_v014 = { workspace = true, features = ["server", "stream", "http1", "http2", "runtime"] }
 libc.workspace = true
 log.workspace = true
+maplit = "1.0.2"
 netif = "0.1.6"
 notify.workspace = true
 once_cell.workspace = true
+opentelemetry = "0.24.0"
+opentelemetry-otlp = { version = "0.17.0", default-features = false, features = ["logs", "http-proto", "reqwest-rustls-webpki-roots"] }
+opentelemetry-semantic-conventions = "0.16.0"
+opentelemetry_sdk = "0.24.1"
 percent-encoding.workspace = true
 regex.workspace = true
+reqwest.workspace = true
 rustyline = { workspace = true, features = ["custom-bindings"] }
 serde.workspace = true
 signal-hook = "0.3.17"

runtime/js/99_main.js

@@ -13,6 +13,7 @@ import {
   op_bootstrap_no_color,
   op_bootstrap_pid,
   op_main_module,
+  op_otel_log,
   op_ppid,
   op_set_format_exception_callback,
   op_snapshot_options,
@@ -61,6 +62,7 @@ import * as version from "ext:runtime/01_version.ts";
 import * as os from "ext:runtime/30_os.js";
 import * as timers from "ext:deno_web/02_timers.js";
 import {
+  Console,
   customInspect,
   getDefaultInspectOptions,
   getStderrNoColor,
@@ -710,6 +712,7 @@ function bootstrapMainRuntime(runtimeOptions, warmup = false) {
       11: mode,
       12: servePort,
       13: serveHost,
+      14: otelEnabled,
     } = runtimeOptions;
 
     if (mode === executionModes.run || mode === executionModes.serve) {
@@ -796,6 +799,15 @@ function bootstrapMainRuntime(runtimeOptions, warmup = false) {
     });
     ObjectSetPrototypeOf(globalThis, Window.prototype);
 
+    if (otelEnabled) {
+      const otelConsole = new Console(op_otel_log);
+      ObjectDefineProperty(
+        globalThis,
+        "console",
+        core.propNonEnumerable(otelConsole),
+      );
+    }
+
     if (inspectFlag) {
       const consoleFromDeno = globalThis.console;
       core.wrapConsole(consoleFromDeno, core.v8Console);
@@ -924,6 +936,10 @@ function bootstrapWorkerRuntime(
       8: shouldDisableDeprecatedApiWarning,
       9: shouldUseVerboseDeprecatedApiWarning,
       10: future,
+      // 11: mode,
+      // 12: servePort,
+      // 13: serveHost,
+      14: otelEnabled,
     } = runtimeOptions;
 
     // TODO(iuioiua): remove in Deno v2. This allows us to dynamically delete
@@ -961,6 +977,15 @@ function bootstrapWorkerRuntime(
     }
     ObjectSetPrototypeOf(globalThis, DedicatedWorkerGlobalScope.prototype);
 
+    if (otelEnabled) {
+      const otelConsole = new Console(op_otel_log);
+      ObjectDefineProperty(
+        globalThis,
+        "console",
+        core.propNonEnumerable(otelConsole),
+      );
+    }
+
     const consoleFromDeno = globalThis.console;
     core.wrapConsole(consoleFromDeno, core.v8Console);
 

runtime/ops/mod.rs

@@ -4,6 +4,7 @@ pub mod bootstrap;
 pub mod fs_events;
 pub mod http;
 pub mod os;
+pub mod otel;
 pub mod permissions;
 pub mod process;
 pub mod runtime;

runtime/ops/os/mod.rs

@@ -174,6 +174,8 @@ fn op_get_exit_code(state: &mut OpState) -> i32 {
 
 #[op2(fast)]
 fn op_exit(state: &mut OpState) {
+  crate::ops::otel::otel_drop_logger(state);
+
   let code = state.borrow::<ExitCode>().get();
   std::process::exit(code)
 }

runtime/ops/otel.rs

@@ -0,0 +1,308 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+
+use crate::tokio_util::create_basic_runtime;
+use anyhow::anyhow;
+use deno_core::futures::channel::mpsc;
+use deno_core::futures::channel::mpsc::UnboundedSender;
+use deno_core::futures::future::BoxFuture;
+use deno_core::futures::stream;
+use deno_core::futures::Stream;
+use deno_core::futures::StreamExt;
+use deno_core::op2;
+use deno_core::OpState;
+use maplit::hashmap;
+use once_cell::sync::Lazy;
+use opentelemetry::logs::LogRecord;
+use opentelemetry::logs::Logger as LoggerTrait;
+use opentelemetry::logs::LoggerProvider;
+use opentelemetry::logs::Severity;
+use opentelemetry::Key;
+use opentelemetry::KeyValue;
+use opentelemetry_otlp::HttpExporterBuilder;
+use opentelemetry_otlp::LogExporterBuilder;
+use opentelemetry_sdk::logs::Logger;
+use opentelemetry_sdk::Resource;
+use opentelemetry_semantic_conventions::resource::SERVICE_NAME;
+use opentelemetry_semantic_conventions::resource::SERVICE_VERSION;
+use serde::Deserialize;
+use serde::Serialize;
+use std::borrow::Cow;
+use std::env;
+use std::fmt::Debug;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+use std::thread;
+use std::time::Duration;
+
+deno_core::extension!(
+  deno_otel,
+  ops = [op_otel_log],
+  options = {
+    otel_config: Option<OtelConfig>, // `None` means OpenTelemetry is disabled.
+  },
+  state = |state, options| {
+    if let Some(otel_config) = options.otel_config {
+      let logger = otel_create_logger(otel_config)
+        .expect("Failed to create OpenTelemetry logger");
+      state.put::<Logger>(logger);
+    }
+  }
+);
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct OtelConfig {
+  pub default_service_name: Cow<'static, str>,
+  pub default_service_version: Cow<'static, str>,
+}
+
+impl Default for OtelConfig {
+  fn default() -> Self {
+    Self {
+      default_service_name: Cow::Borrowed(env!("CARGO_PKG_NAME")),
+      default_service_version: Cow::Borrowed(env!("CARGO_PKG_VERSION")),
+    }
+  }
+}
+
+static OTEL_SHARED_RUNTIME_SPAWN_TASK_TX: Lazy<
+  UnboundedSender<BoxFuture<'static, ()>>,
+> = Lazy::new(otel_create_shared_runtime);
+
+fn otel_create_shared_runtime() -> UnboundedSender<BoxFuture<'static, ()>> {
+  let (spawn_task_tx, mut spawn_task_rx) =
+    mpsc::unbounded::<BoxFuture<'static, ()>>();
+
+  thread::spawn(move || {
+    let rt = create_basic_runtime();
+    rt.block_on(async move {
+      while let Some(task) = spawn_task_rx.next().await {
+        tokio::spawn(task);
+      }
+    });
+  });
+
+  spawn_task_tx
+}
+
+#[derive(Clone, Copy)]
+struct OtelSharedRuntime;
+
+impl opentelemetry_sdk::runtime::Runtime for OtelSharedRuntime {
+  type Interval = Pin<Box<dyn Stream<Item = ()> + Send + 'static>>;
+  type Delay = Pin<Box<tokio::time::Sleep>>;
+
+  fn interval(&self, period: Duration) -> Self::Interval {
+    stream::repeat(())
+      .then(move |_| tokio::time::sleep(period))
+      .boxed()
+  }
+
+  fn spawn(&self, future: BoxFuture<'static, ()>) {
+    (*OTEL_SHARED_RUNTIME_SPAWN_TASK_TX)
+      .unbounded_send(future)
+      .expect("failed to send task to shared OpenTelemetry runtime");
+  }
+
+  fn delay(&self, duration: Duration) -> Self::Delay {
+    Box::pin(tokio::time::sleep(duration))
+  }
+}
+
+impl opentelemetry_sdk::runtime::RuntimeChannel for OtelSharedRuntime {
+  type Receiver<T: Debug + Send> = BatchMessageChannelReceiver<T>;
+  type Sender<T: Debug + Send> = BatchMessageChannelSender<T>;
+
+  fn batch_message_channel<T: Debug + Send>(
+    &self,
+    capacity: usize,
+  ) -> (Self::Sender<T>, Self::Receiver<T>) {
+    let (batch_tx, batch_rx) = tokio::sync::mpsc::channel::<T>(capacity);
+    (batch_tx.into(), batch_rx.into())
+  }
+}
+
+#[derive(Debug)]
+pub struct BatchMessageChannelSender<T: Send> {
+  sender: tokio::sync::mpsc::Sender<T>,
+}
+
+impl<T: Send> From<tokio::sync::mpsc::Sender<T>>
+  for BatchMessageChannelSender<T>
+{
+  fn from(sender: tokio::sync::mpsc::Sender<T>) -> Self {
+    Self { sender }
+  }
+}
+
+impl<T: Send> opentelemetry_sdk::runtime::TrySend
+  for BatchMessageChannelSender<T>
+{
+  type Message = T;
+
+  fn try_send(
+    &self,
+    item: Self::Message,
+  ) -> Result<(), opentelemetry_sdk::runtime::TrySendError> {
+    self.sender.try_send(item).map_err(|err| match err {
+      tokio::sync::mpsc::error::TrySendError::Full(_) => {
+        opentelemetry_sdk::runtime::TrySendError::ChannelFull
+      }
+      tokio::sync::mpsc::error::TrySendError::Closed(_) => {
+        opentelemetry_sdk::runtime::TrySendError::ChannelClosed
+      }
+    })
+  }
+}
+
+pub struct BatchMessageChannelReceiver<T> {
+  receiver: tokio::sync::mpsc::Receiver<T>,
+}
+
+impl<T> From<tokio::sync::mpsc::Receiver<T>>
+  for BatchMessageChannelReceiver<T>
+{
+  fn from(receiver: tokio::sync::mpsc::Receiver<T>) -> Self {
+    Self { receiver }
+  }
+}
+
+impl<T> Stream for BatchMessageChannelReceiver<T> {
+  type Item = T;
+
+  fn poll_next(
+    mut self: Pin<&mut Self>,
+    cx: &mut Context<'_>,
+  ) -> Poll<Option<Self::Item>> {
+    self.receiver.poll_recv(cx)
+  }
+}
+
+fn otel_create_logger(config: OtelConfig) -> anyhow::Result<Logger> {
+  // Parse the `OTEL_EXPORTER_OTLP_PROTOCOL` variable. The opentelemetry_*
+  // crates don't do this automatically. Currently, the only supported protocol
+  // is "http/protobuf".
+  // TODO(piscisaureus): enable GRPC support.
+  let _protocol = match env::var("OTEL_EXPORTER_OTLP_PROTOCOL").as_deref() {
+    Ok(protocol @ "http/protobuf") => protocol,
+    Ok("") | Err(env::VarError::NotPresent) => {
+      return Err(anyhow!("OTEL_EXPORTER_OTLP_PROTOCOL must be set",))
+    }
+    Ok(protocol) => {
+      return Err(anyhow!(
+        "Env var OTEL_EXPORTER_OTLP_PROTOCOL specifies an unsupported protocol: {}",
+        protocol
+      ));
+    }
+    Err(err) => {
+      return Err(anyhow!(
+        "Failed to read env var OTEL_EXPORTER_OTLP_PROTOCOL: {}",
+        err
+      ))
+    }
+  };
+
+  // Verify that `OTEL_EXPORTER_OTLP_ENDPOINT` is set. If unspecified,
+  // `HttpExporterBuilder` will use http://localhost:4317 as the default
+  // endpoint, but this seems not very useful.
+  match env::var("OTEL_EXPORTER_OTLP_ENDPOINT").as_deref() {
+    Ok(endpoint) if !endpoint.is_empty() => {}
+    Ok(_) | Err(env::VarError::NotPresent) => {
+      return Err(anyhow!("OTEL_EXPORTER_OTLP_ENDPOINT must be set",))
+    }
+    Err(err) => {
+      return Err(anyhow!(
+        "Failed to read env var OTEL_EXPORTER_OTLP_ENDPOINT: {}",
+        err
+      ))
+    }
+  };
+
+  // The OTLP endpoint is automatically picked up from the
+  // `OTEL_EXPORTER_OTLP_ENDPOINT` environment variable. Additional headers can
+  // be specified using `OTEL_EXPORTER_OTLP_HEADERS`.
+  let exporter = LogExporterBuilder::Http(
+    HttpExporterBuilder::default().with_http_client(reqwest::Client::new()),
+  );
+
+  // Define the resource attributes that will be attached to all log records.
+  // These attributes are sourced as follows (in order of precedence):
+  //   * The `service.name` attribute from the `OTEL_SERVICE_NAME` env var.
+  //   * Additional attributes from the `OTEL_RESOURCE_ATTRIBUTES` env var.
+  //   * Default attribute values defined here.
+  // TODO(piscisaureus): add more default attributes (e.g. script path).
+  let mut resource = Resource::default();
+  // The default service name assigned by `Resource::default()`, if not
+  // otherwise specified via environment variables, is "unknown_service".
+  // Override this with the current crate name and version.
+  if resource
+    .get(Key::from_static_str(SERVICE_NAME))
+    .filter(|service_name| service_name.as_str() != "unknown_service")
+    .is_none()
+  {
+    resource = resource.merge(&Resource::new(
+      hashmap! {
+        SERVICE_NAME => config.default_service_name,
+        SERVICE_VERSION => config.default_service_version,
+      }
+      .into_iter()
+      .map(|(k, v)| KeyValue::new(k, v)),
+    ))
+  }
+
+  let logging_provider = opentelemetry_otlp::new_pipeline()
+    .logging()
+    .with_exporter(exporter)
+    .with_resource(resource)
+    .install_batch(OtelSharedRuntime)?;
+
+  // Create the `Logger` instance that will be used to emit console logs.
+  // The "console" argument is used to specify the `otel.scope.name` attribute,
+  // which is a standard attribute to instrumentation scope of log records.
+  let logger = logging_provider.logger_builder("console").build();
+
+  Ok(logger)
+}
+
+/// This function is called by the runtime whenever it is about to call
+/// `os::process::exit()`, to ensure that all OpenTelemetry logs are properly
+/// flushed before the process terminates.
+pub fn otel_drop_logger(state: &mut OpState) {
+  let Some(logger) = state.try_take::<Logger>() else {
+    // Since this function is called unconditionaly before `os::process::exit()`,
+    // it is not an error if the logger is not available.
+    return;
+  };
+
+  // When the `Logger` is dropped, the underlying `LoggerProvider` will be
+  // dropped as well. The provider's Drop implementation will flush all logs
+  // and block until the exporter has successfully sent them.
+  drop(logger);
+}
+
+#[op2(fast)]
+fn op_otel_log(
+  state: &mut OpState,
+  #[string] message: String,
+  #[smi] level: i32,
+) {
+  let Some(logger) = state.try_borrow::<Logger>() else {
+    log::error!("op_otel_log: OpenTelemetry Logger not available");
+    return;
+  };
+
+  // Convert the integer log level that ext/console uses to the corresponding
+  // OpenTelemetry log severity.
+  let severity = match level {
+    ..=0 => Severity::Debug,
+    1 => Severity::Info,
+    2 => Severity::Warn,
+    3.. => Severity::Error,
+  };
+
+  let mut log_record = logger.create_log_record();
+  log_record.set_body(message.into());
+  log_record.set_severity_number(severity);
+  log_record.set_severity_text(Cow::Borrowed(severity.name()));
+  logger.emit(log_record);
+}

runtime/snapshot.rs

@@ -263,6 +263,7 @@ pub fn create_runtime_snapshot(
     ),
     ops::fs_events::deno_fs_events::init_ops(),
     ops::os::deno_os::init_ops(Default::default()),
+    ops::otel::deno_otel::init_ops(None),
     ops::permissions::deno_permissions::init_ops(),
     ops::process::deno_process::init_ops(),
     ops::signal::deno_signal::init_ops(),

runtime/web_worker.rs

@@ -504,6 +504,7 @@ impl WebWorker {
       ),
       ops::fs_events::deno_fs_events::init_ops_and_esm(),
       ops::os::deno_os_worker::init_ops_and_esm(),
+      ops::otel::deno_otel::init_ops_and_esm(options.bootstrap.otel_config),
       ops::permissions::deno_permissions::init_ops_and_esm(),
       ops::process::deno_process::init_ops_and_esm(),
       ops::signal::deno_signal::init_ops_and_esm(),

runtime/worker.rs

@@ -430,6 +430,7 @@ impl MainWorker {
       ),
       ops::fs_events::deno_fs_events::init_ops_and_esm(),
       ops::os::deno_os::init_ops_and_esm(exit_code.clone()),
+      ops::otel::deno_otel::init_ops_and_esm(options.bootstrap.otel_config),
       ops::permissions::deno_permissions::init_ops_and_esm(),
       ops::process::deno_process::init_ops_and_esm(),
       ops::signal::deno_signal::init_ops_and_esm(),

runtime/worker_bootstrap.rs

@@ -8,6 +8,8 @@ use std::thread;
 
 use deno_terminal::colors;
 
+use crate::ops::otel::OtelConfig;
+
 /// The execution mode for this worker. Some modes may have implicit behaviour.
 #[derive(Copy, Clone)]
 #[repr(u8)]
@@ -94,6 +96,8 @@ pub struct BootstrapOptions {
   // Used by `deno serve`
   pub serve_port: Option<u16>,
   pub serve_host: Option<String>,
+  // OpenTelemetry output options. If `None`, OpenTelemetry is disabled.
+  pub otel_config: Option<OtelConfig>,
 }
 
 impl Default for BootstrapOptions {
@@ -130,6 +134,7 @@ impl Default for BootstrapOptions {
       mode: WorkerExecutionMode::None,
       serve_port: Default::default(),
       serve_host: Default::default(),
+      otel_config: None,
     }
   }
 }
@@ -173,6 +178,8 @@ struct BootstrapV8<'a>(
   u16,
   // serve host
   Option<&'a str>,
+  // OTEL enabled
+  bool,
 );
 
 impl BootstrapOptions {
@@ -199,6 +206,7 @@ impl BootstrapOptions {
       self.mode as u8 as _,
       self.serve_port.unwrap_or_default(),
       self.serve_host.as_deref(),
+      self.otel_config.is_some(),
     );
 
     bootstrap.serialize(ser).unwrap()