Telemetry

OpenTelemetry tracing, metrics, and logging in the III SDK

Overview

The III SDK provides built-in OpenTelemetry support for distributed tracing, metrics, and logging. When enabled, telemetry data is automatically exported to the III Engine.

Installation

Install the SDK with OpenTelemetry support:

pip install iii-sdk[otel]

This installs:

• opentelemetry-api>=1.25
• opentelemetry-sdk>=1.25

Initialization

init_otel

Initialize OpenTelemetry with automatic engine integration.

from iii import init_otel, OtelConfig

init_otel(OtelConfig(
    service_name="my-service",
    service_version="1.0.0",
    enabled=True
))

config · OtelConfig

OpenTelemetry configuration. If omitted, uses defaults.

loop · asyncio.AbstractEventLoop

Running event loop. When provided, the connection starts immediately. When None, it starts lazily on first use.

OtelConfig

Configuration for OpenTelemetry initialization:

from dataclasses import dataclass

@dataclass
class OtelConfig:
    enabled: bool | None = None
    service_name: str | None = None
    service_version: str | None = None
    service_namespace: str | None = None
    service_instance_id: str | None = None
    engine_ws_url: str | None = None
    fetch_instrumentation_enabled: bool = True
    logs_enabled: bool | None = None
    metrics_enabled: bool = True
    metrics_export_interval_ms: int = 60000

enabled · bool · default: true

Enable OpenTelemetry. Defaults to True unless OTEL_ENABLED=false/0/no/off

service_name · str · default: "iii-python-sdk"

Service name. Uses OTEL_SERVICE_NAME env var if set

service_version · str · default: "unknown"

Service version. Uses SERVICE_VERSION env var if set

service_namespace · str

Service namespace for grouping related services

service_instance_id · str

Unique instance ID. Defaults to a random UUID

engine_ws_url · str · default: "ws://localhost:49134"

III Engine WebSocket URL. Uses III_BRIDGE_URL env var if set

fetch_instrumentation_enabled · bool · default: true

Auto-instrument urllib HTTP calls

logs_enabled · bool · default: true

Enable OpenTelemetry log export

metrics_enabled · bool · default: true

Enable OpenTelemetry metrics export

metrics_export_interval_ms · int · default: 60000

Metrics export interval in milliseconds (60 seconds)

Distributed Tracing

get_tracer

Get the active OpenTelemetry tracer.

from iii import get_tracer
from opentelemetry import trace

tracer = get_tracer()

if tracer:
    with tracer.start_as_current_span("my-operation"):
        # Your code here
        pass

tracer · Tracer | None

The active tracer, or None if OTel is not initialized

Automatic Trace Propagation

Trace context is automatically propagated across function calls:

from iii import init_otel, III, get_tracer
from opentelemetry import trace

init_otel()
tracer = get_tracer()

iii = III("ws://localhost:49134")

async def parent_function(data):
    with tracer.start_as_current_span("process-order"):
        # Trace context is automatically propagated
        payment = await iii.call("payment.process", data)
        return payment

async def payment_function(data):
    # This span is linked to the parent trace
    with tracer.start_as_current_span("charge-card"):
        # Process payment
        return {"status": "success"}

iii.register_function("orders.process", parent_function)
iii.register_function("payment.process", payment_function)

Custom Span Attributes

from iii import get_tracer

tracer = get_tracer()

if tracer:
    with tracer.start_as_current_span("database-query") as span:
        span.set_attribute("db.system", "postgresql")
        span.set_attribute("db.operation", "SELECT")
        span.set_attribute("db.statement", "SELECT * FROM users WHERE id = ?")
        
        # Execute query
        result = execute_query()
        
        span.set_attribute("db.rows_returned", len(result))

HTTP Instrumentation

Urllib HTTP requests are automatically instrumented when fetch_instrumentation_enabled=True:

import urllib.request
from iii import init_otel

init_otel()  # Enables automatic urllib instrumentation

# This request is automatically traced
response = urllib.request.urlopen("https://api.example.com/data")

Spans include attributes:

• http.request.method
• url.full
• server.address
• url.scheme
• url.path
• server.port
• http.response.status_code
• http.request.body.size
• http.response.body.size

Metrics

get_meter

Get the active OpenTelemetry meter.

from iii import get_meter

meter = get_meter()

if meter:
    # Create a counter
    request_counter = meter.create_counter(
        "http.requests",
        description="Number of HTTP requests",
        unit="1"
    )
    
    # Increment counter
    request_counter.add(1, {"method": "GET", "status": "200"})

meter · Meter | None

The active meter, or None if OTel metrics are not initialized

Counter

meter = get_meter()

if meter:
    orders_counter = meter.create_counter(
        "orders.total",
        description="Total number of orders",
        unit="1"
    )
    
    async def create_order(data):
        # Process order
        orders_counter.add(1, {"status": "created"})
        return {"id": "order123"}

Histogram

meter = get_meter()

if meter:
    duration_histogram = meter.create_histogram(
        "order.processing.duration",
        description="Order processing duration",
        unit="ms"
    )
    
    async def process_order(data):
        start = time.time()
        
        # Process order
        
        duration_ms = (time.time() - start) * 1000
        duration_histogram.record(duration_ms, {"status": "success"})

Gauge

import psutil
from iii import get_meter

meter = get_meter()

if meter:
    cpu_gauge = meter.create_observable_gauge(
        "system.cpu.usage",
        callbacks=[lambda options: [(psutil.cpu_percent(), {})]],
        description="CPU usage percentage",
        unit="%"
    )

Logging

Logger

The SDK provides a context-aware logger that emits OpenTelemetry LogRecords:

from iii import get_context

