zlayer-observability 0.14.0

//! Overridable OTLP telemetry seam.
//!
//! ZLayer is a product normal users install, so it MUST NOT depend on the
//! private constellation `zlogging` crate. Telemetry forwarding therefore goes
//! through a trait-based, **overridable** seam:
//!
//! * The **default** provider ([`DefaultOtlpProvider`]) ships spans and
//!   `tracing` events to an OTLP collector over the **OTLP/HTTP-protobuf wire
//!   protocol via the async reqwest client** (NOT gRPC/tonic — the eager tonic
//!   build panicked pre-runtime; NOT the stdout exporter). Export is
//!   **batched/periodic** through `opentelemetry_sdk::runtime::Tokio`, built
//!   INSIDE the daemon's Tokio runtime.
//! * The subscriber installed in the OTLP-on path carries the OTel span layer +
//!   the `opentelemetry-appender-tracing` log layer **ONLY** — it installs no
//!   stdout/console fmt layer, so the daemon's container stdout/stderr stays
//!   clean and telemetry egresses exclusively over OTLP requests. The
//!   console/file subscriber is used only when OTLP is disabled (or on fallback).
//! * BlackLeaf-internal builds register their own provider (e.g. the private
//!   `zlogging` SDK) via [`set_telemetry_provider`] before daemon init; ZLayer
//!   itself carries no `zlogging` dependency.
//!
//! [`init_otlp_in_runtime`] **must be called from inside an active Tokio
//! runtime** (the batch processors use `runtime::Tokio`, and a raw exporter
//! built before the runtime panicked at startup with "there is no reactor
//! running"). On any exporter build error / no endpoint it falls back to the
//! console/file subscriber and never crashes.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};

use opentelemetry::trace::TracerProvider as _;
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
use opentelemetry_http::{Bytes, HttpClient, HttpError};
use opentelemetry_otlp::{Protocol, WithExportConfig, WithHttpConfig};
use opentelemetry_sdk::logs::log_processor_with_async_runtime::BatchLogProcessor;
use opentelemetry_sdk::logs::SdkLoggerProvider;
use opentelemetry_sdk::runtime;
use opentelemetry_sdk::trace::span_processor_with_async_runtime::BatchSpanProcessor;
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::Resource;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::EnvFilter;

use crate::config::{ObservabilityConfig, TracingConfig};
use crate::logging::{self, LogGuard};

/// Guard for the OTLP telemetry activation.
///
/// When OTLP forwarding is active it owns the live SDK tracer/logger providers
/// and flushes + shuts them down on `Drop`. When `init` fell back (build error /
/// no endpoint) it holds the console/file [`LogGuard`] instead. Empty
/// (`Default`) when OTLP is disabled. Hold it for the process lifetime.
#[derive(Default)]
pub struct OtlpGuard {
    /// Live tracer provider — present when OTLP span forwarding is active.
    tracer_provider: Option<SdkTracerProvider>,
    /// Live logger provider — present when OTLP log forwarding is active.
    logger_provider: Option<SdkLoggerProvider>,
    /// Console/file subscriber guard — present when we fell back instead.
    _fallback: Option<LogGuard>,
}

impl Drop for OtlpGuard {
    fn drop(&mut self) {
        // Flush + shut the batch processors down so buffered spans/logs egress
        // before exit. Best-effort: a failed shutdown must never panic on drop.
        if let Some(provider) = &self.tracer_provider {
            let _ = provider.shutdown();
        }
        if let Some(provider) = &self.logger_provider {
            let _ = provider.shutdown();
        }
    }
}

/// Overridable telemetry seam.
///
/// `init` is handed the full [`ObservabilityConfig`] and returns the live
/// [`OtlpGuard`]. The built-in [`DefaultOtlpProvider`] implements the
/// reqwest/HTTP-protobuf batched exporter; BlackLeaf-internal builds register an
/// alternative (e.g. the private `zlogging` SDK) via [`set_telemetry_provider`].
pub trait TelemetryProvider: Send + Sync {
    /// Install telemetry forwarding and return its lifetime guard.
    fn init(&self, cfg: &ObservabilityConfig) -> OtlpGuard;
}

