Django と Celery の構造化イベント監視

import json
import logging
import os
from urllib.request import Request, urlopen
from celery.signals import task_postrun


def report_export_failure():
    try:
        logging.getLogger(__name__).warning("Telemetry export unavailable")
    except Exception:
        pass


@task_postrun.connect(weak=False)
def record_task(sender=None, task_id=None, state=None, **kwargs):
    try:
        status = {"SUCCESS": "success", "FAILURE": "failed",
                  "RETRY": "retry_scheduled"}.get(state)
        if status is None:
            return  # Other states need their own observed lifecycle handlers.
        key = os.getenv("TELEMETRY_API_KEY")
        if not key:
            report_export_failure()
            return
        delivery = sender.request.delivery_info or {}
        payload = {
            "table": "celery_task_attempt_finished",
            "data": {
                "job_name": sender.name,
                "job_id": task_id,
                "queue_name": delivery.get("routing_key"),
                "observed_state": state,
                "status": status,
                "attempt": sender.request.retries + 1,
            },
        }
        request = Request("https://api.telemetry.sh/log",
                          data=json.dumps(payload).encode("utf-8"),
                          headers={"Authorization": key,
                                   "Content-Type": "application/json",
                                   "User-Agent": "telemetry-agent/1.0"},
                          method="POST")
        # Socket timeout, not a hard end-to-end deadline. Export can be lost.
        with urlopen(request, timeout=1.0) as response:
            if not 200 <= response.status < 300:
                report_export_failure()
    except Exception:
        # Do not let export or metadata errors replace the task outcome.
        report_export_failure()