OpenTelemetry in Rust: From Zero to Production Observability

2026-03-12T10:00:01Z INFO  user created user_id=42
2026-03-12T10:00:02Z ERROR database connection failed err="connection refused"

use tracing::{info, warn, error};

info!("user created");
warn!(user_id = 42, "quota almost reached");   // structured key-value fields
error!(err = %e, "database connection failed");

[  handle_request          100ms  ]
  [  auth_check     5ms  ]
  [  db_query       80ms  ]
    [  query_plan   2ms  ]
    [  row_scan     75ms  ]
  [  serialize      3ms  ]

  Your Rust app
  ┌────────────────────────────────────┐
  │  tracing macros (info!, span!, …)  │
  │           │                        │
  │   tracing-subscriber registry      │
  │   ┌───────┴──────────────────────┐ │
  │   │ EnvFilter                    │ │
  │   │ fmt::layer (JSON → stdout)   │ │
  │   │ tracing-opentelemetry layer  │─┼──→ OTLP gRPC (traces)
  │   │ OTelTracingBridge layer      │─┼──→ OTLP gRPC (logs)
  │   └──────────────────────────────┘ │
  │                                    │
  │   global::meter() API              │─→ OTLP gRPC (metrics)
  └────────────────────────────────────┘
              │ :4317
  ┌───────────▼──────────────┐
  │   OTel Collector         │
  │   receivers: [otlp]      │
  │   processors: [batch]    │
  │   exporters:             │
  │     traces  → Jaeger     │
  │     metrics → Prometheus │
  │     logs    → Loki       │
  └──────────────────────────┘

[dependencies]
# Core OTel API (trace + metrics)
opentelemetry = { version = "0.31.0", features = ["trace", "metrics"] }

# OTLP exporter: gRPC transport (tonic), all three signals
opentelemetry-otlp = { version = "0.31.0", features = ["grpc-tonic", "trace", "metrics", "logs"] }

# SDK implementations of the three providers
opentelemetry_sdk = { version = "0.31.0", features = ["rt-tokio", "trace", "metrics"] }

# Bridges tracing log events into OTel log records
opentelemetry-appender-tracing = "0.31.1"

# The tracing subscriber ecosystem
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json"] }

# Converts tracing spans into OTel spans
tracing-opentelemetry = "0.32.1"

pub struct TelemetryConfig {
    pub level: LogLevel,       // fallback log level (overridden by RUST_LOG)
    pub endpoint: String,      // OTLP collector address, e.g. "http://localhost:4317"
    pub tracer_name: String,   // scopes spans from this service in the backend
    pub app_name: String,      // becomes service.name in all telemetry
}

fn resource(app_name: &str) -> Resource {
    use opentelemetry::KeyValue;

    Resource::builder()
        .with_attribute(KeyValue::new("service.name", app_name.to_owned()))
        .build()
}

let span_exporter = SpanExporter::builder()
    .with_tonic()               // gRPC transport via the tonic crate
    .with_endpoint(endpoint)    // "http://localhost:4317"
    .build()?;

let tracer_provider = SdkTracerProvider::builder()
    .with_resource(resource(&cfg.app_name))
    .with_batch_exporter(span_exporter)
    .build();

let tracer = tracer_provider.tracer(cfg.tracer_name.clone());

let metric_exporter = MetricExporter::builder()
    .with_tonic()
    .with_endpoint(endpoint)
    .build()?;

let meter_provider = SdkMeterProvider::builder()
    .with_resource(resource(&cfg.app_name))
    .with_reader(PeriodicReader::builder(metric_exporter).build())
    .build();

opentelemetry::global::set_meter_provider(meter_provider.clone());

let log_exporter = LogExporter::builder()
    .with_tonic()
    .with_endpoint(endpoint)
    .build()?;

let logger_provider = SdkLoggerProvider::builder()
    .with_resource(resource(&cfg.app_name))
    .with_batch_exporter(log_exporter)
    .build();

let env_filter =
    EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| EnvFilter::new(cfg.level.as_str()));

