Telemetry

OpenTelemetry integration for distributed tracing, metrics, and logging

This API requires the otel feature flag. Add features = ["otel"] to your Cargo.toml:

[dependencies]
iii-sdk = { version = "0.4.1", features = ["otel"] }

Overview

The III SDK includes comprehensive OpenTelemetry support for distributed tracing, metrics collection, and structured logging. When the otel feature is enabled, telemetry data is exported to the III Engine over a shared WebSocket connection.

Initialization

init_otel

Initialize OpenTelemetry with the given configuration.

pub async fn init_otel(config: OtelConfig)

config · OtelConfig · required

Configuration for OpenTelemetry including service name, metrics settings, etc.

Example:

use iii_sdk::{III, OtelConfig, init_otel};

let iii = III::new("ws://localhost:49134");

let otel_config = OtelConfig {
    enabled: Some(true),
    service_name: Some("my-service".to_string()),
    service_version: Some("1.0.0".to_string()),
    service_namespace: Some("production".to_string()),
    engine_ws_url: Some("ws://localhost:49134".to_string()),
    metrics_enabled: Some(true),
    metrics_export_interval_ms: Some(60_000),
    logs_enabled: Some(true),
    shutdown_timeout_ms: Some(10_000),
    ..Default::default()
};

iii.set_otel_config(otel_config);
iii.connect().await?; // Automatically initializes OTel

When using III::set_otel_config() and connect(), OpenTelemetry is automatically initialized. You only need to call init_otel() directly if you’re not using the III client.

shutdown_otel

Shutdown OpenTelemetry and flush all pending data.

pub async fn shutdown_otel()

Example:

use iii_sdk::shutdown_otel;

// At application shutdown
shutdown_otel().await;

Always call shutdown_otel() or iii.shutdown_async() before your application exits to ensure all telemetry data is flushed.

flush_otel

Flush all pending telemetry data without shutting down.

pub async fn flush_otel()

Example:

use iii_sdk::flush_otel;

// Periodically flush telemetry
flush_otel().await;

is_initialized

Check if OpenTelemetry has been initialized.

pub fn is_initialized() -> bool

Example:

use iii_sdk::is_initialized;

if is_initialized() {
    println!("OpenTelemetry is active");
}

Configuration

OtelConfig

Configuration structure for OpenTelemetry.

pub struct OtelConfig {
    pub enabled: Option<bool>,
    pub service_name: Option<String>,
    pub service_version: Option<String>,
    pub service_namespace: Option<String>,
    pub service_instance_id: Option<String>,
    pub engine_ws_url: Option<String>,
    pub metrics_enabled: Option<bool>,
    pub metrics_export_interval_ms: Option<u64>,
    pub logs_enabled: Option<bool>,
    pub reconnection_config: Option<ReconnectionConfig>,
    pub shutdown_timeout_ms: Option<u64>,
    pub channel_capacity: Option<usize>,
    pub fetch_instrumentation_enabled: Option<bool>,
}

Field Defaults:

• enabled: true (can be overridden by OTEL_ENABLED env var)
• service_name: "iii-rust-sdk" (can be overridden by OTEL_SERVICE_NAME env var)
• service_version: SDK version from Cargo.toml
• service_instance_id: Random UUID
• engine_ws_url: III client address or ws://localhost:49134
• metrics_enabled: true
• metrics_export_interval_ms: 60000 (1 minute)
• logs_enabled: true
• shutdown_timeout_ms: 10000 (10 seconds)
• channel_capacity: 10000
• fetch_instrumentation_enabled: true

ReconnectionConfig

Configuration for WebSocket reconnection behavior.

pub struct ReconnectionConfig {
    pub initial_delay_ms: u64,
    pub max_delay_ms: u64,
    pub backoff_multiplier: f64,
    pub jitter_factor: f64,
    pub max_retries: Option<u64>,
    pub max_pending_messages: usize,
}

Defaults:

• initial_delay_ms: 1000
• max_delay_ms: 30000
• backoff_multiplier: 2.0
• jitter_factor: 0.3
• max_retries: None (infinite)
• max_pending_messages: 1000

Distributed Tracing

get_tracer

Get a tracer for creating spans manually.

pub fn get_tracer() -> opentelemetry::global::BoxedTracer

