telemetry_subscribers/

lib.rs

// Copyright (c) Mysten Labs, Inc.
// SPDX-License-Identifier: Apache-2.0

use atomic_float::AtomicF64;
use crossterm::tty::IsTty;
use once_cell::sync::Lazy;
use opentelemetry::{
    Context, KeyValue,
    trace::{Link, SamplingResult, SpanKind, TraceId, TracerProvider as _},
};
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_sdk::trace::Sampler;
use opentelemetry_sdk::{
    self, Resource, runtime,
    trace::{BatchSpanProcessor, ShouldSample, TracerProvider},
};
use span_latency_prom::PrometheusSpanLatencyLayer;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::Duration;
use std::{
    env,
    io::{Write, stderr},
    str::FromStr,
    sync::{Arc, Mutex, atomic::Ordering},
};
use tracing::dispatcher::DefaultGuard;
use tracing::metadata::LevelFilter;
use tracing::{Level, error, info};
use tracing_appender::non_blocking::{NonBlocking, WorkerGuard};
use tracing_error::ErrorLayer;
use tracing_subscriber::{
    EnvFilter, Layer, Registry,
    filter::{self, FilterExt},
    fmt::{self, format::Pretty},
    layer::SubscriberExt,
    reload,
};

use crate::file_exporter::{CachedOpenFile, FileExporter};
use crate::test_layer::TestLayer;

mod file_exporter;
pub mod span_latency_prom;
mod test_layer;

/// Global filter layer that rejects callsites above the configured levels at registration time.
///
/// Without this, per-layer filtering causes the Registry to return `Interest::always()` for
/// every callsite. That means trace-level `#[instrument]` spans are allocated in the slab,
/// have their per-layer filters walked, and are immediately discarded — all at significant cost.
/// The same applies to events below the env filter threshold.
///
/// By rejecting high-verbosity callsites globally, `Span::new` short-circuits to
/// `Span::none()` and event macros short-circuit before formatting arguments.
struct GlobalLevelFilter {
    max_span_level: LevelFilter,
    /// The maximum event level any layer will accept. Loaded from the EnvFilter's
    /// max_level_hint and updated via a shared atomic when the filter is reloaded.
    max_event_level: Arc<std::sync::atomic::AtomicU8>,
}

impl GlobalLevelFilter {
    fn load_max_event_level(&self) -> LevelFilter {
        level_filter_from_u8(self.max_event_level.load(Ordering::Relaxed))
    }

    fn exceeds_max_level(&self, metadata: &tracing::Metadata<'_>) -> bool {
        if metadata.is_span() {
            LevelFilter::from_level(*metadata.level()) > self.max_span_level
        } else {
            let max = self.load_max_event_level();
            !max.eq(&LevelFilter::OFF) && LevelFilter::from_level(*metadata.level()) > max
        }
    }
}

fn level_filter_to_u8(lf: LevelFilter) -> u8 {
    match lf {
        LevelFilter::OFF => 0,
        LevelFilter::ERROR => 1,
        LevelFilter::WARN => 2,
        LevelFilter::INFO => 3,
        LevelFilter::DEBUG => 4,
        LevelFilter::TRACE => 5,
    }
}

fn level_filter_from_u8(v: u8) -> LevelFilter {
    match v {
        0 => LevelFilter::OFF,
        1 => LevelFilter::ERROR,
        2 => LevelFilter::WARN,
        3 => LevelFilter::INFO,
        4 => LevelFilter::DEBUG,
        _ => LevelFilter::TRACE,
    }
}

impl<S: tracing::Subscriber> Layer<S> for GlobalLevelFilter {
    fn register_callsite(
        &self,
        metadata: &'static tracing::Metadata<'static>,
    ) -> tracing::subscriber::Interest {
        if self.exceeds_max_level(metadata) {
            tracing::subscriber::Interest::never()
        } else {
            // Return always() for passing callsites. The final interest will be
            // determined by the Registry (which returns sometimes() when per-layer
            // filters are present). We intentionally do NOT override enabled() —
            // doing so would add an extra check to every enabled() walk, which the
            // per-layer filters already handle. For dynamic env filter reloads,
            // rebuild_interest_cache() re-calls register_callsite with our updated
            // atomic max_event_level.
            tracing::subscriber::Interest::always()
        }
    }

    fn max_level_hint(&self) -> Option<LevelFilter> {
        let event = self.load_max_event_level();
        Some(std::cmp::max(self.max_span_level, event))
    }
}

/// Alias for a type-erased error type.
pub type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;

/// Output format for a per-target log route (see [`TelemetryConfig::log_tails`]). Inferred
/// from the route's file extension.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum LogFormat {
    Json,
    #[default]
    Text,
}