/// Process-global override, set at most once (first registration wins).
static TELEMETRY_PROVIDER: OnceLock<Box<dyn TelemetryProvider>> = OnceLock::new();

/// Register an overriding [`TelemetryProvider`] used by [`init_otlp_in_runtime`].
///
/// First-wins: returns `true` if this call installed the provider, `false` if
/// one was already registered. Call before daemon init (see the daemon
/// entrypoint, before `init_observability` / `init_otlp_in_runtime`).
pub fn set_telemetry_provider(provider: Box<dyn TelemetryProvider>) -> bool {
    TELEMETRY_PROVIDER.set(provider).is_ok()
}

/// Whether OTLP forwarding is requested.
///
/// On when `config.enabled` is set, when an endpoint is configured (config or
/// `OTEL_EXPORTER_OTLP_ENDPOINT`), or when `OTEL_TRACES_ENABLED` is truthy. When
/// this returns `true`, `init_observability` defers subscriber installation so
/// that [`init_otlp_in_runtime`] can install the OTel subscriber instead.
#[must_use]
pub fn otlp_is_enabled(config: &TracingConfig) -> bool {
    let env_enabled = std::env::var("OTEL_TRACES_ENABLED").is_ok_and(|v| v == "true" || v == "1");
    config.enabled
        || env_enabled
        || config.otlp_endpoint.is_some()
        || std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok()
}

/// Resolve the OTLP endpoint from config, falling back to the standard env var.
fn resolve_endpoint(config: &TracingConfig) -> Option<String> {
    config
        .otlp_endpoint
        .clone()
        .filter(|s| !s.is_empty())
        .or_else(|| {
            std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
                .ok()
                .filter(|s| !s.is_empty())
        })
}

/// Resolve the service name: `OTEL_SERVICE_NAME` wins, else the configured name.
fn resolve_service_name(config: &TracingConfig) -> String {
    std::env::var("OTEL_SERVICE_NAME")
        .ok()
        .filter(|s| !s.is_empty())
        .unwrap_or_else(|| config.service_name.clone())
}

/// Build the OTLP-on-path filter: honor `RUST_LOG`, else default to `info`.
fn otlp_env_filter() -> EnvFilter {
    EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"))
}

/// Activate OTLP forwarding — **must be called from inside an active Tokio
/// runtime**.
///
/// No-op (empty guard) when OTLP is disabled; the caller's console/file
/// subscriber stays in charge. Otherwise: if a [`TelemetryProvider`] was
/// registered via [`set_telemetry_provider`] it is used; else the built-in
/// [`DefaultOtlpProvider`] (reqwest/HTTP-protobuf, batched via `runtime::Tokio`)
/// runs. On any error or missing endpoint the default provider falls back to the
/// console/file subscriber instead of crashing.
#[must_use]
pub fn init_otlp_in_runtime(config: &ObservabilityConfig) -> OtlpGuard {
    if !otlp_is_enabled(&config.tracing) {
        return OtlpGuard::default();
    }

    if let Some(provider) = TELEMETRY_PROVIDER.get() {
        return provider.init(config);
    }

    DefaultOtlpProvider.init(config)
}

/// The built-in default telemetry provider: OTLP/HTTP-protobuf over the async
/// reqwest client, batched via `runtime::Tokio`.
pub struct DefaultOtlpProvider;

