ext/telemetry/lib.rs - main - githubmirror/deno

ext/telemetry/lib.rs
3 433 строки99 KB
Nathan Whitaker
fix(telemetry): handle reentrant attribute conversion (#36256)
24 июл 2026, 01:43
24 июл 2026, 01:43922b951
// Copyright 2018-2026 the Deno authors. MIT license.

#![expect(
  unexpected_cfgs,
  reason = "allow internal-logs, which is generated by otel_debug! macros, but we don't use it"
)]
#![allow(
  clippy::too_many_arguments,
  reason = "macro expansion causes too many arguments"
)]

use std::borrow::Cow;
use std::cell::RefCell;
use std::collections::HashMap;
use std::ffi::c_void;
use std::fmt::Debug;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicU64;
use std::task::Context;
use std::task::Poll;
use std::thread;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;

use deno_core::GarbageCollected;
use deno_core::OpState;
use deno_core::ToV8;
use deno_core::futures::FutureExt;
use deno_core::futures::Stream;
use deno_core::futures::StreamExt;
use deno_core::futures::channel::mpsc;
use deno_core::futures::channel::mpsc::UnboundedSender;
use deno_core::futures::future::BoxFuture;
use deno_core::op2;
use deno_core::v8;
use deno_core::v8::DataError;
use deno_error::JsError;
use deno_error::JsErrorBox;
use once_cell::sync::Lazy;
use once_cell::sync::OnceCell;
use opentelemetry::Array;
use opentelemetry::InstrumentationScope;
pub use opentelemetry::Key;
pub use opentelemetry::KeyValue;
pub use opentelemetry::StringValue;
pub use opentelemetry::Value;
use opentelemetry::logs::AnyValue;
pub use opentelemetry::logs::LogRecord as LogRecordTrait;
use opentelemetry::logs::Severity;
use opentelemetry::metrics::AsyncInstrumentBuilder;
pub use opentelemetry::metrics::Gauge;
pub use opentelemetry::metrics::Histogram;
use opentelemetry::metrics::InstrumentBuilder;
pub use opentelemetry::metrics::MeterProvider;
pub use opentelemetry::metrics::UpDownCounter;
use opentelemetry::otel_debug;
use opentelemetry::otel_error;
use opentelemetry::trace::Event;
use opentelemetry::trace::Link;
use opentelemetry::trace::SpanContext;
use opentelemetry::trace::SpanId;
use opentelemetry::trace::SpanKind;
use opentelemetry::trace::Status as SpanStatus;
use opentelemetry::trace::TraceFlags;
use opentelemetry::trace::TraceId;
use opentelemetry::trace::TraceState;
use opentelemetry_otlp::HttpExporterBuilder;
use opentelemetry_otlp::Protocol;
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_otlp::WithHttpConfig;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::error::OTelSdkResult;
use opentelemetry_sdk::logs::LogProcessor;
pub use opentelemetry_sdk::logs::SdkLogRecord as LogRecord;
use opentelemetry_sdk::logs::log_processor_with_async_runtime::BatchLogProcessor;
use opentelemetry_sdk::metrics::ManualReader;
use opentelemetry_sdk::metrics::SdkMeterProvider;
use opentelemetry_sdk::metrics::Temporality;
use opentelemetry_sdk::metrics::exporter::PushMetricExporter;
use opentelemetry_sdk::metrics::reader::MetricReader;
use opentelemetry_sdk::trace::IdGenerator;
use opentelemetry_sdk::trace::RandomIdGenerator;
use opentelemetry_sdk::trace::SpanData;
use opentelemetry_sdk::trace::SpanEvents;
use opentelemetry_sdk::trace::SpanLinks;
use opentelemetry_sdk::trace::SpanProcessor as _;
use opentelemetry_sdk::trace::span_processor_with_async_runtime::BatchSpanProcessor;
use opentelemetry_semantic_conventions::resource::PROCESS_RUNTIME_NAME;
use opentelemetry_semantic_conventions::resource::PROCESS_RUNTIME_VERSION;
use opentelemetry_semantic_conventions::resource::TELEMETRY_SDK_LANGUAGE;
use opentelemetry_semantic_conventions::resource::TELEMETRY_SDK_NAME;
use opentelemetry_semantic_conventions::resource::TELEMETRY_SDK_VERSION;
use serde::Deserialize;
use serde::Serialize;
use sys_traits::EnvVar;
use sys_traits::FsRead;
use thiserror::Error;
use tokio::sync::oneshot;
use tokio::task::JoinSet;

mod console_exporter;
mod grpc_exporter;

deno_core::extension!(
  deno_telemetry,
  ops = [
    op_otel_collect_isolate_metrics,
    op_otel_enable_isolate_metrics,
    op_otel_log,
    op_otel_log_foreign,
    op_otel_span_attribute1,
    op_otel_span_attribute2,
    op_otel_span_attribute3,
    op_otel_span_add_link,
    op_otel_span_update_name,
    op_otel_metric_attribute3,
    op_otel_metric_record0,
    op_otel_metric_record1,
    op_otel_metric_record2,
    op_otel_metric_record3,
    op_otel_metric_observable_record0,
    op_otel_metric_observable_record1,
    op_otel_metric_observable_record2,
    op_otel_metric_observable_record3,
    op_otel_metric_wait_to_observe,
    op_otel_metric_observation_done,
  ],
  objects = [OtelTracer, OtelMeter, OtelSpan],
  lazy_loaded_js = ["telemetry.ts", "util.ts"],
);

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct OtelRuntimeConfig {
  pub runtime_name: Cow<'static, str>,
  pub runtime_version: Cow<'static, str>,
}

#[derive(Default, Debug, Clone, Serialize, Deserialize)]
pub struct OtelConfig {
  pub tracing_enabled: bool,
  pub metrics_enabled: bool,
  pub console: OtelConsoleConfig,
  pub deterministic_prefix: Option<u8>,
  pub propagators: std::collections::HashSet<OtelPropagators>,
}

impl OtelConfig {
  pub fn as_v8(&self) -> Box<[u8]> {
    let mut data = vec![
      self.tracing_enabled as u8,
      self.metrics_enabled as u8,
      self.console as u8,
    ];

    data.extend(self.propagators.iter().map(|propagator| *propagator as u8));

    data.into_boxed_slice()
  }
}

#[derive(
  Default, Debug, Clone, Copy, Serialize, Deserialize, Eq, PartialEq, Hash,
)]
#[repr(u8)]
pub enum OtelPropagators {
  TraceContext = 0,
  Baggage = 1,
  #[default]
  None = 2,
}

#[derive(
  Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize,
)]
#[repr(u8)]
pub enum OtelConsoleConfig {
  #[default]
  Ignore = 0,
  Capture = 1,
  Replace = 2,
}

static OTEL_SHARED_RUNTIME_SPAWN_TASK_TX: Lazy<
  UnboundedSender<BoxFuture<'static, ()>>,
> = Lazy::new(otel_create_shared_runtime);

static OTEL_PRE_COLLECT_CALLBACKS: Lazy<
  Mutex<Vec<oneshot::Sender<oneshot::Sender<()>>>>,
> = Lazy::new(Default::default);

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 = tokio::runtime::Builder::new_current_thread()
      .enable_io()
      .enable_time()
      // This limits the number of threads for blocking operations (like for
      // synchronous fs ops) or CPU bound tasks like when we run dprint in
      // parallel for deno fmt.
      // The default value is 512, which is an unhelpfully large thread pool. We
      // don't ever want to have more than a couple dozen threads.
      .max_blocking_threads(if cfg!(windows) {
        // on windows, tokio uses blocking tasks for child process IO, make sure
        // we have enough available threads for other tasks to run
        4 * std::thread::available_parallelism()
          .map(|n| n.get())
          .unwrap_or(8)
      } else {
        32
      })
      .build()
      .unwrap();

    rt.block_on(async move {
      while let Some(task) = spawn_task_rx.next().await {
        tokio::spawn(task);
      }
    });
  });

  spawn_task_tx
}

#[derive(Clone, Copy)]
pub struct OtelSharedRuntime;

impl hyper::rt::Executor<BoxFuture<'static, ()>> for OtelSharedRuntime {
  fn execute(&self, fut: BoxFuture<'static, ()>) {
    (*OTEL_SHARED_RUNTIME_SPAWN_TASK_TX)
      .unbounded_send(fut)
      .expect("failed to send task to shared OpenTelemetry runtime");
  }
}

impl opentelemetry_sdk::runtime::Runtime for OtelSharedRuntime {
  fn spawn<F>(&self, future: F)
  where
    F: std::future::Future<Output = ()> + Send + 'static,
  {
    (*OTEL_SHARED_RUNTIME_SPAWN_TASK_TX)
      .unbounded_send(future.boxed())
      .expect("failed to send task to shared OpenTelemetry runtime");
  }

  fn delay(
    &self,
    duration: Duration,
  ) -> impl std::future::Future<Output = ()> + Send + 'static {
    tokio::time::sleep(duration)
  }
}

/// Mint a blank [`LogRecord`]. In opentelemetry 0.32 `SdkLogRecord` no longer
/// has a public constructor; records are created through a logger. We keep a
/// process-wide factory logger purely to produce empty records, which are then
/// populated and emitted with an explicit instrumentation scope via the log
/// processor.
fn new_log_record() -> LogRecord {
  use opentelemetry::logs::Logger as _;
  use opentelemetry::logs::LoggerProvider as _;

  static LOG_RECORD_FACTORY: Lazy<opentelemetry_sdk::logs::SdkLogger> =
    Lazy::new(|| {
      opentelemetry_sdk::logs::SdkLoggerProvider::builder()
        .build()
        .logger("deno")
    });
  LOG_RECORD_FACTORY.create_log_record()
}

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)
  }
}

enum DenoPeriodicReaderMessage {
  Register(std::sync::Weak<opentelemetry_sdk::metrics::Pipeline>),
  Export,
  ForceFlush(oneshot::Sender<OTelSdkResult>),
  Shutdown(oneshot::Sender<OTelSdkResult>),
}

#[derive(Debug)]
struct DenoPeriodicReader {
  tx: tokio::sync::mpsc::Sender<DenoPeriodicReaderMessage>,
  temporality: Temporality,
}

impl MetricReader for DenoPeriodicReader {
  fn register_pipeline(
    &self,
    pipeline: std::sync::Weak<opentelemetry_sdk::metrics::Pipeline>,
  ) {
    let _ = self
      .tx
      .try_send(DenoPeriodicReaderMessage::Register(pipeline));
  }

  fn collect(
    &self,
    _rm: &mut opentelemetry_sdk::metrics::data::ResourceMetrics,
  ) -> OTelSdkResult {
    unreachable!("collect should not be called on DenoPeriodicReader");
  }

  fn force_flush(&self) -> OTelSdkResult {
    let (tx, rx) = oneshot::channel();
    let _ = self.tx.try_send(DenoPeriodicReaderMessage::ForceFlush(tx));
    deno_core::futures::executor::block_on(rx).unwrap()?;
    Ok(())
  }

  fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
    let (tx, rx) = oneshot::channel();
    let _ = self.tx.try_send(DenoPeriodicReaderMessage::Shutdown(tx));
    deno_core::futures::executor::block_on(rx).unwrap()?;
    Ok(())
  }

  fn temporality(
    &self,
    _kind: opentelemetry_sdk::metrics::InstrumentKind,
  ) -> Temporality {
    self.temporality
  }
}

const METRIC_EXPORT_INTERVAL_NAME: &str = "OTEL_METRIC_EXPORT_INTERVAL";
const DEFAULT_INTERVAL: Duration = Duration::from_secs(60);

