Observability with OpenTelemetry

Overview

The III SDK includes built-in OpenTelemetry support for traces, metrics, and logs. All telemetry data is exported to the III Engine via a shared WebSocket connection using OTLP JSON format.

OpenTelemetry is enabled by default. To disable, set OTEL_ENABLED=false or { otel: { enabled: false } } in init options.

Quick Start

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  otel: {
    enabled: true,                     // Default: true
    serviceName: 'my-service',         // Default: 'iii-node'
    serviceVersion: '1.0.0',           // Default: 'unknown'
    metricsEnabled: true,              // Default: true
    metricsExportIntervalMs: 60000,    // Default: 60s
    fetchInstrumentationEnabled: true  // Default: true (auto-instruments fetch)
  }
})

// Traces, metrics, and logs are automatically exported!

from iii import III, InitOptions
from iii.telemetry_types import OtelConfig

iii = III('ws://localhost:49134', InitOptions(
    otel=OtelConfig(
        enabled=True,                     # Default: True
        service_name='my-service',        # Default: 'iii-python-sdk'
        service_version='1.0.0',          # Default: 'unknown'
        metrics_enabled=True,             # Default: True
        metrics_export_interval_ms=60000, # Default: 60s
        fetch_instrumentation_enabled=True # Default: True (auto-instruments urllib)
    )
))

# Traces, metrics, and logs are automatically exported!

Configuration

Service Identity

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  otel: {
    serviceName: 'payment-service',       // Required for grouping
    serviceVersion: '2.1.0',              // Useful for rollback correlation
    serviceNamespace: 'production',       // Environment/namespace
    serviceInstanceId: 'pod-abc-123'      // Defaults to auto-generated UUID
  }
})

from iii import III, InitOptions
from iii.telemetry_types import OtelConfig

iii = III('ws://localhost:49134', InitOptions(
    otel=OtelConfig(
        service_name='payment-service',       # Required for grouping
        service_version='2.1.0',              # Useful for rollback correlation
        service_namespace='production',       # Environment/namespace
        service_instance_id='pod-abc-123'     # Defaults to auto-generated UUID
    )
))

Environment Variables

You can also configure OpenTelemetry via environment variables:

OTEL_ENABLED=true
OTEL_SERVICE_NAME=my-service
SERVICE_VERSION=1.2.3
SERVICE_NAMESPACE=staging
SERVICE_INSTANCE_ID=worker-5
III_BRIDGE_URL=ws://engine:49134
OTEL_METRICS_ENABLED=true

Distributed Tracing

Automatic Tracing

All function invocations are automatically traced with parent-child span relationships:

import { getContext } from 'iii-sdk'

iii.registerFunction({ id: 'orders::create' }, async (input) => {
  const { logger, trace } = getContext()
  
  // This function call is automatically a child span
  const user = await iii.call('users::get', { id: input.userId })
  
  // Another child span
  const inventory = await iii.call('inventory::reserve', { sku: input.sku })
  
  logger.info('Order created', { orderId: input.id })
  return { success: true }
})

// Trace hierarchy:
// orders::create (parent)
// ├── users::get (child)
// └── inventory::reserve (child)

from iii import get_context

async def create_order(input_data: dict) -> dict:
    ctx = get_context()
    
    # This function call is automatically a child span
    user = await iii.call('users::get', {'id': input_data['user_id']})
    
    # Another child span
    inventory = await iii.call('inventory::reserve', {'sku': input_data['sku']})
    
    ctx.logger.info('Order created', {'order_id': input_data['id']})
    return {'success': True}

iii.register_function('orders::create', create_order)

# Trace hierarchy:
# orders::create (parent)
# ├── users::get (child)
# └── inventory::reserve (child)

Custom Spans

Create custom spans for fine-grained tracing:

import { withSpan, SpanKind, getContext } from 'iii-sdk'

iii.registerFunction({ id: 'analytics::report' }, async (input) => {
  const { trace } = getContext()
  
  // Add custom attributes to the function span
  trace?.setAttribute('report.type', input.type)
  trace?.setAttribute('report.date_range', input.dateRange)
  
  // Create a custom span for database query
  const data = await withSpan(
    'query-analytics-db',
    { kind: SpanKind.CLIENT },
    async (span) => {
      span.setAttribute('db.system', 'postgresql')
      span.setAttribute('db.statement', 'SELECT * FROM analytics WHERE ...')
      
      const result = await db.query('SELECT ...')
      span.setAttribute('db.rows_returned', result.length)
      
      return result
    }
  )
  
  // Create a custom span for aggregation
  const aggregated = await withSpan(
    'aggregate-data',
    { kind: SpanKind.INTERNAL },
    async (span) => {
      span.addEvent('Starting aggregation', { rows: data.length })
      const result = processData(data)
      span.addEvent('Aggregation complete', { buckets: result.length })
      return result
    }
  )
  
  return aggregated
})

