feat: Add OpenTelemetry instrumentation support (#67)

Introduce OpenTelemetry instrumentation for distributed tracing, enabling monitoring of job processing and enqueue operations.

Usage is auto-enabled when the Litestar OpenTelemetry Plugin is found.

Author: cofin

README.md

@@ -6,18 +6,28 @@
 pip install litestar-saq
 ```
 
+For OpenTelemetry support:
+
+```shell
+pip install litestar-saq[otel]
+```
+
 ## Usage
 
 Here is a basic application that demonstrates how to use the plugin.
 
 ```python
-from __future__ import annotations
-
 from litestar import Litestar
-
 from litestar_saq import QueueConfig, SAQConfig, SAQPlugin
 
-saq = SAQPlugin(config=SAQConfig(use_server_lifespan=True, queue_configs=[QueueConfig(name="samples", dsn="redis://localhost:6379/0")]))
+saq = SAQPlugin(
+    config=SAQConfig(
+        use_server_lifespan=True,
+        queue_configs=[
+            QueueConfig(name="samples", dsn="redis://localhost:6379/0")
+        ],
+    )
+)
 app = Litestar(plugins=[saq])
 ```
 
@@ -47,16 +57,59 @@ If you are starting the process for only specific queues and still want to read
 import os
 from saq import Queue
 
-
 def get_queue_directly(queue_name: str, redis_url: str) -> Queue:
     return Queue.from_url(redis_url, name=queue_name)
 
 redis_url = os.getenv("REDIS_URL")
 queue = get_queue_directly("queue-in-other-process", redis_url)
+
 # Get queue info
 info = await queue.info(jobs=True)
+
 # Enqueue new task
-queue.enqueue(
-    ....
+await queue.enqueue("task_name", arg1="value1")
+```
+
+## Monitored Jobs
+
+For long-running tasks, use the `monitored_job` decorator to automatically send heartbeats and prevent SAQ from marking jobs as stuck:
+
+```python
+from litestar_saq import monitored_job
+
+@monitored_job()  # Auto-calculates interval from job.heartbeat
+async def long_running_task(ctx):
+    await process_large_dataset()
+    return {"status": "complete"}
+
+@monitored_job(interval=30.0)  # Explicit 30-second interval
+async def train_model(ctx, model_id: str):
+    for epoch in range(100):
+        await train_epoch(model)
+    return {"model_id": model_id}
+```
+
+## OpenTelemetry Integration
+
+litestar-saq supports optional OpenTelemetry instrumentation for distributed tracing.
+
+### Configuration
+
+```python
+from litestar_saq import SAQConfig, QueueConfig
+
+config = SAQConfig(
+    queue_configs=[QueueConfig(dsn="redis://localhost:6379/0")],
+    enable_otel=None,  # Auto-detect (default) - enabled if OTEL installed AND Litestar OpenTelemetryPlugin is active
+    # enable_otel=True,  # Force enable (raises error if not installed)
+    # enable_otel=False,  # Force disable
 )
 ```
+
+When enabled, the plugin creates:
+
+- **CONSUMER spans** for job processing
+- **PRODUCER spans** for job enqueue operations
+- **Automatic context propagation** across process boundaries
+
+Spans follow [OpenTelemetry messaging semantic conventions](https://opentelemetry.io/docs/specs/semconv/messaging/) with `messaging.system = "saq"`.

examples/basic.py

@@ -16,6 +16,11 @@
 class SampleController(Controller):
     @get(path="/samples")
     async def samples_queue_info(self, task_queues: TaskQueues) -> QueueInfo:
+        """Get information about the samples queue.
+
+        Returns:
+            Queue information including pending jobs and workers.
+        """
         queue = task_queues.get("samples")
         return await queue.info()
 

examples/hooks.py

@@ -37,6 +37,9 @@ async def example_task(_ctx: Context, *, name: str) -> dict[str, str]:
 
     This task demonstrates a simple async operation with timing hooks
     that will automatically log execution duration.
+
+    Returns:
+        Dictionary with greeting message.
     """
     await asyncio.sleep(0.5)  # Simulate work
     return {"message": f"Hello, {name}!"}
@@ -48,13 +51,20 @@ async def slow_task(_ctx: Context, *, duration: float = 2.0) -> dict[str, float]
     Args:
         _ctx: SAQ context.
         duration: How long to sleep (default 2 seconds).
+
+    Returns:
+        Dictionary with the sleep duration.
     """
     await asyncio.sleep(duration)
     return {"slept_for": duration}
 
 
 async def scheduled_task(_ctx: Context) -> dict[str, str]:
-    """Example scheduled task that runs on cron."""
+    """Example scheduled task that runs on cron.
+
+    Returns:
+        Dictionary with execution status.
+    """
     return {"status": "scheduled task executed"}
 
 
@@ -63,7 +73,11 @@ class TaskController(Controller):
 
     @get(path="/queue-info")
     async def get_queue_info(self, task_queues: TaskQueues) -> QueueInfo:
-        """Get information about the task queue."""
+        """Get information about the task queue.
+
+        Returns:
+            Queue information including pending jobs and workers.
+        """
         queue = task_queues.get("default")
         return await queue.info()
 

examples/postgres.py

@@ -42,6 +42,11 @@ async def system_task(_: Context) -> None:
 class SampleController(Controller):
     @get(path="/samples")
     async def samples_queue_info(self, task_queues: TaskQueues) -> QueueInfo:
+        """Get information about the samples queue.
+
+        Returns:
+            Queue information including pending jobs and workers.
+        """
         queue = task_queues.get("samples")
         return await queue.info()
 

litestar_saq/__init__.py

@@ -12,8 +12,11 @@
     timing_before_process,
 )
 from litestar_saq.plugin import SAQPlugin
+from litestar_saq.typing import OPENTELEMETRY_INSTALLED
 
 __all__ = (
+    # OpenTelemetry
+    "OPENTELEMETRY_INSTALLED",
     "CronJob",
     "Job",
     "PostgresQueueOptions",