tutorial

How to Monitor Dramatiq with Vigilmon

Dramatiq is a Python background task processing library built for reliability — but silent worker crashes, growing message queues, and dead-letter accumulation can halt your processing without a single alert. Learn how to monitor queue health, worker liveness, and failure rates with Vigilmon.

Dramatiq is a Python background job library designed to be simple, reliable, and observable. It supports both Redis and RabbitMQ as brokers and ships with middleware for retries, rate limiting, time limits, and age limits out of the box. It's a popular alternative to Celery in Django and Flask applications that want predictable retry semantics without Celery's configuration complexity.

But Dramatiq's reliability features only protect individual messages — they can't tell you that all your workers crashed, that your dead-letter queue is growing, or that a particular task hasn't run in six hours. Vigilmon gives you external visibility into Dramatiq health through HTTP monitors for queue health endpoints and heartbeat monitors for worker processes.


Why Dramatiq Needs External Monitoring

Dramatiq's middleware catches many failures per message, but doesn't monitor infrastructure. External monitoring with Vigilmon adds:

  • Proactive alerting when worker processes die and aren't restarted by your process supervisor
  • Queue depth threshold checks before message backlogs grow out of control
  • Dead-letter queue monitoring to catch messages that have exhausted all retries
  • Broker connectivity checks — Dramatiq can't process without Redis or RabbitMQ
  • Heartbeat monitoring for periodic tasks that must prove they are executing on schedule

Step 1: Build a Dramatiq Health Endpoint

With Redis broker

Create a Flask or FastAPI health endpoint that queries Redis queue state:

# health/dramatiq_health.py
import os
import redis
from flask import Flask, jsonify

app = Flask(__name__)

redis_client = redis.Redis(
    host=os.environ.get("REDIS_HOST", "localhost"),
    port=int(os.environ.get("REDIS_PORT", 6379)),
    password=os.environ.get("REDIS_PASSWORD"),
    decode_responses=True,
)

QUEUE_NAMES        = os.environ.get("DRAMATIQ_QUEUES", "default").split(",")
WAITING_THRESHOLD  = int(os.environ.get("DQ_WAITING_THRESHOLD", 500))
DLQ_THRESHOLD      = int(os.environ.get("DQ_DLQ_THRESHOLD", 20))


@app.get("/health/dramatiq")
def dramatiq_health():
    issues = []
    queue_stats = {}

    # Check Redis connectivity
    try:
        redis_client.ping()
    except redis.ConnectionError as e:
        return jsonify({"status": "down", "reason": "redis_unreachable", "error": str(e)}), 503

    for queue in QUEUE_NAMES:
        # Dramatiq stores messages in a Redis list keyed as `dramatiq:<queue>`
        main_key = f"dramatiq:{queue}"
        dlq_key  = f"dramatiq:{queue}.DQ"

        waiting = redis_client.llen(main_key)
        dlq     = redis_client.llen(dlq_key)

        queue_stats[queue] = {"waiting": waiting, "dead_letter": dlq}

        if waiting > WAITING_THRESHOLD:
            issues.append(f"queue_{queue}_waiting_{waiting}")
        if dlq > DLQ_THRESHOLD:
            issues.append(f"queue_{queue}_dlq_{dlq}")

    if issues:
        return jsonify({"status": "degraded", "issues": issues, "queues": queue_stats}), 503

    return jsonify({"status": "ok", "queues": queue_stats})


if __name__ == "__main__":
    app.run(port=int(os.environ.get("HEALTH_PORT", 8080)))

With RabbitMQ broker

# health/dramatiq_health_amqp.py
import os
import requests
from flask import Flask, jsonify

app = Flask(__name__)

RABBITMQ_API = os.environ.get("RABBITMQ_API_URL", "http://localhost:15672")
RABBITMQ_USER = os.environ.get("RABBITMQ_USER", "guest")
RABBITMQ_PASS = os.environ.get("RABBITMQ_PASS", "guest")
QUEUE_NAMES   = os.environ.get("DRAMATIQ_QUEUES", "default").split(",")
WAITING_THRESHOLD = int(os.environ.get("DQ_WAITING_THRESHOLD", 500))


@app.get("/health/dramatiq")
def dramatiq_health():
    issues = []
    queue_stats = {}

    for queue in QUEUE_NAMES:
        try:
            resp = requests.get(
                f"{RABBITMQ_API}/api/queues/%2F/{queue}",
                auth=(RABBITMQ_USER, RABBITMQ_PASS),
                timeout=5,
            )
            data = resp.json()
            waiting  = data.get("messages_ready", 0)
            dlq_name = f"{queue}.DQ"
            queue_stats[queue] = {"waiting": waiting}

            if waiting > WAITING_THRESHOLD:
                issues.append(f"queue_{queue}_waiting_{waiting}")
        except Exception as e:
            issues.append(f"queue_{queue}_unreachable")

    if issues:
        return jsonify({"status": "degraded", "issues": issues, "queues": queue_stats}), 503

    return jsonify({"status": "ok", "queues": queue_stats})

