使用 Django and Celery structured event monitoring 发送并验证事件
观察一次 Celery 任务尝试，保留其原始结果，并在创建图表前核对记录的状态。
- 1
选择尝试
Celery 工作进程结果
- 2
定义合同
job_name、job_id、queue_name、observed_state 和 status
- 3
记录尝试状态
此示例将 task_postrun 的 SUCCESS、FAILURE 和 RETRY 记录为尝试状态；重试不是最终失败。
- 4
核实证据
运行已知任务并查询 celery_task_attempt_finished，检查观察到的尝试状态；重试行不是最终结果。
开始之前
开始之前
- 服务器端TELEMETRY_API_KEY
- 工作进程中已安装 Celery
- 在 Celery 工作进程启动时导入此模块
交货设置
安装并初始化服务器端
此 Celery 工作进程模块通过 Python 标准库 HTTP 发送事件，无需安装或初始化 Telemetry SDK。 将摄取凭据保留在浏览器包、客户端可见的环境变量、源代码控制、日志和异常消息之外。
- 1每个工作进程加载信号模块一次，导出错误不得改变任务结果。
- 2在成功、失败、重试或超时边界处添加结果事件。
- 3在启用警报之前发送受控装置并检查存储的行。
片段
从一个结构化事件开始
在工作流程完成、失败或重试的位置添加此形状。然后从真实的字段构建仪表板。
Django and Celery structured event monitoring事件
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()事件结构
job_name、job_id、queue_name、observed_state 和 status
所观察执行的 attempt 为 retries + 1
仅包含工作进程元数据；导出前请检查任务和队列标识符
检查配置
检查站 1
此示例将 task_postrun 的 SUCCESS、FAILURE 和 RETRY 记录为尝试状态；重试不是最终失败。
检查站 2
重试时保留逻辑任务 ID，并将 retries + 1 记录为尝试序号。
检查站 3
在每个工作进程中导入此模块一次。它不记录 Django 请求，也不测量任务耗时。
检查站 4
导出尽力而为，套接字超时为一秒；它会增加工作进程延迟，未实现严格的总耗时上限、持久化投递或导出重试。
检查站 5
不要将任务参数、结果、异常文本、URL 或凭据写入事件。工作进程丢失或终止可能阻止 task_postrun，因此缺失事件的状态未知。
验证
证明事件已到达
在演练已知的成功和失败案例后运行此命令。如果您的最终事件契约与代码片段不同，请替换后备表名称。
Django and Celery structured event monitoring 验证查询
SELECT *
FROM celery_task_attempt_finished
ORDER BY timestamp_utc DESC
LIMIT 20;实施参考
在启用新的生产路径之前，请检查事件合同、数据安全指南和上游主要文档。
在哪里记录事件
保持结果事件小且可恢复
该模式提供了
- 除了上游工作流程之外，还有一个有界的、SQL 就绪的结果。
- 用于仪表板、警报和跨事件关联的稳定字段。
- 用于验证成功、失败、重试和超时行为的夹具驱动路径。
该模式不提供
- OTLP 导出器、自动收集管道或详细跟踪和诊断日志的替代品。
- 仅因为有效负载包含事件 ID，所以仅传送一次。
- 收集原始提供商有效负载、用户内容、凭证或受监管数据的权限。
事件模式示例
此工作流的事件模式
在将查询或代码片段适应生产之前，请检查行粒度、发出边界、所需类型、隐私类、示例有效负载和验证清单。
相关产品功能
继续此工作流程 告警
将经过审查的可靠性查询提升到拥有的阈值和响应工作流程中。
相关 SQL 查询示例
更多 SQL 示例
针对此工作流程中的结构化字段运行查询，检查示例结果，并将有用的答案转换为仪表板或警报。
按实施系列浏览
比较相关集成模式
与此集成配对的模板
更多集成