impl DenoPeriodicReader {
  fn new<E: PushMetricExporter + 'static>(
    sys: &impl TelemetrySys,
    exporter: E,
  ) -> Self {
    let interval = sys
      .env_var(METRIC_EXPORT_INTERVAL_NAME)
      .ok()
      .and_then(|v| {
        v.parse()
          .map(Duration::from_millis)
          .ok()
          .filter(|d| !d.is_zero())
      })
      .unwrap_or(DEFAULT_INTERVAL);

    let (tx, mut rx) = tokio::sync::mpsc::channel(256);

    let temporality = PushMetricExporter::temporality(&exporter);

    let worker = async move {
      let inner = ManualReader::builder()
        .with_temporality(PushMetricExporter::temporality(&exporter))
        .build();

      let collect_and_export = |collect_observed: bool| {
        let inner = &inner;
        let exporter = &exporter;
        async move {
          let mut resource_metrics =
            opentelemetry_sdk::metrics::data::ResourceMetrics::default();
          if collect_observed {
            let callbacks = {
              let mut callbacks = OTEL_PRE_COLLECT_CALLBACKS.lock().unwrap();
              std::mem::take(&mut *callbacks)
            };
            let mut futures = JoinSet::new();
            for callback in callbacks {
              let (tx, rx) = oneshot::channel();
              if let Ok(()) = callback.send(tx) {
                futures.spawn(rx);
              }
            }
            while futures.join_next().await.is_some() {}
          }
          inner.collect(&mut resource_metrics)?;
          if resource_metrics.scope_metrics().next().is_none() {
            return Ok(());
          }
          exporter.export(&resource_metrics).await?;
          Ok(())
        }
      };

      let mut ticker = tokio::time::interval(interval);
      ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
      ticker.tick().await;

      loop {
        let message = tokio::select! {
          _ = ticker.tick() => DenoPeriodicReaderMessage::Export,
          message = rx.recv() => if let Some(message) = message {
            message
          } else {
            break;
          },
        };

        match message {
          DenoPeriodicReaderMessage::Register(new_pipeline) => {
            inner.register_pipeline(new_pipeline);
          }
          DenoPeriodicReaderMessage::Export => {
            otel_debug!(
                name: "DenoPeriodicReader.ExportTriggered",
                message = "Export message received.",
            );
            if let Err(err) = collect_and_export(true).await {
              otel_error!(
                name: "DenoPeriodicReader.ExportFailed",
                message = "Failed to export metrics",
                reason = format!("{}", err));
            }
          }
          DenoPeriodicReaderMessage::ForceFlush(sender) => {
            otel_debug!(
                name: "DenoPeriodicReader.ForceFlushCalled",
                message = "Flush message received.",
            );
            let res = collect_and_export(false).await;
            if let Err(send_error) = sender.send(res) {
              otel_debug!(
                  name: "DenoPeriodicReader.Flush.SendResultError",
                  message = "Failed to send flush result.",
                  reason = format!("{:?}", send_error),
              );
            }
          }
          DenoPeriodicReaderMessage::Shutdown(sender) => {
            otel_debug!(
                name: "DenoPeriodicReader.ShutdownCalled",
                message = "Shutdown message received",
            );
            let res = collect_and_export(false).await;
            let _ = exporter.shutdown();
            if let Err(send_error) = sender.send(res) {
              otel_debug!(
                  name: "DenoPeriodicReader.Shutdown.SendResultError",
                  message = "Failed to send shutdown result",
                  reason = format!("{:?}", send_error),
              );
            }
            break;
          }
        }
      }
    };

    (*OTEL_SHARED_RUNTIME_SPAWN_TASK_TX)
      .unbounded_send(worker.boxed())
      .expect("failed to send task to shared OpenTelemetry runtime");

    DenoPeriodicReader { tx, temporality }
  }
}

mod hyper_client {
  use std::fmt::Debug;
  use std::pin::Pin;
  use std::task::Poll;
  use std::time::Duration;

  use deno_net::tunnel::TunnelConnection;
  use deno_net::tunnel::TunnelStream;
  use deno_net::tunnel::get_tunnel;
  use deno_tls::SocketUse;
  use deno_tls::TlsKey;
  use deno_tls::TlsKeys;
  use deno_tls::create_client_config;
  use deno_tls::load_certs;
  use deno_tls::load_private_keys;
  use http_body_util::BodyExt;
  use http_body_util::Full;
  use hyper::Uri;
  use hyper_rustls::HttpsConnector;
  use hyper_rustls::MaybeHttpsStream;
  use hyper_util::client::legacy::Client;
  use hyper_util::client::legacy::connect::Connected;
  use hyper_util::client::legacy::connect::HttpConnector;
  use hyper_util::rt::TokioIo;
  use opentelemetry_http::Bytes;
  use opentelemetry_http::HttpError;
  use opentelemetry_http::Request;
  use opentelemetry_http::Response;
  use opentelemetry_http::ResponseExt;
  use sys_traits::FsRead;
  use tokio::net::TcpStream;
  #[cfg(any(
    target_os = "android",
    target_os = "linux",
    target_os = "macos"
  ))]
  use tokio_vsock::VsockAddr;
  #[cfg(any(
    target_os = "android",
    target_os = "linux",
    target_os = "macos"
  ))]
  use tokio_vsock::VsockStream;

  use super::OtelSharedRuntime;

  #[derive(Debug, thiserror::Error)]
  enum Error {
    #[error(transparent)]
    StdIo(#[from] std::io::Error),
    #[error(transparent)]
    Box(#[from] Box<dyn std::error::Error + Send + Sync>),
    #[error(transparent)]
    Tunnel(#[from] deno_net::tunnel::Error),
  }

  #[derive(Debug, Clone)]
  enum Connector {
    Http(HttpsConnector<HttpConnector>),
    Tunnel(TunnelConnection),
    #[cfg(any(
      target_os = "android",
      target_os = "linux",
      target_os = "macos"
    ))]
    Vsock(VsockAddr),
  }

  #[allow(clippy::large_enum_variant, reason = "TODO: investigate")]
  #[pin_project::pin_project(project = IOProj)]
  enum IO {
    Tls(#[pin] TokioIo<MaybeHttpsStream<TokioIo<TcpStream>>>),
    Tunnel(#[pin] TunnelStream),
    #[cfg(any(
      target_os = "android",
      target_os = "linux",
      target_os = "macos"
    ))]
    Vsock(#[pin] VsockStream),
  }

  impl tokio::io::AsyncRead for IO {
    fn poll_read(
      self: std::pin::Pin<&mut Self>,
      cx: &mut std::task::Context<'_>,
      buf: &mut tokio::io::ReadBuf<'_>,
    ) -> Poll<std::io::Result<()>> {
      match self.project() {
        IOProj::Tls(stream) => stream.poll_read(cx, buf),
        IOProj::Tunnel(stream) => stream.poll_read(cx, buf),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IOProj::Vsock(stream) => stream.poll_read(cx, buf),
      }
    }
  }

  impl tokio::io::AsyncWrite for IO {
    fn poll_write(
      self: std::pin::Pin<&mut Self>,
      cx: &mut std::task::Context<'_>,
      buf: &[u8],
    ) -> Poll<Result<usize, std::io::Error>> {
      match self.project() {
        IOProj::Tls(stream) => stream.poll_write(cx, buf),
        IOProj::Tunnel(stream) => stream.poll_write(cx, buf),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IOProj::Vsock(stream) => stream.poll_write(cx, buf),
      }
    }

    fn poll_flush(
      self: std::pin::Pin<&mut Self>,
      cx: &mut std::task::Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
      match self.project() {
        IOProj::Tls(stream) => stream.poll_flush(cx),
        IOProj::Tunnel(stream) => stream.poll_flush(cx),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IOProj::Vsock(stream) => stream.poll_flush(cx),
      }
    }

    fn poll_shutdown(
      self: std::pin::Pin<&mut Self>,
      cx: &mut std::task::Context<'_>,
    ) -> Poll<Result<(), std::io::Error>> {
      match self.project() {
        IOProj::Tls(stream) => stream.poll_shutdown(cx),
        IOProj::Tunnel(stream) => stream.poll_shutdown(cx),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IOProj::Vsock(stream) => stream.poll_shutdown(cx),
      }
    }

    fn is_write_vectored(&self) -> bool {
      match self {
        IO::Tls(stream) => stream.is_write_vectored(),
        IO::Tunnel(stream) => stream.is_write_vectored(),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IO::Vsock(stream) => stream.is_write_vectored(),
      }
    }

    fn poll_write_vectored(
      self: std::pin::Pin<&mut Self>,
      cx: &mut std::task::Context<'_>,
      bufs: &[std::io::IoSlice<'_>],
    ) -> Poll<Result<usize, std::io::Error>> {
      match self.project() {
        IOProj::Tls(stream) => stream.poll_write_vectored(cx, bufs),
        IOProj::Tunnel(stream) => stream.poll_write_vectored(cx, bufs),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        IOProj::Vsock(stream) => stream.poll_write_vectored(cx, bufs),
      }
    }
  }

  impl hyper_util::client::legacy::connect::Connection for IO {
    fn connected(&self) -> Connected {
      match self {
        Self::Tls(stream) => stream.connected(),
        Self::Tunnel(_) => Connected::new().proxy(true),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        Self::Vsock(_) => Connected::new().proxy(true),
      }
    }
  }

  impl tower_service::Service<Uri> for Connector {
    type Response = TokioIo<IO>;
    type Error = Error;
    type Future = Pin<
      Box<
        dyn std::future::Future<Output = Result<Self::Response, Self::Error>>
          + Send,
      >,
    >;

    fn poll_ready(
      &mut self,
      cx: &mut std::task::Context<'_>,
    ) -> Poll<Result<(), Self::Error>> {
      match self {
        Self::Http(c) => c.poll_ready(cx).map_err(Into::into),
        Self::Tunnel(_) => Poll::Ready(Ok(())),
        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        Self::Vsock(_) => Poll::Ready(Ok(())),
      }
    }

    fn call(&mut self, dst: Uri) -> Self::Future {
      let this = self.clone();
      Box::pin(async move {
        match this {
          Self::Http(mut connector) => {
            let stream = connector.call(dst).await?;
            Ok(TokioIo::new(IO::Tls(TokioIo::new(stream))))
          }
          Self::Tunnel(listener) => {
            let stream = listener.create_agent_stream().await?;
            Ok(TokioIo::new(IO::Tunnel(stream)))
          }
          #[cfg(any(
            target_os = "android",
            target_os = "linux",
            target_os = "macos"
          ))]
          Self::Vsock(addr) => {
            let stream = VsockStream::connect(addr).await?;
            Ok(TokioIo::new(IO::Vsock(stream)))
          }
        }
      })
    }
  }

  const DEFAULT_OTEL_EXPORTER_OTLP_TIMEOUT: Duration = Duration::from_secs(10);

  fn parse_otlp_timeout() -> Duration {
    match std::env::var("OTEL_EXPORTER_OTLP_TIMEOUT") {
      Ok(val) => match val.parse::<u64>() {
        Ok(millis) if millis > 0 => Duration::from_millis(millis),
        _ => DEFAULT_OTEL_EXPORTER_OTLP_TIMEOUT,
      },
      Err(_) => DEFAULT_OTEL_EXPORTER_OTLP_TIMEOUT,
    }
  }

  #[derive(Debug, Clone)]
  pub struct HyperClient {
    inner: Client<Connector, Full<Bytes>>,
    timeout: Duration,
  }

  impl HyperClient {
    fn build_connector(
      sys: &impl FsRead,
    ) -> deno_core::anyhow::Result<Connector> {
      let connector = if let Some(tunnel) = get_tunnel() {
        Connector::Tunnel(tunnel.clone())
      } else if let Ok(addr) = std::env::var("OTEL_DENO_VSOCK") {
        #[cfg(not(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        )))]
        {
          let _ = addr;
          deno_core::anyhow::bail!("vsock is not supported on this platform")
        }

        #[cfg(any(
          target_os = "android",
          target_os = "linux",
          target_os = "macos"
        ))]
        {
          let Some((cid, port)) = addr.split_once(':') else {
            deno_core::anyhow::bail!("invalid vsock addr");
          };
          let cid = if cid == "-1" { u32::MAX } else { cid.parse()? };
          let port = port.parse()?;
          let addr = VsockAddr::new(cid, port);
          Connector::Vsock(addr)
        }
      } else {
        let ca_certs = match std::env::var("OTEL_EXPORTER_OTLP_CERTIFICATE") {
          Ok(path) => vec![sys.fs_read(path)?.into_owned()],
          _ => vec![],
        };

        let keys = match (
          std::env::var("OTEL_EXPORTER_OTLP_CLIENT_KEY"),
          std::env::var("OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE"),
        ) {
          (Ok(key_path), Ok(cert_path)) => {
            let key = sys.fs_read(key_path)?;
            let cert = sys.fs_read(cert_path)?;

            let certs = load_certs(&mut std::io::Cursor::new(cert))?;
            let key =
              load_private_keys(&key)?.into_iter().next().ok_or_else(|| {
                deno_core::anyhow::anyhow!(
                  "no private key found in OTEL_EXPORTER_OTLP_CLIENT_KEY file"
                )
              })?;

            TlsKeys::Static(TlsKey(certs, key))
          }
          _ => TlsKeys::Null,
        };

        let tls_config =
          create_client_config(deno_tls::TlsClientConfigOptions {
            root_cert_store: None,
            ca_certs,
            unsafely_ignore_certificate_errors: None,
            unsafely_disable_hostname_verification: false,
            cert_chain_and_key: keys,
            socket_use: SocketUse::Http,
          })?;
        let mut http_connector = HttpConnector::new();
        http_connector.enforce_http(false);
        let connector = HttpsConnector::from((http_connector, tls_config));
        Connector::Http(connector)
      };

      Ok(connector)
    }

    pub fn new(sys: &impl FsRead) -> deno_core::anyhow::Result<Self> {
      let connector = Self::build_connector(sys)?;
      Ok(Self::from_connector(connector, false))
    }

    /// Create a client configured for gRPC (HTTP/2 enforced).
    pub fn new_h2(sys: &impl FsRead) -> deno_core::anyhow::Result<Self> {
      let connector = Self::build_connector(sys)?;
      Ok(Self::from_connector(connector, true))
    }

    fn from_connector(connector: Connector, http2_only: bool) -> Self {
      let mut builder = Client::builder(OtelSharedRuntime);
      if http2_only {
        builder.http2_only(true);
      }
      Self {
        inner: builder.build(connector),
        timeout: parse_otlp_timeout(),
      }
    }
  }

  impl HyperClient {
    pub fn timeout(&self) -> Duration {
      self.timeout
    }

    /// Send a gRPC request, preserving HTTP/2 trailers for grpc-status.
    /// Returns (response_headers, trailers).
    pub async fn grpc_request(
      &self,
      request: Request<Vec<u8>>,
    ) -> Result<
      (hyper::http::response::Parts, Option<hyper::HeaderMap>),
      Box<dyn std::error::Error + Send + Sync>,
    > {
      let (parts, body) = request.into_parts();
      let request = Request::from_parts(parts, Full::from(body));
      let result = tokio::time::timeout(self.timeout, async {
        let response = self.inner.request(request).await?;
        let (parts, body) = response.into_parts();
        let collected = http_body_util::Limited::new(body, 1024 * 1024)
          .collect()
          .await?;
        let trailers = collected.trailers().cloned();
        Ok::<_, Box<dyn std::error::Error + Send + Sync>>((parts, trailers))
      })
      .await
      .map_err(|_| -> Box<dyn std::error::Error + Send + Sync> {
        Box::new(std::io::Error::new(
          std::io::ErrorKind::TimedOut,
          format!("OTEL export timed out after {}ms", self.timeout.as_millis()),
        ))
      })??;
      Ok(result)
    }
  }

  #[async_trait::async_trait]
  impl opentelemetry_http::HttpClient for HyperClient {
    async fn send_bytes(
      &self,
      request: Request<Bytes>,
    ) -> Result<Response<Bytes>, HttpError> {
      let (parts, body) = request.into_parts();
      let request = Request::from_parts(parts, Full::new(body));
      let response = tokio::time::timeout(self.timeout, async {
        let response = self.inner.request(request).await?;
        let (parts, body) = response.into_parts();
        let body = http_body_util::Limited::new(body, 1024 * 1024)
          .collect()
          .await?
          .to_bytes();
        Ok::<_, HttpError>(Response::from_parts(parts, body))
      })
      .await
      .map_err(|_| {
        std::io::Error::new(
          std::io::ErrorKind::TimedOut,
          format!("OTEL export timed out after {}ms", self.timeout.as_millis()),
        )
      })??;
      Ok(response.error_for_status()?)
    }
  }
}