Example:

use iii_sdk::get_tracer;
use opentelemetry::trace::{Tracer, SpanKind};

let tracer = get_tracer();
let span = tracer
    .span_builder("my_operation")
    .with_kind(SpanKind::Internal)
    .start(&tracer);

// Do work

span.end();

with_span

Execute a function within a traced span with automatic error handling.

pub async fn with_span<F, Fut, T>(
    name: &str,
    traceparent: Option<&str>,
    kind: Option<SpanKind>,
    f: F,
) -> Result<T, Box<dyn std::error::Error + Send + Sync>>

name · &str · required

Name of the span

traceparent · Option<&str>

W3C traceparent header to set parent context

kind · Option<SpanKind>

Span kind (defaults to Internal)

f · F · required

Async function to execute within the span

Example:

use iii_sdk::{with_span, SpanKind};

let result = with_span(
    "process_order",
    None,
    Some(SpanKind::Internal),
    || async {
        // Your code here
        Ok("processed")
    }
).await?;

Automatic Trace Propagation

Trace context is automatically propagated across function calls:

use iii_sdk::{III, with_span, SpanKind};
use serde_json::json;

let iii = III::new("ws://localhost:49134");
iii.connect().await?;

// Parent span
let result = with_span(
    "handle_request",
    None,
    Some(SpanKind::Server),
    || async {
        // This call will be a child span
        let user = iii.call("user.get", json!({ "id": "123" })).await?;
        
        // This call will also be a child span
        let order = iii.call("order.create", json!({ "user": user })).await?;
        
        Ok(order)
    }
).await?;

Trace Context Functions

current_trace_id

Get the current trace ID.

pub fn current_trace_id() -> Option<String>

current_span_id

Get the current span ID.

pub fn current_span_id() -> Option<String>

Example:

use iii_sdk::{current_trace_id, current_span_id};

if let Some(trace_id) = current_trace_id() {
    println!("Trace ID: {}", trace_id);
}

if let Some(span_id) = current_span_id() {
    println!("Span ID: {}", span_id);
}

inject_traceparent

Inject current trace context into a W3C traceparent header.

pub fn inject_traceparent() -> Option<String>

extract_traceparent

Extract trace context from a W3C traceparent header.

pub fn extract_traceparent(traceparent: &str) -> OtelContext

Example:

use iii_sdk::{inject_traceparent, extract_traceparent};

// Inject for outbound request
if let Some(traceparent) = inject_traceparent() {
    // Add to HTTP headers
    headers.insert("traceparent", traceparent);
}

// Extract from inbound request
let traceparent = headers.get("traceparent").unwrap();
let context = extract_traceparent(traceparent);

Baggage

Baggage allows you to propagate key-value pairs across service boundaries.

set_baggage_entry

Set a baggage entry.

pub fn set_baggage_entry(key: &str, value: &str) -> OtelContext

get_baggage_entry

Get a baggage entry.

pub fn get_baggage_entry(key: &str) -> Option<String>

get_all_baggage

Get all baggage entries.

pub fn get_all_baggage() -> HashMap<String, String>

Example:

use iii_sdk::{set_baggage_entry, get_baggage_entry, get_all_baggage};

// Set baggage
let cx = set_baggage_entry("user_id", "123");
let _guard = cx.attach();

// Get baggage
if let Some(user_id) = get_baggage_entry("user_id") {
    println!("User ID: {}", user_id);
}

// Get all
let all = get_all_baggage();
for (key, value) in all {
    println!("{}: {}", key, value);
}

Metrics

get_meter

Get a meter for creating metrics.

pub fn get_meter() -> opentelemetry::metrics::Meter

Example:

use iii_sdk::get_meter;

let meter = get_meter();

// Create a counter
let counter = meter.u64_counter("requests_total")
    .with_description("Total number of requests")
    .init();

counter.add(1, &[]);

// Create a histogram
let histogram = meter.f64_histogram("request_duration_seconds")
    .with_description("Request duration in seconds")
    .init();

let start = std::time::Instant::now();
// ... do work ...
let duration = start.elapsed().as_secs_f64();
histogram.record(duration, &[]);

Metric Types

OpenTelemetry provides several metric types:

Counter:

let counter = get_meter().u64_counter("operation_count").init();
counter.add(1, &[]);