tracing_subscriber::registry()
    .with(env_filter)
    .with(tracing_subscriber::fmt::layer().json())
    .with(tracing_opentelemetry::layer().with_tracer(tracer))
    .with(
        opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge::new(
            &logger_provider,
        ),
    )
    .init();

pub struct Shutdown {
    tracer_provider: SdkTracerProvider,
    meter_provider: SdkMeterProvider,
    logger_provider: SdkLoggerProvider,
}

impl Drop for Shutdown {
    fn drop(&mut self) {
        let _ = self.tracer_provider.shutdown();
        let _ = self.meter_provider.shutdown();
        let _ = self.logger_provider.shutdown();
    }
}

let _shutdown = init_telemetry(&config.telemetry)?;
//  ^
//  The leading underscore suppresses "unused variable" warnings, but
//  crucially the binding is NOT dropped immediately (unlike `let _ = ...`
//  which would drop it at the end of the statement).

use tracing::{info, warn, error, debug};

info!("application started");
warn!(user_id = 42, "quota almost reached");
error!(err = %e, "payment failed");
debug!(payload = ?body, "received request"); // ? = Debug format

use tracing::instrument;

#[instrument]
async fn process_job(job_id: u64) {
    info!("processing started");
    do_work().await;
    info!("processing complete");
}

#[instrument(skip(password), fields(user = %user.id))]
async fn login(user: &User, password: &str) { … }

use tracing::{info_span, Instrument};

let span = info_span!("send_email", recipient = %email, template = "welcome");
send_email_inner(payload).instrument(span).await;

use opentelemetry::global;

// --- Counter (monotonically increasing, e.g. total requests processed) ---
let meter = global::meter("payment-service");
let counter = meter.u64_counter("payments.processed").build();
counter.add(1, &[]);

// --- Histogram (distribution of values, e.g. latency) ---
let histogram = meter.f64_histogram("payment.duration_ms").build();
histogram.record(latency_ms, &[]);

// --- With attributes (dimensions/labels for slicing in dashboards) ---
use opentelemetry::KeyValue;
counter.add(1, &[
    KeyValue::new("payment.method", "card"),
    KeyValue::new("payment.status", "success"),
]);

services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.123.0
    ports:
      - "4317:4317" # OTLP gRPC  ← your app sends here
      - "4318:4318" # OTLP HTTP
      - "8888:8888" # Collector self-metrics

  jaeger:
    image: jaegertracing/jaeger:2.6.0
    environment:
      COLLECTOR_OTLP_ENABLED: true
    ports:
      - "16686:16686" # Jaeger UI → http://localhost:16686

  postgres: …
  redis: …

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317

processors:
  batch:
    timeout: 1s
    send_batch_size: 1024

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317 # internal Docker network name
    tls:
      insecure: true
  debug:
    verbosity: normal

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [otlp/jaeger, debug]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [debug]
    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [debug]

make up      # starts docker-compose (Collector, Jaeger, Postgres, Redis)
make dev     # cargo run with CONFIG_PATH=configs/config.yaml

use opentelemetry_sdk::trace::BatchConfigBuilder;

let batch_config = BatchConfigBuilder::default()
    .with_max_queue_size(8192)
    .with_max_export_batch_size(1024)
    .build();

SdkTracerProvider::builder()
    .with_batch_exporter(span_exporter)
    // apply batch_config via the builder...
    .build();

RUST_LOG=rustapp=debug,sqlx=warn,tokio=error

let _shutdown = init_telemetry(&config.telemetry)?;  // CORRECT lives to end of main
let _ = init_telemetry(&config.telemetry)?;          // WRONG dropped immediately

info!("message")                   // plain log
info!(key = value, "message")      // structured log with fields
#[instrument]                      // auto-span the whole function
#[instrument(skip(secret))]        // auto-span, hide a field
info_span!("name", key = value)    // manual span
.instrument(span)                  // attach span to a future