#[derive(Debug)]
pub struct OtelGlobals {
  pub span_processor: BatchSpanProcessor<OtelSharedRuntime>,
  pub log_processor: BatchLogProcessor<OtelSharedRuntime>,
  pub id_generator: DenoIdGenerator,
  pub meter_provider: SdkMeterProvider,
  pub builtin_instrumentation_scope: InstrumentationScope,
  pub span_event_count_limit: usize,
  pub span_attribute_count_limit: usize,
  pub span_attribute_value_length_limit: Option<usize>,
  pub sampler: Sampler,
  pub config: OtelConfig,
}

impl OtelGlobals {
  pub fn has_tracing(&self) -> bool {
    self.config.tracing_enabled
  }

  pub fn has_metrics(&self) -> bool {
    self.config.metrics_enabled
  }
}

/// Default maximum number of events per span, per the OpenTelemetry SDK spec
/// for `OTEL_SPAN_EVENT_COUNT_LIMIT`.
const DEFAULT_SPAN_EVENT_COUNT_LIMIT: usize = 128;

/// Resolve the span event count limit from `OTEL_SPAN_EVENT_COUNT_LIMIT`,
/// falling back to the spec default of 128.
///
/// See <https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#span-limits>
fn span_event_count_limit_from_env(sys: &impl TelemetrySys) -> usize {
  sys
    .env_var("OTEL_SPAN_EVENT_COUNT_LIMIT")
    .ok()
    .and_then(|v| v.trim().parse::<usize>().ok())
    .unwrap_or(DEFAULT_SPAN_EVENT_COUNT_LIMIT)
}

/// The effective span event count limit from the initialized globals, falling
/// back to the spec default if telemetry is not yet initialized.
fn span_event_count_limit() -> usize {
  OTEL_GLOBALS
    .get()
    .map(|g| g.span_event_count_limit)
    .unwrap_or(DEFAULT_SPAN_EVENT_COUNT_LIMIT)
}

/// Default maximum number of attributes per span (and per span event / link),
/// as defined by the OpenTelemetry SDK spec for `OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT`
/// (and `OTEL_ATTRIBUTE_COUNT_LIMIT`).
const DEFAULT_ATTRIBUTE_COUNT_LIMIT: usize = 128;

/// Resolve the span attribute count limit from the environment, honoring
/// `OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT` and falling back to the general
/// `OTEL_ATTRIBUTE_COUNT_LIMIT`, then to the spec default of 128.
///
/// See <https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#attribute-limits>
fn span_attribute_count_limit_from_env(sys: &impl TelemetrySys) -> usize {
  [
    "OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT",
    "OTEL_ATTRIBUTE_COUNT_LIMIT",
  ]
  .into_iter()
  .find_map(|name| {
    sys
      .env_var(name)
      .ok()
      .and_then(|v| v.trim().parse::<usize>().ok())
  })
  .unwrap_or(DEFAULT_ATTRIBUTE_COUNT_LIMIT)
}

/// The effective span attribute count limit from the initialized globals,
/// falling back to the spec default if telemetry is not yet initialized.
fn attribute_count_limit() -> usize {
  OTEL_GLOBALS
    .get()
    .map(|g| g.span_attribute_count_limit)
    .unwrap_or(DEFAULT_ATTRIBUTE_COUNT_LIMIT)
}

/// Resolve the maximum attribute value length from the environment, honoring
/// `OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT` and falling back to the general
/// `OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT`. The default is no limit, and per the
/// spec non-positive values are invalid and treated as no limit.
///
/// See <https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#attribute-limits>
fn span_attribute_value_length_limit_from_env(
  sys: &impl TelemetrySys,
) -> Option<usize> {
  [
    "OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT",
    "OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT",
  ]
  .into_iter()
  .find_map(|name| sys.env_var(name).ok().map(|v| v.trim().to_string()))
  .and_then(|v| v.parse::<usize>().ok())
  .filter(|&limit| limit > 0)
}

/// The effective attribute value length limit from the initialized globals.
/// Returns `None` (no limit) when unset or telemetry is not yet initialized.
fn attribute_value_length_limit() -> Option<usize> {
  OTEL_GLOBALS
    .get()
    .and_then(|g| g.span_attribute_value_length_limit)
}

/// Truncate a single string attribute value to at most `limit` characters,
/// per `OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT`.
fn truncate_string_value(value: &mut StringValue, limit: usize) {
  if value.as_str().chars().count() > limit {
    let truncated: String = value.as_str().chars().take(limit).collect();
    *value = StringValue::from(truncated);
  }
}

/// Apply the configured attribute value length limit to a built attribute
/// value. String values and the elements of string-array values are truncated
/// to `limit` characters; other value types are unaffected. A `None` limit is
/// a no-op.
fn truncate_attr_value(value: &mut Value, limit: Option<usize>) {
  let Some(limit) = limit else {
    return;
  };
  match value {
    Value::String(s) => truncate_string_value(s, limit),
    Value::Array(Array::String(arr)) => {
      for s in arr.iter_mut() {
        truncate_string_value(s, limit);
      }
    }
    _ => {}
  }
}

/// Head-based trace sampler configured via the `OTEL_TRACES_SAMPLER` and
/// `OTEL_TRACES_SAMPLER_ARG` environment variables.
///
/// See <https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#general-sdk-configuration>
#[derive(Debug, Clone, Copy)]
pub struct Sampler {
  /// When set, the sampling decision of a valid parent span takes precedence
  /// over `root` (the `parentbased_*` variants).
  parent_based: bool,
  /// The sampler consulted when there is no parent (or `parent_based` is
  /// false).
  root: RootSampler,
}

#[derive(Debug, Clone, Copy)]
enum RootSampler {
  AlwaysOn,
  AlwaysOff,
  TraceIdRatio(f64),
}

impl Default for Sampler {
  fn default() -> Self {
    // When `OTEL_TRACES_SAMPLER` is unset, preserve Deno's historical behavior
    // of recording and sampling every span.
    Sampler {
      parent_based: false,
      root: RootSampler::AlwaysOn,
    }
  }
}

impl Sampler {
  fn from_env(
    sys: &impl TelemetrySys,
  ) -> Result<Self, deno_core::anyhow::Error> {
    let Ok(value) = sys.env_var("OTEL_TRACES_SAMPLER") else {
      return Ok(Self::default());
    };
    let value = value.trim();
    if value.is_empty() {
      return Ok(Self::default());
    }
    // `OTEL_TRACES_SAMPLER_ARG` is the sampling probability for the
    // `traceidratio` samplers, in the range [0, 1]. It defaults to 1.0 and is
    // ignored by the other samplers.
    let ratio = || -> f64 {
      sys
        .env_var("OTEL_TRACES_SAMPLER_ARG")
        .ok()
        .and_then(|s| s.trim().parse::<f64>().ok())
        .unwrap_or(1.0)
        .clamp(0.0, 1.0)
    };
    let sampler = match value {
      "always_on" => Sampler {
        parent_based: false,
        root: RootSampler::AlwaysOn,
      },
      "always_off" => Sampler {
        parent_based: false,
        root: RootSampler::AlwaysOff,
      },
      "traceidratio" => Sampler {
        parent_based: false,
        root: RootSampler::TraceIdRatio(ratio()),
      },
      "parentbased_always_on" => Sampler {
        parent_based: true,
        root: RootSampler::AlwaysOn,
      },
      "parentbased_always_off" => Sampler {
        parent_based: true,
        root: RootSampler::AlwaysOff,
      },
      "parentbased_traceidratio" => Sampler {
        parent_based: true,
        root: RootSampler::TraceIdRatio(ratio()),
      },
      other => {
        return Err(deno_core::anyhow::anyhow!(
          "Env var OTEL_TRACES_SAMPLER specifies an unsupported sampler: {}",
          other
        ));
      }
    };
    Ok(sampler)
  }

  /// Returns whether a span with the given `trace_id` should be sampled
  /// (recorded and exported), given its `parent` span context if any.
  fn should_sample(
    &self,
    parent: Option<&SpanContext>,
    trace_id: TraceId,
  ) -> bool {
    let parent_decision = parent.and_then(|parent| {
      (self.parent_based && parent.is_valid()).then(|| parent.is_sampled())
    });
    if let Some(sampled) = parent_decision {
      return sampled;
    }
    match self.root {
      RootSampler::AlwaysOn => true,
      RootSampler::AlwaysOff => false,
      RootSampler::TraceIdRatio(ratio) => {
        if ratio >= 1.0 {
          return true;
        }
        if ratio <= 0.0 {
          return false;
        }
        // Matches the opentelemetry-rust `TraceIdRatioBased` sampler: derive a
        // deterministic value in [0, 2^63) from the lower 64 bits of the trace
        // id and keep the span when it falls below the probability bound.
        let bytes = trace_id.to_bytes();
        let low = u64::from_be_bytes(bytes[8..16].try_into().unwrap());
        let value = low >> 1;
        let bound = (ratio * (1u64 << 63) as f64) as u64;
        value < bound
      }
    }
  }
}

