Metadata-Version: 2.4
Name: otel-messagequeue-exporter
Version: 0.1.2
Summary: OpenTelemetry span exporters for AWS SQS and Azure Service Bus
Author-email: Abhishek Bhat <abhishek.bhat@lyzr.ai>
License: MIT
Project-URL: Homepage, https://github.com/NeuralgoLyzr/otel-messagequeue-exporter
Project-URL: Documentation, https://github.com/NeuralgoLyzr/otel-messagequeue-exporter#readme
Project-URL: Repository, https://github.com/NeuralgoLyzr/otel-messagequeue-exporter
Project-URL: Bug Tracker, https://github.com/NeuralgoLyzr/otel-messagequeue-exporter/issues
Keywords: opentelemetry,tracing,observability,telemetry
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.8.1
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: opentelemetry-api>=1.20.0
Requires-Dist: opentelemetry-sdk>=1.20.0
Requires-Dist: opentelemetry-proto>=1.20.0
Requires-Dist: protobuf>=4.0.0
Provides-Extra: aws
Requires-Dist: boto3>=1.28.0; extra == "aws"
Provides-Extra: azure
Requires-Dist: azure-servicebus>=7.11.0; extra == "azure"
Provides-Extra: all
Requires-Dist: boto3>=1.28.0; extra == "all"
Requires-Dist: azure-servicebus>=7.11.0; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-cov>=4.0.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: flake8>=6.0.0; extra == "dev"
Requires-Dist: mypy>=1.0.0; extra == "dev"
Dynamic: license-file

# 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

```bash
# 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

```python
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

```python
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`:

```python
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:

```python
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:

```bash
# 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

```python
# 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

```bash
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
