otel-messagequeue-exporter 0.1.2

OpenTelemetry span exporters for AWS SQS and Azure Service Bus

otel-messagequeue-exporter

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

Built for async-first frameworks like FastAPI. Includes a custom AsyncSpanProcessor, 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 otel-messagequeue-exporter

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

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

# All exporters
pip install otel-messagequeue-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 otel_messagequeue_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 otel_messagequeue_exporter import AzureServiceBusSpanExporter, AsyncSpanProcessor

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

processor = AsyncSpanProcessor(exporter=exporter)
# Same lifespan pattern as above

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.

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

Parameter	Type	Default	Description
queue_url	str	required	AWS SQS queue URL
region_name	str	"us-east-1"	AWS region
encoding	str	"otlp_proto"	"otlp_proto" or "otlp_json"
aws_access_key_id	str	None	AWS credentials (for dev; use IAM roles in prod)
aws_secret_access_key	str	None	AWS credentials (for dev)
wal_enabled	bool	False	Enable Write-Ahead Log
wal_file_path	str	None	WAL file path (default: .otel_wal/sqs_exporter.wal)
wal_max_size	int	67108864	WAL file size in bytes (default: 64MB)
s3_bucket	str	None	S3 bucket for large payloads
s3_prefix	str	"otel-traces/"	S3 key prefix
large_payload_threshold_kb	int	250	Size threshold (KB) to trigger S3 upload
flush_interval_ms	int	5000	Flush interval in milliseconds
max_batch_size	int	512	Flush when this many spans are pending

AzureServiceBusSpanExporter

Parameter	Type	Default	Description
connection_string	str	required	Azure Service Bus connection string
queue_name	str	required	Queue name
encoding	str	"otlp_proto"	"otlp_proto" or "otlp_json"
servicebus_namespace	str	None	Namespace (for logging)
wal_enabled	bool	False	Enable Write-Ahead Log
wal_file_path	str	None	WAL file path (default: .otel_wal/azure_exporter.wal)
wal_max_size	int	67108864	WAL file size in bytes (default: 64MB)
flush_interval_ms	int	5000	Flush interval in milliseconds
max_batch_size	int	512	Flush when this many spans are pending

AsyncSpanProcessor

Parameter	Type	Default	Description
exporter	SpanExporter	required	The span exporter to use
max_queue_size	int	1000	Max spans in the asyncio.Queue before dropping

Encoding Formats

Format	Size	Speed	Use case
otlp_proto	~121 KB / 500 spans	Faster	Production (default)
otlp_json	~243 KB / 500 spans	Slightly slower	Debugging, human readability

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):

Benchmark	Result
WAL write_batch() speedup	10x faster than write() loop at 1024 spans
Micro-batch effectiveness	62.5x fewer export() calls (32 vs 2000)
Full pipeline (WAL mode)	~7,400 spans/sec
Full pipeline (in-memory)	~136,000 spans/sec
Sustained throughput (3s)	~4,200 spans/sec, 0 drops, ~529 spans/SQS call

Graceful Shutdown

# FastAPI lifespan (recommended)
@asynccontextmanager
async def lifespan(app):
    await processor.start()
    yield
    await processor.shutdown()  # Drains queue, flushes WAL, closes connections

# Sync shutdown
import atexit
atexit.register(lambda: (exporter.force_flush(), exporter.shutdown()))

Development

git clone https://github.com/NeuralgoLyzr/otel-messagequeue-exporter.git
cd otel-messagequeue-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