pub static OTEL_GLOBALS: OnceCell<OtelGlobals> = OnceCell::new();

#[sys_traits::auto_impl]
pub trait TelemetrySys: EnvVar + FsRead {}

pub fn init(
  sys: &impl TelemetrySys,
  rt_config: OtelRuntimeConfig,
  config: OtelConfig,
) -> deno_core::anyhow::Result<()> {
  if !config.metrics_enabled
    && !config.tracing_enabled
    && config.console == OtelConsoleConfig::Ignore
  {
    return Ok(());
  }

  // Parse the `OTEL_EXPORTER_OTLP_PROTOCOL` variable. The opentelemetry_*
  // crates don't do this automatically.
  // `opentelemetry_otlp::Protocol::Grpc` is feature-gated behind the
  // (tonic-based) `grpc-tonic` feature, which we don't enable because we ship
  // our own gRPC framing. Track the gRPC choice with a separate flag so the
  // `protocol` value is only ever the HTTP variant passed to the OTLP HTTP
  // exporter builder.
  let protocol_var = sys.env_var("OTEL_EXPORTER_OTLP_PROTOCOL");
  let (use_console_exporter, use_grpc, protocol) = match protocol_var.as_deref()
  {
    Ok("console") => (true, false, Protocol::HttpBinary),
    Ok("http/protobuf") | Ok("") | Err(std::env::VarError::NotPresent) => {
      (false, false, Protocol::HttpBinary)
    }
    Ok("http/json") => (false, false, Protocol::HttpJson),
    Ok("grpc") => (false, true, Protocol::HttpBinary),
    Ok(protocol) => {
      return Err(deno_core::anyhow::anyhow!(
        "Env var OTEL_EXPORTER_OTLP_PROTOCOL specifies an unsupported protocol: {}",
        protocol
      ));
    }
    Err(err) => {
      return Err(deno_core::anyhow::anyhow!(
        "Failed to read env var OTEL_EXPORTER_OTLP_PROTOCOL: {}",
        err
      ));
    }
  };

  // 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).
  // The base resource picks up `service.name`, the `telemetry.sdk.*`
  // attributes and any `OTEL_RESOURCE_ATTRIBUTES`/`OTEL_SERVICE_NAME` values.
  let base_resource = Resource::builder().build();
  let sdk_language = base_resource
    .get(&Key::new(TELEMETRY_SDK_LANGUAGE))
    .unwrap();
  let sdk_name = base_resource.get(&Key::new(TELEMETRY_SDK_NAME)).unwrap();
  let sdk_version =
    base_resource.get(&Key::new(TELEMETRY_SDK_VERSION)).unwrap();

  // Add the runtime name and version to the resource attributes. Also override
  // the `telemetry.sdk` attributes to include the Deno runtime.
  let resource = Resource::builder()
    .with_attributes([
      KeyValue::new(PROCESS_RUNTIME_NAME, rt_config.runtime_name),
      KeyValue::new(PROCESS_RUNTIME_VERSION, rt_config.runtime_version.clone()),
      KeyValue::new(TELEMETRY_SDK_LANGUAGE, format!("deno-{sdk_language}")),
      KeyValue::new(TELEMETRY_SDK_NAME, format!("deno-{sdk_name}")),
      KeyValue::new(
        TELEMETRY_SDK_VERSION,
        format!("{}-{}", rt_config.runtime_version, sdk_version),
      ),
    ])
    .build();

  // 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 temporality_preference = sys
    .env_var("OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE")
    .ok()
    .map(|s| s.to_lowercase());
  let temporality = match temporality_preference.as_deref() {
    None | Some("cumulative") => Temporality::Cumulative,
    Some("delta") => Temporality::Delta,
    Some("lowmemory") => Temporality::LowMemory,
    Some(other) => {
      return Err(deno_core::anyhow::anyhow!(
        "Invalid value for OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE: {}",
        other
      ));
    }
  };

  let (span_processor, meter_provider, log_processor) = if use_console_exporter
  {
    let span_exporter = console_exporter::ConsoleSpanExporter::new();
    let mut span_processor =
      BatchSpanProcessor::builder(span_exporter, OtelSharedRuntime).build();
    span_processor.set_resource(&resource);

    let metric_exporter =
      console_exporter::ConsoleMetricExporter::new(temporality);
    let metric_reader = DenoPeriodicReader::new(sys, metric_exporter);
    let meter_provider = SdkMeterProvider::builder()
      .with_reader(metric_reader)
      .with_resource(resource.clone())
      .build();

    let log_exporter = console_exporter::ConsoleLogExporter::new();
    let mut log_processor =
      BatchLogProcessor::builder(log_exporter, OtelSharedRuntime).build();
    log_processor.set_resource(&resource);

    (span_processor, meter_provider, log_processor)
  } else if use_grpc {
    let client = hyper_client::HyperClient::new_h2(sys)?;

    let span_exporter = grpc_exporter::GrpcSpanExporter::new(client.clone());
    let mut span_processor =
      BatchSpanProcessor::builder(span_exporter, OtelSharedRuntime).build();
    span_processor.set_resource(&resource);

    let metric_exporter =
      grpc_exporter::GrpcMetricExporter::new(client.clone(), temporality);
    let metric_reader = DenoPeriodicReader::new(sys, metric_exporter);
    let meter_provider = SdkMeterProvider::builder()
      .with_reader(metric_reader)
      .with_resource(resource.clone())
      .build();

    let log_exporter = grpc_exporter::GrpcLogExporter::new(client);
    let mut log_processor =
      BatchLogProcessor::builder(log_exporter, OtelSharedRuntime).build();
    log_processor.set_resource(&resource);

    (span_processor, meter_provider, log_processor)
  } else {
    let client = hyper_client::HyperClient::new(sys)?;

    let span_exporter = HttpExporterBuilder::default()
      .with_http_client(client.clone())
      .with_protocol(protocol)
      .build_span_exporter()?;
    let mut span_processor =
      BatchSpanProcessor::builder(span_exporter, OtelSharedRuntime).build();
    span_processor.set_resource(&resource);

    let metric_exporter = HttpExporterBuilder::default()
      .with_http_client(client.clone())
      .with_protocol(protocol)
      .build_metrics_exporter(temporality)?;
    let metric_reader = DenoPeriodicReader::new(sys, metric_exporter);
    let meter_provider = SdkMeterProvider::builder()
      .with_reader(metric_reader)
      .with_resource(resource.clone())
      .build();

    let log_exporter = HttpExporterBuilder::default()
      .with_http_client(client)
      .with_protocol(protocol)
      .build_log_exporter()?;
    let mut log_processor =
      BatchLogProcessor::builder(log_exporter, OtelSharedRuntime).build();
    log_processor.set_resource(&resource);

    (span_processor, meter_provider, log_processor)
  };

  let builtin_instrumentation_scope =
    opentelemetry::InstrumentationScope::builder("deno")
      .with_version(rt_config.runtime_version.clone())
      .build();

  let id_generator = if let Some(prefix) = config.deterministic_prefix {
    DenoIdGenerator::deterministic(prefix)
  } else {
    DenoIdGenerator::random()
  };

  let span_event_count_limit = span_event_count_limit_from_env(sys);
  let span_attribute_count_limit = span_attribute_count_limit_from_env(sys);
  let span_attribute_value_length_limit =
    span_attribute_value_length_limit_from_env(sys);
  let sampler = Sampler::from_env(sys)?;

  OTEL_GLOBALS
    .set(OtelGlobals {
      log_processor,
      span_processor,
      id_generator,
      meter_provider,
      builtin_instrumentation_scope,
      span_event_count_limit,
      span_attribute_count_limit,
      span_attribute_value_length_limit,
      sampler,
      config,
    })
    .map_err(|_| deno_core::anyhow::anyhow!("failed to set otel globals"))?;

  deno_signals::before_exit(before_exit);
  deno_net::tunnel::disable_before_exit();

  Ok(())
}

fn before_exit() {
  log::trace!("deno_telemetry::before_exit");

  let Some(OtelGlobals {
    span_processor: spans,
    log_processor: logs,
    meter_provider,
    ..
  }) = OTEL_GLOBALS.get()
  else {
    return;
  };

  let r = spans.shutdown();
  log::trace!("spans={:?}", r);

  let r = logs.shutdown();
  log::trace!("logs={:?}", r);

  let r = meter_provider.shutdown();
  log::trace!("meters={:?}", r);

  deno_net::tunnel::before_exit();
}

pub fn handle_log(record: &log::Record) {
  use log::Level;

  let Some(OtelGlobals {
    log_processor: logs,
    builtin_instrumentation_scope,
    ..
  }) = OTEL_GLOBALS.get()
  else {
    return;
  };

  let mut log_record = new_log_record();

  let now = SystemTime::now();
  log_record.set_timestamp(now);
  log_record.set_observed_timestamp(now);
  log_record.set_severity_number(match record.level() {
    Level::Error => Severity::Error,
    Level::Warn => Severity::Warn,
    Level::Info => Severity::Info,
    Level::Debug => Severity::Debug,
    Level::Trace => Severity::Trace,
  });
  log_record.set_severity_text(record.level().as_str());
  log_record.set_body(record.args().to_string().into());
  log_record.set_target(record.metadata().target().to_string());

  struct Visitor<'s>(&'s mut LogRecord);

  impl<'kvs> log::kv::VisitSource<'kvs> for Visitor<'_> {
    fn visit_pair(
      &mut self,
      key: log::kv::Key<'kvs>,
      value: log::kv::Value<'kvs>,
    ) -> Result<(), log::kv::Error> {
      #[allow(clippy::manual_map, reason = "unhelpful clippy error")]
      let value = if let Some(v) = value.to_bool() {
        Some(AnyValue::Boolean(v))
      } else if let Some(v) = value.to_borrowed_str() {
        Some(AnyValue::String(v.to_owned().into()))
      } else if let Some(v) = value.to_f64() {
        Some(AnyValue::Double(v))
      } else if let Some(v) = value.to_i64() {
        Some(AnyValue::Int(v))
      } else {
        None
      };

      if let Some(value) = value {
        let key = Key::from(key.as_str().to_owned());
        self.0.add_attribute(key, value);
      }

      Ok(())
    }
  }

  let _ = record.key_values().visit(&mut Visitor(&mut log_record));

  logs.emit(&mut log_record, builtin_instrumentation_scope);
}

#[derive(Debug)]
pub enum DenoIdGenerator {
  Random(RandomIdGenerator),
  Deterministic {
    next_trace_id: AtomicU64,
    next_span_id: AtomicU64,
  },
}

impl IdGenerator for DenoIdGenerator {
  fn new_trace_id(&self) -> TraceId {
    match self {
      Self::Random(generator) => generator.new_trace_id(),
      Self::Deterministic { next_trace_id, .. } => {
        let id =
          next_trace_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let bytes = id.to_be_bytes();
        let bytes = [
          0, 0, 0, 0, 0, 0, 0, 0, bytes[0], bytes[1], bytes[2], bytes[3],
          bytes[4], bytes[5], bytes[6], bytes[7],
        ];
        TraceId::from_bytes(bytes)
      }
    }
  }

  fn new_span_id(&self) -> SpanId {
    match self {
      Self::Random(generator) => generator.new_span_id(),
      Self::Deterministic { next_span_id, .. } => {
        let id =
          next_span_id.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        SpanId::from_bytes(id.to_be_bytes())
      }
    }
  }
}

impl DenoIdGenerator {
  fn random() -> Self {
    Self::Random(RandomIdGenerator::default())
  }

  fn deterministic(prefix: u8) -> Self {
    let prefix = u64::from(prefix) << 56;
    Self::Deterministic {
      next_trace_id: AtomicU64::new(prefix + 1),
      next_span_id: AtomicU64::new(prefix + 1),
    }
  }
}

fn parse_trace_id(
  scope: &mut v8::PinScope<'_, '_>,
  trace_id: v8::Local<'_, v8::Value>,
) -> TraceId {
  if let Ok(string) = trace_id.try_cast() {
    let value_view = v8::ValueView::new(scope, string);
    match value_view.data() {
      v8::ValueViewData::OneByte(bytes) => {
        TraceId::from_hex(&String::from_utf8_lossy(bytes))
          .unwrap_or(TraceId::INVALID)
      }

      _ => TraceId::INVALID,
    }
  } else if let Ok(uint8array) = trace_id.try_cast::<v8::Uint8Array>() {
    let data = uint8array.data();
    let byte_length = uint8array.byte_length();
    if byte_length != 16 {
      return TraceId::INVALID;
    }
    // SAFETY: We have ensured that the byte length is 16, so it is safe to
    // cast the data to an array of 16 bytes.
    let bytes = unsafe { &*(data as *const u8 as *const [u8; 16]) };
    TraceId::from_bytes(*bytes)
  } else {
    TraceId::INVALID
  }
}

