RunSignal for Python

Cron job and background job monitoring for Python, on OpenTelemetry. Get an alert when a Celery task or Celery beat schedule, an APScheduler job or a cron script is missed, fails, runs too long or processes nothing, and when a consumer or worker loop stops making progress: the failures no HTTP status code describes. No hosted service and no ping URLs: the alerts live in the backend you already use.

RunSignal records runs, items, heartbeats and the age of runs still in progress as OpenTelemetry metrics, with the same names as every other RunSignal SDK, so the runsignal CLI provisions the same alerts for Python services as for .NET, Go, Java, Node.js and Rust ones: New Relic alerts and dashboards, or Prometheus rules and a Grafana dashboard. The main README covers the concepts, the alerts and the backends.

Installation

pip install runsignal                # core, depends only on opentelemetry-api
pip install "runsignal[celery]"      # with Celery 5.3+
pip install "runsignal[apscheduler]" # with APScheduler 3.10+

Requires Python 3.9 or later. Install and configure the OpenTelemetry SDK and an exporter as usual (opentelemetry-sdk, opentelemetry-exporter-otlp).

Quick start

from runsignal import default_registry

invoices = default_registry.workload(
    "nightly-invoice",
    schedule="0 2 * * *",
    max_duration="30m",      # or seconds, or a timedelta
    min_processed=100,
    severity="critical",
)

with invoices.run():
    for order in unbilled_orders():
        create_invoice(order)
        invoices.processed()

Or as a decorator, for sync and async functions:

from runsignal import monitor

@monitor("nightly-invoice", schedule="0 2 * * *", max_duration="30m")
def generate_invoices():
    for order in unbilled_orders():
        create_invoice(order)
        generate_invoices.workload.processed()

default_registry records on the global OpenTelemetry providers, which forward to the SDK once it is configured, so it works even when it is created by an import first. For explicit providers, use WorkloadRegistry(meter_provider=..., tracer_provider=...).

Recording work

error.type is the exception's error_type attribute if it has one (such as "statement_missing"), otherwise its fully qualified class name (ValueError, requests.exceptions.ConnectionError).

Celery

from celery import Celery
from celery.schedules import crontab
from runsignal.celery import instrument, workload_of

app = Celery("billing")
app.conf.beat_schedule = {
    "nightly-invoice": {"task": "billing.tasks.generate_invoices", "schedule": crontab(hour=2, minute=0)},
}
instrument(app, workloads={"billing.tasks.generate_invoices": {"max_duration": "30m", "severity": "critical"}})

@app.task(bind=True)
def generate_invoices(self):
    for order in unbilled_orders():
        create_invoice(order)
        workload_of(self).processed()

• What is recorded: every task execution, with no code in your tasks: success, failure with the exception's type (each retry counts as a failed run), cancellation when a task is revoked, and duration. Celery's own celery.* tasks are skipped.
• Naming: tasks are named after their task name (billing.tasks.generate_invoices).
• Celery beat: instrument(app) registers every task in beat_schedule with its crontab as the schedule, so a periodic task that never runs gets a missed alert. Interval schedules get no schedule.
• Manifest: the worker writes it when it is ready if RUNSIGNAL_MANIFEST_PATH is set. In CI, import the app and call default_registry.write_manifest_from_env("billing") instead; no broker is needed.
• Prefork workers: configure the OpenTelemetry SDK in worker_process_init, as for any OpenTelemetry instrumentation of Celery.

APScheduler

from apscheduler.schedulers.background import BackgroundScheduler
from runsignal.apscheduler import instrument

scheduler = BackgroundScheduler()
scheduler.add_job(generate_invoices, "cron", hour=2, minute=0, id="nightly-invoice")
instrument(scheduler, workloads={"nightly-invoice": {"max_duration": "30m", "severity": "critical"}})
scheduler.start()

• What is recorded: every run of every job, including jobs added after instrument, for all schedulers (BackgroundScheduler, BlockingScheduler, AsyncIOScheduler, ...).
• Naming: jobs are named after the id you gave them, or their name when APScheduler generated the id.
• Schedules: cron triggers become the workload's schedule, with APScheduler's Monday = 0 converted to cron's numbering. Triggers that use year, week or days such as 1st mon get no schedule, with a warning.
• Persistent job stores (SQLAlchemy, Redis, MongoDB) cannot store a wrapped function: decorate the job function with @default_registry.monitor("nightly-invoice") instead. instrument still registers the schedule.

Cron scripts and loops

A script run by cron, a Kubernetes CronJob or systemd timer is one run per invocation:

with default_registry.workload("nightly-invoice", schedule="0 2 * * *").run():
    main()

The missed alert needs a long-running process to report since_success, so for short-lived scripts declare the workload in the service that exports continuously, or rely on the failed, overrun and low output alerts. Long-running loops send heartbeats:

sync = default_registry.workload("stock-sync", expect_every="5m")
while True:
    sync_stock()
    sync.beat()

Manifest

default_registry.write_manifest_from_env("billing-service")  # writes to RUNSIGNAL_MANIFEST_PATH when it is set

Then provision the alerts from CI with the runsignal CLI.

Exporting

New Relic needs delta temporality. With cumulative, a stopped workload never looks stopped:

export OTEL_EXPORTER_OTLP_ENDPOINT=https://otlp.eu01.nr-data.net:4317   # US: https://otlp.nr-data.net:4317
export OTEL_EXPORTER_OTLP_HEADERS=api-key=<your license key>
export OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE=delta

Prometheus 3 receives OTLP at /api/v1/otlp/v1/metrics with the default (cumulative) temporality. Generate its rules with runsignal export prometheus and a dashboard with runsignal export grafana.

Development

python -m venv .venv && .venv/bin/pip install -e ".[celery,apscheduler]" opentelemetry-sdk pytest
.venv/bin/pytest

The tests include the cross-language conformance vectors in spec/conformance.

Links

Specification · Changelog · Other languages