brokkr-utils::telemetry

Rust

Brokkr Telemetry Module

This module provides OpenTelemetry-based distributed tracing for Brokkr components.

Features

• OTLP export to any OpenTelemetry-compatible collector
• Configurable sampling rate
• Integration with the tracing crate for instrumentation

Usage

use brokkr_utils::telemetry;
use brokkr_utils::config::ResolvedTelemetry;

let config = ResolvedTelemetry {
    enabled: true,
    otlp_endpoint: "http://localhost:4317".to_string(),
    service_name: "brokkr-broker".to_string(),
    sampling_rate: 0.1,
};

telemetry::init(&config)?;

// Use tracing macros for instrumentation
tracing::info!("Application started");

Enums

enum TelemetryError

pub

Error type for telemetry initialization

Variants

• ExporterError - Failed to create OTLP exporter
• TracerError - Failed to initialize tracer
• SubscriberError - Failed to set global subscriber

Functions

fn set_​log_​level

pub

fn set_log_level(level: &str) -> Result<bool, String>

Set the live log filter to level (for example “debug”).

Returns Ok(false) and changes nothing when there is no live filter to change: init did not run, or RUST_LOG set the filter.

Source

pub fn set_log_level(level: &str) -> Result<bool, String> {
    match LOG_FILTER.get() {
        Some(handle) => apply_log_level(handle, level).map(|()| true),
        None => Ok(false),
    }
}

fn init

pub

fn init(
    config: &ResolvedTelemetry,
    log_level: &str,
    log_format: &str,
) -> Result<(), TelemetryError>

Initialize OpenTelemetry tracing with the given configuration.

If telemetry is disabled in the config, this function sets up a basic tracing subscriber without OpenTelemetry export.

Parameters:

Name	Type	Description
config	&ResolvedTelemetry	Resolved telemetry configuration (from Telemetry::for_broker() or for_agent())
log_level	&str	Log level filter string (e.g., “info”, “debug”)
log_format	&str	Log format (“text” or “json”)

Returns:

• Ok(()) on success
• Err(TelemetryError) if initialization fails

Source

pub fn init(
    config: &ResolvedTelemetry,
    log_level: &str,
    log_format: &str,
) -> Result<(), TelemetryError> {
    let (env_filter, handle) = reloadable_filter(log_level);
    if let Some(handle) = handle {
        // A second `init` in one process fails at `try_init` below, so the
        // first handle is the right one to keep.
        let _ = LOG_FILTER.set(handle);
    }

    if !config.enabled {
        // Telemetry disabled - just set up basic tracing subscriber
        let subscriber = tracing_subscriber::registry().with(env_filter);

        if log_format.eq_ignore_ascii_case("json") {
            subscriber
                .with(tracing_subscriber::fmt::layer().json())
                .try_init()
                .map_err(|e| TelemetryError::SubscriberError(e.to_string()))?;
        } else {
            subscriber
                .with(tracing_subscriber::fmt::layer())
                .try_init()
                .map_err(|e| TelemetryError::SubscriberError(e.to_string()))?;
        }

        return Ok(());
    }

    // Create OTLP exporter
    let exporter = opentelemetry_otlp::SpanExporter::builder()
        .with_tonic()
        .with_endpoint(&config.otlp_endpoint)
        .build()
        .map_err(|e| TelemetryError::ExporterError(e.to_string()))?;

    // Create sampler based on sampling rate
    let sampler = if config.sampling_rate >= 1.0 {
        Sampler::AlwaysOn
    } else if config.sampling_rate <= 0.0 {
        Sampler::AlwaysOff
    } else {
        Sampler::TraceIdRatioBased(config.sampling_rate)
    };

    // Create tracer provider with resource attributes
    let tracer_provider = opentelemetry_sdk::trace::TracerProvider::builder()
        .with_batch_exporter(exporter, runtime::Tokio)
        .with_sampler(sampler)
        .with_resource(Resource::new(vec![
            KeyValue::new(
                opentelemetry_semantic_conventions::resource::SERVICE_NAME,
                config.service_name.clone(),
            ),
            KeyValue::new(
                opentelemetry_semantic_conventions::resource::SERVICE_VERSION,
                env!("CARGO_PKG_VERSION"),
            ),
        ]))
        .build();

    // Get tracer from provider
    let tracer = tracer_provider.tracer(config.service_name.clone());

    // Set global tracer provider
    opentelemetry::global::set_tracer_provider(tracer_provider);

    // Create OpenTelemetry tracing layer
    let otel_layer = tracing_opentelemetry::layer().with_tracer(tracer);

    // Build subscriber with OpenTelemetry layer
    let subscriber = tracing_subscriber::registry()
        .with(env_filter)
        .with(otel_layer);

    if log_format.eq_ignore_ascii_case("json") {
        subscriber
            .with(tracing_subscriber::fmt::layer().json())
            .try_init()
            .map_err(|e| TelemetryError::SubscriberError(e.to_string()))?;
    } else {
        subscriber
            .with(tracing_subscriber::fmt::layer())
            .try_init()
            .map_err(|e| TelemetryError::SubscriberError(e.to_string()))?;
    }

    Ok(())
}

fn shutdown

pub

fn shutdown()

Shutdown OpenTelemetry, flushing any pending traces.

Should be called during graceful shutdown to ensure all traces are exported.

Source

pub fn shutdown() {
    opentelemetry::global::shutdown_tracer_provider();
}