from iii import get_context
from opentelemetry import trace
from opentelemetry.trace import SpanKind

async def analytics_report(input_data: dict) -> dict:
    ctx = get_context()
    tracer = trace.get_tracer(__name__)
    
    # Add custom attributes to the function span
    if ctx.trace:
        ctx.trace.set_attribute('report.type', input_data['type'])
        ctx.trace.set_attribute('report.date_range', input_data['date_range'])
    
    # Create a custom span for database query
    with tracer.start_as_current_span('query-analytics-db', kind=SpanKind.CLIENT) as span:
        span.set_attribute('db.system', 'postgresql')
        span.set_attribute('db.statement', 'SELECT * FROM analytics WHERE ...')
        
        result = await db.query('SELECT ...')
        span.set_attribute('db.rows_returned', len(result))
        data = result
    
    # Create a custom span for aggregation
    with tracer.start_as_current_span('aggregate-data', kind=SpanKind.INTERNAL) as span:
        span.add_event('Starting aggregation', {'rows': len(data)})
        aggregated = process_data(data)
        span.add_event('Aggregation complete', {'buckets': len(aggregated)})
    
    return aggregated

iii.register_function('analytics::report', analytics_report)

HTTP Client Tracing

HTTP requests are automatically instrumented:

// Node.js: fetch is auto-instrumented by default
iii.registerFunction({ id: 'external::fetch-user' }, async (input) => {
  // Automatically creates a CLIENT span with rich attributes:
  // - http.request.method: GET
  // - url.full: https://api.example.com/users/123
  // - server.address: api.example.com
  // - http.response.status_code: 200
  const response = await fetch(`https://api.example.com/users/${input.id}`)
  return response.json()
})

// Disable if needed:
const iii = init('ws://localhost:49134', {
  otel: { fetchInstrumentationEnabled: false }
})

import urllib.request
import json

# Python: urllib is auto-instrumented by default
async def fetch_user(input_data: dict) -> dict:
    # Automatically creates a CLIENT span with rich attributes:
    # - http.request.method: GET
    # - url.full: https://api.example.com/users/123
    # - server.address: api.example.com
    # - http.response.status_code: 200
    with urllib.request.urlopen(f"https://api.example.com/users/{input_data['id']}") as response:
        return json.loads(response.read().decode())

iii.register_function('external::fetch-user', fetch_user)

# Disable if needed:
iii = III('ws://localhost:49134', InitOptions(
    otel={'fetch_instrumentation_enabled': False}
))

Third-Party Instrumentations

import { PrismaInstrumentation } from '@prisma/instrumentation'
import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  otel: {
    instrumentations: [
      new PrismaInstrumentation()  // Auto-trace all Prisma queries
    ]
  }
})

# Python: Use standard OpenTelemetry instrumentations
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentation
from iii import III, InitOptions

# Initialize instrumentation before III
SQLAlchemyInstrumentation().instrument()

iii = III('ws://localhost:49134', InitOptions(
    otel={'enabled': True}
))

Metrics

Automatic Metrics

The SDK automatically reports worker metrics:

iii.worker.cpu_usage - CPU usage percentage
iii.worker.memory_usage - Memory usage in bytes
iii.worker.active_invocations - Number of active function calls

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  enableMetricsReporting: true,  // Default: true
  otel: {
    metricsEnabled: true,         // Default: true
    metricsExportIntervalMs: 30000 // Export every 30s
  }
})

from iii import III, InitOptions

iii = III('ws://localhost:49134', InitOptions(
    enable_metrics_reporting=True,  # Default: True
    otel={'metrics_enabled': True, 'metrics_export_interval_ms': 30000}
))

Custom Metrics

import { getMeter } from 'iii-sdk'

const meter = getMeter()
if (meter) {
  // Counter: monotonically increasing value
  const orderCounter = meter.createCounter('orders.created', {
    description: 'Total number of orders created'
  })
  
  // Histogram: distribution of values
  const orderValueHistogram = meter.createHistogram('orders.value', {
    description: 'Order value distribution',
    unit: 'USD'
  })
  
  // UpDownCounter: value that can increase or decrease
  const activeOrdersGauge = meter.createUpDownCounter('orders.active', {
    description: 'Number of active orders'
  })
  
  iii.registerFunction({ id: 'orders::create' }, async (input) => {
    orderCounter.add(1, { region: input.region })
    orderValueHistogram.record(input.total, { currency: input.currency })
    activeOrdersGauge.add(1)
    
    // ... create order logic
    
    return { success: true }
  })
}