fn parse_span_id(
  scope: &mut v8::PinScope<'_, '_>,
  span_id: v8::Local<'_, v8::Value>,
) -> SpanId {
  if let Ok(string) = span_id.try_cast() {
    let value_view = v8::ValueView::new(scope, string);
    match value_view.data() {
      v8::ValueViewData::OneByte(bytes) => {
        SpanId::from_hex(&String::from_utf8_lossy(bytes))
          .unwrap_or(SpanId::INVALID)
      }
      _ => SpanId::INVALID,
    }
  } else if let Ok(uint8array) = span_id.try_cast::<v8::Uint8Array>() {
    let data = uint8array.data();
    let byte_length = uint8array.byte_length();
    if byte_length != 8 {
      return SpanId::INVALID;
    }
    // SAFETY: We have ensured that the byte length is 8, so it is safe to
    // cast the data to an array of 8 bytes.
    let bytes = unsafe { &*(data as *const u8 as *const [u8; 8]) };
    SpanId::from_bytes(*bytes)
  } else {
    SpanId::INVALID
  }
}

macro_rules! attr_raw {
  ($scope:ident, $name:expr, $value:expr) => {{
    let name = if let Ok(name) = $name.try_cast() {
      let view = v8::ValueView::new($scope, name);
      match view.data() {
        v8::ValueViewData::OneByte(bytes) => {
          Some(String::from_utf8_lossy(bytes).into_owned())
        }
        v8::ValueViewData::TwoByte(bytes) => {
          Some(String::from_utf16_lossy(bytes))
        }
      }
    } else {
      None
    };
    let value = 'value: {
      if let Ok(string) = $value.try_cast::<v8::String>() {
        break 'value Some(Value::String(StringValue::from({
          let x = v8::ValueView::new($scope, string);
          match x.data() {
            v8::ValueViewData::OneByte(bytes) => {
              String::from_utf8_lossy(bytes).into_owned()
            }
            v8::ValueViewData::TwoByte(bytes) => {
              String::from_utf16_lossy(bytes)
            }
          }
        })));
      }
      if let Ok(number) = $value.try_cast::<v8::Number>() {
        break 'value Some(Value::F64(number.value()));
      }
      if let Ok(boolean) = $value.try_cast::<v8::Boolean>() {
        break 'value Some(Value::Bool(boolean.is_true()));
      }
      if let Ok(bigint) = $value.try_cast::<v8::BigInt>() {
        let (i64_value, _lossless) = bigint.i64_value();
        break 'value Some(Value::I64(i64_value));
      }
      if let Ok(array) = $value.try_cast::<v8::Array>() {
        let len = array.length();
        if len == 0 {
          break 'value Some(Value::Array(Array::String(vec![])));
        }
        let Some(first) = array.get_index($scope, 0) else {
          return;
        };
        if first.is_string() {
          let mut vec = Vec::with_capacity(len as usize);
          for i in 0..len {
            let Some(element) = array.get_index($scope, i) else {
              return;
            };
            if let Ok(s) = element.try_cast::<v8::String>() {
              let view = v8::ValueView::new($scope, s);
              vec.push(StringValue::from(match view.data() {
                v8::ValueViewData::OneByte(bytes) => {
                  String::from_utf8_lossy(bytes).into_owned()
                }
                v8::ValueViewData::TwoByte(bytes) => {
                  String::from_utf16_lossy(bytes)
                }
              }));
            }
          }
          break 'value Some(Value::Array(Array::String(vec)));
        }
        if first.is_number() {
          let mut vec = Vec::with_capacity(len as usize);
          for i in 0..len {
            let Some(element) = array.get_index($scope, i) else {
              return;
            };
            if let Ok(n) = element.try_cast::<v8::Number>() {
              vec.push(n.value());
            }
          }
          break 'value Some(Value::Array(Array::F64(vec)));
        }
        if first.is_boolean() {
          let mut vec = Vec::with_capacity(len as usize);
          for i in 0..len {
            let Some(element) = array.get_index($scope, i) else {
              return;
            };
            if let Ok(b) = element.try_cast::<v8::Boolean>() {
              vec.push(b.is_true());
            }
          }
          break 'value Some(Value::Array(Array::Bool(vec)));
        }
        if first.is_big_int() {
          let mut vec = Vec::with_capacity(len as usize);
          for i in 0..len {
            let Some(element) = array.get_index($scope, i) else {
              return;
            };
            if let Ok(b) = element.try_cast::<v8::BigInt>() {
              let (i64_value, _lossless) = b.i64_value();
              vec.push(i64_value);
            }
          }
          break 'value Some(Value::Array(Array::I64(vec)));
        }
      }
      None
    };
    if let (Some(name), Some(value)) = (name, value) {
      Some(KeyValue::new(name, value))
    } else {
      None
    }
  }};
}

fn push_parsed_attr(
  attributes: &mut Vec<KeyValue>,
  dropped_attributes_count: &mut u32,
  limit: usize,
  value_length_limit: Option<usize>,
  attr: Option<KeyValue>,
) {
  // Enforce the configured per-element attribute count limit
  // (`OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT`): once the limit is reached, further
  // attributes are dropped and counted, matching the OpenTelemetry SDK spec.
  if attributes.len() >= limit {
    *dropped_attributes_count += 1;
  } else if let Some(mut kv) = attr {
    // Enforce `OTEL_(SPAN_)ATTRIBUTE_VALUE_LENGTH_LIMIT`: string values (and
    // string-array elements) longer than the limit are truncated.
    truncate_attr_value(&mut kv.value, value_length_limit);
    attributes.push(kv);
  } else {
    *dropped_attributes_count += 1;
  }
}

/// Convert the integer log level that ext/console uses to the corresponding
/// OpenTelemetry log severity.
fn severity_from_level(level: i32) -> Severity {
  match level {
    ..=0 => Severity::Debug,
    1 => Severity::Info,
    2 => Severity::Warn,
    3 | 5.. => Severity::Error,
    4 => Severity::Trace,
  }
}

#[op2(fast)]
fn op_otel_log<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  message: v8::Local<'s, v8::Value>,
  #[smi] level: i32,
  span: v8::Local<'s, v8::Value>,
  #[string] exception_type: String,
  #[string] exception_message: String,
  #[string] exception_stacktrace: String,
) {
  let Some(OtelGlobals {
    log_processor,
    builtin_instrumentation_scope,
    ..
  }) = OTEL_GLOBALS.get()
  else {
    return;
  };

  let severity = severity_from_level(level);

  let mut log_record = new_log_record();
  let now = SystemTime::now();
  log_record.set_timestamp(now);
  log_record.set_observed_timestamp(now);
  let Ok(message) = message.try_cast() else {
    return;
  };
  log_record.set_body(owned_string(scope, message).into());
  log_record.set_severity_number(severity);
  log_record.set_severity_text(severity.name());

  // console.warn (level 2) and console.error (level 3) write to stderr,
  // everything else writes to stdout.
  let iostream = if level == 2 || level == 3 {
    "stderr"
  } else {
    "stdout"
  };
  log_record.add_attribute("log.iostream", iostream);

  if let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  {
    let state = span.0.borrow();
    match &**state {
      OtelSpanState::Recording(span) => {
        log_record.set_trace_context(
          span.span_context.trace_id(),
          span.span_context.span_id(),
          Some(span.span_context.trace_flags()),
        );
      }
      OtelSpanState::Done(span_context) => {
        log_record.set_trace_context(
          span_context.trace_id(),
          span_context.span_id(),
          Some(span_context.trace_flags()),
        );
      }
    }
  }

  otel_log_add_exception_attributes(
    &mut log_record,
    exception_type,
    exception_message,
    exception_stacktrace,
  );

  log_processor.emit(&mut log_record, builtin_instrumentation_scope);
}

fn otel_log_add_exception_attributes(
  log_record: &mut LogRecord,
  exception_type: String,
  exception_message: String,
  exception_stacktrace: String,
) {
  if !exception_type.is_empty() {
    log_record.add_attribute(
      Key::from_static_str("exception.type"),
      AnyValue::String(exception_type.into()),
    );
  }
  if !exception_message.is_empty() {
    log_record.add_attribute(
      Key::from_static_str("exception.message"),
      AnyValue::String(exception_message.into()),
    );
  }
  if !exception_stacktrace.is_empty() {
    log_record.add_attribute(
      Key::from_static_str("exception.stacktrace"),
      AnyValue::String(exception_stacktrace.into()),
    );
  }
}

#[op2(fast)]
fn op_otel_log_foreign(
  scope: &mut v8::PinScope<'_, '_>,
  #[string] message: String,
  #[smi] level: i32,
  trace_id: v8::Local<'_, v8::Value>,
  span_id: v8::Local<'_, v8::Value>,
  #[smi] trace_flags: u8,
  #[string] exception_type: String,
  #[string] exception_message: String,
  #[string] exception_stacktrace: String,
) {
  let Some(OtelGlobals {
    log_processor,
    builtin_instrumentation_scope,
    ..
  }) = OTEL_GLOBALS.get()
  else {
    return;
  };

  let severity = severity_from_level(level);

  let trace_id = parse_trace_id(scope, trace_id);
  let span_id = parse_span_id(scope, span_id);

  let mut log_record = new_log_record();

  let now = SystemTime::now();
  log_record.set_timestamp(now);
  log_record.set_observed_timestamp(now);
  log_record.set_body(message.into());
  log_record.set_severity_number(severity);
  log_record.set_severity_text(severity.name());

  let iostream = if level == 2 || level == 3 {
    "stderr"
  } else {
    "stdout"
  };
  log_record.add_attribute("log.iostream", iostream);

  if trace_id != TraceId::INVALID && span_id != SpanId::INVALID {
    log_record.set_trace_context(
      trace_id,
      span_id,
      Some(TraceFlags::new(trace_flags)),
    );
  }

  otel_log_add_exception_attributes(
    &mut log_record,
    exception_type,
    exception_message,
    exception_stacktrace,
  );

  log_processor.emit(&mut log_record, builtin_instrumentation_scope);
}

pub fn report_event(name: &'static str, data: impl std::fmt::Display) {
  let Some(OtelGlobals {
    log_processor,
    builtin_instrumentation_scope,
    ..
  }) = OTEL_GLOBALS.get()
  else {
    return;
  };

  let mut log_record = new_log_record();

  log_record.set_observed_timestamp(SystemTime::now());
  log_record.set_event_name(name);
  log_record.set_severity_number(Severity::Trace);
  log_record.set_severity_text(Severity::Trace.name());
  log_record.set_body(format!("{data}").into());

  log_processor.emit(&mut log_record, builtin_instrumentation_scope);
}

fn owned_string<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  string: v8::Local<'s, v8::String>,
) -> String {
  let x = v8::ValueView::new(scope, string);
  match x.data() {
    v8::ValueViewData::OneByte(bytes) => {
      String::from_utf8_lossy(bytes).into_owned()
    }
    v8::ValueViewData::TwoByte(bytes) => String::from_utf16_lossy(bytes),
  }
}

struct OtelTracer(InstrumentationScope);

// SAFETY: we're sure this can be GCed
unsafe impl deno_core::GarbageCollected for OtelTracer {
  fn trace(&self, _visitor: &mut deno_core::v8::cppgc::Visitor) {}

  fn get_name(&self) -> &'static std::ffi::CStr {
    c"OtelTracer"
  }
}

#[op2]
impl OtelTracer {
  #[constructor]
  #[cppgc]
  fn new(
    #[string] name: String,
    #[string] version: Option<String>,
    #[string] schema_url: Option<String>,
  ) -> OtelTracer {
    let mut builder = opentelemetry::InstrumentationScope::builder(name);
    if let Some(version) = version {
      builder = builder.with_version(version);
    }
    if let Some(schema_url) = schema_url {
      builder = builder.with_schema_url(schema_url);
    }
    let scope = builder.build();
    OtelTracer(scope)
  }

