lyzr-otel-mq-exporter

Export OpenTelemetry traces to AWS SQS and Azure Service Bus in OTLP format (Protobuf or JSON).

Ships two span processors — PersistSpanProcessor (thread-based; writes each span to the WAL inside on_end() for crash-safe delivery, works in any runtime) and AsyncSpanProcessor (asyncio bridge for FastAPI/Starlette) — plus an mmap-backed Write-Ahead Log for guaranteed delivery and an S3 Extended Client for payloads exceeding SQS's 256KB limit.

Installation

# Base
pip install lyzr-otel-mq-exporter

# With AWS support
pip install lyzr-otel-mq-exporter[aws]

# With Azure support
pip install lyzr-otel-mq-exporter[azure]

# All exporters
pip install lyzr-otel-mq-exporter[all]

Quick Start — FastAPI + SQS

from contextlib import asynccontextmanager
from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource, SERVICE_NAME

from lyzr_otel_mq_exporter import SQSSpanExporter, AsyncSpanProcessor

resource = Resource.create({SERVICE_NAME: "my-fastapi-service"})
provider = TracerProvider(resource=resource)
trace.set_tracer_provider(provider)

exporter = SQSSpanExporter(
    queue_url="https://sqs.us-east-1.amazonaws.com/123456789/traces",
    region_name="us-east-1",
    encoding="otlp_proto",
    wal_enabled=True,
    flush_interval_ms=5000,
    max_batch_size=512,
)

processor = AsyncSpanProcessor(exporter=exporter, max_queue_size=2000)
provider.add_span_processor(processor)

@asynccontextmanager
async def lifespan(app):
    await processor.start()
    yield
    await processor.shutdown()

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("handle_request"):
        return {"status": "ok"}

Quick Start — Azure Service Bus

from lyzr_otel_mq_exporter import AzureServiceBusSpanExporter, PersistSpanProcessor

exporter = AzureServiceBusSpanExporter(
    connection_string="Endpoint=sb://namespace.servicebus.windows.net/;SharedAccessKeyName=...",
    queue_name="traces",
    encoding="otlp_proto",
    wal_enabled=True,
)

# Crash-safe, works in any runtime. Use AsyncSpanProcessor instead if you
# specifically want the asyncio bridge on an event loop.
processor = PersistSpanProcessor(exporter, schedule_delay_millis=5000)
provider.add_span_processor(processor)

Azure AD / Managed Identity (keyless)

Recommended for production — no SAS keys in config. Provide a fully_qualified_namespace instead of a connection string; when credential is omitted, DefaultAzureCredential resolves managed identity, workload identity, environment service principals, or az login automatically. Requires the azure-identity package (included in the [azure] extra).

from lyzr_otel_mq_exporter import AzureServiceBusSpanExporter

# Keyless — DefaultAzureCredential (managed identity in Azure, az login locally)
exporter = AzureServiceBusSpanExporter(
    fully_qualified_namespace="namespace.servicebus.windows.net",
    queue_name="traces",
    wal_enabled=True,
)

# Or pass an explicit credential (not closed by the exporter — you own it):
from azure.identity import ManagedIdentityCredential
exporter = AzureServiceBusSpanExporter(
    fully_qualified_namespace="namespace.servicebus.windows.net",
    queue_name="traces",
    credential=ManagedIdentityCredential(client_id="<user-assigned-mi-client-id>"),
)

Provide either connection_string or fully_qualified_namespace, not both. credential is only valid with fully_qualified_namespace.

Choosing a span processor

