polyc_runtime/

observability.rs

//! Logging + traces + panic discipline (PRD §11).
//!
//! All logs go to **stderr** (stdout is reserved for program data). Format
//! is selected by `RUST_LOG_FORMAT` (`json` in prod, pretty otherwise);
//! verbosity by `RUST_LOG` via `EnvFilter`.
//!
//! When `OTEL_EXPORTER_OTLP_ENDPOINT` is set, the tracing subscriber also
//! gains an OpenTelemetry layer that batches spans and ships them via OTLP
//! gRPC to the configured collector. `OTEL_SERVICE_NAME` overrides the
//! service name (defaults to the caller's argument). Dev runs without
//! that env var pay zero overhead — the layer is not constructed.
//!
//! Spans propagate through the in-process `tokio` runtime via
//! `tracing-opentelemetry`; cross-process propagation across the
//! `connectrpc` boundary uses the W3C `traceparent` header — wire it through
//! the dialer when running in production.

use std::io;
use std::sync::OnceLock;

use opentelemetry::KeyValue;
use opentelemetry::global;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry_otlp::SpanExporter;
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::propagation::TraceContextPropagator;
use opentelemetry_sdk::trace::SdkTracerProvider;
use tokio::sync::broadcast;
use tracing_subscriber::fmt::MakeWriter;
use tracing_subscriber::{EnvFilter, fmt, prelude::*};

/// How many recent log lines the in-process broadcast ring buffers for a slow
/// subscriber before it starts dropping the oldest. The local dashboard's log
/// stream is a live tail, not an audit trail, so lagging is fine.
const LOG_BROADCAST_CAPACITY: usize = 512;

/// Process-wide sender for the live log stream, set once by [`init`]. A
/// `OnceLock` (not passed through every call site) so any in-process surface —
/// the local dashboard's log route in particular — can `subscribe` without the
/// binary threading a handle down to it.
static LOG_BROADCAST: OnceLock<broadcast::Sender<String>> = OnceLock::new();

/// Subscribe to the live process-log stream.
///
/// Returns a receiver that yields one formatted log line per emitted `tracing`
/// event (the same text written to stderr), or `None` when [`init`] has not run
/// yet. A receiver that falls behind the bounded ring drops the oldest lines
/// rather than stalling the writer.
#[must_use]
pub fn subscribe_logs() -> Option<broadcast::Receiver<String>> {
    LOG_BROADCAST.get().map(broadcast::Sender::subscribe)
}

/// A `MakeWriter` that tees each formatted log line to stderr and to the live
/// broadcast. Cloning shares the same sender.
#[derive(Clone)]
struct BroadcastTee {
    tx: broadcast::Sender<String>,
}

impl<'a> MakeWriter<'a> for BroadcastTee {
    type Writer = TeeLine;

    fn make_writer(&'a self) -> Self::Writer {
        TeeLine {
            buf: Vec::new(),
            tx: self.tx.clone(),
        }
    }
}

/// One event's worth of formatted bytes: written straight through to stderr and
/// accumulated so the whole line can be broadcast when the writer drops (the
/// `fmt` layer makes a fresh writer per event and drops it once written).
struct TeeLine {
    buf: Vec<u8>,
    tx: broadcast::Sender<String>,
}

impl io::Write for TeeLine {
    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
        // stdout is reserved for program data; all logs go to stderr.
        io::stderr().write_all(data)?;
        self.buf.extend_from_slice(data);
        Ok(data.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        io::stderr().flush()
    }
}

impl Drop for TeeLine {
    fn drop(&mut self) {
        if self.buf.is_empty() {
            return;
        }
        // A lossy decode keeps a stray non-UTF-8 byte from dropping the line.
        let line = String::from_utf8_lossy(&self.buf).trim_end().to_owned();
        if !line.is_empty() {
            // `send` errs only when there are no receivers; the buffered ring
            // means that is the sole failure mode, and it is not one to log.
            let _ = self.tx.send(line);
        }
    }
}

/// Handle returned from [`init`] — keep it alive for the process lifetime;
/// `drop` flushes traces before exit. `SdkTracerProvider::shutdown` blocks
/// until exporters drain.
pub struct ShutdownGuard {
    provider: Option<SdkTracerProvider>,
}