from iii.telemetry import get_meter

meter = get_meter()
if meter:
    # Counter: monotonically increasing value
    order_counter = meter.create_counter(
        'orders.created',
        description='Total number of orders created'
    )
    
    # Histogram: distribution of values
    order_value_histogram = meter.create_histogram(
        'orders.value',
        description='Order value distribution',
        unit='USD'
    )
    
    # UpDownCounter: value that can increase or decrease
    active_orders_gauge = meter.create_up_down_counter(
        'orders.active',
        description='Number of active orders'
    )
    
    async def create_order(input_data: dict) -> dict:
        order_counter.add(1, {'region': input_data['region']})
        order_value_histogram.record(input_data['total'], {'currency': input_data['currency']})
        active_orders_gauge.add(1)
        
        # ... create order logic
        
        return {'success': True}
    
    iii.register_function('orders::create', create_order)

Logs

See the Context-Aware Logging guide for details on using the Logger API.

Telemetry Export

Separate WebSocket Connection

Telemetry uses a dedicated WebSocket connection separate from the main III connection. This ensures telemetry export doesn’t interfere with function invocations.

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  // Main connection config
  reconnectionConfig: { maxRetries: -1 },
  
  // Telemetry connection config (independent)
  otel: {
    engineWsUrl: 'ws://telemetry-endpoint:49134',  // Can be different!
    reconnectionConfig: {
      maxRetries: 5,      // Less critical than main connection
      maxDelayMs: 60000   // Higher delay tolerance
    }
  }
})

from iii import III, InitOptions, ReconnectionConfig

iii = III('ws://localhost:49134', InitOptions(
    # Main connection config
    reconnection_config=ReconnectionConfig(max_retries=-1),
    
    # Telemetry connection config (independent)
    otel={
        'engine_ws_url': 'ws://telemetry-endpoint:49134',  # Can be different!
        'reconnection_config': {
            'max_retries': 5,      # Less critical than main connection
            'max_delay_ms': 60000  # Higher delay tolerance
        }
    }
))

OTLP JSON Format

Telemetry is exported using OTLP JSON over WebSocket with binary frame prefixes:

Traces: OTLP + JSON payload
Metrics: MTRC + JSON payload
Logs: LOGS + JSON payload

Shutdown

import { shutdownOtel } from 'iii-sdk'

// Graceful shutdown flushes all pending telemetry
process.on('SIGTERM', async () => {
  await iii.shutdown()  // Automatically calls shutdownOtel()
  process.exit(0)
})

// Or shutdown OTel separately:
await shutdownOtel()

from iii.telemetry import shutdown_otel_async
import signal

async def shutdown(signum, frame):
    await iii.shutdown()  # Automatically calls shutdown_otel_async()

signal.signal(signal.SIGTERM, lambda s, f: asyncio.create_task(shutdown(s, f)))

# Or shutdown OTel separately:
await shutdown_otel_async()

Best Practices

Observability Checklist:

✅ Set meaningful serviceName and serviceVersion
✅ Use serviceNamespace to differentiate environments (dev, staging, prod)
✅ Add custom span attributes for business-critical operations
✅ Use structured logging with context (see Context Logging)
✅ Create custom metrics for domain-specific KPIs
✅ Use semantic conventions for standard operations (HTTP, DB, etc.)
✅ Configure separate reconnection for telemetry (less aggressive than main connection)
✅ Always flush telemetry on shutdown (await iii.shutdown())

Get Started

Core Concepts

Guides

Observability with OpenTelemetry

Overview

Quick Start

Configuration

Service Identity

Environment Variables

Distributed Tracing

Automatic Tracing

Custom Spans

HTTP Client Tracing

Third-Party Instrumentations

Metrics

Automatic Metrics

Custom Metrics

Logs

Telemetry Export

Separate WebSocket Connection

OTLP JSON Format

Shutdown

Best Practices

​Overview

​Quick Start

​Configuration

​Service Identity

​Environment Variables

​Distributed Tracing

​Automatic Tracing

​Custom Spans

​HTTP Client Tracing

​Third-Party Instrumentations

​Metrics

​Automatic Metrics

​Custom Metrics

​Logs

​Telemetry Export

​Separate WebSocket Connection

​OTLP JSON Format

​Shutdown

​Best Practices

Overview

Quick Start

Configuration

Service Identity

Environment Variables

Distributed Tracing

Automatic Tracing

Custom Spans

HTTP Client Tracing

Third-Party Instrumentations

Metrics

Automatic Metrics

Custom Metrics

Logs

Telemetry Export

Separate WebSocket Connection

OTLP JSON Format

Shutdown

Best Practices