Recipe: Custom Metrics Sink¶

Soniq emits two events for each job run: record_job_start when the worker claims a job, and record_job_end when the handler returns or raises. Pass a MetricsSink to Soniq(metrics_sink=...) to send those events to Prometheus, statsd or another monitoring system.

Protocol¶

# soniq/observability/metrics.py
from typing import Optional, Protocol, runtime_checkable


@runtime_checkable
class MetricsSink(Protocol):
    async def record_job_start(
        self,
        *,
        job_id: str,
        job_name: str,
        queue: str,
        attempt: int,
    ) -> None: ...

    async def record_job_end(
        self,
        *,
        job_id: str,
        job_name: str,
        queue: str,
        status: str,
        duration_s: float,
        error: Optional[str] = None,
    ) -> None: ...

status is one of done (handler returned), failed (handler raised; retries remaining), dead_letter (retries exhausted, retry policy declined, or backend contract violation), snoozed (handler returned Snooze(...); the job does not use a retry), or reclaimed (the handler ran so long that its job was requeued and claimed by another worker, so this run's outcome was dropped).

Built-in: Prometheus¶

pip install soniq installs prometheus_client. Wire the sink like this:

from prometheus_client import start_http_server

from soniq import Soniq
from soniq.observability import PrometheusMetricsSink

soniq = Soniq(
    database_url="postgresql://localhost/myapp",
    metrics_sink=PrometheusMetricsSink(),
)

# Expose /metrics on http://localhost:9090
start_http_server(9090)

Metrics exported (default prefix soniq):

soniq_jobs_started_total{queue, job_name} (Counter)
soniq_jobs_completed_total{queue, job_name, status} (Counter)
soniq_job_duration_seconds{queue, job_name, status} (Histogram)
soniq_jobs_in_progress{queue, job_name} (Gauge)

The Histogram buckets cover typical background-job latencies (5ms, 10ms, ..., 60s, 120s, 300s, 600s, +Inf).

Custom prefix or registry¶

from prometheus_client import CollectorRegistry

# Keep Soniq's metrics out of the default registry.
reg = CollectorRegistry()
sink = PrometheusMetricsSink(registry=reg, prefix="myapp_jobs")
soniq = Soniq(metrics_sink=sink)

The sink does not start an HTTP server, push to a gateway, or scrape its own state - you do those yourself. Use the prometheus_client helpers (start_http_server, push_to_gateway) that match your scrape model.

Custom sink: forward to your platform¶

For Datadog, OpenTelemetry, statsd or any other system, implement the protocol directly:

import time
from soniq import Soniq


class StatsdMetricsSink:
    """Emit per-job timings and counters to a statsd server."""

    def __init__(self, statsd_client):
        self._statsd = statsd_client

    async def record_job_start(self, *, job_id, job_name, queue, attempt):
        self._statsd.incr(
            "soniq.jobs.started",
            tags=[f"queue:{queue}", f"job:{job_name}"],
        )

    async def record_job_end(
        self, *, job_id, job_name, queue, status, duration_s, error=None
    ):
        tags = [f"queue:{queue}", f"job:{job_name}", f"status:{status}"]
        self._statsd.incr("soniq.jobs.completed", tags=tags)
        self._statsd.timing(
            "soniq.jobs.duration",
            duration_s * 1000,  # statsd convention: milliseconds
            tags=tags,
        )


import statsd  # or whatever client your platform ships
soniq = Soniq(metrics_sink=StatsdMetricsSink(statsd.StatsClient()))

Combining with the noop default¶

The default sink is NoopMetricsSink. It emits nothing and adds no overhead. Metrics are opt-in: if you pass None or omit metrics_sink=, Soniq emits no metrics.

Caveats¶

The sink is invoked by the worker, not by the enqueuer. await soniq.enqueue(...) does not call into the sink. If you need enqueue-time metrics, instrument your application code directly.
If a sink method raises, the error propagates and fails the job. To make metrics best-effort, wrap your sink in try/except:

class BestEffortSink:
    def __init__(self, inner):
        self._inner = inner

    async def record_job_start(self, **kw):
        try:
            await self._inner.record_job_start(**kw)
        except Exception:
            pass  # don't fail jobs because metrics broke

    async def record_job_end(self, **kw):
        try:
            await self._inner.record_job_end(**kw)
        except Exception:
            pass