  #[static_method]
  #[cppgc]
  fn builtin() -> Result<OtelTracer, JsErrorBox> {
    let OtelGlobals {
      builtin_instrumentation_scope,
      ..
    } = OTEL_GLOBALS
      .get()
      .ok_or_else(|| JsErrorBox::generic("otel not initialized"))?;
    Ok(OtelTracer(builtin_instrumentation_scope.clone()))
  }

  #[cppgc]
  fn start_span<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    #[cppgc] parent: Option<&OtelSpan>,
    name: v8::Local<'s, v8::Value>,
    #[smi] span_kind: u8,
    start_time: Option<f64>,
    #[smi] attribute_count: usize,
  ) -> Result<OtelSpan, JsErrorBox> {
    let OtelGlobals {
      id_generator,
      sampler,
      ..
    } = OTEL_GLOBALS
      .get()
      .ok_or_else(|| JsErrorBox::generic("otel not initialized"))?;
    let parent_span_id;
    let trace_id;
    let trace_state;
    let parent_span_context;
    match parent {
      Some(parent) => {
        let parent = parent.0.borrow();
        let ctx = match &**parent {
          OtelSpanState::Recording(span) => &span.span_context,
          OtelSpanState::Done(span_context) => span_context,
        };
        trace_id = ctx.trace_id();
        trace_state = ctx.trace_state().clone();
        parent_span_id = ctx.span_id();
        parent_span_context = Some(ctx.clone());
      }
      None => {
        trace_id = id_generator.new_trace_id();
        trace_state = TraceState::NONE;
        parent_span_id = SpanId::INVALID;
        parent_span_context = None;
      }
    }
    let sampled = sampler.should_sample(parent_span_context.as_ref(), trace_id);
    let span_context = SpanContext::new(
      trace_id,
      id_generator.new_span_id(),
      if sampled {
        TraceFlags::SAMPLED
      } else {
        TraceFlags::default()
      },
      false,
      trace_state,
    );
    if !sampled {
      // The span is not sampled: keep its context for propagation, but do not
      // record or export it.
      return Ok(OtelSpan(Rc::new(RefCell::new(Box::new(
        OtelSpanState::Done(span_context),
      )))));
    }
    let name = owned_string(
      scope,
      name
        .try_cast()
        .map_err(|e: DataError| JsErrorBox::generic(e.to_string()))?,
    );
    let span_kind = match span_kind {
      0 => SpanKind::Internal,
      1 => SpanKind::Server,
      2 => SpanKind::Client,
      3 => SpanKind::Producer,
      4 => SpanKind::Consumer,
      _ => return Err(JsErrorBox::generic("invalid span kind")),
    };
    let start_time = start_time
      .map(|start_time| {
        SystemTime::UNIX_EPOCH
          .checked_add(std::time::Duration::from_secs_f64(start_time / 1000.0))
          .ok_or_else(|| JsErrorBox::generic("invalid start time"))
      })
      .unwrap_or_else(|| Ok(SystemTime::now()))?;
    let span_data = SpanData {
      span_context,
      parent_span_id,
      parent_span_is_remote: false,
      span_kind,
      name: Cow::Owned(name),
      start_time,
      end_time: SystemTime::UNIX_EPOCH,
      attributes: Vec::with_capacity(attribute_count),
      dropped_attributes_count: 0,
      status: SpanStatus::Unset,
      events: SpanEvents::default(),
      links: SpanLinks::default(),
      instrumentation_scope: self.0.clone(),
    };
    Ok(OtelSpan(Rc::new(RefCell::new(Box::new(
      OtelSpanState::Recording(span_data),
    )))))
  }

  #[cppgc]
  fn start_span_foreign<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    parent_trace_id: v8::Local<'s, v8::Value>,
    parent_span_id: v8::Local<'s, v8::Value>,
    #[smi] parent_trace_flags: u8,
    name: v8::Local<'s, v8::Value>,
    #[smi] span_kind: u8,
    start_time: Option<f64>,
    #[smi] attribute_count: usize,
  ) -> Result<OtelSpan, JsErrorBox> {
    let parent_trace_id = parse_trace_id(scope, parent_trace_id);
    if parent_trace_id == TraceId::INVALID {
      return Err(JsErrorBox::generic("invalid trace id"));
    };
    let parent_span_id = parse_span_id(scope, parent_span_id);
    if parent_span_id == SpanId::INVALID {
      return Err(JsErrorBox::generic("invalid span id"));
    };
    let OtelGlobals {
      id_generator,
      sampler,
      ..
    } = OTEL_GLOBALS
      .get()
      .ok_or_else(|| JsErrorBox::generic("otel not initialized"))?;
    // Reconstruct the remote parent context so `parentbased_*` samplers honor
    // the upstream sampling decision carried in the propagated trace flags.
    let parent_context = SpanContext::new(
      parent_trace_id,
      parent_span_id,
      TraceFlags::new(parent_trace_flags),
      true,
      TraceState::NONE,
    );
    let sampled = sampler.should_sample(Some(&parent_context), parent_trace_id);
    let span_context = SpanContext::new(
      parent_trace_id,
      id_generator.new_span_id(),
      if sampled {
        TraceFlags::SAMPLED
      } else {
        TraceFlags::default()
      },
      false,
      TraceState::NONE,
    );
    if !sampled {
      return Ok(OtelSpan(Rc::new(RefCell::new(Box::new(
        OtelSpanState::Done(span_context),
      )))));
    }
    let name = owned_string(
      scope,
      name
        .try_cast()
        .map_err(|e: DataError| JsErrorBox::generic(e.to_string()))?,
    );
    let span_kind = match span_kind {
      0 => SpanKind::Internal,
      1 => SpanKind::Server,
      2 => SpanKind::Client,
      3 => SpanKind::Producer,
      4 => SpanKind::Consumer,
      _ => return Err(JsErrorBox::generic("invalid span kind")),
    };
    let start_time = start_time
      .map(|start_time| {
        SystemTime::UNIX_EPOCH
          .checked_add(std::time::Duration::from_secs_f64(start_time / 1000.0))
          .ok_or_else(|| JsErrorBox::generic("invalid start time"))
      })
      .unwrap_or_else(|| Ok(SystemTime::now()))?;
    let span_data = SpanData {
      span_context,
      parent_span_id,
      parent_span_is_remote: false,
      span_kind,
      name: Cow::Owned(name),
      start_time,
      end_time: SystemTime::UNIX_EPOCH,
      attributes: Vec::with_capacity(attribute_count),
      dropped_attributes_count: 0,
      status: SpanStatus::Unset,
      events: SpanEvents::default(),
      links: SpanLinks::default(),
      instrumentation_scope: self.0.clone(),
    };
    Ok(OtelSpan(Rc::new(RefCell::new(Box::new(
      OtelSpanState::Recording(span_data),
    )))))
  }
}

#[derive(ToV8)]
struct JsSpanContext {
  trace_id: Box<str>,
  span_id: Box<str>,
  trace_flags: u8,
}

#[derive(Debug, Error, JsError)]
#[error("OtelSpan cannot be constructed.")]
#[class(type)]
struct OtelSpanCannotBeConstructedError;

#[derive(Debug, Error, JsError)]
#[error("invalid span status code")]
#[class(type)]
struct InvalidSpanStatusCodeError;

// Rc-wrapped so the span can be shared between JS (via cppgc) and the HTTP
// record (for copying attributes to metrics). The inner Box is kept to keep
// the cppgc-traced struct small (see https://github.com/denoland/rusty_v8/issues/1676).
#[derive(Debug, Clone)]
pub struct OtelSpan(pub Rc<RefCell<Box<OtelSpanState>>>);

#[derive(Debug)]
#[allow(clippy::large_enum_variant, reason = "TODO: investigate")]
pub enum OtelSpanState {
  Recording(SpanData),
  Done(SpanContext),
}

// SAFETY: we're sure this can be GCed
unsafe impl deno_core::GarbageCollected for OtelSpan {
  fn trace(&self, _visitor: &mut deno_core::v8::cppgc::Visitor) {}

  fn get_name(&self) -> &'static std::ffi::CStr {
    c"OtelSpan"
  }
}

#[op2]
impl OtelSpan {
  #[constructor]
  #[cppgc]
  fn new() -> Result<OtelSpan, OtelSpanCannotBeConstructedError> {
    Err(OtelSpanCannotBeConstructedError)
  }

  fn span_context(&self) -> JsSpanContext {
    let state = self.0.borrow();
    let span_context = match &**state {
      OtelSpanState::Recording(span) => &span.span_context,
      OtelSpanState::Done(span_context) => span_context,
    };
    JsSpanContext {
      trace_id: format!("{:?}", span_context.trace_id()).into(),
      span_id: format!("{:?}", span_context.span_id()).into(),
      trace_flags: span_context.trace_flags().to_u8(),
    }
  }

  #[fast]
  fn set_status<'s>(
    &self,
    #[smi] status: u8,
    #[string] error_description: String,
  ) -> Result<(), InvalidSpanStatusCodeError> {
    let mut state = self.0.borrow_mut();
    let OtelSpanState::Recording(span) = &mut **state else {
      return Ok(());
    };
    span.status = match status {
      0 => SpanStatus::Unset,
      1 => SpanStatus::Ok,
      2 => SpanStatus::Error {
        description: Cow::Owned(error_description),
      },
      _ => return Err(InvalidSpanStatusCodeError),
    };
    Ok(())
  }

  #[fast]
  fn add_event(&self, #[string] name: String, start_time: f64) -> u32 {
    let start_time = if start_time.is_nan() {
      SystemTime::now()
    } else {
      SystemTime::UNIX_EPOCH
        .checked_add(Duration::from_secs_f64(start_time / 1000.0))
        .unwrap()
    };
    let limit = span_event_count_limit();
    let mut state = self.0.borrow_mut();
    let OtelSpanState::Recording(span) = &mut **state else {
      return 0;
    };
    // Enforce OTEL_SPAN_EVENT_COUNT_LIMIT: once the limit is reached, drop
    // further events and count them in droppedEventsCount.
    if span.events.events.len() >= limit {
      span.events.dropped_count += 1;
      return 0;
    }
    span
      .events
      .events
      .push(Event::new(name, start_time, vec![], 0));
    span.events.events.len() as u32
  }

  #[fast]
  fn drop_event(&self) {
    let mut state = self.0.borrow_mut();
    match &mut **state {
      OtelSpanState::Recording(span) => {
        span.events.dropped_count += 1;
      }
      OtelSpanState::Done(_) => {}
    }
  }

  #[fast]
  fn end(&self, end_time: f64) {
    let end_time = if end_time.is_nan() {
      SystemTime::now()
    } else {
      SystemTime::UNIX_EPOCH
        .checked_add(Duration::from_secs_f64(end_time / 1000.0))
        .unwrap()
    };

    let mut state = self.0.borrow_mut();
    if let OtelSpanState::Recording(span) = &mut **state {
      let span_context = span.span_context.clone();
      if let OtelSpanState::Recording(mut span) = *std::mem::replace(
        &mut *state,
        Box::new(OtelSpanState::Done(span_context)),
      ) {
        span.end_time = end_time;
        let Some(OtelGlobals { span_processor, .. }) = OTEL_GLOBALS.get()
        else {
          return;
        };
        span_processor.on_end(span);
      }
    }
  }
}

fn span_attributes(
  span: &mut SpanData,
  location: u32,
  target: u32,
) -> Option<(&mut Vec<KeyValue>, &mut u32)> {
  match location {
    // SELF
    0 => Some((&mut span.attributes, &mut span.dropped_attributes_count)),
    // EVENT
    1 => target
      .checked_sub(1)
      .and_then(|index| span.events.events.get_mut(index as usize))
      .map(|event| {
        (&mut event.attributes, &mut event.dropped_attributes_count)
      }),
    // LINK
    2 => target
      .checked_sub(1)
      .and_then(|index| span.links.links.get_mut(index as usize))
      .map(|link| (&mut link.attributes, &mut link.dropped_attributes_count)),
    _ => None,
  }
}

fn should_parse_span_attribute(
  span: &OtelSpan,
  location: u32,
  target: u32,
  limit: usize,
) -> bool {
  let mut state = span.0.borrow_mut();
  let OtelSpanState::Recording(span) = &mut **state else {
    return false;
  };
  let Some((attributes, dropped_attributes_count)) =
    span_attributes(span, location, target)
  else {
    return false;
  };
  if attributes.len() >= limit {
    *dropped_attributes_count += 1;
    false
  } else {
    true
  }
}