meter.u64_counter("name").build()       // count up (requests, jobs)
meter.f64_histogram("name").build()     // distribution (latency, size)
meter.i64_up_down_counter("name").build() // goes up and down (queue depth)
meter.f64_gauge("name").build()         // current value (temperature, ratio)

// src/services/user_service.rs
use tracing::{debug, error, info, instrument, warn};
use uuid::Uuid;

pub struct UserService { /* … db pool, etc. */ }

impl UserService {
    /// Creates a new user.
    /// The #[instrument] macro wraps the whole function in a span named
    /// "create_user" and records `email` as a span attribute automatically.
    /// Every log inside is nested under that span in Jaeger.
    #[instrument(skip(self), fields(service = "user"))]
    pub async fn create_user(&self, email: &str) -> Result<Uuid, anyhow::Error> {
        // Plain info log shows up in stdout JSON and in the OTel log backend.
        info!(email, "creating user");

        // Structured fields come before the message string.
        // `%` means Display format, `?` means Debug format.
        debug!(email, attempt = 1, "querying db");

        let user_id = Uuid::now_v7();

        match self.db_insert(user_id, email).await {
            Ok(_) => {
                // Attach key-value context to the log line.
                info!(user_id = %user_id, email, "user created successfully");
                Ok(user_id)
            }
            Err(e) => {
                // `%e` formats the error with its Display impl.
                error!(err = %e, email, "failed to insert user");
                Err(e)
            }
        }
    }

    async fn db_insert(&self, _id: Uuid, _email: &str) -> Result<(), anyhow::Error> {
        Ok(()) // placeholder
    }
}

// src/services/payment_service.rs
use tracing::{error, info, instrument, warn};

pub struct PaymentService;

impl PaymentService {
    /// `skip(self, card_number)`: hides sensitive fields from the span.
    /// `fields(…)`: adds custom attributes that are not function arguments.
    #[instrument(
        skip(self, card_number),
        fields(
            payment.method = "card",
            payment.currency = %currency,
        )
    )]
    pub async fn charge(
        &self,
        user_id: u64,
        amount_cents: u64,
        currency: &str,
        card_number: &str,   // skipped never appears in telemetry
    ) -> Result<String, anyhow::Error> {
        info!(user_id, amount_cents, "initiating charge");

        // Simulate calling a downstream payment provider.
        let result = self.call_provider(amount_cents).await;

        match result {
            Ok(ref txn_id) => {
                // Attach the transaction ID to the current span's fields.
                tracing::Span::current().record("payment.txn_id", txn_id.as_str());
                info!(txn_id, "charge successful");
                result
            }
            Err(ref e) => {
                warn!(err = %e, user_id, "charge failed");
                result
            }
        }
    }

    async fn call_provider(&self, _amount: u64) -> Result<String, anyhow::Error> {
        Ok("txn_abc123".to_string()) // placeholder
    }
}

// src/jobs/email_job.rs
use tracing::{info, info_span, Instrument};

pub async fn send_welcome_emails(user_ids: Vec<u64>) {
    for user_id in user_ids {
        // Build a span with custom attributes, then attach it to the future.
        let span = info_span!(
            "send_welcome_email",
            user.id = user_id,
            email.template = "welcome_v2",
        );

        async move {
            info!(user_id, "sending email");
            // … actual send logic …
            info!(user_id, "email sent");
        }
        .instrument(span)
        .await;
    }
}

// src/handlers/checkout.rs
use opentelemetry::trace::Status;
use tracing::{error, instrument};
use tracing_opentelemetry::OpenTelemetrySpanExt; // needed for set_status

#[instrument]
pub async fn checkout_handler(order_id: u64) -> Result<(), anyhow::Error> {
    let result = process_order(order_id).await;

    if let Err(ref e) = result {
        // Mark the current span as ERROR so Jaeger colors it red.
        tracing::Span::current()
            .set_status(Status::error(e.to_string()));
        error!(err = %e, order_id, "checkout failed");
    }

    result
}

async fn process_order(_id: u64) -> Result<(), anyhow::Error> {
    Ok(())
}