Histogram:

let histogram = get_meter().f64_histogram("latency").init();
histogram.record(0.123, &[]);

Gauge (via UpDownCounter):

let gauge = get_meter().i64_up_down_counter("active_connections").init();
gauge.add(1, &[]); // connection opened
gauge.add(-1, &[]); // connection closed

HTTP Instrumentation

execute_traced_request

Execute an HTTP request with automatic tracing.

pub async fn execute_traced_request(
    request: reqwest::Request,
) -> Result<reqwest::Response, reqwest::Error>

request · reqwest::Request · required

HTTP request to execute

Example:

use iii_sdk::execute_traced_request;

let client = reqwest::Client::new();
let request = client
    .get("https://api.example.com/users")
    .build()?;

// Automatically creates a CLIENT span and injects trace context
let response = execute_traced_request(request).await?;
println!("Status: {}", response.status());

HTTP instrumentation automatically injects traceparent and baggage headers into outbound requests.

Logging

When the otel feature is enabled, the Logger automatically emits OpenTelemetry LogRecords:

use iii_sdk::{III, get_context};
use serde_json::json;

let iii = III::new("ws://localhost:49134");

iii.register_function("process", |input| async move {
    let ctx = get_context();
    
    // These logs are exported via OpenTelemetry
    ctx.logger.info("Processing started", None);
    ctx.logger.debug("Input data", Some(input.clone()));
    
    // Logs include trace context automatically
    ctx.logger.warn("Warning message", Some(json!({ "details": "..." })));
    ctx.logger.error("Error occurred", None);
    
    Ok(json!({ "status": "ok" }))
});

Log Levels:

• logger.debug(message, data)
• logger.info(message, data)
• logger.warn(message, data)
• logger.error(message, data)

Complete Example

use iii_sdk::{
    III, OtelConfig, get_tracer, get_meter, with_span,
    SpanKind, current_trace_id, set_baggage_entry,
};
use opentelemetry::trace::Tracer;
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Configure OpenTelemetry
    let otel_config = OtelConfig {
        enabled: Some(true),
        service_name: Some("order-service".to_string()),
        service_version: Some("1.0.0".to_string()),
        metrics_enabled: Some(true),
        logs_enabled: Some(true),
        ..Default::default()
    };
    
    let iii = III::new("ws://localhost:49134");
    iii.set_otel_config(otel_config);
    iii.connect().await?;
    
    // Set up metrics
    let meter = get_meter();
    let order_counter = meter.u64_counter("orders_created").init();
    let latency_histogram = meter.f64_histogram("order_latency").init();
    
    // Register function with tracing
    iii.register_function("order.create", move |input| {
        let order_counter = order_counter.clone();
        let latency_histogram = latency_histogram.clone();
        
        async move {
            let start = std::time::Instant::now();
            
            // Set baggage for this operation
            let cx = set_baggage_entry("tenant_id", "tenant-123");
            let _guard = cx.attach();
            
            if let Some(trace_id) = current_trace_id() {
                println!("Processing order in trace: {}", trace_id);
            }
            
            // Process order
            let result = json!({ "order_id": "order-456" });
            
            // Record metrics
            order_counter.add(1, &[]);
            latency_histogram.record(start.elapsed().as_secs_f64(), &[]);
            
            Ok(result)
        }
    });
    
    // Call function with tracing
    let result = with_span(
        "create_order_flow",
        None,
        Some(SpanKind::Server),
        || async {
            iii.call("order.create", json!({
                "items": ["item1", "item2"]
            })).await
        }
    ).await?;
    
    println!("Order created: {}", result);
    
    // Flush and shutdown
    iii.shutdown_async().await;
    
    Ok(())
}

Resource Attributes

The SDK automatically adds these resource attributes:

• service.name: Service name from config or OTEL_SERVICE_NAME
• service.version: Service version from config or SERVICE_VERSION
• service.instance.id: Service instance ID (UUID)
• service.namespace: Optional namespace from config
• telemetry.sdk.name: "iii-rust-sdk"
• telemetry.sdk.language: "rust"
• telemetry.sdk.version: SDK version

See Also

• Context API - Access logger in function handlers
• Client API - Configure OTel when creating III client
• Invocation API - Automatic trace propagation in function calls