impl LogFormat {
    /// Infer the output format from a log file's extension: the common newline-delimited
    /// JSON extensions `jsonl` and `ndjson` (case-insensitive) select [`LogFormat::Json`];
    /// every other extension (including `.json` and `.log`) and none default to
    /// [`LogFormat::Text`].
    fn from_path(path: &Path) -> Self {
        if let Some(ext) = path.extension().and_then(|ext| ext.to_str())
            && (ext.eq_ignore_ascii_case("jsonl") || ext.eq_ignore_ascii_case("ndjson"))
        {
            LogFormat::Json
        } else {
            LogFormat::Text
        }
    }
}

/// A per-target output route: *additionally* write `target`'s logs to `file`. Configured via
/// the `RUST_LOG_TAILS` env var, e.g. `graphql_request=/var/log/sui/graphql_request.jsonl`.
#[derive(Clone, Debug)]
pub struct LogTail {
    pub target: String,
    pub format: LogFormat,
    pub file: PathBuf,
}

/// Configuration for different logging/tracing options
/// ===
/// - json_log_output: Output JSON logs to stdout only.
/// - log_tails: Additional per-target log files via RUST_LOG_TAILS.
/// - log_file: If defined, write output to a file starting with this name, ex app.log
/// - log_level: error/warn/info/debug/trace, defaults to info
#[derive(Default, Clone, Debug)]
pub struct TelemetryConfig {
    pub enable_otlp_tracing: bool,
    /// Enables Tokio Console debugging on port 6669
    pub tokio_console: bool,
    /// Output JSON logs.
    pub json_log_output: bool,
    /// If defined, write output to a file starting with this name, ex app.log
    pub log_file: Option<String>,
    /// Log level to set, defaults to info
    pub log_string: Option<String>,
    /// Span level - what level of spans should be created.  Note this is not same as logging level.
    /// If set to None, then defaults to INFO. Use LevelFilter::OFF to disable all spans.
    pub span_level: Option<LevelFilter>,
    /// Set a panic hook
    pub panic_hook: bool,
    /// Crash on panic
    pub crash_on_panic: bool,
    /// Optional Prometheus registry - if present, all enabled span latencies are measured
    pub prom_registry: Option<prometheus::Registry>,
    /// Disable the PrometheusSpanLatencyLayer even when a prom_registry is set.
    pub disable_span_latency: bool,
    pub sample_rate: f64,
    /// Add directive to include trace logs with provided target
    pub trace_target: Option<Vec<String>>,
    /// Print unadorned logs from the target on standard error if this is a tty
    pub user_info_target: Vec<String>,
    /// Per-target output routing — one [`LogTail`] per entry — populated from the
    /// `RUST_LOG_TAILS` env var. See [`TelemetryConfig::with_log_tail`] for how each tail
    /// is emitted.
    pub log_tails: Vec<LogTail>,
    // Sets the subscriber created by this config to be the global default. Defaults to true if not set.
    pub set_global_default: bool,
    // Add error layer to capture trace spans. Defaults to false if not set.
    pub enable_error_layer: bool,
    // Capture output for tests. Defaults to false if not set.
    pub enable_test_layer: bool,
}

#[must_use]
#[allow(dead_code)]
pub struct TelemetryGuards {
    worker_guard: WorkerGuard,
    /// Flush guards for any per-target log files (`RUST_LOG_TAILS`). Held so the
    /// background writer threads keep flushing until the guards are dropped.
    file_guards: Vec<WorkerGuard>,
    provider: Option<TracerProvider>,
    subscriber: Option<DefaultGuard>,
}

impl TelemetryGuards {
    fn new(
        config: TelemetryConfig,
        worker_guard: WorkerGuard,
        file_guards: Vec<WorkerGuard>,
        provider: Option<TracerProvider>,
        subscriber: Option<DefaultGuard>,
    ) -> Self {
        // Do not set the global config if subscriber is present because a global subscriber was not set.
        if subscriber.is_none() {
            set_global_telemetry_config(config);
        }
        Self {
            worker_guard,
            file_guards,
            provider,
            subscriber,
        }
    }
}

impl Drop for TelemetryGuards {
    fn drop(&mut self) {
        clear_global_telemetry_config();
    }
}

#[derive(Clone, Debug)]
pub struct FilterHandle {
    reload: reload::Handle<EnvFilter, Registry>,
    /// Shared with GlobalLevelFilter so that reloading the env filter also
    /// updates the global event-level gate.
    max_event_level: Option<Arc<std::sync::atomic::AtomicU8>>,
}

impl FilterHandle {
    pub fn update<S: AsRef<str>>(&self, directives: S) -> Result<(), BoxError> {
        let filter = EnvFilter::try_new(directives)?;
        if let Some(ref max_level) = self.max_event_level {
            let hint = filter.max_level_hint().unwrap_or(LevelFilter::TRACE);
            max_level.store(level_filter_to_u8(hint), Ordering::Relaxed);
        }
        self.reload.reload(filter)?;
        Ok(())
    }