Step 2: Add Heartbeat Pings to Actors

HTTP queue checks confirm broker health. Heartbeats confirm specific actors are actually executing:

# tasks/billing.py
import os
import requests
import dramatiq
from dramatiq.brokers.redis import RedisBroker

broker = RedisBroker(host=os.environ.get("REDIS_HOST", "localhost"))
dramatiq.set_broker(broker)

HEARTBEAT_URL = os.environ.get("VIGILMON_INVOICE_HEARTBEAT")


@dramatiq.actor(queue_name="billing", max_retries=3, time_limit=60_000)
def generate_invoice(subscription_id: str):
    _do_generate_invoice(subscription_id)

    # Ping heartbeat only on success
    if HEARTBEAT_URL:
        try:
            requests.get(HEARTBEAT_URL, timeout=5)
        except Exception:
            pass  # don't fail the job if the heartbeat fails

For periodic tasks via APScheduler or django-dramatiq:

# tasks/periodic.py
import os
import requests
import dramatiq
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()


@dramatiq.actor(queue_name="default")
def nightly_report():
    _run_report()

    heartbeat_url = os.environ.get("VIGILMON_NIGHTLY_REPORT_HEARTBEAT")
    if heartbeat_url:
        requests.get(heartbeat_url, timeout=5)


@scheduler.scheduled_job("cron", hour=2, minute=0)
def enqueue_nightly_report():
    nightly_report.send()


if __name__ == "__main__":
    scheduler.start()

For high-frequency actors, use a counter to avoid pinging on every message:

import threading

_counter_lock   = threading.Lock()
_job_counter    = 0
HEARTBEAT_EVERY = int(os.environ.get("HEARTBEAT_EVERY", 25))


@dramatiq.actor(queue_name="emails")
def send_email(user_id: str, template: str):
    global _job_counter
    _deliver_email(user_id, template)

    with _counter_lock:
        _job_counter += 1
        should_ping = _job_counter % HEARTBEAT_EVERY == 0

    if should_ping and (url := os.environ.get("VIGILMON_EMAIL_HEARTBEAT")):
        requests.get(url, timeout=5)

Step 3: Configure Vigilmon HTTP Monitors

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose HTTP / HTTPS
  3. Set the URL to your Dramatiq health endpoint: https://your-app.example.com/health/dramatiq
  4. Set the check interval to 1 minute
  5. Under Expected response, configure:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 3000ms
  6. Under Alert channels, assign your Slack or email integration
  7. Save the monitor

Add monitors for each concern:

| Monitor URL | Purpose | Interval | |---|---|---| | /health/dramatiq | Queue depth, dead-letter count, broker connectivity | 1 min |


Step 4: Configure Heartbeat Monitors for Workers

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Set the name: dramatiq-nightly-report
  3. Set the expected interval matching your schedule (e.g. 24 hours)
  4. Set the grace period: 30 minutes
  5. Save — copy the heartbeat URL: https://vigilmon.online/heartbeat/abc123xyz
  6. Set the corresponding environment variable in production

Create one heartbeat monitor per critical actor or periodic task:

| Actor | Heartbeat Interval | Grace Period | |---|---|---| | nightly_report | 24 hours | 30 min | | hourly_sync | 1 hour | 10 min | | generate_invoice | 5 min | 10 min |


Step 5: Alert Routing

| Monitor | Alert Channel | Priority | |---|---|---| | Queue health /health/dramatiq | Slack + PagerDuty | P1 | | Actor heartbeats | Slack + PagerDuty | P1 |

Key thresholds to tune for your workload:

  • Waiting threshold: set per queue — high-volume email queues tolerate larger backlogs than low-volume billing queues
  • Dead-letter threshold: keep low — dead-letter messages have exhausted all retries and represent messages that will never be processed without manual intervention
  • Broker connectivity: always alert immediately — there is no Dramatiq without its broker
  • Grace periods: add meaningful slack to heartbeat intervals — a few minutes of scheduler jitter shouldn't trigger a 3 AM page

Summary

Dramatiq failures are silent by design — a worker process that exits cleanly leaves messages in the queue without anyone noticing. External monitoring with Vigilmon catches problems before they become incidents:

| Monitor Type | What It Covers | |---|---| | HTTP monitor on /health/dramatiq | Queue depth, dead-letter count, broker health | | Heartbeat monitor | Specific actor execution on schedule |

Get started free at vigilmon.online — your first Dramatiq monitor is running in under two minutes.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →