//! Telemetry bootstrap: structured logs, plus optional OpenTelemetry export.
//!
//! [`init`] installs the process wide [`tracing`] subscriber. It always writes
//! human readable records to stdout, and when it is explicitly opted into it
//! mirrors the same spans to an OpenTelemetry collector over OTLP/HTTP. Jaeger
//! ingests OTLP natively, so nothing more than a collector is needed on the
//! other side:
//!
//! ```shell
//! $ docker run --rm -p 16686:16686 -p 4318:4318 jaegertracing/jaeger:2
//! $ OTEL_TRACES_EXPORTER=otlp cargo run
//! ```
//!
//! # Configuration
//!
//! Everything is driven by the environment, so the collector can be changed
//! without touching the code.
//!
//! | Variable | Effect | Default |
//! |---|---|---|
//! | `RUST_LOG` | Filter directives for the console output | [`DEFAULT_CONSOLE_FILTER`] |
//! | `OTEL_TRACES_EXPORTER` | Export spans when the list contains `otlp` | unset: console only |
//! | `OTEL_SDK_DISABLED` | `true` forces the exporter off, whatever the above says | `false` |
//! | `OTEL_TRACES_FILTER` | Filter directives for the exported spans | [`DEFAULT_SPANS_FILTER`] |
//! | `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | Which traces to keep, see [`sampler`] | `parentbased_always_on` |
//! | `OTEL_SERVICE_NAME` | `service.name` resource attribute | crate name |
//! | `OTEL_RESOURCE_ATTRIBUTES` | Extra resource attributes, e.g. `deployment.environment.name=develop` | unset |
//! | `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, `OTEL_EXPORTER_OTLP_ENDPOINT` | Collector address, in that order of precedence. The second is a base URL and gains the [`SIGNAL_PATH`]; the first is used verbatim. | [`DEFAULT_ENDPOINT`] |
//! | `OTEL_EXPORTER_OTLP_TIMEOUT` | Export timeout, in milliseconds | SDK default |
//! | `OTEL_EXPORTER_OTLP_HEADERS` | Extra headers, `key=value` pairs separated by commas | unset |
//!
//! The console and the exporter are filtered independently on purpose: turning
//! `RUST_LOG` down to `warn` to quiet the terminal should not silently stop
//! traces from reaching the collector.
use std::env;
use actix_web::body::MessageBody;
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_web::http::StatusCode;
use actix_web::{Error, error::ResponseError};
use opentelemetry::global;
use opentelemetry::trace::TracerProvider as _;
use opentelemetry::{KeyValue, Value};
use opentelemetry_otlp::{SpanExporter, WithExportConfig};
use opentelemetry_sdk::Resource;
use opentelemetry_sdk::error::OTelSdkResult;
use opentelemetry_sdk::trace::{
BatchSpanProcessor, Sampler, SdkTracer, SdkTracerProvider, SpanData,
};
use tokio::runtime::{Builder, Runtime};
use tracing::{Span, debug, info};
use tracing_actix_web::{RootSpanBuilder, TracingLogger, root_span};
use tracing_opentelemetry::OpenTelemetryLayer;
use tracing_subscriber::EnvFilter;
use tracing_subscriber::layer::{Layer, SubscriberExt};
use tracing_subscriber::util::SubscriberInitExt;
/// Comma separated list of exporters to use, per the OpenTelemetry SDK spec.
const TRACES_EXPORTER_VAR: &str = "OTEL_TRACES_EXPORTER";
/// Kills the whole SDK regardless of [`TRACES_EXPORTER_VAR`].
const SDK_DISABLED_VAR: &str = "OTEL_SDK_DISABLED";
/// Filter directives for the exported spans.
const SPANS_FILTER_VAR: &str = "OTEL_TRACES_FILTER";
/// Which traces to record, and at which rate.
const SAMPLER_VAR: &str = "OTEL_TRACES_SAMPLER";
const SAMPLER_ARG_VAR: &str = "OTEL_TRACES_SAMPLER_ARG";
/// `service.name`, the resource attribute a collector groups traces by.
const SERVICE_NAME_VAR: &str = "OTEL_SERVICE_NAME";
/// Collector address, signal specific first as the spec prescribes.
const TRACES_ENDPOINT_VAR: &str = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT";
const ENDPOINT_VAR: &str = "OTEL_EXPORTER_OTLP_ENDPOINT";
/// OTLP/HTTP port a stock Jaeger or collector container listens on.
const DEFAULT_ENDPOINT: &str = "http://localhost:4318";
/// Path the OTLP/HTTP spec puts the trace signal on.
const SIGNAL_PATH: &str = "v1/traces";
/// Applied to stdout when `RUST_LOG` is unset. The chatty crates stay quiet so
/// that what remains is worth reading.
const DEFAULT_CONSOLE_FILTER: &str = "info,actix_server=warn,sqlx=warn,sea_orm=warn,frogue=debug";
/// Applied to the exporter when [`SPANS_FILTER_VAR`] is unset.
///
/// `debug` rather than the console's `info`: the spans that make a trace worth
/// reading - the database round trips, the password hashing - sit at `debug`
/// so they do not clutter the terminal, and the exporter has a filter of its
/// own precisely so that turning the console down does not silence them.
const DEFAULT_SPANS_FILTER: &str = "sea_orm=info,html5ever=warn,debug";
/// Reports a problem with the telemetry setup itself.
///
/// This module runs before the subscriber exists and after it is being torn
/// down, so these notes go straight to stderr rather than through `tracing`,
/// which is also where an operator will be looking for them.
fn note(message: impl std::fmt::Display) {
eprintln!("frogue telemetry: {message}");
}
/// Installs the global subscriber and returns the handle that owns the pipeline.
///
/// The returned [`Telemetry`] must outlive the server: dropping it shuts the
/// tracer provider down, which is what flushes the spans still sitting in the
/// batch queue.
#[must_use = "telemetry is torn down as soon as the guard is dropped"]
pub fn init() -> Telemetry {
let console = tracing_subscriber::fmt::layer()
.with_target(true)
.with_writer(std::io::stderr)
.with_filter(filter_from_env("RUST_LOG", DEFAULT_CONSOLE_FILTER));
// Resolved before the subscriber goes up, so that a bad configuration can
// only be reported on stderr.
let otlp = OtlpExporter::from_env();
let spans_filter = filter_from_env(SPANS_FILTER_VAR, DEFAULT_SPANS_FILTER);
// The two branches cannot be collapsed behind an `Option`: the filter bound
// to the OpenTelemetry layer is the subscriber it ends up on, and that type
// is only known once the layer has been placed in the stack.
match &otlp {
Some(exporter) => tracing_subscriber::registry()
.with(console)
.with(OpenTelemetryLayer::new(exporter.tracer()).with_filter(spans_filter))
.init(),
None => tracing_subscriber::registry().with(console).init(),
}
match &otlp {
Some(exporter) => info!(endpoint = %exporter.endpoint, "exporting spans over OTLP"),
None => info!("span export disabled, spans are only written to the console"),
}
Telemetry {
_otlp: otlp,
request_logger: TracingLogger::new(),
}
}
/// The process wide telemetry pipeline.
///
/// Kept alive by `main` so that spans recorded while the server is shutting
/// down still have somewhere to go.
pub struct Telemetry {
/// Held for its `Drop`, which drains the exporter. `None` when export is off.
_otlp: Option<OtlpExporter>,
request_logger: TracingLogger<ErrorAwareRootSpanBuilder>,
}
impl Telemetry {
/// Middleware opening a span per incoming request, closing it once the
/// response body has been streamed out.
///
/// Cheap to [`Clone`], so hand out one per actix worker.
#[must_use]
pub fn request_logger(&self) -> TracingLogger<ErrorAwareRootSpanBuilder> {
self.request_logger.clone()
}
}
/// The root span per request, like [`tracing_actix_web::DefaultRootSpanBuilder`]
/// but with one correction: a `5xx` built by a handler is an error.
///
/// The default only considers a response failed when it carries an attached
/// error, so a handler that answers a bare `HttpResponse::InternalServerError()`
/// (which is how every failure path in this crate reports a database problem)
/// would be recorded as `otel.status_code=OK`. Marking the status from the
/// response code makes those traces come out red in Jaeger, where the operator
/// looks for them.
pub struct ErrorAwareRootSpanBuilder;
impl RootSpanBuilder for ErrorAwareRootSpanBuilder {
fn on_request_start(request: &ServiceRequest) -> Span {
root_span!(level = tracing::Level::INFO, request)
}
fn on_request_end<B: MessageBody>(span: Span, outcome: &Result<ServiceResponse<B>, Error>) {
match outcome {
Ok(response) => {
if let Some(error) = response.response().error() {
record_failure(span, response.status(), error.as_response_error());
} else {
let code: i32 = response.response().status().as_u16().into();
span.record("http.status_code", code);
if response.response().status().is_server_error() {
span.record("otel.status_code", "ERROR");
} else {
span.record("otel.status_code", "OK");
}
}
}
Err(error) => {
let response_error = error.as_response_error();
record_failure(span, response_error.status_code(), response_error);
}
}
}
}
/// Records the details of a failed request on its root span.
///
/// Copied from the default builder, which keeps the function private; the two
/// have to agree on the field names for a collector to make sense of them.
fn record_failure(span: Span, status_code: StatusCode, response_error: &dyn ResponseError) {
// Pre-formatting the errors is a workaround for
// https://github.com/tokio-rs/tracing/issues/1565.
let message = format!("{response_error}");
let details = format!("{response_error:?}");
span.record("exception.message", tracing::field::display(message));
span.record("exception.details", tracing::field::display(details));
let code: i32 = status_code.as_u16().into();
span.record("http.status_code", code);
// A client error is the caller's mistake and leaves the span green; an
// error the server is responsible for is what the operator needs to see.
if status_code.is_client_error() {
span.record("otel.status_code", "OK");
} else {
span.record("otel.status_code", "ERROR");
}
}
/// OTLP/HTTP span pipeline.
struct OtlpExporter {
/// The provider the layer records into. Dropping it flushes and closes the
/// HTTP client, so it must not be dropped while the layer is still live.
provider: SdkTracerProvider,
/// Resolved collector address, reported at startup.
endpoint: String,
}
impl OtlpExporter {
/// Builds the pipeline, or returns `None` when export is not wanted.
fn from_env() -> Option<Self> {
if is_truthy(SDK_DISABLED_VAR) {
debug!(
var = SDK_DISABLED_VAR,
"span export disabled by the environment"
);
return None;
}
if !lists_exporter(TRACES_EXPORTER_VAR, "otlp") {
return None;
}
let endpoint = endpoint();
// Exporting needs a tokio reactor to make progress, and the batch
// processor drains batches on a bare OS thread that has none -- so the
// exporter gets wrapped in a runtime of our own. `ExportOnRuntime`
// documents the details.
let runtime = match Builder::new_current_thread().enable_all().build() {
Ok(runtime) => runtime,
Err(err) => {
note(format!(
"could not start the span export runtime, carrying on without it: {err}"
));
return None;
}
};
let exporter = {
// Enter the runtime while the client is built, so that anything it
// captures here refers to the runtime that will later drive it.
let _guard = runtime.enter();
match SpanExporter::builder()
.with_http()
.with_endpoint(&endpoint)
.build()
{
Ok(exporter) => exporter,
Err(err) => {
note(format!(
"could not build the OTLP exporter, carrying on without it: {err}"
));
return None;
}
}
};
let processor = BatchSpanProcessor::builder(ExportOnRuntime { exporter, runtime }).build();
let provider = SdkTracerProvider::builder()
.with_sampler(sampler())
.with_resource(resource())
.with_span_processor(processor)
.build();
// Also publishes the provider globally, so libraries instrumented
// outside of `tracing` can pick it up.
global::set_tracer_provider(provider.clone());
Some(Self { provider, endpoint })
}
/// The tracer the OpenTelemetry layer records through.
///
/// Named after the crate so that a collector receiving spans from several
/// services can tell Frogue apart from the libraries it depends on.
fn tracer(&self) -> SdkTracer {
self.provider.tracer(env!("CARGO_PKG_NAME"))
}
}
impl Drop for OtlpExporter {
fn drop(&mut self) {
// `shutdown` drains the batch queue before closing the channel, so
// simply dropping the provider would lose whatever is still pending.
if let Err(err) = self.provider.shutdown() {
note(format!("could not shut the tracer provider down: {err}"));
}
}
}
/// Gives the batch processor an exporter it can drive.
///
/// The stable [`BatchSpanProcessor`] queues and batches spans on a thread of
/// its own, then awaits each export with `futures_executor::block_on`. That
/// drives the future but supplies no reactor, and an asynchronous HTTP client
/// needs one, so the request would sit there until the export timed out.
///
/// Wrapping the exporter sidesteps that: every batch is handed to a runtime
/// owned here and awaited on the calling thread. The processor only ever calls
/// `export` from its own plain thread, never from a runtime thread, so
/// [`Runtime::block_on`] -- which panics on a runtime thread -- is exactly as
/// safe to call as the blocking wait it replaces.
#[derive(Debug)]
struct ExportOnRuntime {
exporter: SpanExporter,
/// A current-thread runtime: `block_on` drives it from the waiting thread,
/// so it needs no worker threads of its own.
runtime: Runtime,
}
impl opentelemetry_sdk::trace::SpanExporter for ExportOnRuntime {
async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
self.runtime.block_on(self.exporter.export(batch))
}
fn force_flush(&self) -> OTelSdkResult {
self.exporter.force_flush()
}
fn shutdown(&self) -> OTelSdkResult {
self.exporter.shutdown()
}
fn set_resource(&mut self, resource: &Resource) {
self.exporter.set_resource(resource);
}
}
/// Describes the process to the collector.
fn resource() -> Resource {
let service_name =
non_empty(SERVICE_NAME_VAR).unwrap_or_else(|| env!("CARGO_PKG_NAME").to_owned());
// `Resource::builder` already folds in `OTEL_RESOURCE_ATTRIBUTES` and
// `OTEL_SERVICE_NAME`; repeating the name here only supplies the fallback
// for when neither is set, as the merge order lets the later value win.
Resource::builder()
.with_service_name(service_name)
.with_attributes([KeyValue::new(
"service.version",
Value::from(env!("CARGO_PKG_VERSION")),
)])
.build()
}
/// Resolves [`SAMPLER_VAR`] and [`SAMPLER_ARG_VAR`] as the OpenTelemetry SDK
/// spec describes, falling back to `parentbased_always_on`.
fn sampler() -> Sampler {
let requested = env::var(SAMPLER_VAR).unwrap_or_default();
let requested = requested.trim();
// `parentbased_*` keeps whatever decision the incoming span already made
// and only defers to the inner sampler for root spans.
let (name, parent_based) = match requested.strip_prefix("parentbased_") {
Some(inner) => (inner, true),
// An unset variable means the spec default, which is parent based.
None if requested.is_empty() => ("always_on", true),
None => (requested, false),
};
let sampler = match name {
"always_on" => Sampler::AlwaysOn,
"always_off" => Sampler::AlwaysOff,
"traceidratio" => ratio()
.map(Sampler::TraceIdRatioBased)
.unwrap_or_else(missing_ratio),
unsupported => {
if !unsupported.is_empty() {
note(format!(
"unsupported {SAMPLER_VAR}={unsupported:?}, keeping every trace"
));
}
Sampler::AlwaysOn
}
};
if parent_based {
Sampler::ParentBased(Box::new(sampler))
} else {
sampler
}
}
/// The fraction of traces to keep, as given by [`SAMPLER_ARG_VAR`].
fn ratio() -> Option<f64> {
non_empty(SAMPLER_ARG_VAR)?.parse().ok()
}
/// Stands in for a ratio sampler that was asked for without a usable rate.
fn missing_ratio() -> Sampler {
note(format!(
"{SAMPLER_ARG_VAR} is missing or not a number, keeping every trace"
));
Sampler::AlwaysOn
}
/// Resolves the collector address.
///
/// Follows the OTLP spec for the HTTP transport: the signal specific variable
/// is a complete URL and is used verbatim, while the general one is a base URL
/// that gains the signal path.
fn endpoint() -> String {
if let Some(endpoint) = non_empty(TRACES_ENDPOINT_VAR) {
return endpoint;
}
let base = non_empty(ENDPOINT_VAR).unwrap_or_else(|| DEFAULT_ENDPOINT.to_string());
if base.ends_with('/') {
format!("{base}{SIGNAL_PATH}")
} else {
format!("{base}/{SIGNAL_PATH}")
}
}
/// Builds a filter from `var`, falling back to `default` when it is unset or
/// malformed so that a typo never silences the process entirely.
fn filter_from_env(var: &str, default: &str) -> EnvFilter {
let Some(directives) = non_empty(var) else {
return EnvFilter::new(default);
};
match EnvFilter::builder().parse(directives.as_str()) {
Ok(filter) => filter,
Err(err) => {
note(format!(
"could not parse {var}={directives:?}, using {default:?}: {err}"
));
EnvFilter::new(default)
}
}
}
/// Whether `var` holds a comma separated list naming `exporter`.
fn lists_exporter(var: &str, exporter: &str) -> bool {
non_empty(var).is_some_and(|list| {
list.split(',')
.any(|entry| entry.trim().eq_ignore_ascii_case(exporter))
})
}
/// Whether `var` holds one of the spellings the spec accepts for `true`.
fn is_truthy(var: &str) -> bool {
non_empty(var).is_some_and(|value| value.eq_ignore_ascii_case("true") || value == "1")
}
/// Reads `var`, treating unset, empty and blank as absent.
fn non_empty(var: &str) -> Option<String> {
env::var(var)
.ok()
.map(|value| value.trim().to_owned())
.filter(|value| !value.is_empty())
}