impl TelemetryProvider for DefaultOtlpProvider {
    fn init(&self, config: &ObservabilityConfig) -> OtlpGuard {
        // Enabled but nowhere to send: install the normal console/file subscriber
        // (`init_observability` deferred it on our behalf).
        let Some(endpoint) = resolve_endpoint(&config.tracing) else {
            return OtlpGuard {
                tracer_provider: None,
                logger_provider: None,
                _fallback: logging::init_logging_inner(&config.logging).ok(),
            };
        };

        match build_otlp(&endpoint, config) {
            Ok(guard) => guard,
            Err(e) => {
                // Never let telemetry init take the daemon down. Print on the real
                // fd 2 (reaches the journal even before any subscriber is up), then
                // fall back to the console/file subscriber.
                eprintln!("OTLP telemetry init failed: {e}; falling back to console/file logging");
                OtlpGuard {
                    tracer_provider: None,
                    logger_provider: None,
                    _fallback: logging::init_logging_inner(&config.logging).ok(),
                }
            }
        }
    }
}

/// Build the per-signal OTLP/HTTP URL from a base endpoint.
///
/// CRUX (verified against `opentelemetry-otlp` 0.31's
/// `exporter::http::resolve_http_endpoint`): a **programmatic**
/// `.with_endpoint(value)` is used **verbatim — the per-signal path is NOT
/// appended**. From the crate source:
///
/// ```text
/// // programmatic configuration overrides any value set via environment variables
/// if let Some(provider_endpoint) = provided_endpoint.filter(|s| !s.is_empty()) {
///     provider_endpoint.parse() ...   // used as-is, no "/v1/traces" appended
/// }
/// ```
///
/// (and its test `test_http_exporter_endpoint`: *"if builder endpoint is set, it
/// should not add signal path"*). Only the bare `OTEL_EXPORTER_OTLP_ENDPOINT`
/// **env** var gets `/v1/{traces,logs}` appended — but `init_observability`
/// passes the resolved endpoint **programmatically**, so we must append the
/// per-signal path ourselves or every export POSTs to the bare base (which the
/// collector does not route — the bug this fixes).
///
/// `signal_path` is e.g. `/v1/traces` or `/v1/logs`. Robust to a trailing slash
/// on `base`, and idempotent if `base` already carries the per-signal suffix
/// (so a fully-qualified `…/otel/v1/logs` is left untouched).
fn signal_endpoint(base: &str, signal_path: &str) -> String {
    let trimmed = base.trim_end_matches('/');
    if trimmed.ends_with(signal_path) {
        trimmed.to_string()
    } else {
        format!("{trimmed}{signal_path}")
    }
}

/// A thin [`HttpClient`] wrapper that prints the failing OTLP endpoint to fd 2
/// **exactly once**, then delegates to the inner reqwest client.
///
/// The default exporter swallows export errors, and in the OTLP-on path a
/// `tracing::warn!` would route back through the OTLP appender (circular and
/// invisible). So on the first failed export — including any non-2xx response,
/// which the `reqwest::Client` `HttpClient` impl maps to `Err` via
/// `error_for_status` — we write a single diagnostic line to the real stderr
/// (which reaches the journal / daemon file log). One-shot, never per-export.
#[derive(Debug, Clone)]
struct DiagnosticHttpClient {
    inner: reqwest::Client,
    reported: Arc<AtomicBool>,
}

#[async_trait::async_trait]
impl HttpClient for DiagnosticHttpClient {
    async fn send_bytes(
        &self,
        request: http::Request<Bytes>,
    ) -> Result<http::Response<Bytes>, HttpError> {
        let uri = request.uri().clone();
        match self.inner.send_bytes(request).await {
            Ok(response) => Ok(response),
            Err(err) => {
                if !self.reported.swap(true, Ordering::Relaxed) {
                    // Real fd 2 (NOT tracing::warn!, which would loop through the
                    // OTLP appender). One-shot: prints once for the daemon log.
                    eprintln!(
                        "[zlayer-observability] OTLP export to {uri} failed: {err} \
                         — telemetry is being dropped; verify OTEL_EXPORTER_OTLP_ENDPOINT \
                         (expected a collector base like https://host/otel). \
                         This notice prints once."
                    );
                }
                Err(err)
            }
        }
    }
}

