Observable Instruments - OpenTelemetry Rust

Observable instruments (also called asynchronous instruments) report measurements via callbacks that are invoked during metric collection. They’re ideal for values that are expensive to compute, read from external sources, or already being tracked elsewhere.

When to Use Observable Instruments

Use observable instruments when:

Reading from system APIs or sensors
Values are expensive to compute
Measurements should happen on a schedule (not inline)
Multiple instruments need the same source data
You don’t control when collection happens

Common examples:

CPU time from OS APIs
Memory usage from system calls
Process statistics
Sensor readings
External resource metrics

Observable instruments use callbacks that run during metric collection. The SDK determines when to call your callback, typically during export intervals.

Available Observable Instruments

ObservableCounter

For monotonically increasing values:

pub struct ObservableCounter<T> {
    _marker: std::marker::PhantomData<T>,
}

Supports: u64, f64

ObservableUpDownCounter

For values that can increase or decrease:

pub struct ObservableUpDownCounter<T> {
    _marker: std::marker::PhantomData<T>,
}

Supports: i64, f64

ObservableGauge

For current/instantaneous values:

pub struct ObservableGauge<T> {
    _marker: std::marker::PhantomData<T>,
}

Supports: u64, i64, f64

AsyncInstrument Trait

All observable instruments implement the AsyncInstrument trait:

pub trait AsyncInstrument<T>: Send + Sync {
    fn observe(&self, measurement: T, attributes: &[KeyValue]);
}

The observe method is called within your callback to report measurements.

Callback Type

Callbacks are defined as:

pub type Callback<T> = Box<dyn Fn(&dyn AsyncInstrument<T>) + Send + Sync>;

Your callback receives an observer that implements AsyncInstrument<T>, which you use to report measurements.

Creating Observable Instruments

Observable instruments are created using the builder pattern with with_callback:

ObservableCounter

use opentelemetry::{global, KeyValue};

let meter = global::meter("my-app");

// u64 observable counter
let _observable_counter = meter
    .u64_observable_counter("cpu_time")
    .with_description("CPU time used by the process")
    .with_unit("ms")
    .with_callback(|observer| {
        let cpu_time = get_process_cpu_time();  // Your function
        observer.observe(cpu_time, &[]);
    })
    .build();

// f64 observable counter
let _observable_counter = meter
    .f64_observable_counter("bytes_processed")
    .with_callback(|observer| {
        let bytes = get_total_bytes_processed();
        observer.observe(bytes, &[]);
    })
    .build();

ObservableUpDownCounter

let _observable_updown = meter
    .i64_observable_up_down_counter("active_connections")
    .with_description("Number of active connections")
    .with_callback(|observer| {
        let count = get_active_connection_count();
        observer.observe(count, &[]);
    })
    .build();

ObservableGauge

let _observable_gauge = meter
    .f64_observable_gauge("cpu_usage")
    .with_description("Current CPU usage percentage")
    .with_unit("%")
    .with_callback(|observer| {
        let usage = get_current_cpu_usage();
        observer.observe(usage, &[KeyValue::new("core", "0")]);
    })
    .build();

Complete Example

use opentelemetry::{global, KeyValue};
use opentelemetry_sdk::metrics::SdkMeterProvider;
use opentelemetry_sdk::Resource;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

// Simulated system state
struct SystemStats {
    cpu_time: AtomicU64,
    memory_used: AtomicU64,
    connections: AtomicU64,
}

impl SystemStats {
    fn new() -> Self {
        SystemStats {
            cpu_time: AtomicU64::new(0),
            memory_used: AtomicU64::new(0),
            connections: AtomicU64::new(0),
        }
    }
}