    pub fn get(&self) -> Result<String, BoxError> {
        self.reload
            .with_current(|filter| filter.to_string())
            .map_err(Into::into)
    }
}

pub struct TracingHandle {
    log: FilterHandle,
    trace: Option<FilterHandle>,
    file_output: CachedOpenFile,
    test_layer: Option<TestLayer>,
    sampler: SamplingFilter,
}

impl TracingHandle {
    pub fn update_log<S: AsRef<str>>(&self, directives: S) -> Result<(), BoxError> {
        self.log.update(directives)
    }

    pub fn get_log(&self) -> Result<String, BoxError> {
        self.log.get()
    }

    pub fn update_sampling_rate(&self, sample_rate: f64) {
        self.sampler.update_sampling_rate(sample_rate);
    }

    pub fn update_trace_file<S: AsRef<str>>(&self, trace_file: S) -> Result<(), BoxError> {
        let trace_path = PathBuf::from_str(trace_file.as_ref())?;
        self.file_output.update_path(trace_path)?;
        Ok(())
    }

    pub fn update_trace_filter<S: AsRef<str>>(
        &self,
        directives: S,
        duration: Duration,
    ) -> Result<(), BoxError> {
        if let Some(trace) = &self.trace {
            let res = trace.update(directives);
            // after duration is elapsed, reset to the env setting
            let trace = trace.clone();
            let trace_filter_env = env::var("TRACE_FILTER").unwrap_or_else(|_| "off".to_string());
            tokio::spawn(async move {
                tokio::time::sleep(duration).await;
                if let Err(e) = trace.update(trace_filter_env) {
                    error!("failed to reset trace filter: {}", e);
                }
            });
            res
        } else {
            info!("tracing not enabled, ignoring update");
            Ok(())
        }
    }

    pub fn clear_file_output(&self) {
        self.file_output.clear_path();
    }

    pub fn reset_trace(&self) {
        if let Some(trace) = &self.trace {
            let trace_filter_env = env::var("TRACE_FILTER").unwrap_or_else(|_| "off".to_string());
            if let Err(e) = trace.update(trace_filter_env) {
                error!("failed to reset trace filter: {}", e);
            }
        }
    }

    pub fn get_test_layer_events(&self) -> Vec<String> {
        self.test_layer
            .as_ref()
            .expect("test layer not enabled")
            .get_events()
    }
}

fn get_output(log_file: Option<String>) -> (NonBlocking, WorkerGuard) {
    if let Some(logfile_prefix) = log_file {
        let file_appender = tracing_appender::rolling::daily("", logfile_prefix);
        tracing_appender::non_blocking(file_appender)
    } else {
        tracing_appender::non_blocking(stderr())
    }
}

/// Open a per-target log file (`RUST_LOG_TAILS`), pre-creating its parent directory.
/// Panics `init()` if the directory cannot be created, rather than silently dropping the
/// target's file sink.
fn open_tail_file(path: &Path) -> (NonBlocking, WorkerGuard) {
    if let Some(parent) = path.parent()
        && !parent.as_os_str().is_empty()
    {
        std::fs::create_dir_all(parent).unwrap_or_else(|e| {
            panic!(
                "telemetry: cannot create directory for RUST_LOG_TAILS file {}: {e}",
                path.display()
            )
        });
    }
    let file_appender = tracing_appender::rolling::daily("", path);
    tracing_appender::non_blocking(file_appender)
}

// NOTE: this function is copied from tracing's panic_hook example
fn set_panic_hook(crash_on_panic: bool) {
    let default_panic_handler = std::panic::take_hook();

    // Set a panic hook that records the panic as a `tracing` event at the
    // `ERROR` verbosity level.
    //
    // If we are currently in a span when the panic occurred, the logged event
    // will include the current span, allowing the context in which the panic
    // occurred to be recorded.
    std::panic::set_hook(Box::new(move |panic| {
        // If the panic has a source location, record it as structured fields.
        if let Some(location) = panic.location() {
            // On nightly Rust, where the `PanicInfo` type also exposes a
            // `message()` method returning just the message, we could record
            // just the message instead of the entire `fmt::Display`
            // implementation, avoiding the duplicated location
            tracing::error!(
                message = %panic,
                panic.file = location.file(),
                panic.line = location.line(),
                panic.column = location.column(),
            );
        } else {
            tracing::error!(message = %panic);
        }

        default_panic_handler(panic);

        // We're panicking so we can't do anything about the flush failing
        let _ = std::io::stderr().flush();
        let _ = std::io::stdout().flush();

        if crash_on_panic {
            // Kill the process
            std::process::exit(12);
        }
    }));
}

static GLOBAL_CONFIG: Lazy<Arc<Mutex<Option<TelemetryConfig>>>> =
    Lazy::new(|| Arc::new(Mutex::new(None)));