fn push_span_attribute(
  span: &OtelSpan,
  location: u32,
  target: u32,
  limit: usize,
  value_length_limit: Option<usize>,
  attr: Option<KeyValue>,
) {
  let mut state = span.0.borrow_mut();
  if let OtelSpanState::Recording(span) = &mut **state {
    let Some((attributes, dropped_attributes_count)) =
      span_attributes(span, location, target)
    else {
      return;
    };
    push_parsed_attr(
      attributes,
      dropped_attributes_count,
      limit,
      value_length_limit,
      attr,
    );
  }
}

#[op2(fast, reentrant)]
fn op_otel_span_attribute1<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  span: v8::Local<'_, v8::Value>,
  #[smi] location: u32,
  #[smi] target: u32,
  key: v8::Local<'s, v8::Value>,
  value: v8::Local<'s, v8::Value>,
) {
  let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  else {
    return;
  };
  let limit = attribute_count_limit();
  let value_length_limit = attribute_value_length_limit();
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key, value);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
}

#[op2(fast, reentrant)]
fn op_otel_span_attribute2<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  span: v8::Local<'_, v8::Value>,
  #[smi] location: u32,
  #[smi] target: u32,
  key1: v8::Local<'s, v8::Value>,
  value1: v8::Local<'s, v8::Value>,
  key2: v8::Local<'s, v8::Value>,
  value2: v8::Local<'s, v8::Value>,
) {
  let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  else {
    return;
  };
  let limit = attribute_count_limit();
  let value_length_limit = attribute_value_length_limit();
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key1, value1);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key2, value2);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_span_attribute3<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  span: v8::Local<'_, v8::Value>,
  #[smi] location: u32,
  #[smi] target: u32,
  key1: v8::Local<'s, v8::Value>,
  value1: v8::Local<'s, v8::Value>,
  key2: v8::Local<'s, v8::Value>,
  value2: v8::Local<'s, v8::Value>,
  key3: v8::Local<'s, v8::Value>,
  value3: v8::Local<'s, v8::Value>,
) {
  let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  else {
    return;
  };
  let limit = attribute_count_limit();
  let value_length_limit = attribute_value_length_limit();
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key1, value1);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key2, value2);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
  if should_parse_span_attribute(&span, location, target, limit) {
    let attr = attr_raw!(scope, key3, value3);
    push_span_attribute(
      &span,
      location,
      target,
      limit,
      value_length_limit,
      attr,
    );
  }
}

#[op2(fast)]
fn op_otel_span_update_name<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  span: v8::Local<'s, v8::Value>,
  name: v8::Local<'s, v8::Value>,
) {
  let Ok(name) = name.try_cast() else {
    return;
  };
  let name = owned_string(scope, name);
  let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  else {
    return;
  };
  let mut state = span.0.borrow_mut();
  if let OtelSpanState::Recording(span) = &mut **state {
    span.name = Cow::Owned(name)
  }
}

#[op2(fast)]
fn op_otel_span_add_link<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  span: v8::Local<'s, v8::Value>,
  trace_id: v8::Local<'s, v8::Value>,
  span_id: v8::Local<'s, v8::Value>,
  #[smi] trace_flags: u8,
  is_remote: bool,
  #[smi] dropped_attributes_count: u32,
) -> u32 {
  let trace_id = parse_trace_id(scope, trace_id);
  if trace_id == TraceId::INVALID {
    return 0;
  };
  let span_id = parse_span_id(scope, span_id);
  if span_id == SpanId::INVALID {
    return 0;
  };
  let span_context = SpanContext::new(
    trace_id,
    span_id,
    TraceFlags::new(trace_flags),
    is_remote,
    TraceState::NONE,
  );

  let Some(span) =
    deno_core::_ops::try_unwrap_cppgc_object::<OtelSpan>(scope, span)
  else {
    return 0;
  };
  let mut state = span.0.borrow_mut();
  if let OtelSpanState::Recording(span) = &mut **state {
    span.links.links.push(Link::new(
      span_context,
      vec![],
      dropped_attributes_count,
    ));
    span.links.links.len() as u32
  } else {
    0
  }
}

struct OtelMeter(opentelemetry::metrics::Meter);

// SAFETY: we're sure this can be GCed
unsafe impl deno_core::GarbageCollected for OtelMeter {
  fn trace(&self, _visitor: &mut deno_core::v8::cppgc::Visitor) {}

  fn get_name(&self) -> &'static std::ffi::CStr {
    c"OtelMeter"
  }
}

#[op2]
impl OtelMeter {
  #[constructor]
  #[cppgc]
  fn new(
    #[string] name: String,
    #[string] version: Option<String>,
    #[string] schema_url: Option<String>,
  ) -> Result<OtelMeter, JsErrorBox> {
    let mut builder = opentelemetry::InstrumentationScope::builder(name);
    if let Some(version) = version {
      builder = builder.with_version(version);
    }
    if let Some(schema_url) = schema_url {
      builder = builder.with_schema_url(schema_url);
    }
    let scope = builder.build();
    let meter = OTEL_GLOBALS
      .get()
      .ok_or_else(|| JsErrorBox::generic("otel not initialized"))?
      .meter_provider
      .meter_with_scope(scope);
    Ok(OtelMeter(meter))
  }

  #[cppgc]
  fn create_counter<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_instrument(
      |name| self.0.f64_counter(name),
      |i| Instrument::Counter(i.build()),
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }

  #[cppgc]
  fn create_up_down_counter<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_instrument(
      |name| self.0.f64_up_down_counter(name),
      |i| Instrument::UpDownCounter(i.build()),
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }

  #[cppgc]
  fn create_gauge<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_instrument(
      |name| self.0.f64_gauge(name),
      |i| Instrument::Gauge(i.build()),
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }

  #[cppgc]
  fn create_histogram<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
    #[scoped] boundaries: Option<Vec<f64>>,
  ) -> Result<Instrument, JsErrorBox> {
    let name = owned_string(
      scope,
      name
        .try_cast()
        .map_err(|e: DataError| JsErrorBox::generic(e.to_string()))?,
    );
    let mut builder = self.0.f64_histogram(name);
    if !description.is_null_or_undefined() {
      let description = owned_string(
        scope,
        description
          .try_cast()
          .map_err(|e: DataError| JsErrorBox::generic(e.to_string()))?,
      );
      builder = builder.with_description(description);
    };
    if !unit.is_null_or_undefined() {
      let unit = owned_string(
        scope,
        unit
          .try_cast()
          .map_err(|e: DataError| JsErrorBox::generic(e.to_string()))?,
      );
      builder = builder.with_unit(unit);
    };
    if let Some(boundaries) = boundaries {
      builder = builder.with_boundaries(boundaries);
    }

    Ok(Instrument::Histogram(builder.build()))
  }

  #[cppgc]
  fn create_observable_counter<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_async_instrument(
      |name| self.0.f64_observable_counter(name),
      |i| {
        i.build();
      },
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }

  #[cppgc]
  fn create_observable_up_down_counter<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_async_instrument(
      |name| self.0.f64_observable_up_down_counter(name),
      |i| {
        i.build();
      },
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }

  #[cppgc]
  fn create_observable_gauge<'s>(
    &self,
    scope: &mut v8::PinScope<'s, '_>,
    name: v8::Local<'s, v8::Value>,
    description: v8::Local<'s, v8::Value>,
    unit: v8::Local<'s, v8::Value>,
  ) -> Result<Instrument, JsErrorBox> {
    create_async_instrument(
      |name| self.0.f64_observable_gauge(name),
      |i| {
        i.build();
      },
      scope,
      name,
      description,
      unit,
    )
    .map_err(|e| JsErrorBox::generic(e.to_string()))
  }
}

enum Instrument {
  Counter(opentelemetry::metrics::Counter<f64>),
  UpDownCounter(UpDownCounter<f64>),
  Gauge(opentelemetry::metrics::Gauge<f64>),
  Histogram(Histogram<f64>),
  Observable(Arc<Mutex<HashMap<Vec<KeyValue>, f64>>>),
}

// SAFETY: we're sure this can be GCed
unsafe impl GarbageCollected for Instrument {
  fn trace(&self, _visitor: &mut deno_core::v8::cppgc::Visitor) {}

  fn get_name(&self) -> &'static std::ffi::CStr {
    c"Instrument"
  }
}