async def my_function(data):
    ctx = get_context()
    
    ctx.logger.info("Processing request", data={"user_id": data["user_id"]})
    
    try:
        # Process data
        result = process(data)
        ctx.logger.info("Request processed successfully")
        return result
    except Exception as e:
        ctx.logger.error("Processing failed", data={"error": str(e)})
        raise

Log Levels

ctx = get_context()

ctx.logger.debug("Debug information", data={"details": "..."})
ctx.logger.info("Informational message", data={"status": "ok"})
ctx.logger.warn("Warning message", data={"threshold": 90})
ctx.logger.error("Error message", data={"error": "Something went wrong"})

Log records include:

• Timestamp
• Severity level
• Message body
• Function name (if available)
• Trace context (span ID, trace ID)
• Custom attributes

Fallback to Python Logging

If OTel is not initialized, logs fallback to standard Python logging:

import logging

logging.basicConfig(level=logging.INFO)

# Without OTel, this uses Python's logging module
ctx = get_context()
ctx.logger.info("This is logged via Python logging")

Shutdown

shutdown_otel

Shut down OpenTelemetry synchronously (best-effort):

from iii import shutdown_otel

shutdown_otel()

shutdown_otel_async

Shut down OpenTelemetry and await WebSocket connection close:

from iii import shutdown_otel_async

await shutdown_otel_async()

is_initialized

Check if OpenTelemetry has been initialized:

from iii import is_initialized

if is_initialized():
    print("OTel is active")
else:
    print("OTel is not initialized")

initialized · bool

True if OTel has been successfully initialized

Example: Full Observability

import asyncio
import time
from iii import (
    III,
    init_otel,
    OtelConfig,
    get_tracer,
    get_meter,
    get_context,
    shutdown_otel_async,
)
from opentelemetry import trace

# Initialize OTel
init_otel(OtelConfig(
    service_name="order-service",
    service_version="1.0.0",
    service_namespace="ecommerce",
    metrics_enabled=True,
    logs_enabled=True,
))

tracer = get_tracer()
meter = get_meter()

# Create metrics
orders_counter = meter.create_counter(
    "orders.total",
    description="Total orders processed",
    unit="1"
)

processing_time = meter.create_histogram(
    "orders.processing_time",
    description="Order processing time",
    unit="ms"
)

iii = III("ws://localhost:49134")

async def process_order(data):
    ctx = get_context()
    start = time.time()
    
    with tracer.start_as_current_span("process-order") as span:
        order_id = data["order_id"]
        span.set_attribute("order.id", order_id)
        
        ctx.logger.info("Processing order", data={"order_id": order_id})
        
        try:
            # Validate inventory
            with tracer.start_as_current_span("validate-inventory"):
                inventory = await iii.call("inventory.check", {
                    "product_id": data["product_id"],
                    "quantity": data["quantity"]
                })
                
                if not inventory["available"]:
                    raise ValueError("Out of stock")
            
            # Process payment
            with tracer.start_as_current_span("process-payment"):
                payment = await iii.call("payment.charge", {
                    "amount": data["amount"],
                    "customer_id": data["customer_id"]
                })
                span.set_attribute("payment.id", payment["id"])
            
            # Record metrics
            duration_ms = (time.time() - start) * 1000
            orders_counter.add(1, {"status": "success"})
            processing_time.record(duration_ms, {"status": "success"})
            
            ctx.logger.info("Order processed successfully", data={
                "order_id": order_id,
                "duration_ms": duration_ms
            })
            
            return {
                "order_id": order_id,
                "status": "confirmed",
                "payment_id": payment["id"]
            }
        
        except Exception as e:
            duration_ms = (time.time() - start) * 1000
            orders_counter.add(1, {"status": "failed"})
            processing_time.record(duration_ms, {"status": "failed"})
            
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            span.record_exception(e)
            
            ctx.logger.error("Order processing failed", data={
                "order_id": order_id,
                "error": str(e)
            })
            
            raise

iii.register_function("orders.process", process_order)

async def main():
    await iii.connect()
    
    try:
        result = await iii.call("orders.process", {
            "order_id": "order123",
            "product_id": "prod456",
            "quantity": 2,
            "amount": 99.99,
            "customer_id": "cust789"
        })
        print(f"Order result: {result}")
    except Exception as e:
        print(f"Order failed: {e}")
    finally:
        await iii.shutdown()
        await shutdown_otel_async()

if __name__ == "__main__":
    asyncio.run(main())

Environment Variables

The SDK respects these environment variables:

• OTEL_ENABLED: Set to false, 0, no, or off to disable OTel
• OTEL_SERVICE_NAME: Default service name
• SERVICE_VERSION: Default service version
• III_BRIDGE_URL: III Engine WebSocket URL (default: ws://localhost:49134)

export OTEL_SERVICE_NAME=my-service
export SERVICE_VERSION=2.0.0
export III_BRIDGE_URL=ws://engine.example.com:49134

python app.py

Best Practices

1. Initialize early: Call init_otel() before connecting to the III Engine
2. Use context: Access logger via get_context() for automatic tracing
3. Meaningful names: Use descriptive span names and metric names
4. Attributes: Add relevant attributes to spans for filtering and analysis
5. Error handling: Always set span status and record exceptions
6. Cleanup: Call shutdown_otel_async() on graceful shutdown
7. Sampling: Use OTel’s built-in sampling for high-volume services

Integration with III Engine

Telemetry data is automatically exported to the III Engine via WebSocket:

• Traces: Exported via EngineSpanExporter
• Metrics: Exported via EngineMetricsExporter every 60 seconds
• Logs: Exported via EngineLogExporter

The engine aggregates telemetry from all workers and provides a unified observability view.