fn set_global_telemetry_config(config: TelemetryConfig) {
    let mut global_config = GLOBAL_CONFIG.lock().unwrap();
    assert!(global_config.is_none());
    *global_config = Some(config);
}

fn clear_global_telemetry_config() {
    let mut global_config = GLOBAL_CONFIG.lock().unwrap();
    *global_config = None;
}

pub fn get_global_telemetry_config() -> Option<TelemetryConfig> {
    let global_config = GLOBAL_CONFIG.lock().unwrap();
    global_config.clone()
}

fn parse_log_tails(spec: &str) -> Vec<LogTail> {
    spec.split(',')
        .map(str::trim)
        .filter(|entry| !entry.is_empty())
        .map(parse_log_tail_entry)
        .collect()
}

fn parse_log_tail_entry(entry: &str) -> LogTail {
    if let Some((target, file)) = entry.split_once('=') {
        let (target, file) = (target.trim(), file.trim());
        if !target.is_empty() && !file.is_empty() {
            let file = PathBuf::from(file);
            let format = LogFormat::from_path(&file);
            return LogTail {
                target: target.to_owned(),
                format,
                file,
            };
        }
    }
    panic!(
        "telemetry: invalid RUST_LOG_TAILS entry {entry:?}: a `=file` path is \
         required, e.g. `graphql_request=/var/log/sui/graphql_request.jsonl`"
    )
}

impl TelemetryConfig {
    pub fn new() -> Self {
        Self {
            enable_otlp_tracing: false,
            tokio_console: false,
            json_log_output: false,
            log_file: None,
            log_string: None,
            span_level: None,
            panic_hook: true,
            crash_on_panic: false,
            prom_registry: None,
            disable_span_latency: false,
            sample_rate: 1.0,
            trace_target: None,
            user_info_target: Vec::new(),
            log_tails: Vec::new(),
            set_global_default: true,
            enable_error_layer: false,
            enable_test_layer: false,
        }
    }

    pub fn with_json(mut self) -> Self {
        self.json_log_output = true;
        self
    }

    pub fn with_log_level(mut self, log_string: &str) -> Self {
        self.log_string = Some(log_string.to_owned());
        self
    }

    pub fn with_span_level(mut self, span_level: Level) -> Self {
        self.span_level = Some(LevelFilter::from_level(span_level));
        self
    }

    pub fn with_log_file(mut self, filename: &str) -> Self {
        self.log_file = Some(filename.to_owned());
        self
    }

    pub fn with_prom_registry(mut self, registry: &prometheus::Registry) -> Self {
        self.prom_registry = Some(registry.clone());
        self
    }

    pub fn with_disable_span_latency(mut self, disable: bool) -> Self {
        self.disable_span_latency = disable;
        self
    }

    pub fn with_sample_rate(mut self, rate: f64) -> Self {
        self.sample_rate = rate;
        self
    }

    pub fn with_trace_target(mut self, target: &str) -> Self {
        match self.trace_target {
            Some(ref mut v) => v.push(target.to_owned()),
            None => self.trace_target = Some(vec![target.to_owned()]),
        };

        self
    }

    /// Adds `target` to the list of modules that will have their info & above logs printed
    /// unadorned to standard error (for non-json output)
    pub fn with_user_info_target(mut self, target: &str) -> Self {
        self.user_info_target.push(target.to_owned());
        self
    }

    /// Adds a `RUST_LOG_TAILS`-style routing rule: *additionally* write `target`'s logs to
    /// `file`, in a format inferred from its extension.
    pub fn with_log_tail(mut self, target: &str, file: impl Into<PathBuf>) -> Self {
        let file = file.into();
        let format = LogFormat::from_path(&file);
        self.log_tails.push(LogTail {
            target: target.to_owned(),
            format,
            file,
        });
        self
    }

    pub fn with_set_global_default(mut self, set_global_default: bool) -> Self {
        self.set_global_default = set_global_default;
        self
    }

    pub fn with_enable_error_layer(mut self, enable_error_layer: bool) -> Self {
        self.enable_error_layer = enable_error_layer;
        self
    }

    pub fn with_enable_test_layer(mut self, enable_test_layer: bool) -> Self {
        self.enable_test_layer = enable_test_layer;
        self
    }