fn init_meter_provider() -> SdkMeterProvider {
    let exporter = opentelemetry_stdout::MetricExporterBuilder::default().build();
    let provider = SdkMeterProvider::builder()
        .with_periodic_exporter(exporter)
        .with_resource(
            Resource::builder()
                .with_service_name("observable-example")
                .build(),
        )
        .build();
    global::set_meter_provider(provider.clone());
    provider
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let meter_provider = init_meter_provider();
    let meter = global::meter("system-monitor");
    
    let stats = Arc::new(SystemStats::new());

    // Observable counter for cumulative CPU time
    let stats_clone = stats.clone();
    let _cpu_time = meter
        .u64_observable_counter("process_cpu_time")
        .with_description("Total CPU time used")
        .with_unit("ms")
        .with_callback(move |observer| {
            let time = stats_clone.cpu_time.load(Ordering::Relaxed);
            observer.observe(time, &[]);
        })
        .build();

    // Observable gauge for current memory usage
    let stats_clone = stats.clone();
    let _memory = meter
        .u64_observable_gauge("process_memory_usage")
        .with_description("Current memory usage")
        .with_unit("By")
        .with_callback(move |observer| {
            let memory = stats_clone.memory_used.load(Ordering::Relaxed);
            observer.observe(memory, &[]);
        })
        .build();

    // Observable gauge for active connections
    let stats_clone = stats.clone();
    let _connections = meter
        .u64_observable_gauge("active_connections")
        .with_callback(move |observer| {
            let count = stats_clone.connections.load(Ordering::Relaxed);
            observer.observe(count, &[]);
        })
        .build();

    // Simulate system activity
    stats.cpu_time.store(12500, Ordering::Relaxed);
    stats.memory_used.store(1024 * 1024 * 512, Ordering::Relaxed);
    stats.connections.store(42, Ordering::Relaxed);

    // Metrics will be collected automatically during export
    meter_provider.shutdown()?;
    Ok(())
}

Example from Source

From the metrics-basic example:

// Create a ObservableCounter instrument and register a callback
let _observable_counter = meter
    .u64_observable_counter("my_observable_counter")
    .with_description("My observable counter example description")
    .with_unit("myunit")
    .with_callback(|observer| {
        observer.observe(
            100,
            &[
                KeyValue::new("mykey1", "myvalue1"),
                KeyValue::new("mykey2", "myvalue2"),
            ],
        )
    })
    .build();

Multiple Observations in One Callback

A single callback can report multiple measurements with different attributes:

let _cpu_usage = meter
    .f64_observable_gauge("cpu_usage_per_core")
    .with_callback(|observer| {
        // Report usage for each CPU core
        observer.observe(45.2, &[KeyValue::new("core", "0")]);
        observer.observe(52.1, &[KeyValue::new("core", "1")]);
        observer.observe(38.7, &[KeyValue::new("core", "2")]);
        observer.observe(61.3, &[KeyValue::new("core", "3")]);
    })
    .build();

Multiple Callbacks

You can register multiple callbacks for the same instrument:

let _gauge = meter
    .f64_observable_gauge("temperature")
    .with_callback(|observer| {
        observer.observe(22.5, &[KeyValue::new("sensor", "cpu")]);
    })
    .with_callback(|observer| {
        observer.observe(35.0, &[KeyValue::new("sensor", "gpu")]);
    })
    .build();

Capturing State in Callbacks

Use closures to capture state:

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

let request_count = Arc::new(AtomicU64::new(0));

// Increment in your application code
let counter_clone = request_count.clone();
request_count.fetch_add(1, Ordering::Relaxed);

// Report in observable instrument
let _observable = meter
    .u64_observable_counter("requests_total")
    .with_callback(move |observer| {
        let count = counter_clone.load(Ordering::Relaxed);
        observer.observe(count, &[]);
    })
    .build();

Synchronous vs. Observable

Choose the right approach for your use case:

// Record as events happen
let counter = meter.u64_counter("requests").build();

fn handle_request() {
    counter.add(1, &[KeyValue::new("endpoint", "/api")]);
    // Process request...
}

Instrument Type Comparison

Instrument	Synchronous	Observable	Use Case
Counter	`Counter`	`ObservableCounter`	Monotonically increasing
UpDownCounter	`UpDownCounter`	`ObservableUpDownCounter`	Can increase or decrease
Gauge	`Gauge`	`ObservableGauge`	Current/instantaneous value
Histogram	`Histogram`	(none)	Value distributions

There is no ObservableHistogram. Histograms are only available as synchronous instruments.

Callback Best Practices

Keep Callbacks Fast

Callbacks should complete quickly:

// Good - fast operation
let _gauge = meter
    .u64_observable_gauge("cache_size")
    .with_callback(|observer| {
        let size = CACHE.len();  // O(1) operation
        observer.observe(size as u64, &[]);
    })
    .build();

