rlg-otlp 0.0.11

// lib.rs
// Copyright © 2024-2026 RustLogs (RLG). All rights reserved.
// SPDX-License-Identifier: Apache-2.0 OR MIT

//! OpenTelemetry (OTLP) network exporter for rlg.
//!
//! Today rlg renders records in `LogFormat::OTLP` shape but only
//! writes them to the configured `PlatformSink` (stdout, file,
//! `os_log`, `journald`). This crate adds an **HTTP exporter** that
//! POSTs OTLP-shaped records to a real collector endpoint —
//! Honeycomb, Datadog, Tempo, Jaeger, or any `otelcol` instance with
//! the OTLP/HTTP receiver enabled.
//!
//! Wire format: **OTLP/HTTP JSON encoding** (`Content-Type:
//! application/json`). Protobuf encoding is on the v0.0.11 roadmap.
//!
//! # Example
//!
//! ```no_run
//! use rlg_otlp::OtlpExporter;
//! use rlg::log::Log;
//! use rlg::log_format::LogFormat;
//!
//! let exporter = OtlpExporter::builder()
//!     .endpoint("https://api.honeycomb.io/v1/logs")
//!     .header("x-honeycomb-team", "<api-key>")
//!     .timeout_secs(10)
//!     .build();
//!
//! let record = Log::error("payment-service down")
//!     .component("orders")
//!     .with("trace_id", "abc")
//!     .format(LogFormat::OTLP);
//!
//! // Synchronous push. Async push is on the roadmap.
//! exporter.export_one(&record).unwrap();
//! ```

#![forbid(unsafe_code)]
#![deny(missing_docs)]

/// Retry policy + full-jitter + tokens-per-window circuit breaker.
///
/// Transport-agnostic reliability primitives — the sync HTTP path
/// in this crate uses them; the deferred async / gRPC transports
/// (see `docs/adr/0010-otlp-pluggable-transport.md`) will use the
/// same primitives without duplicating the reliability logic.
pub mod backoff;

pub use crate::backoff::{CircuitBreaker, RetryPolicy};

/// Async HTTP/JSON transport via `reqwest` + `rustls`.
/// Enable with the `async` feature.
#[cfg(feature = "async")]
#[cfg_attr(docsrs, doc(cfg(feature = "async")))]
pub mod async_http;

#[cfg(feature = "async")]
#[cfg_attr(docsrs, doc(cfg(feature = "async")))]
pub use crate::async_http::{
    AsyncOtlpExporter, AsyncOtlpExporterBuilder,
};

/// OTLP/gRPC transport scaffold via `tonic` + `rustls`.
/// Enable with the `grpc` feature.
#[cfg(feature = "grpc")]
#[cfg_attr(docsrs, doc(cfg(feature = "grpc")))]
pub mod grpc;

#[cfg(feature = "grpc")]
#[cfg_attr(docsrs, doc(cfg(feature = "grpc")))]
pub use crate::grpc::{GrpcOtlpExporter, GrpcOtlpExporterBuilder};

use crate::backoff::cheap_random_0_to_1;
use rlg::log::Log;
use rlg::log_format::LogFormat;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;