impl Drop for ShutdownGuard {
    fn drop(&mut self) {
        if let Some(provider) = self.provider.take() {
            // Best-effort: log to stderr and continue if flush fails.
            if let Err(err) = provider.shutdown() {
                eprintln!("opentelemetry shutdown failed: {err}");
            }
        }
    }
}

/// Initialise the global tracing subscriber + (optional) `OTel` layer.
///
/// Also installs the panic hook. Call once, early in `main`. The returned
/// guard flushes traces on drop — keep it alive for the process lifetime.
#[must_use]
pub fn init(service_name: &str) -> ShutdownGuard {
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
    let json = std::env::var("RUST_LOG_FORMAT").is_ok_and(|v| v.eq_ignore_ascii_case("json"));

    let (tracer, provider) = build_otel_tracer(service_name);
    let otel_layer = tracer.map(|t| tracing_opentelemetry::layer().with_tracer(t));

    // Install the W3C `traceparent` propagator unconditionally. The format
    // half is cheap (a unit struct), and installing it always means the
    // dialer's `inject_*` calls and the server's `extract_*` calls behave
    // consistently whether or not OTLP export is enabled — without it, the
    // global getter returns a no-op propagator and `traceparent` headers
    // silently disappear, which is the worst-of-both-worlds debug state.
    // See `crate::propagation` for the matching inject/extract helpers.
    global::set_text_map_propagator(TraceContextPropagator::new());

    // Tee every formatted log line to stderr and to a process-wide broadcast the
    // local dashboard tails. The ring is bounded, so a stalled subscriber never
    // slows logging. Setting the sender always (not only when subscribed) means
    // the very first lines are already flowing before anyone connects.
    let (log_tx, _seed) = broadcast::channel(LOG_BROADCAST_CAPACITY);
    let _ = LOG_BROADCAST.set(log_tx.clone());
    let tee = BroadcastTee { tx: log_tx };

    let registry = tracing_subscriber::registry().with(filter).with(otel_layer);
    if json {
        let fmt_layer = fmt::layer().json().flatten_event(true).with_writer(tee);
        registry.with(fmt_layer).init();
    } else {
        let fmt_layer = fmt::layer().with_writer(tee);
        registry.with(fmt_layer).init();
    }

    install_panic_hook();
    ShutdownGuard { provider }
}

/// Build the OpenTelemetry tracer (the thing the tracing layer wraps), or
/// `(None, None)` if no OTLP endpoint is configured. Also installs the
/// global tracer provider so libraries that talk to
/// `opentelemetry::global` see it.
///
/// Caveat: "the global" means *this* otel version's static. The lock
/// currently carries a second otel stack (0.31, pinned by
/// commonware-runtime — see the lockstep comment in the workspace
/// Cargo.toml) whose own `global` stays the no-op default; only libraries
/// emitting through the shared `tracing` facade are version-proof.
fn build_otel_tracer(
    service_name: &str,
) -> (
    Option<opentelemetry_sdk::trace::Tracer>,
    Option<SdkTracerProvider>,
) {
    if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err() {
        return (None, None);
    }
    let resolved_name = std::env::var("OTEL_SERVICE_NAME")
        .ok()
        .unwrap_or_else(|| service_name.to_owned());
    let exporter = match SpanExporter::builder().with_tonic().build() {
        Ok(exp) => exp,
        Err(err) => {
            eprintln!("opentelemetry OTLP exporter build failed: {err}; continuing without traces");
            return (None, None);
        }
    };
    let resource = Resource::builder()
        .with_attribute(KeyValue::new("service.name", resolved_name.clone()))
        .build();
    let provider = SdkTracerProvider::builder()
        .with_batch_exporter(exporter)
        .with_resource(resource)
        .build();
    let tracer = provider.tracer(resolved_name);
    global::set_tracer_provider(provider.clone());
    (Some(tracer), Some(provider))
}

/// Emit a single structured `error` event on panic, then defer to the previous
/// hook (which, under `panic = "abort"`, terminates the process).
fn install_panic_hook() {
    let previous = std::panic::take_hook();
    std::panic::set_hook(Box::new(move |info| {
        let location = info
            .location()
            .map_or_else(|| "unknown".to_owned(), ToString::to_string);
        tracing::error!(panic = %info, location = %location, "process panicked");
        previous(info);
    }));
}