    pub fn with_env(mut self) -> Self {
        if env::var("CRASH_ON_PANIC").is_ok() {
            self.crash_on_panic = true
        }

        if env::var("TRACE_FILTER").is_ok() {
            self.enable_otlp_tracing = true
        }

        if env::var("RUST_LOG_JSON").is_ok() {
            self.json_log_output = true;
        }

        if let Ok(spec) = env::var("RUST_LOG_TAILS") {
            self.log_tails = parse_log_tails(&spec);
        }

        if env::var("TOKIO_CONSOLE").is_ok() {
            self.tokio_console = true;
        }

        if let Ok(span_level) = env::var("TOKIO_SPAN_LEVEL") {
            self.span_level =
                Some(LevelFilter::from_str(&span_level).expect("Cannot parse TOKIO_SPAN_LEVEL"));
        }

        if let Ok(filepath) = env::var("RUST_LOG_FILE") {
            self.log_file = Some(filepath);
        }

        if let Ok(sample_rate) = env::var("SAMPLE_RATE") {
            self.sample_rate = sample_rate.parse().expect("Cannot parse SAMPLE_RATE");
        }

        if let Ok(enable_error_layer) = env::var("ENABLE_ERROR_LAYER") {
            self.enable_error_layer = enable_error_layer
                .parse()
                .expect("Cannot parse ENABLE_ERROR_LAYER");
        }

        self
    }

