Telemetry

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

Initialize early: Call init_otel() before connecting to the III Engine
Use context: Access logger via get_context() for automatic tracing
Meaningful names: Use descriptive span names and metric names
Attributes: Add relevant attributes to spans for filtering and analysis
Error handling: Always set span status and record exceptions
Cleanup: Call shutdown_otel_async() on graceful shutdown
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.

Getting Started

Core APIs

Advanced

Overview

Installation

Initialization

init_otel

OtelConfig

Distributed Tracing

get_tracer

Automatic Trace Propagation

Custom Span Attributes

HTTP Instrumentation

Metrics

get_meter

Counter

Histogram

Gauge

Logging

Logger

Log Levels

Fallback to Python Logging

Shutdown

shutdown_otel

shutdown_otel_async

is_initialized

Example: Full Observability

Environment Variables

Best Practices

Integration with III Engine

Build docs developers (and LLMs) love

Getting Started

Core APIs

Advanced

Documentation Index

​Overview

​Installation

​Initialization

​init_otel

​OtelConfig

​Distributed Tracing

​get_tracer

​Automatic Trace Propagation

​Custom Span Attributes

​HTTP Instrumentation

​Metrics

​get_meter

​Counter

​Histogram

​Gauge

​Logging

​Logger

​Log Levels

​Fallback to Python Logging

​Shutdown

​shutdown_otel

​shutdown_otel_async

​is_initialized

​Example: Full Observability

​Environment Variables

​Best Practices

​Integration with III Engine

Build docs developers (and LLMs) love

Overview

Installation

Initialization

init_otel

OtelConfig

Distributed Tracing

get_tracer

Automatic Trace Propagation

Custom Span Attributes

HTTP Instrumentation

Metrics

get_meter

Counter

Histogram

Gauge

Logging

Logger

Log Levels

Fallback to Python Logging

Shutdown

shutdown_otel

shutdown_otel_async

is_initialized

Example: Full Observability

Environment Variables

Best Practices

Integration with III Engine