// Bad - slow operation
let _gauge = meter
    .u64_observable_gauge("expensive_metric")
    .with_callback(|observer| {
        let value = expensive_computation();  // Avoid!
        observer.observe(value, &[]);
    })
    .build();

Avoid Blocking I/O

Don’t perform blocking operations in callbacks:

// Bad - blocking I/O
let _gauge = meter
    .u64_observable_gauge("file_size")
    .with_callback(|observer| {
        let size = std::fs::metadata("file.txt").unwrap().len();  // Avoid!
        observer.observe(size, &[]);
    })
    .build();

// Good - pre-computed value
let file_size = Arc::new(AtomicU64::new(0));

// Update in background
let size_clone = file_size.clone();
std::thread::spawn(move || {
    loop {
        if let Ok(metadata) = std::fs::metadata("file.txt") {
            size_clone.store(metadata.len(), Ordering::Relaxed);
        }
        std::thread::sleep(Duration::from_secs(60));
    }
});

// Report in callback
let _gauge = meter
    .u64_observable_gauge("file_size")
    .with_callback(move |observer| {
        observer.observe(file_size.load(Ordering::Relaxed), &[]);
    })
    .build();

Handle Errors Gracefully

let _gauge = meter
    .f64_observable_gauge("temperature")
    .with_callback(|observer| {
        match read_temperature_sensor() {
            Ok(temp) => observer.observe(temp, &[]),
            Err(_) => {}
            // Don't observe on error, or use a default value
        }
    })
    .build();

Instrument Lifecycle

Keep observable instruments in scope! The callback will not be invoked if the instrument is dropped:

// Bad - instrument dropped immediately
meter.u64_observable_counter("my_counter")
    .with_callback(|observer| observer.observe(100, &[]))
    .build();
// Callback will never be called!

// Good - keep instrument in scope
let _my_counter = meter
    .u64_observable_counter("my_counter")
    .with_callback(|observer| observer.observe(100, &[]))
    .build();
// Callback will be called during collection

Common Patterns

System Metrics

let _cpu = meter
    .f64_observable_gauge("system_cpu_usage")
    .with_unit("%")
    .with_callback(|observer| {
        if let Ok(usage) = get_cpu_usage() {
            observer.observe(usage, &[]);
        }
    })
    .build();

let _memory = meter
    .u64_observable_gauge("system_memory_used")
    .with_unit("By")
    .with_callback(|observer| {
        if let Ok(used) = get_memory_used() {
            observer.observe(used, &[]);
        }
    })
    .build();

Process Statistics

let _uptime = meter
    .u64_observable_counter("process_uptime")
    .with_unit("s")
    .with_callback(|observer| {
        let uptime = get_process_uptime_seconds();
        observer.observe(uptime, &[]);
    })
    .build();

Next Steps

Counter

Learn about synchronous Counter instruments

Gauge

Learn about synchronous Gauge instruments

Meters

Create and configure Meters

Views

Customize metric aggregation

Documentation Index

​When to Use Observable Instruments

​Available Observable Instruments

​ObservableCounter

​ObservableUpDownCounter

​ObservableGauge

​AsyncInstrument Trait

​Callback Type

​Creating Observable Instruments

​ObservableCounter

​ObservableUpDownCounter

​ObservableGauge

​Complete Example

​Example from Source

​Multiple Observations in One Callback

​Multiple Callbacks

​Capturing State in Callbacks

​Synchronous vs. Observable

​Instrument Type Comparison

​Callback Best Practices

​Keep Callbacks Fast

​Avoid Blocking I/O

​Handle Errors Gracefully

​Instrument Lifecycle

​Common Patterns

​System Metrics

​Process Statistics

​Next Steps

Counter

Gauge

Meters

Views

When to Use Observable Instruments

Available Observable Instruments

ObservableCounter

ObservableUpDownCounter

ObservableGauge

AsyncInstrument Trait

Callback Type

Creating Observable Instruments

ObservableCounter

ObservableUpDownCounter

ObservableGauge

Complete Example

Example from Source

Multiple Observations in One Callback

Multiple Callbacks

Capturing State in Callbacks

Synchronous vs. Observable

Instrument Type Comparison

Callback Best Practices

Keep Callbacks Fast

Avoid Blocking I/O

Handle Errors Gracefully

Instrument Lifecycle

Common Patterns

System Metrics

Process Statistics

Next Steps