Metadata-Version: 2.4
Name: lyzr-otel-mq-exporter
Version: 0.1.0
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/lyzr-otel-mq-exporter
Project-URL: Documentation, https://github.com/NeuralgoLyzr/lyzr-otel-mq-exporter#readme
Project-URL: Repository, https://github.com/NeuralgoLyzr/lyzr-otel-mq-exporter
Project-URL: Bug Tracker, https://github.com/NeuralgoLyzr/lyzr-otel-mq-exporter/issues
Keywords: opentelemetry,tracing,observability,telemetry
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
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"
Requires-Dist: azure-identity>=1.15.0; extra == "azure"
Provides-Extra: all
Requires-Dist: boto3>=1.28.0; extra == "all"
Requires-Dist: azure-servicebus>=7.11.0; extra == "all"
Requires-Dist: azure-identity>=1.15.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

# 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

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

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

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

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

| Processor | Runtime | What `on_end()` does | Crash-safe? |
|-----------|---------|----------------------|-------------|
| **`PersistSpanProcessor`** | any (sync or async) | persists to the exporter (WAL write) **synchronously**, then a background thread flushes | ✅ with `wal_enabled=True` |
| `AsyncSpanProcessor` | asyncio only (FastAPI/Starlette) | enqueues to an asyncio queue; a background task persists + flushes | ⚠️ in-memory window before the WAL write |
| OTel `BatchSpanProcessor` | any | enqueues to an in-memory queue; a background thread persists + flushes | ⚠️ in-memory window |

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)

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

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

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

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

```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` | `None` | SAS connection string (mutually exclusive with `fully_qualified_namespace`) |
| `queue_name` | `str` | required | Queue name |
| `encoding` | `str` | `"otlp_proto"` | `"otlp_proto"` or `"otlp_json"` |
| `servicebus_namespace` | `str` | `None` | Namespace (for logging) |
| `fully_qualified_namespace` | `str` | `None` | Service Bus FQDN for Azure AD auth (mutually exclusive with `connection_string`) |
| `credential` | `TokenCredential` | `None` | azure-identity credential for Azure AD auth; only valid with `fully_qualified_namespace`. Defaults to `DefaultAzureCredential()` when a namespace is given without one |
| `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 |

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

### PersistSpanProcessor

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `exporter` | `SpanExporter` | required | Exporter exposing `persist()` / `flush_to_queue()` (SQS or Azure) |
| `max_export_batch_size` | `int` | `512` | Wake the worker to flush once this many spans are pending |
| `schedule_delay_millis` | `int` | `5000` | Maximum time between flushes |
| `block_on_full` | `bool` | `True` | Block `on_end()` when the WAL is full (backpressure) instead of dropping |
| `full_block_timeout_millis` | `int` | `None` | Cap how long `on_end()` may block before dropping (safety valve); `None` = block indefinitely |

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

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