How can I monitor PowerMTA queue sizes by domain and set up alerts for stuck emails?

curl -fsS -u "monitor:REDACTED" \
  "http://127.0.0.1:8080/queues?format=json" \
  > /var/tmp/pmta-queues.json

#!/usr/bin/env python3
import base64
import json
import os
import sys
from urllib.request import Request, urlopen

URL = os.environ["PMTA_QUEUES_URL"]
USER = os.environ.get("PMTA_USER", "")
PASSWORD = os.environ.get("PMTA_PASS", "")
STATE_FILE = os.environ.get(
    "PMTA_QUEUE_STATE", "/var/tmp/pmta-queue-watch-state.json"
)

LIMITS = {
    "gmail.com": 8000,
    "google.rollup": 12000,
    "hotmail.com": 5000,
    "outlook.com": 5000,
    "microsoft.rollup": 9000,
}

REQUIRED_READINGS = 3
MIN_REMAINING_RATIO = 0.95


def pick(row, names, default=None):
    for name in names:
        if name in row and row[name] not in ("", None):
            return row[name]
    return default


def as_int(value):
    return int(str(value).replace(",", ""))


def queue_rows(payload):
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict):
        for key in ("queues", "data", "records"):
            value = payload.get(key)
            if isinstance(value, list):
                return value
            if isinstance(value, dict):
                rows = queue_rows(value)
                if rows:
                    return rows
    return []


def fetch_json():
    request = Request(URL)
    if USER and PASSWORD:
        raw = f"{USER}:{PASSWORD}".encode()
        token = base64.b64encode(raw).decode()
        request.add_header("Authorization", f"Basic {token}")
    with urlopen(request, timeout=10) as response:
        return json.load(response)


def load_state():
    try:
        with open(STATE_FILE, encoding="utf-8") as handle:
            return json.load(handle)
    except FileNotFoundError:
        return {}


def save_state(state):
    temporary = f"{STATE_FILE}.tmp"
    with open(temporary, "w", encoding="utf-8") as handle:
        json.dump(state, handle)
    os.replace(temporary, STATE_FILE)


def main():
    payload = fetch_json()
    rows = queue_rows(payload)
    if not rows:
        raise RuntimeError("PowerMTA returned no queue rows; verify JSON mapping")

    previous = load_state()
    current = {}
    new_alerts = []

    for row in rows:
        queue = str(pick(row, ("name", "queue", "queueName"), "")).lower()
        queued_value = pick(row, ("rcpt", "recipients", "queued"))
        if not queue or queued_value is None:
            continue

        domain = queue.split("/", 1)[0]
        limit = LIMITS.get(domain)
        if limit is None:
            continue

        queued = as_int(queued_value)
        old = previous.get(queue, {})
        old_queued = as_int(old.get("queued", 0))
        high = queued >= limit
        not_draining = old_queued > 0 and queued >= old_queued * MIN_REMAINING_RATIO

        if high:
            streak = int(old.get("streak", 0)) + 1 if not_draining else 1
        else:
            streak = 0

        active = streak >= REQUIRED_READINGS
        already_alerted = bool(old.get("alerted", False))
        current[queue] = {
            "queued": queued,
            "streak": streak,
            "alerted": active,
        }

        if active and not already_alerted:
            new_alerts.append(
                f"{queue}: {queued} queued, high and not draining "
                f"for {streak} readings"
            )

    save_state(current)

    if new_alerts:
        print("\n".join(new_alerts))
        return 2

    print("ok")
    return 0


try:
    raise SystemExit(main())
except Exception as error:
    print(f"PowerMTA queue monitor failed: {error}", file=sys.stderr)
    raise SystemExit(1)

#!/bin/sh
export PMTA_QUEUES_URL="http://127.0.0.1:8080/queues?format=json"
export PMTA_USER="monitor"
export PMTA_PASS="REDACTED"

umask 077
/usr/local/bin/pmta_queue_watch.py > /var/tmp/pmta-queue.out 2>&1
status=$?

if [ "$status" -ne 0 ]; then
  mail -s "PowerMTA queue monitor alert" ops@example.com \
    < /var/tmp/pmta-queue.out
fi

exit "$status"

pmta show queues --errors --mode-events gmail.com/*
pmta show disabled sources gmail.com/*
pmta show settings gmail.com/*
pmta show status

<mx-rollup-list>
  mx *.l.google.com google.rollup rollup-by-ip
  mx *.google.com google.rollup rollup-by-ip
  mx *.googlemail.com google.rollup rollup-by-ip
  mx *.mail.protection.outlook.com microsoft.rollup rollup-by-ip
</mx-rollup-list>

<domain hotmail.com>
  notify-of-delay-every 1h
</domain>