// src/metrics.rs
use opentelemetry::{
    global,
    metrics::{Counter, Histogram, Meter, UpDownCounter},
    KeyValue,
};
use std::sync::OnceLock;

/// All metrics for the HTTP layer.
pub struct HttpMetrics {
    /// Total number of HTTP requests received.
    pub requests_total: Counter<u64>,

    /// Number of requests currently in-flight.
    pub requests_in_flight: UpDownCounter<i64>,

    /// Request duration in milliseconds (histogram = p50/p95/p99).
    pub request_duration_ms: Histogram<f64>,

    /// Response body size in bytes.
    pub response_bytes: Histogram<u64>,
}

impl HttpMetrics {
    pub fn new() -> Self {
        // "http-server" is the instrumentation scope visible in your backend
        // to distinguish metrics from different components.
        let meter: Meter = global::meter("http-server");

        Self {
            requests_total: meter
                .u64_counter("http.requests.total")
                .with_description("Total HTTP requests received")
                .build(),

            requests_in_flight: meter
                .i64_up_down_counter("http.requests.in_flight")
                .with_description("Requests currently being processed")
                .build(),

            request_duration_ms: meter
                .f64_histogram("http.request.duration_ms")
                .with_description("Request latency in milliseconds")
                // Optional: explicit histogram bucket boundaries.
                // Default buckets work for most cases.
                .build(),

            response_bytes: meter
                .u64_histogram("http.response.size_bytes")
                .with_description("Response body size in bytes")
                .build(),
        }
    }
}

// Global singleton initialised once after `init_telemetry()`.
static HTTP_METRICS: OnceLock<HttpMetrics> = OnceLock::new();

pub fn http_metrics() -> &'static HttpMetrics {
    HTTP_METRICS.get_or_init(HttpMetrics::new)
}

// src/handlers/orders.rs
use std::time::Instant;
use opentelemetry::KeyValue;
use crate::metrics::http_metrics;

pub async fn list_orders_handler(/* … */) {
    let m = http_metrics();

    // Dimensions (attributes) let you slice metrics in dashboards.
    // e.g. "show me requests grouped by method and route"
    let attrs = &[
        KeyValue::new("http.method", "GET"),
        KeyValue::new("http.route", "/orders"),
    ];

    // Track in-flight requests (goes up when request starts, down when done).
    m.requests_in_flight.add(1, attrs);
    let start = Instant::now();

    let result = fetch_orders_from_db().await;

    let elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;

    // Record latency in the histogram this feeds p50/p95/p99 in Grafana.
    m.request_duration_ms.record(elapsed_ms, attrs);

    // Increment total counter with a status dimension.
    let status = if result.is_ok() { "200" } else { "500" };
    m.requests_total.add(1, &[
        KeyValue::new("http.method", "GET"),
        KeyValue::new("http.route", "/orders"),
        KeyValue::new("http.status_code", status),
    ]);

    m.requests_in_flight.add(-1, attrs);
}

async fn fetch_orders_from_db() -> Result<Vec<()>, anyhow::Error> {
    Ok(vec![])
}

// src/main.rs
mod handlers;
mod metrics;
mod services;
mod utils;

use std::env;

use anyhow::Result;
use tracing::info;

use crate::utils::config::load_config;
use crate::utils::telemetry::init_telemetry;

#[tokio::main]
async fn main() -> Result<()> {
    let config_path =
        env::var("CONFIG_PATH").unwrap_or_else(|_| "configs/config.yaml".to_string());

    let config = load_config(&config_path)?;

    // IMPORTANT: `_shutdown` must stay alive until the end of main.
    // When it is dropped, all three providers flush their buffers.
    // `let _ = ...` would drop it immediately do NOT do that.
    let _shutdown = init_telemetry(&config.telemetry)?;

    // Warm up the metrics instruments now that the global meter is registered.
    // This ensures the OnceLock is populated before any request comes in.
    let _ = metrics::http_metrics();

    info!(config_path, "application started");

    // … start your axum server, spawn background tasks, etc. …

    Ok(())
    // _shutdown is dropped here → providers flush → process exits cleanly
}