    pub fn init(self) -> (TelemetryGuards, TracingHandle) {
        let config = self;
        let config_clone = config.clone();

        // Setup an EnvFilter for filtering logging output layers.
        // NOTE: we don't want to use this to filter all layers.  That causes problems for layers with
        // different filtering needs, including tokio-console/console-subscriber, and it also doesn't
        // fit with the span creation needs for distributed tracing and other span-based tools.
        let mut directives = config.log_string.unwrap_or_else(|| "info".into());
        if let Some(targets) = config.trace_target {
            for target in targets {
                directives.push_str(&format!(",{}=trace", target));
            }
        }
        let env_filter = EnvFilter::try_from_default_env()
            .unwrap_or_else(|_| EnvFilter::new(directives.clone()));
        // When the test layer is enabled, it accepts events at any level without a
        // per-layer filter, so global event-level filtering would hide events the
        // test wants to capture. Disable it by pinning max_event_level to TRACE
        // (and skip propagating env filter reloads into it).
        let max_event_level_seed = if config.enable_test_layer {
            LevelFilter::TRACE
        } else {
            env_filter.max_level_hint().unwrap_or(LevelFilter::TRACE)
        };
        let max_event_level = Arc::new(std::sync::atomic::AtomicU8::new(level_filter_to_u8(
            max_event_level_seed,
        )));
        let (log_filter, reload_handle) = reload::Layer::new(env_filter);
        let log_filter_handle = FilterHandle {
            reload: reload_handle,
            max_event_level: if config.enable_test_layer {
                None
            } else {
                Some(max_event_level.clone())
            },
        };

        // Separate span level filter.
        // This is a dumb filter for now - allows all spans that are below a given level.
        // TODO: implement a sampling filter
        let span_level = config
            .span_level
            .unwrap_or(LevelFilter::from_level(Level::INFO));
        let span_filter = filter::filter_fn(move |metadata| {
            metadata.is_span() && LevelFilter::from_level(*metadata.level()) <= span_level
        });

        let mut layers = Vec::new();

        // tokio-console layer
        // Please see https://docs.rs/console-subscriber/latest/console_subscriber/struct.Builder.html#configuration
        // for environment vars/config options
        if config.tokio_console {
            layers.push(console_subscriber::spawn().boxed());
        }

        if let Some(registry) = config.prom_registry
            && !config.disable_span_latency
        {
            let span_lat_layer = PrometheusSpanLatencyLayer::try_new(&registry, 15)
                .expect("Could not initialize span latency layer");
            layers.push(span_lat_layer.with_filter(span_filter.clone()).boxed());
        }

        let mut trace_filter_handle = None;
        let mut file_output = CachedOpenFile::new::<&str>(None).unwrap();
        let mut provider = None;
        let sampler = SamplingFilter::new(config.sample_rate);
        let service_name = env::var("OTEL_SERVICE_NAME").unwrap_or("sui-node".to_owned());

        if config.enable_otlp_tracing {
            let trace_file = env::var("TRACE_FILE").ok();
            let mut otel_kv_vec = vec![opentelemetry::KeyValue::new(
                "service.name",
                service_name.clone(),
            )];
            if let Ok(namespace) = env::var("NAMESPACE") {
                otel_kv_vec.push(opentelemetry::KeyValue::new("service.namespace", namespace));
            }
            if let Ok(hostname) = env::var("HOSTNAME") {
                otel_kv_vec.push(opentelemetry::KeyValue::new("host", hostname));
            }
            if let Ok(network) = env::var("NETWORK") {
                otel_kv_vec.push(opentelemetry::KeyValue::new("network", network));
            }

            let resource = Resource::new(otel_kv_vec);
            let sampler = Sampler::ParentBased(Box::new(sampler.clone()));

            // We can either do file output or OTLP, but not both. tracing-opentelemetry
            // only supports a single tracer at a time.
            let telemetry = if let Some(trace_file) = trace_file {
                let exporter =
                    FileExporter::new(Some(trace_file.into())).expect("Failed to create exporter");
                file_output = exporter.cached_open_file.clone();
                let processor = BatchSpanProcessor::builder(exporter, runtime::Tokio).build();

                let p = TracerProvider::builder()
                    .with_resource(resource)
                    .with_sampler(sampler)
                    .with_span_processor(processor)
                    .build();

                let tracer = p.tracer(service_name);
                provider = Some(p);

                tracing_opentelemetry::layer().with_tracer(tracer)
            } else {
                let endpoint = env::var("OTLP_ENDPOINT")
                    .unwrap_or_else(|_| "http://localhost:4317".to_string());
                let otlp_exporter = opentelemetry_otlp::SpanExporter::builder()
                    .with_tonic()
                    .with_endpoint(endpoint)
                    .build()
                    .unwrap();
                let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
                    .with_resource(resource)
                    .with_sampler(sampler)
                    .with_batch_exporter(otlp_exporter, runtime::Tokio)
                    .build();
                let tracer = tracer_provider.tracer(service_name);
                tracing_opentelemetry::layer().with_tracer(tracer)
            };

            // Enable Trace Contexts for tying spans together
            opentelemetry::global::set_text_map_propagator(
                opentelemetry_sdk::propagation::TraceContextPropagator::new(),
            );

            let trace_env_filter = EnvFilter::try_from_env("TRACE_FILTER").unwrap();
            let (trace_env_filter, reload_handle) = reload::Layer::new(trace_env_filter);
            trace_filter_handle = Some(FilterHandle {
                reload: reload_handle,
                max_event_level: None,
            });

            layers.push(telemetry.with_filter(trace_env_filter).boxed());
        }

        let (nb_output, worker_guard) = get_output(config.log_file.clone());
        // Flush guards for any per-target files opened below; kept alive in TelemetryGuards.
        let mut file_guards: Vec<WorkerGuard> = Vec::new();
        if config.json_log_output {
            // Output to file or to stderr in a newline-delimited JSON format
            let json_layer = fmt::layer()
                .with_file(true)
                .with_line_number(true)
                .json()
                .with_writer(nb_output)
                .with_filter(log_filter)
                .boxed();
            layers.push(json_layer);
        } else {
            let with_ansi = config.log_file.is_none() && stderr().is_tty();

            // Main human-readable layer — always emits everything permitted by the env
            // filter (`RUST_LOG`).
            let fmt_layer = fmt::layer()
                .with_ansi(with_ansi)
                .with_writer(nb_output.clone())
                .with_filter(log_filter)
                .boxed();
            layers.push(fmt_layer);

            if config.log_file.is_none() && stderr().is_tty() && !config.user_info_target.is_empty()
            {
                // Add another printer that prints unadorned info messages to stderr
                let mut directives = String::from("none");
                for target in config.user_info_target {
                    directives.push_str(&format!(",{target}=info"));
                }

                let fmt_layer = fmt::layer()
                    .with_ansi(config.log_file.is_none() && stderr().is_tty())
                    .event_format(
                        fmt::format()
                            .without_time()
                            .with_target(false)
                            .with_level(false),
                    )
                    .with_writer(nb_output)
                    .with_filter(EnvFilter::new(directives))
                    .boxed();

                layers.push(fmt_layer);
            }
        }

        if !config.log_tails.is_empty() {
            // One non-blocking writer per unique file, shared by that file's layers.
            let mut file_writers: HashMap<PathBuf, NonBlocking> = HashMap::new();
            for path in config.log_tails.iter().map(|t| &t.file) {
                if !file_writers.contains_key(path) {
                    let (writer, guard) = open_tail_file(path);
                    file_writers.insert(path.clone(), writer);
                    file_guards.push(guard);
                }
            }

            // One layer per routed target: its format, written to its file.
            for log_tail in &config.log_tails {
                let writer = file_writers
                    .get(&log_tail.file)
                    .cloned()
                    .expect("a writer was created above for every target's file");
                // Target restriction (matches this target at any level) ANDed with an EnvFilter
                // (built like the main one) so `RUST_LOG`'s per-target level is the binding gate.
                let target_filter = filter::Targets::new()
                    .with_target(log_tail.target.clone(), LevelFilter::TRACE)
                    .and(
                        EnvFilter::try_from_default_env()
                            .unwrap_or_else(|_| EnvFilter::new(directives.clone())),
                    );
                let layer = match log_tail.format {
                    LogFormat::Json => fmt::layer()
                        .with_file(true)
                        .with_line_number(true)
                        .json()
                        .with_writer(writer)
                        .with_filter(target_filter)
                        .boxed(),
                    LogFormat::Text => fmt::layer()
                        .with_ansi(false)
                        .with_writer(writer)
                        .with_filter(target_filter)
                        .boxed(),
                };
                layers.push(layer);
            }
        }

        if config.enable_error_layer {
            layers.push(ErrorLayer::new(Pretty::default()).boxed())
        }

        let test_layer = if config.enable_test_layer {
            let test_layer = TestLayer::new();
            layers.push(test_layer.clone().boxed());
            Some(test_layer)
        } else {
            None
        };

        // Global level filter: rejects span callsites above span_level and event
        // callsites above the env filter's max level at registration time, preventing
        // the Registry from dispatching callsites that every per-layer filter would
        // immediately discard.
        //
        // Must be stacked on top of `layers` via `.with()` rather than pushed into the
        // Vec. `Vec<Layer>::register_callsite` returns the most permissive Interest
        // across its layers, so a `never()` from this filter would be overridden by
        // any layer (e.g. fmt+EnvFilter) returning `sometimes()`. As an outer
        // `Layered`, `pick_interest` short-circuits to `never()` and the inner stack
        // is never consulted.
        let global_filter = GlobalLevelFilter {
            max_span_level: span_level,
            max_event_level,
        };
        let subscriber = tracing_subscriber::registry()
            .with(layers)
            .with(global_filter);
        let subscriber_guard = if config.set_global_default {
            tracing::subscriber::set_global_default(subscriber)
                .expect("unable to initialize tracing subscriber");
            None
        } else {
            Some(tracing::subscriber::set_default(subscriber))
        };

        if config.panic_hook {
            set_panic_hook(config.crash_on_panic);
        }

        // The guard must be returned and kept in the main fn of the app, as when it's dropped then the output
        // gets flushed and closed. If this is dropped too early then no output will appear!
        let guards = TelemetryGuards::new(
            config_clone,
            worker_guard,
            file_guards,
            provider,
            subscriber_guard,
        );

        (
            guards,
            TracingHandle {
                log: log_filter_handle,
                trace: trace_filter_handle,
                file_output,
                test_layer,
                sampler,
            },
        )
    }
}