Reach for PersistSpanProcessor when you want the strongest guarantee: the span is written to the WAL before on_end() returns, so it survives a process crash. It runs a background thread (like OTel's BatchSpanProcessor), so it works in any runtime — Gunicorn/Uvicorn workers, Celery, plain scripts — not just async.

Quick Start — PersistSpanProcessor (crash-safe)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

from lyzr_otel_mq_exporter import SQSSpanExporter, PersistSpanProcessor

provider = TracerProvider()
trace.set_tracer_provider(provider)

exporter = SQSSpanExporter(
    queue_url="https://sqs.us-east-1.amazonaws.com/123456789/traces",
    wal_enabled=True,               # required for the crash-safe guarantee
)

processor = PersistSpanProcessor(
    exporter,
    max_export_batch_size=512,      # flush once this many spans are pending...
    schedule_delay_millis=5000,     # ...or after this long, whichever first
    block_on_full=True,             # backpressure (don't drop) when the WAL is full
)
provider.add_span_processor(processor)

# No start() needed — the worker thread starts on construction.
with trace.get_tracer(__name__).start_as_current_span("do-work"):
    ...

provider.shutdown()                 # drains WAL, flushes, closes

Under FastAPI it needs no event loop — just drain on shutdown:

@asynccontextmanager
async def lifespan(app):
    yield
    provider.shutdown()             # PersistSpanProcessor.shutdown(): drain + flush + close WAL

Quick Start — Sync (BatchSpanProcessor)

For non-async applications (Django, Flask, scripts), use the standard BatchSpanProcessor:

from opentelemetry.sdk.trace.export import BatchSpanProcessor

exporter = SQSSpanExporter(
    queue_url="https://sqs.us-east-1.amazonaws.com/123456789/traces",
    wal_enabled=True,
)
processor = BatchSpanProcessor(exporter)
provider.add_span_processor(processor)

Architecture

on_end(span) -> asyncio.Queue -> micro-batch (up to 64 spans)
                                      |
                           run_in_executor
                                      |
                              exporter.export(batch)
                                      |
                         serialize each span -> WAL write_batch()
                          (single lock, single mmap flush)
                                      |
                         check flush conditions:
                           - time since last flush > interval?
                           - WAL pending count >= max_batch_size?
                                      |  (if yes)
                         merge all pending WAL entries -> single OTLP message
                                      |
                         send to SQS/Azure Service Bus -> mark delivered

Key design: Each span is durable on disk the moment it arrives (WAL write). Flushing to the queue happens separately — on interval or when the pending count hits the threshold. This gives maximum crash safety with batched network I/O.

How It Works

AsyncSpanProcessor

A thin async bridge between OpenTelemetry's sync on_end() callback and the async world. All batching and flush logic lives in the exporter.

• Micro-batching: After getting the first span from the queue, drains up to 63 more that are already waiting. This reduces thread pool submissions by up to 64x under load.
• Idle flush: When no spans arrive for 1 second, calls export([]) to give the exporter a chance to flush pending WAL entries.

PersistSpanProcessor

A thread-based processor (modeled on OpenTelemetry's BatchSpanProcessor) that moves the persist step into on_end():

• Synchronous persist: on_end() calls exporter.persist([span]) before returning. With wal_enabled=True the span is on disk (WAL) at that moment, so it survives a process crash. A daemon thread flushes pending WAL entries to the backend on schedule_delay_millis, or early once max_export_batch_size spans are pending.
• Backpressure: when the WAL is full (e.g. a sustained backend outage), on_end() blocks until the worker frees space (block_on_full=True, default) instead of dropping spans. Set block_on_full=False to drop and count, or full_block_timeout_millis to cap the block as a safety valve.
• Any runtime: uses an OS thread, so it works in sync apps (Flask, Celery, Gunicorn) as well as async — no event loop required.
• Without wal_enabled, it degrades to a best-effort in-memory buffer, like BatchSpanProcessor. Durability is a property of the exporter (wal_enabled), not the processor.

Exporters (SQS / Azure Service Bus)

Two modes of operation:

WAL mode (wal_enabled=True):

1. Each span is serialized and written to WAL immediately via write_batch() (single file lock + single mmap flush for the whole micro-batch)
2. Pending WAL entries are merged into a single OTLP message and sent as one SQS/Azure API call
3. On success, all entries are marked delivered. On transient failure, entries stay in WAL for retry.

In-memory mode (wal_enabled=False, default):

1. Spans are buffered in a list
2. When the buffer reaches max_batch_size or flush_interval_ms elapses, the batch is serialized and sent
3. On crash, in-memory spans are lost

Write-Ahead Log (WAL)

mmap-backed durable storage with per-operation file locking for multi-process safety (Gunicorn, Uvicorn, Celery workers can share a single WAL file).

• Level 2 durability: Process crash safe. Each write_batch() does a single mmap.flush() after all entries are written.
• CRC32 per entry: Detects corruption without invalidating the entire file
• Auto-compaction: Triggered when >50% of entries are delivered, or when space runs out
• Crash recovery: On startup, scans for orphan entries past the write offset

S3 Extended Client (SQS only)

When a merged payload exceeds the threshold (default 250KB), it's uploaded to S3 and a reference is sent via SQS:

exporter = SQSSpanExporter(
    queue_url="...",
    s3_bucket="my-traces-bucket",
    s3_prefix="otel-traces/",
    large_payload_threshold_kb=250,
)

The SQS message includes payload_location=s3 and s3_bucket in message attributes. Your consumer reads the attribute to decide whether to fetch from S3 or read inline.

Configuration Reference

SQSSpanExporter

AzureServiceBusSpanExporter

Provide either connection_string or fully_qualified_namespace. Azure AD auth requires the azure-identity package (bundled in the [azure] extra).

PersistSpanProcessor

AsyncSpanProcessor

Encoding Formats

Both formats are compatible with the OpenTelemetry Collector's SQS and Azure Service Bus receivers.

Benchmarks

Run the benchmarks:

# WAL write() vs write_batch() comparison
uv run python benchmarks/bench_wal.py

# Full end-to-end pipeline benchmarks
uv run python benchmarks/bench_pipeline.py

Results on Apple Silicon (M-series):

Graceful Shutdown

# AsyncSpanProcessor (asyncio) — needs start()/shutdown() on the event loop
@asynccontextmanager
async def lifespan(app):
    await processor.start()
    yield
    await processor.shutdown()   # drains queue, flushes WAL, closes connections

# PersistSpanProcessor (threaded) — no start(); drain via the provider
@asynccontextmanager
async def lifespan(app):
    yield
    provider.shutdown()          # cascades to processor.shutdown(): drain + flush + close WAL

# Sync apps / scripts (either processor)
import atexit
atexit.register(provider.shutdown)

Development

git clone https://github.com/NeuralgoLyzr/lyzr-otel-mq-exporter.git
cd lyzr-otel-mq-exporter

# Install with dev dependencies
uv sync --all-extras

# Run tests
uv run pytest

# Run benchmarks
uv run python benchmarks/bench_pipeline.py

License

MIT