fn create_instrument<'a, 'b, T>(
  cb: impl FnOnce(String) -> InstrumentBuilder<'b, T>,
  cb2: impl FnOnce(InstrumentBuilder<'b, T>) -> Instrument,
  scope: &mut v8::PinScope<'a, '_>,
  name: v8::Local<'a, v8::Value>,
  description: v8::Local<'a, v8::Value>,
  unit: v8::Local<'a, v8::Value>,
) -> Result<Instrument, v8::DataError> {
  let name = owned_string(scope, name.try_cast()?);
  let mut builder = cb(name);
  if !description.is_null_or_undefined() {
    let description = owned_string(scope, description.try_cast()?);
    builder = builder.with_description(description);
  };
  if !unit.is_null_or_undefined() {
    let unit = owned_string(scope, unit.try_cast()?);
    builder = builder.with_unit(unit);
  };

  Ok(cb2(builder))
}

fn create_async_instrument<'a, 'b, T>(
  cb: impl FnOnce(String) -> AsyncInstrumentBuilder<'b, T, f64>,
  cb2: impl FnOnce(AsyncInstrumentBuilder<'b, T, f64>),
  scope: &mut v8::PinScope<'a, '_>,
  name: v8::Local<'a, v8::Value>,
  description: v8::Local<'a, v8::Value>,
  unit: v8::Local<'a, v8::Value>,
) -> Result<Instrument, DataError> {
  let name = owned_string(scope, name.try_cast()?);
  let mut builder = cb(name);
  if !description.is_null_or_undefined() {
    let description = owned_string(scope, description.try_cast()?);
    builder = builder.with_description(description);
  };
  if !unit.is_null_or_undefined() {
    let unit = owned_string(scope, unit.try_cast()?);
    builder = builder.with_unit(unit);
  };

  let data_share = Arc::new(Mutex::new(HashMap::new()));
  let data_share_: Arc<Mutex<HashMap<Vec<KeyValue>, f64>>> = data_share.clone();
  builder = builder.with_callback(move |i| {
    let data = {
      let mut data = data_share_.lock().unwrap();
      std::mem::take(&mut *data)
    };
    for (attributes, value) in data {
      i.observe(value, &attributes);
    }
  });
  cb2(builder);

  Ok(Instrument::Observable(data_share))
}

struct MetricAttributes {
  attributes: Vec<KeyValue>,
}

#[op2(fast)]
fn op_otel_metric_record0(
  state: &mut OpState,
  #[cppgc] instrument: &Instrument,
  value: f64,
) {
  let values = state.try_take::<MetricAttributes>();
  let attributes = match &values {
    Some(values) => &*values.attributes,
    None => &[],
  };
  match instrument {
    Instrument::Counter(counter) => counter.add(value, attributes),
    Instrument::UpDownCounter(counter) => counter.add(value, attributes),
    Instrument::Gauge(gauge) => gauge.record(value, attributes),
    Instrument::Histogram(histogram) => histogram.record(value, attributes),
    _ => {}
  }
}

#[op2(fast, reentrant)]
fn op_otel_metric_record1(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let mut values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let attributes = match &mut values {
    Some(values) => {
      if let Some(kv) = attr1 {
        values.attributes.reserve_exact(1);
        values.attributes.push(kv);
      }
      &*values.attributes
    }
    None => match attr1 {
      Some(kv1) => &[kv1] as &[KeyValue],
      None => &[],
    },
  };
  match &*instrument {
    Instrument::Counter(counter) => counter.add(value, attributes),
    Instrument::UpDownCounter(counter) => counter.add(value, attributes),
    Instrument::Gauge(gauge) => gauge.record(value, attributes),
    Instrument::Histogram(histogram) => histogram.record(value, attributes),
    _ => {}
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_metric_record2(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
  key2: v8::Local<'_, v8::Value>,
  value2: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let mut values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let attr2 = attr_raw!(scope, key2, value2);
  let attributes = match &mut values {
    Some(values) => {
      values.attributes.reserve_exact(2);
      if let Some(kv1) = attr1 {
        values.attributes.push(kv1);
      }
      if let Some(kv2) = attr2 {
        values.attributes.push(kv2);
      }
      &*values.attributes
    }
    None => match (attr1, attr2) {
      (Some(kv1), Some(kv2)) => &[kv1, kv2] as &[KeyValue],
      (Some(kv1), None) => &[kv1],
      (None, Some(kv2)) => &[kv2],
      (None, None) => &[],
    },
  };
  match &*instrument {
    Instrument::Counter(counter) => counter.add(value, attributes),
    Instrument::UpDownCounter(counter) => counter.add(value, attributes),
    Instrument::Gauge(gauge) => gauge.record(value, attributes),
    Instrument::Histogram(histogram) => histogram.record(value, attributes),
    _ => {}
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_metric_record3(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
  key2: v8::Local<'_, v8::Value>,
  value2: v8::Local<'_, v8::Value>,
  key3: v8::Local<'_, v8::Value>,
  value3: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let mut values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let attr2 = attr_raw!(scope, key2, value2);
  let attr3 = attr_raw!(scope, key3, value3);
  let attributes = match &mut values {
    Some(values) => {
      values.attributes.reserve_exact(3);
      if let Some(kv1) = attr1 {
        values.attributes.push(kv1);
      }
      if let Some(kv2) = attr2 {
        values.attributes.push(kv2);
      }
      if let Some(kv3) = attr3 {
        values.attributes.push(kv3);
      }
      &*values.attributes
    }
    None => match (attr1, attr2, attr3) {
      (Some(kv1), Some(kv2), Some(kv3)) => &[kv1, kv2, kv3] as &[KeyValue],
      (Some(kv1), Some(kv2), None) => &[kv1, kv2],
      (Some(kv1), None, Some(kv3)) => &[kv1, kv3],
      (None, Some(kv2), Some(kv3)) => &[kv2, kv3],
      (Some(kv1), None, None) => &[kv1],
      (None, Some(kv2), None) => &[kv2],
      (None, None, Some(kv3)) => &[kv3],
      (None, None, None) => &[],
    },
  };
  match &*instrument {
    Instrument::Counter(counter) => counter.add(value, attributes),
    Instrument::UpDownCounter(counter) => counter.add(value, attributes),
    Instrument::Gauge(gauge) => gauge.record(value, attributes),
    Instrument::Histogram(histogram) => histogram.record(value, attributes),
    _ => {}
  }
}

#[op2(fast)]
fn op_otel_metric_observable_record0(
  state: &mut OpState,
  #[cppgc] instrument: &Instrument,
  value: f64,
) {
  let values = state.try_take::<MetricAttributes>();
  let attributes = values.map(|attr| attr.attributes).unwrap_or_default();
  if let Instrument::Observable(data_share) = instrument {
    let mut data = data_share.lock().unwrap();
    data.insert(attributes, value);
  }
}

#[op2(fast, reentrant)]
fn op_otel_metric_observable_record1(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let mut attributes = values
    .map(|mut attr| {
      attr.attributes.reserve_exact(1);
      attr.attributes
    })
    .unwrap_or_else(|| Vec::with_capacity(1));
  if let Some(kv1) = attr1 {
    attributes.push(kv1);
  }
  if let Instrument::Observable(data_share) = &*instrument {
    let mut data = data_share.lock().unwrap();
    data.insert(attributes, value);
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_metric_observable_record2(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
  key2: v8::Local<'_, v8::Value>,
  value2: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let attr2 = attr_raw!(scope, key2, value2);
  let mut attributes = values
    .map(|mut attr| {
      attr.attributes.reserve_exact(2);
      attr.attributes
    })
    .unwrap_or_else(|| Vec::with_capacity(2));
  if let Some(kv1) = attr1 {
    attributes.push(kv1);
  }
  if let Some(kv2) = attr2 {
    attributes.push(kv2);
  }
  if let Instrument::Observable(data_share) = &*instrument {
    let mut data = data_share.lock().unwrap();
    data.insert(attributes, value);
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_metric_observable_record3(
  state: Rc<RefCell<OpState>>,
  scope: &mut v8::PinScope<'_, '_>,
  instrument: v8::Local<'_, v8::Value>,
  value: f64,
  key1: v8::Local<'_, v8::Value>,
  value1: v8::Local<'_, v8::Value>,
  key2: v8::Local<'_, v8::Value>,
  value2: v8::Local<'_, v8::Value>,
  key3: v8::Local<'_, v8::Value>,
  value3: v8::Local<'_, v8::Value>,
) {
  let Some(instrument) = deno_core::_ops::try_unwrap_cppgc_object::<Instrument>(
    &mut *scope,
    instrument,
  ) else {
    return;
  };
  let values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let attr1 = attr_raw!(scope, key1, value1);
  let attr2 = attr_raw!(scope, key2, value2);
  let attr3 = attr_raw!(scope, key3, value3);
  let mut attributes = values
    .map(|mut attr| {
      attr.attributes.reserve_exact(3);
      attr.attributes
    })
    .unwrap_or_else(|| Vec::with_capacity(3));
  if let Some(kv1) = attr1 {
    attributes.push(kv1);
  }
  if let Some(kv2) = attr2 {
    attributes.push(kv2);
  }
  if let Some(kv3) = attr3 {
    attributes.push(kv3);
  }
  if let Instrument::Observable(data_share) = &*instrument {
    let mut data = data_share.lock().unwrap();
    data.insert(attributes, value);
  }
}

#[allow(clippy::too_many_arguments, reason = "op")]
#[op2(fast, reentrant)]
fn op_otel_metric_attribute3<'s>(
  scope: &mut v8::PinScope<'s, '_>,
  state: Rc<RefCell<OpState>>,
  #[smi] capacity: u32,
  key1: v8::Local<'s, v8::Value>,
  value1: v8::Local<'s, v8::Value>,
  key2: v8::Local<'s, v8::Value>,
  value2: v8::Local<'s, v8::Value>,
  key3: v8::Local<'s, v8::Value>,
  value3: v8::Local<'s, v8::Value>,
) {
  let values = {
    let mut state = state.borrow_mut();
    state.try_take::<MetricAttributes>()
  };
  let mut attributes = values
    .map(|mut values| {
      values.attributes.reserve_exact(
        (capacity as usize).saturating_sub(values.attributes.capacity()),
      );
      values.attributes
    })
    .unwrap_or_else(|| Vec::with_capacity(capacity as usize));
  let attr1 = attr_raw!(scope, key1, value1);
  let attr2 = attr_raw!(scope, key2, value2);
  let attr3 = attr_raw!(scope, key3, value3);
  if let Some(kv1) = attr1 {
    attributes.push(kv1);
  }
  if let Some(kv2) = attr2 {
    attributes.push(kv2);
  }
  if let Some(kv3) = attr3 {
    attributes.push(kv3);
  }
  state.borrow_mut().put(MetricAttributes { attributes });
}

struct ObservationDone(oneshot::Sender<()>);

#[op2]
async fn op_otel_metric_wait_to_observe(state: Rc<RefCell<OpState>>) -> bool {
  let (tx, rx) = oneshot::channel();
  {
    OTEL_PRE_COLLECT_CALLBACKS
      .lock()
      .expect("mutex poisoned")
      .push(tx);
  }
  match rx.await {
    Ok(done) => {
      state.borrow_mut().put(ObservationDone(done));
      true
    }
    _ => false,
  }
}

#[op2(fast)]
fn op_otel_metric_observation_done(state: &mut OpState) {
  if let Some(ObservationDone(done)) = state.try_take::<ObservationDone>() {
    let _ = done.send(());
  }
}

struct GcMetricDataInner {
  start: Instant,
  duration: Histogram<f64>,
}

struct GcMetricData(RefCell<GcMetricDataInner>);

impl GcMetricData {
  extern "C" fn prologue_callback(
    isolate: v8::UnsafeRawIsolatePtr,
    _gc_type: v8::GCType,
    _flags: v8::GCCallbackFlags,
    _data: *mut c_void,
  ) {
    // SAFETY: Isolate is valid during callback
    let isolate =
      unsafe { v8::Isolate::from_raw_isolate_ptr_unchecked(isolate) };
    let Some(this) = isolate.get_slot::<Self>() else {
      return;
    };
    this.0.borrow_mut().start = Instant::now();
  }

  extern "C" fn epilogue_callback(
    isolate: v8::UnsafeRawIsolatePtr,
    gc_type: v8::GCType,
    _flags: v8::GCCallbackFlags,
    _data: *mut c_void,
  ) {
    // SAFETY: Isolate is valid during callback
    let isolate =
      unsafe { v8::Isolate::from_raw_isolate_ptr_unchecked(isolate) };
    let Some(this) = isolate.get_slot::<Self>() else {
      return;
    };
    let this = this.0.borrow_mut();

    let elapsed = this.start.elapsed();

    // https://opentelemetry.io/docs/specs/semconv/runtime/v8js-metrics/#metric-v8jsgcduration
    let gc_type = KeyValue::new(
      "v8js.gc.type",
      match gc_type {
        v8::GCType::kGCTypeScavenge => "minor",
        v8::GCType::kGCTypeMinorMarkSweep => "minor",
        v8::GCType::kGCTypeMarkSweepCompact => "major",
        v8::GCType::kGCTypeIncrementalMarking => "incremental",
        v8::GCType::kGCTypeProcessWeakCallbacks => "weakcb",
        _ => return,
      },
    );

    this.duration.record(elapsed.as_secs_f64(), &[gc_type]);
  }
}

#[derive(Clone)]
struct HeapMetricData {
  heap_limit: Gauge<u64>,
  heap_size: Gauge<u64>,
  available_size: Gauge<u64>,
  physical_size: Gauge<u64>,
}

#[op2(fast)]
fn op_otel_enable_isolate_metrics(scope: &mut v8::PinScope<'_, '_>) {
  if scope.get_slot::<GcMetricData>().is_some() {
    return;
  }

  let Some(globals) = OTEL_GLOBALS.get() else {
    return;
  };
  let meter = globals.meter_provider.meter("v8js");

  // https://opentelemetry.io/docs/specs/semconv/runtime/v8js-metrics/#metric-v8jsgcduration
  let duration = meter
    .f64_histogram("v8js.gc.duration")
    .with_unit("S")
    .with_description("Garbage collection duration")
    .with_boundaries(vec![0.01, 0.1, 1.0, 10.0])
    .build();

  scope.set_slot(GcMetricData(RefCell::new(GcMetricDataInner {
    start: Instant::now(),
    duration,
  })));

  scope.add_gc_prologue_callback(
    GcMetricData::prologue_callback,
    std::ptr::null_mut(),
    v8::GCType::kGCTypeAll,
  );

  scope.add_gc_epilogue_callback(
    GcMetricData::epilogue_callback,
    std::ptr::null_mut(),
    v8::GCType::kGCTypeAll,
  );

  let heap_limit = meter
    .u64_gauge("v8js.memory.heap.limit")
    .with_unit("By")
    .with_description("Total heap memory size pre-allocated.")
    .build();
  let heap_size = meter
    .u64_gauge("v8js.memory.heap.size")
    .with_unit("By")
    .with_description("Heap Memory size allocated.")
    .build();
  let available_size = meter
    .u64_gauge("v8js.memory.space.available_size")
    .with_unit("By")
    .with_description("Heap space available size.")
    .build();
  let physical_size = meter
    .u64_gauge("v8js.memory.space.physical_size")
    .with_unit("By")
    .with_description("Committed size of a heap space.")
    .build();

  scope.set_slot(HeapMetricData {
    heap_limit,
    heap_size,
    available_size,
    physical_size,
  });
}

#[op2(fast)]
fn op_otel_collect_isolate_metrics(scope: &mut v8::PinScope<'_, '_>) {
  let Some(data) = scope.get_slot::<HeapMetricData>() else {
    return;
  };
  let data = data.clone();
  for i in 0..scope.get_number_of_data_slots() {
    let Some(space) = scope.get_heap_space_statistics(i as _) else {
      continue;
    };
    // SAFETY: api has wrong lifetime, 'static is correct:
    // https://github.com/denoland/rusty_v8/pull/1744
    let space_name: &'static std::ffi::CStr =
      unsafe { std::ffi::CStr::from_ptr(space.space_name().as_ptr()) };
    let Ok(space_name) = space_name.to_str() else {
      continue;
    };
    let attributes = [KeyValue::new("v8js.heap.space.name", space_name)];
    data.heap_limit.record(space.space_size() as _, &attributes);
    data
      .heap_size
      .record(space.space_used_size() as _, &attributes);
    data
      .available_size
      .record(space.space_available_size() as _, &attributes);
    data
      .physical_size
      .record(space.physical_space_size() as _, &attributes);
  }
}