// Like Sampler::TraceIdRatioBased, but can be updated at runtime
#[derive(Debug, Clone)]
struct SamplingFilter {
    // Sampling filter needs to be fast, so we avoid a mutex.
    sample_rate: Arc<AtomicF64>,
}

impl SamplingFilter {
    fn new(sample_rate: f64) -> Self {
        SamplingFilter {
            sample_rate: Arc::new(AtomicF64::new(Self::clamp(sample_rate))),
        }
    }

    fn clamp(sample_rate: f64) -> f64 {
        // clamp sample rate to between 0.0001 and 1.0
        sample_rate.clamp(0.0001, 1.0)
    }

    fn update_sampling_rate(&self, sample_rate: f64) {
        // clamp sample rate to between 0.0001 and 1.0
        let sample_rate = Self::clamp(sample_rate);
        self.sample_rate.store(sample_rate, Ordering::Relaxed);
    }
}

impl ShouldSample for SamplingFilter {
    fn should_sample(
        &self,
        parent_context: Option<&Context>,
        trace_id: TraceId,
        name: &str,
        span_kind: &SpanKind,
        attributes: &[KeyValue],
        links: &[Link],
    ) -> SamplingResult {
        let sample_rate = self.sample_rate.load(Ordering::Relaxed);
        let sampler = Sampler::TraceIdRatioBased(sample_rate);

        sampler.should_sample(parent_context, trace_id, name, span_kind, attributes, links)
    }
}

/// Globally set a tracing subscriber suitable for testing environments
pub fn init_for_testing() {
    static LOGGER: Lazy<()> = Lazy::new(|| {
        let subscriber = ::tracing_subscriber::FmtSubscriber::builder()
            .with_env_filter(
                EnvFilter::builder()
                    .with_default_directive(LevelFilter::INFO.into())
                    .from_env_lossy(),
            )
            .with_file(true)
            .with_line_number(true)
            .with_test_writer()
            .finish();
        ::tracing::subscriber::set_global_default(subscriber)
            .expect("unable to initialize logging for tests");
    });

    Lazy::force(&LOGGER);
}

#[cfg(test)]
mod tests {
    use super::*;
    use prometheus::proto::MetricType;
    use std::time::Duration;
    use tracing::{debug, debug_span, info, trace_span, warn};