/// Build the batched OTLP/HTTP-protobuf exporters and install the OTel-only
/// subscriber (span layer + log-appender layer; **no** stdout/console layer).
fn build_otlp(
    endpoint: &str,
    config: &ObservabilityConfig,
) -> Result<OtlpGuard, opentelemetry_otlp::ExporterBuildError> {
    let service_name = resolve_service_name(&config.tracing);
    let resource = Resource::builder().with_service_name(service_name).build();

    // Per-signal full URLs (see `signal_endpoint`): with base
    // `https://logging.blackleafcloud.com/otel`, spans POST to `…/otel/v1/traces`
    // and logs to `…/otel/v1/logs` — matching the working Go runner's path.
    let traces_endpoint = signal_endpoint(endpoint, "/v1/traces");
    let logs_endpoint = signal_endpoint(endpoint, "/v1/logs");

    // One instrumented reqwest client, shared by both exporters: rustls/native
    // roots (matching the `reqwest-rustls` feature, no OpenSSL), and a one-shot
    // fd-2 notice on the first export failure. Mirror the upstream exporter's
    // build-or-default fallback so a client error never aborts telemetry init.
    let http_client = DiagnosticHttpClient {
        inner: reqwest::Client::builder()
            .build()
            .unwrap_or_else(|_| reqwest::Client::new()),
        reported: Arc::new(AtomicBool::new(false)),
    };

    // Span exporter: OTLP/HTTP-protobuf over reqwest, batched via runtime::Tokio.
    let span_exporter = opentelemetry_otlp::SpanExporter::builder()
        .with_http()
        .with_http_client(http_client.clone())
        .with_endpoint(&traces_endpoint)
        .with_protocol(Protocol::HttpBinary)
        .build()?;
    let span_processor = BatchSpanProcessor::builder(span_exporter, runtime::Tokio).build();
    let tracer_provider = SdkTracerProvider::builder()
        .with_resource(resource.clone())
        .with_span_processor(span_processor)
        .build();

    // Log exporter: OTLP/HTTP-protobuf over reqwest, batched via runtime::Tokio.
    let log_exporter = opentelemetry_otlp::LogExporter::builder()
        .with_http()
        .with_http_client(http_client)
        .with_endpoint(&logs_endpoint)
        .with_protocol(Protocol::HttpBinary)
        .build()?;
    let log_processor = BatchLogProcessor::builder(log_exporter, runtime::Tokio).build();
    let logger_provider = SdkLoggerProvider::builder()
        .with_resource(resource)
        .with_log_processor(log_processor)
        .build();

    // Make the tracer provider global so W3C context propagation (the
    // `propagation` module) shares the same provider.
    opentelemetry::global::set_tracer_provider(tracer_provider.clone());

    // OTLP-ON SUBSCRIBER: OTel span layer + log-appender layer ONLY. No console /
    // file fmt layer, so the daemon's container stdout/stderr stays clean and
    // telemetry egresses exclusively over OTLP requests.
    let span_layer = tracing_opentelemetry::layer().with_tracer(tracer_provider.tracer("zlayer"));
    let log_layer = OpenTelemetryTracingBridge::new(&logger_provider);
    tracing_subscriber::registry()
        .with(otlp_env_filter())
        .with(span_layer)
        .with(log_layer)
        .init();

    Ok(OtlpGuard {
        tracer_provider: Some(tracer_provider),
        logger_provider: Some(logger_provider),
        _fallback: None,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::TracingConfig;

    #[test]
    fn test_disabled_when_unconfigured() {
        // NOTE: assumes OTEL_* env is unset in the test environment.
        let config = TracingConfig {
            enabled: false,
            otlp_endpoint: None,
            ..Default::default()
        };
        if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err()
            && std::env::var("OTEL_TRACES_ENABLED").is_err()
        {
            assert!(!otlp_is_enabled(&config));
        }
    }

    #[test]
    fn test_enabled_via_config_endpoint() {
        let config = TracingConfig {
            enabled: false,
            otlp_endpoint: Some("https://logging.blackleafcloud.com/otel".to_string()),
            ..Default::default()
        };
        assert!(otlp_is_enabled(&config));
        assert_eq!(
            resolve_endpoint(&config).as_deref(),
            Some("https://logging.blackleafcloud.com/otel")
        );
    }

    #[test]
    fn test_signal_endpoint_appends_per_signal_path() {
        // Base endpoint -> per-signal path appended (the bug fix).
        assert_eq!(
            signal_endpoint("https://logging.blackleafcloud.com/otel", "/v1/traces"),
            "https://logging.blackleafcloud.com/otel/v1/traces"
        );
        assert_eq!(
            signal_endpoint("https://logging.blackleafcloud.com/otel", "/v1/logs"),
            "https://logging.blackleafcloud.com/otel/v1/logs"
        );
        // Trailing slash on the base is tolerated (no doubled slash).
        assert_eq!(
            signal_endpoint("https://logging.blackleafcloud.com/otel/", "/v1/logs"),
            "https://logging.blackleafcloud.com/otel/v1/logs"
        );
        // Already-qualified per-signal URL is left untouched (idempotent).
        assert_eq!(
            signal_endpoint(
                "https://logging.blackleafcloud.com/otel/v1/traces",
                "/v1/traces"
            ),
            "https://logging.blackleafcloud.com/otel/v1/traces"
        );
    }

    #[test]
    fn test_resolve_service_name_falls_back_to_config() {
        let config = TracingConfig {
            service_name: "zlayer-test".to_string(),
            ..Default::default()
        };
        // Only assert the config fallback when OTEL_SERVICE_NAME is unset.
        if std::env::var("OTEL_SERVICE_NAME").is_err() {
            assert_eq!(resolve_service_name(&config), "zlayer-test");
        }
    }

    #[test]
    #[allow(clippy::used_underscore_binding)] // intentionally reads the held `_fallback` guard field
    fn test_disabled_returns_empty_guard() {
        let config = ObservabilityConfig::default();
        if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err()
            && std::env::var("OTEL_TRACES_ENABLED").is_err()
        {
            let guard = init_otlp_in_runtime(&config);
            assert!(guard.tracer_provider.is_none());
            assert!(guard.logger_provider.is_none());
            assert!(guard._fallback.is_none());
        }
    }

    /// A registered provider overrides the built-in default exporter.
    ///
    /// Registering a dummy provider proves the seam routes through it (and never
    /// builds a real OTLP exporter / installs a subscriber). `OnceLock` is
    /// first-wins and process-global; this is the only test that registers one,
    /// and the only one that drives `init_otlp_in_runtime` with OTLP enabled.
    #[test]
    fn test_registered_provider_is_used() {
        use std::sync::atomic::{AtomicBool, Ordering};

        static CALLED: AtomicBool = AtomicBool::new(false);

        struct DummyProvider;
        impl TelemetryProvider for DummyProvider {
            fn init(&self, _cfg: &ObservabilityConfig) -> OtlpGuard {
                CALLED.store(true, Ordering::SeqCst);
                OtlpGuard::default()
            }
        }

        assert!(
            set_telemetry_provider(Box::new(DummyProvider)),
            "first registration should win"
        );

        let config = ObservabilityConfig {
            tracing: TracingConfig {
                enabled: true,
                otlp_endpoint: Some("https://example.invalid/otel".to_string()),
                ..Default::default()
            },
            ..Default::default()
        };

        let guard = init_otlp_in_runtime(&config);
        assert!(
            CALLED.load(Ordering::SeqCst),
            "registered provider must run"
        );
        // The dummy returned an empty guard, not the default exporter path.
        assert!(guard.tracer_provider.is_none());
        assert!(guard.logger_provider.is_none());
    }
}