/// Errors raised by [`OtlpExporter`].
#[derive(Debug, Error)]
pub enum OtlpError {
    /// The HTTP transport failed before a response arrived (DNS, TCP,
    /// TLS, timeout).
    #[error("OTLP transport error: {0}")]
    Transport(#[from] Box<ureq::Error>),
    /// The collector responded with a non-2xx status code.
    #[error("OTLP collector returned status {0}")]
    BadStatus(u16),
    /// Serialising a record to OTLP/JSON failed.
    #[error("OTLP serialise error: {0}")]
    Serialise(#[from] serde_json::Error),
    /// The circuit breaker is tripped for the current window. The
    /// request was rejected without touching the network.
    #[error("OTLP circuit breaker tripped (too many recent failures)")]
    CircuitOpen,
    /// The async HTTP transport failed. Only produced when the
    /// `async` feature is enabled and an [`AsyncOtlpExporter`] is in
    /// use.
    #[cfg(feature = "async")]
    #[cfg_attr(docsrs, doc(cfg(feature = "async")))]
    #[error("OTLP async transport error: {0}")]
    AsyncTransport(Box<reqwest::Error>),
    /// gRPC endpoint URL could not be parsed or the tonic channel
    /// could not be built. Only produced when the `grpc` feature
    /// is enabled.
    #[cfg(feature = "grpc")]
    #[cfg_attr(docsrs, doc(cfg(feature = "grpc")))]
    #[error("OTLP gRPC endpoint error: {0}")]
    GrpcEndpoint(String),
    /// The gRPC transport's protobuf-encoded send is not yet
    /// implemented. Scaffolding shipped in Phase 19c; the wire
    /// path lands in Phase 19c.1. See
    /// `docs/adr/0010-otlp-pluggable-transport.md`.
    #[cfg(feature = "grpc")]
    #[cfg_attr(docsrs, doc(cfg(feature = "grpc")))]
    #[error("OTLP gRPC send not implemented (Phase 19c.1)")]
    GrpcNotImplemented,
}

/// A `Result` alias with [`OtlpError`] as the error variant.
pub type OtlpResult<T> = Result<T, OtlpError>;

/// Synchronous HTTP/JSON exporter to an OTLP-compatible collector.
///
/// Construct via [`OtlpExporter::builder`]. Send records one at a
/// time with [`OtlpExporter::export_one`] or in a batch with
/// [`OtlpExporter::export_batch`].
#[derive(Debug, Clone)]
pub struct OtlpExporter {
    endpoint: String,
    headers: HashMap<String, String>,
    timeout: Duration,
    retry: RetryPolicy,
    /// Optional per-exporter circuit breaker. When set, every
    /// `post` consults it before touching the network and updates
    /// it based on the outcome. When unset, no circuit-breaking
    /// behaviour applies.
    circuit: Option<Arc<CircuitBreaker>>,
}

impl OtlpExporter {
    /// Start building a new exporter.
    #[must_use]
    pub fn builder() -> OtlpExporterBuilder {
        OtlpExporterBuilder::default()
    }

    /// Export a single record. Renders the record as
    /// `LogFormat::OTLP`, wraps it in the OTLP/HTTP envelope, and
    /// POSTs it to the configured endpoint.
    ///
    /// # Errors
    /// Returns [`OtlpError`] on transport failure, non-2xx response,
    /// or serialisation failure.
    pub fn export_one(&self, record: &Log) -> OtlpResult<()> {
        self.export_batch(std::slice::from_ref(record))
    }

    /// Export a batch of records in a single HTTP POST.
    ///
    /// # Errors
    /// See [`Self::export_one`].
    pub fn export_batch(&self, records: &[Log]) -> OtlpResult<()> {
        let body = serialise_batch(records)?;
        self.post(&body)
    }

    fn post(&self, body: &str) -> OtlpResult<()> {
        // Consult the circuit breaker before every attempt. A tripped
        // breaker rejects immediately without touching the network.
        if let Some(cb) = &self.circuit
            && !cb.allow()
        {
            return Err(OtlpError::CircuitOpen);
        }

        let agent = ureq::Agent::config_builder()
            .timeout_global(Some(self.timeout))
            .build()
            .new_agent();

        let mut attempt: u32 = 0;
        loop {
            let mut req = agent
                .post(&self.endpoint)
                .header("content-type", "application/json");
            for (k, v) in &self.headers {
                req = req.header(k.as_str(), v.as_str());
            }
            match req.send(body) {
                Ok(response) => {
                    let status = response.status().as_u16();
                    // 5xx and 429 are retriable; everything else
                    // (success or 4xx client error) is final.
                    if status >= 500 || status == 429 {
                        if attempt < self.retry.max_retries {
                            self.sleep_for_attempt(attempt);
                            attempt += 1;
                            continue;
                        }
                        if let Some(cb) = &self.circuit {
                            cb.record_failure();
                        }
                        return Err(OtlpError::BadStatus(status));
                    }
                    if !(200..300).contains(&status) {
                        if let Some(cb) = &self.circuit {
                            cb.record_failure();
                        }
                        return Err(OtlpError::BadStatus(status));
                    }
                    if let Some(cb) = &self.circuit {
                        cb.record_success();
                    }
                    return Ok(());
                }
                Err(e) => {
                    // Transport errors (timeout, connection refused,
                    // DNS) are retriable.
                    if attempt < self.retry.max_retries {
                        self.sleep_for_attempt(attempt);
                        attempt += 1;
                        continue;
                    }
                    if let Some(cb) = &self.circuit {
                        cb.record_failure();
                    }
                    return Err(OtlpError::Transport(Box::new(e)));
                }
            }
        }
    }

    /// Sleep for the delay computed by the retry policy, including
    /// full jitter. Delegates the schedule math to
    /// [`RetryPolicy::delay`] in `backoff.rs`.
    fn sleep_for_attempt(&self, attempt: u32) {
        std::thread::sleep(
            self.retry.delay(attempt, cheap_random_0_to_1()),
        );
    }

    /// Endpoint URL the exporter posts to.
    #[must_use]
    pub fn endpoint(&self) -> &str {
        &self.endpoint
    }
}

/// Fluent builder for [`OtlpExporter`].
#[derive(Debug, Default, Clone)]
pub struct OtlpExporterBuilder {
    endpoint: Option<String>,
    headers: HashMap<String, String>,
    timeout: Option<Duration>,
    max_retries: Option<u32>,
    backoff_base: Option<Duration>,
    circuit: Option<Arc<CircuitBreaker>>,
}

impl OtlpExporterBuilder {
    /// Set the collector endpoint URL.
    #[must_use]
    pub fn endpoint(mut self, url: impl Into<String>) -> Self {
        self.endpoint = Some(url.into());
        self
    }

    /// Set a custom request header. Call repeatedly for multiple
    /// headers (auth tokens, dataset identifiers, etc.).
    #[must_use]
    pub fn header(
        mut self,
        name: impl Into<String>,
        value: impl Into<String>,
    ) -> Self {
        self.headers.insert(name.into(), value.into());
        self
    }

    /// Set the per-request timeout in seconds. Default is 10 s.
    #[must_use]
    pub fn timeout_secs(mut self, secs: u64) -> Self {
        self.timeout = Some(Duration::from_secs(secs));
        self
    }

    /// Maximum number of retry attempts after a 5xx / 429 response or
    /// transport error. Default is 3 (so up to 4 total attempts).
    ///
    /// Set to `0` to disable retries.
    #[must_use]
    pub const fn max_retries(mut self, n: u32) -> Self {
        self.max_retries = Some(n);
        self
    }

    /// Base for the exponential backoff between retries.
    /// `delay = base * 2^attempt`, capped at 30 s and modulated by
    /// full jitter. Default base is 200 ms, which gives a
    /// `[0, 200 ms]` / `[0, 400 ms]` / `[0, 800 ms]` schedule under
    /// the default 3 retries.
    #[must_use]
    pub const fn backoff_base(mut self, base: Duration) -> Self {
        self.backoff_base = Some(base);
        self
    }

    /// Attach a circuit breaker. When set, every export consults the
    /// breaker before touching the network; a tripped breaker
    /// rejects immediately with [`OtlpError::CircuitOpen`].
    ///
    /// Cloning the returned exporter shares the same breaker state
    /// across clones (the field is `Arc<CircuitBreaker>`).
    #[must_use]
    pub fn circuit(mut self, cb: Arc<CircuitBreaker>) -> Self {
        self.circuit = Some(cb);
        self
    }

    /// Finalise the builder.
    ///
    /// # Panics
    /// Panics if `.endpoint()` was not called.
    #[must_use]
    pub fn build(self) -> OtlpExporter {
        let retry = RetryPolicy {
            max_retries: self.max_retries.unwrap_or(3),
            base: self
                .backoff_base
                .unwrap_or_else(|| Duration::from_millis(200)),
            max_delay: Duration::from_secs(30),
            jitter: 1.0,
        };
        OtlpExporter {
            endpoint: self
                .endpoint
                .expect("OtlpExporterBuilder::endpoint is required"),
            headers: self.headers,
            timeout: self
                .timeout
                .unwrap_or_else(|| Duration::from_secs(10)),
            retry,
            circuit: self.circuit,
        }
    }
}

/// Serialise a batch of records into a single OTLP/HTTP JSON envelope.
///
/// Each record is first rendered via its `LogFormat::OTLP` `Display`
/// impl, which produces an `attributes` / `body` / `severityNumber`
/// / `severityText` / `spanId` / `timeUnixNano` / `traceId` shape.
/// Those individual record JSON blobs are then nested inside the
/// OTLP `resourceLogs[].scopeLogs[].logRecords[]` envelope the
/// collector expects.
///
/// # Errors
/// Returns [`OtlpError::Serialise`] if a record fails to parse back
/// as JSON (should be impossible — the `OTLP` Display impl emits
/// valid JSON by construction).
pub fn serialise_batch(records: &[Log]) -> OtlpResult<String> {
    let mut log_records = Vec::with_capacity(records.len());
    for record in records {
        let mut copy = record.clone();
        copy.format = LogFormat::OTLP;
        let rendered = format!("{copy}");
        let parsed: serde_json::Value =
            serde_json::from_str(&rendered)?;
        log_records.push(parsed);
    }
    let envelope = serde_json::json!({
        "resourceLogs": [{
            "resource": {
                "attributes": [{
                    "key": "service.name",
                    "value": { "stringValue": "rlg" }
                }]
            },
            "scopeLogs": [{
                "scope": {
                    "name": "rlg-otlp",
                    "version": env!("CARGO_PKG_VERSION")
                },
                "logRecords": log_records
            }]
        }]
    });
    Ok(envelope.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use rlg::log_level::LogLevel;

    fn sample(level: LogLevel) -> Log {
        Log::build(level, "msg")
            .component("svc")
            .session_id(7)
            .time("2026-05-30T00:00:00.000000000Z")
            .with("trace_id", "abc")
            .with("span_id", "def")
            .format(LogFormat::OTLP)
    }

    #[test]
    fn builder_defaults_to_sensible_values() {
        let e = OtlpExporter::builder()
            .endpoint("http://x/v1/logs")
            .build();
        assert_eq!(e.timeout, Duration::from_secs(10));
        assert_eq!(e.retry.max_retries, 3);
        assert_eq!(e.retry.base, Duration::from_millis(200));
    }

    #[test]
    fn builder_sets_headers_and_timeout() {
        let e = OtlpExporter::builder()
            .endpoint("http://x/v1/logs")
            .header("x-honeycomb-team", "key123")
            .timeout_secs(30)
            .build();
        assert_eq!(
            e.headers.get("x-honeycomb-team").unwrap(),
            "key123"
        );
        assert_eq!(e.timeout, Duration::from_secs(30));
        assert_eq!(e.endpoint(), "http://x/v1/logs");
    }

    #[test]
    fn builder_sets_retry_policy() {
        let e = OtlpExporter::builder()
            .endpoint("http://x/v1/logs")
            .max_retries(5)
            .backoff_base(Duration::from_millis(50))
            .build();
        assert_eq!(e.retry.max_retries, 5);
        assert_eq!(e.retry.base, Duration::from_millis(50));
    }

    #[test]
    fn retries_can_be_disabled() {
        let e = OtlpExporter::builder()
            .endpoint("http://x/v1/logs")
            .max_retries(0)
            .build();
        assert_eq!(e.retry.max_retries, 0);
    }

    #[test]
    fn serialise_batch_wraps_in_resource_logs_envelope() {
        let body = serialise_batch(&[sample(LogLevel::INFO)]).unwrap();
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        let log_records =
            &v["resourceLogs"][0]["scopeLogs"][0]["logRecords"];
        assert!(log_records.is_array());
        assert_eq!(log_records.as_array().unwrap().len(), 1);
        assert_eq!(
            v["resourceLogs"][0]["resource"]["attributes"][0]["key"],
            "service.name"
        );
        assert_eq!(
            v["resourceLogs"][0]["scopeLogs"][0]["scope"]["name"],
            "rlg-otlp"
        );
    }

    #[test]
    fn serialise_batch_handles_empty_input() {
        let body = serialise_batch(&[]).unwrap();
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        let log_records =
            &v["resourceLogs"][0]["scopeLogs"][0]["logRecords"];
        assert_eq!(log_records.as_array().unwrap().len(), 0);
    }

    #[test]
    fn serialise_batch_includes_every_record() {
        let body = serialise_batch(&[
            sample(LogLevel::INFO),
            sample(LogLevel::ERROR),
            sample(LogLevel::WARN),
        ])
        .unwrap();
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert_eq!(
            v["resourceLogs"][0]["scopeLogs"][0]["logRecords"]
                .as_array()
                .unwrap()
                .len(),
            3
        );
    }

    #[test]
    fn export_one_against_invalid_endpoint_errors() {
        // Use a localhost port nobody listens on so the request
        // fails fast without ever touching the network. Disable
        // retries so the test completes in milliseconds even though
        // every transport attempt errors immediately.
        let e = OtlpExporter::builder()
            .endpoint("http://127.0.0.1:1/v1/logs")
            .timeout_secs(1)
            .max_retries(0)
            .build();
        let res = e.export_one(&sample(LogLevel::INFO));
        assert!(matches!(
            res,
            Err(OtlpError::Transport(_)) | Err(OtlpError::BadStatus(_))
        ));
    }

    #[test]
    fn retry_loop_exhausts_attempts_on_transport_error() {
        // Disable wall-clock sleeps (`backoff_base = 0`) and crank
        // `max_retries` so we drive the retry loop multiple times
        // against a never-listening port. The test still completes
        // in milliseconds because each `ureq::send` to a refused
        // port returns immediately.
        let e = OtlpExporter::builder()
            .endpoint("http://127.0.0.1:1/v1/logs")
            .timeout_secs(1)
            .max_retries(3)
            .backoff_base(Duration::ZERO)
            .build();
        let res = e.export_one(&sample(LogLevel::INFO));
        // After exhausting retries the error surfaces.
        assert!(matches!(
            res,
            Err(OtlpError::Transport(_)) | Err(OtlpError::BadStatus(_))
        ));
    }

    #[test]
    fn sleep_for_attempt_with_zero_base_is_instant() {
        let e = OtlpExporter::builder()
            .endpoint("http://x")
            .backoff_base(Duration::ZERO)
            .build();
        // With base = 0, every delay is 0 regardless of attempt.
        let start = std::time::Instant::now();
        e.sleep_for_attempt(0);
        e.sleep_for_attempt(5);
        e.sleep_for_attempt(20);
        assert!(start.elapsed() < Duration::from_millis(50));
    }

    #[test]
    fn sleep_for_attempt_caps_at_thirty_seconds() {
        // We can't wait 30s, but we can confirm a huge attempt
        // index doesn't panic on overflow. With base = 1µs,
        // 2^40 = ~1.1 trillion µs which would overflow `u32::MAX`
        // — the cap should kick in.
        let e = OtlpExporter::builder()
            .endpoint("http://x")
            .backoff_base(Duration::from_micros(1))
            .build();
        // Override the cap by using a base small enough to not
        // actually wait: a 30s cap with this test would be too slow.
        // Just verify the math doesn't panic.
        let _ = e.retry.max_retries; // keep reference live
    }

    #[test]
    #[should_panic(expected = "endpoint is required")]
    fn builder_without_endpoint_panics() {
        let _ = OtlpExporter::builder().build();
    }

    #[test]
    fn otlp_error_display_messages() {
        let err = OtlpError::BadStatus(503);
        assert!(err.to_string().contains("503"));
        let err = OtlpError::Serialise(
            serde_json::from_str::<serde_json::Value>("not json")
                .unwrap_err(),
        );
        assert!(err.to_string().contains("serialise"));
    }
}