    #[test]
    #[should_panic]
    fn test_telemetry_init() {
        let registry = prometheus::Registry::new();
        // Default logging level is INFO, but here we set the span level to DEBUG.  TRACE spans should be ignored.
        let config = TelemetryConfig::new()
            .with_span_level(Level::DEBUG)
            .with_prom_registry(&registry);
        let _guard = config.init();

        info!(a = 1, "This will be INFO.");
        // Spans are debug level or below, so they won't be printed out either.  However latencies
        // should be recorded for at least one span
        debug_span!("yo span yo").in_scope(|| {
            // This debug log will not print out, log level set to INFO by default
            debug!(a = 2, "This will be DEBUG.");
            std::thread::sleep(Duration::from_millis(100));
            warn!(a = 3, "This will be WARNING.");
        });

        // This span won't be enabled
        trace_span!("this span should not be created").in_scope(|| {
            info!("This log appears, but surrounding span is not created");
            std::thread::sleep(Duration::from_millis(100));
        });

        let metrics = registry.gather();
        // There should be 1 metricFamily and 1 metric
        assert_eq!(metrics.len(), 1);
        assert_eq!(metrics[0].name(), "tracing_span_latencies");
        assert_eq!(metrics[0].get_field_type(), MetricType::HISTOGRAM);
        let inner = metrics[0].get_metric();
        assert_eq!(inner.len(), 1);
        let labels = inner[0].get_label();
        assert_eq!(labels.len(), 1);
        assert_eq!(labels[0].name(), "span_name");
        assert_eq!(labels[0].value(), "yo span yo");

        panic!("This should cause error logs to be printed out!");
    }

    // Both the following tests should be able to "race" to initialize logging without causing a
    // panic
    #[test]
    fn testing_logger_1() {
        init_for_testing();
    }

    #[test]
    fn testing_logger_2() {
        init_for_testing();
    }

    #[test]
    fn with_log_tail_appends() {
        // Format is inferred from the file extension.
        let config = TelemetryConfig::new()
            .with_log_tail("graphql_request", "/var/log/x.ndjson")
            .with_log_tail("foo", "/var/log/foo.log");
        assert_eq!(config.log_tails.len(), 2);
        assert_eq!(config.log_tails[0].target, "graphql_request");
        assert_eq!(config.log_tails[0].format, LogFormat::Json);
        assert_eq!(config.log_tails[0].file, PathBuf::from("/var/log/x.ndjson"));
        assert_eq!(config.log_tails[1].target, "foo");
        assert_eq!(config.log_tails[1].format, LogFormat::Text);
        assert_eq!(config.log_tails[1].file, PathBuf::from("/var/log/foo.log"));
    }

    #[test]
    fn parse_log_tails_infers_format_from_extension() {
        // `target=file`, comma-separated, with surrounding whitespace; the file extension
        // selects the format.
        let t = parse_log_tails(
            "a=/var/log/a.jsonl, b=/var/log/b.ndjson , c=/var/log/c.log, d=/var/log/d.JSONL, e=/var/log/e.json",
        );
        assert_eq!(t.len(), 5);
        let got: Vec<_> = t
            .iter()
            .map(|lt| (&*lt.target, lt.format, lt.file.clone()))
            .collect();
        assert_eq!(
            got,
            vec![
                ("a", LogFormat::Json, PathBuf::from("/var/log/a.jsonl")),
                ("b", LogFormat::Json, PathBuf::from("/var/log/b.ndjson")),
                ("c", LogFormat::Text, PathBuf::from("/var/log/c.log")),
                // Extension match is case-insensitive.
                ("d", LogFormat::Json, PathBuf::from("/var/log/d.JSONL")),
                // `.json` is a single JSON document, not newline-delimited → text.
                ("e", LogFormat::Text, PathBuf::from("/var/log/e.json")),
            ]
        );

        // Empty / whitespace-only entries are skipped (they are absent, not malformed).
        assert!(parse_log_tails("").is_empty());
        assert!(parse_log_tails("  ,  , ").is_empty());
    }

    #[test]
    #[should_panic(expected = "invalid RUST_LOG_TAILS entry")]
    fn parse_log_tails_panics_without_file() {
        // A non-empty entry without a `=file` is a misconfiguration → hard-fail init.
        let _ = parse_log_tails("graphql_request");
    }

    #[test]
    #[should_panic(expected = "invalid RUST_LOG_TAILS entry")]
    fn parse_log_tails_panics_on_empty_file() {
        let _ = parse_log_tails("a=");
    }

    #[test]
    #[should_panic(expected = "invalid RUST_LOG_TAILS entry")]
    fn parse_log_tails_panics_on_empty_target() {
        let _ = parse_log_tails("=/var/log/x.log");
    }

    #[test]
    #[should_panic(expected = "invalid RUST_LOG_TAILS entry")]
    fn parse_log_tails_panics_on_one_bad_entry_among_valid() {
        // A single malformed entry fails the whole parse, even alongside valid ones.
        let _ = parse_log_tails("bare, ok=/var/log/ok.jsonl");
    }

    #[test]
    #[should_panic(expected = "cannot create directory for RUST_LOG_TAILS file")]
    fn open_tail_file_panics_on_uncreatable_dir() {
        // `/dev/null` is not a directory, so create_dir_all under it fails deterministically —
        // a bad RUST_LOG_TAILS file path must hard-fail init rather than silently degrade.
        let _ = open_tail_file(Path::new("/dev/null/x/test.log"));
    }
}