funboost-observability

# Funboost 可观测性配置 ## 概述 Funboost 通过 **Mixin 机制**（`consumer_override_cls` / `publisher_override_cls`）和 **BoosterParams 内置字段**，为任务队列提供生产级可观测性能力，无需修改 broker 源码。 **核心原则：** - 监控/追踪/告警均通过 Mixin 或 BoosterParams 声明式配置 - Mixin 专属参数统一放在 `user_options` 字典中 - 各 Mixin 位于 `funboost/contrib/override_publisher_consumer_cls/`，**不在** `from funboost import ...` 顶层导出 ## 适用场景 | 需求 | 方案 | |------|------| | Grafana 运维大盘 | `PrometheusConsumerMixin` | | 跨队列/跨服务链路追踪 | `AutoOtelConsumerMixin` + `AutoOtelPublisherMixin` | | 连续失败/错误率即时告警 | `AlertNotifierConsumerMixin` | | 周期内限制总执行次数 | `PeriodicQuotaConsumerMixin` | | 任务状态/结果持久化 + Web 查询 | `function_result_status_persistance_conf` | | 分布式聚合告警（无需 Mixin） | `MongoAlertMonitor`（需开启 `is_save_status=True`） | ## 导入路径速查 ```python # 顶层导出（funboost/__init__.py） from funboost import ( boost, BoosterParams, BrokerEnum, FunctionResultStatusPersistanceConfig, # 函数结果持久化配置 ) # Mixin 需从 contrib 子模块导入 from funboost.contrib.override_publisher_consumer_cls.funboost_promethus_mixin import ( PrometheusConsumerMixin, PrometheusPublisherMixin, PrometheusBoosterParams, PrometheusPushGatewayBoosterParams, start_prometheus_http_server, ) from funboost.contrib.override_publisher_consumer_cls.funboost_otel_mixin import ( AutoOtelConsumerMixin, AutoOtelPublisherMixin, OtelBoosterParams, ) from funboost.contrib.override_publisher_consumer_cls.alert_notifier_mixin import ( AlertNotifierConsumerMixin, AlertNotifierBoosterParams, ) from funboost.contrib.override_publisher_consumer_cls.periodic_quota_mixin import ( PeriodicQuotaConsumerMixin, PeriodicQuotaBoosterParams, ) from funboost.core.mongo_alert_monitor import MongoAlertMonitor # 基于 MongoDB 的聚合告警 ``` --- ## 1. Prometheus 指标监控 ### 工作原理 - `PrometheusPublisherMixin`：在 `_after_publish` 钩子记录 `funboost_publish_total` - `PrometheusConsumerMixin`：在 `_both_sync_and_aio_frame_custom_record_process_info_func` 钩子记录任务计数、耗时、重试、队列积压 ### 两种部署模式 | 模式 | 适用场景 | 关键步骤 | |------|----------|----------| | HTTP Server（单进程） | 单消费者进程 | 调用 `start_prometheus_http_server(port=8000)`，Prometheus 主动拉取 `/metrics` | | Push Gateway（多进程） | `multi_process_consume` 等多进程 | 配置 `user_options` 中的 Pushgateway 地址，后台线程定期推送 | ### 指标说明 | 指标名 | 类型 | Labels | 说明 | |--------|------|--------|------| | `funboost_task_total` | Counter | queue, status | 任务计数（status: success/fail/requeue/dlx） | | `funboost_task_latency_seconds` | Histogram | queue | 任务执行耗时 | | `funboost_task_retries_total` | Counter | queue | 重试次数 | | `funboost_queue_msg_count` | Gauge | queue | 队列剩余消息数 | | `funboost_publish_total` | Counter | queue | 发布消息计数 | ### user_options 配置（Push Gateway 模式） | 键名 | 类型 | 默认值 | 说明 | |------|------|--------|------| | `prometheus_pushgateway_url` | str | **必填** | Pushgateway 地址，如 `'localhost:9091'` | | `prometheus_push_interval` | float | `10.0` | 推送间隔（秒） | | `prometheus_job_name` | str | `'funboost'` | Prometheus job 名称 | ### 代码示例 ```python # ── 方式1：HTTP Server 模式（单进程） ── from funboost import boost from funboost.contrib.override_publisher_consumer_cls.funboost_promethus_mixin import ( PrometheusBoosterParams, start_prometheus_http_server, ) start_prometheus_http_server(port=8000) # 访问 http://0.0.0.0:8000/metrics @boost(PrometheusBoosterParams(queue_name='my_task')) def my_task(x): return x * 2 if __name__ == '__main__': my_task.consume() my_task.push(10) # ── 方式2：Push Gateway 模式（多进程推荐） ── from funboost import boost from funboost.contrib.override_publisher_consumer_cls.funboost_promethus_mixin import ( PrometheusPushGatewayBoosterParams, ) @boost(PrometheusPushGatewayBoosterParams( queue_name='my_task', user_options={ 'prometheus_pushgateway_url': 'localhost:9091', 'prometheus_push_interval': 10.0, 'prometheus_job_name': 'my_app', }, )) def my_task_mp(x): return x * 2 if __name__ == '__main__': my_task_mp.multi_process_consume(4) ``` ### 手动指定 Mixin（不用预配置 Params 类） ```python from funboost import boost, BoosterParams, BrokerEnum from funboost.contrib.override_publisher_consumer_cls.funboost_promethus_mixin import ( PrometheusConsumerMixin, PrometheusPublisherMixin, start_prometheus_http_server, ) start_prometheus_http_server(port=8000) @boost(BoosterParams( queue_name='my_task', broker_kind=BrokerEnum.REDIS, consumer_override_cls=PrometheusConsumerMixin, publisher_override_cls=PrometheusPublisherMixin, )) def my_task(x): return x * 2 ``` --- ## 2. OpenTelemetry 链路追踪 ### 工作原理 - `AutoOtelPublisherMixin`：重写 `_execute_publish`，创建 PRODUCER span，将 trace context **注入**到 `msg['extra']['otel_context']` - `AutoOtelConsumerMixin`：重写 `_run` / `_async_run`，从消息 **提取** context，创建 CONSUMER span 作为子节点 - 支持 W3C Trace Context 规范，可对接 Jaeger / Zipkin / SkyWalking 等 OTel 兼容后端 - `aio_publish` 场景自动处理跨线程 context 丢失问题 ### 前置条件 **必须在消费启动前初始化 OpenTelemetry**（Mixin 本身不包含 exporter 配置）： ```python from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter def init_opentelemetry(service_name: str = "my-funboost-app"): resource = Resource.create({"service.name": service_name}) provider = TracerProvider(resource=resource) otlp_exporter = OTLPSpanExporter(endpoint="localhost:4317", insecure=True) provider.add_span_processor(BatchSpanProcessor(otlp_exporter)) trace.set_tracer_provider(provider) ``` 启动 Jaeger（可选，本地调试）： ```sh docker run -d --name jaeger \ -e COLLECTOR_OTLP_ENABLED=true \ -p 16686:16686 \ -p 4317:4317 \ -p 4318:4318 \ jaegertracing/all-in-one:latest ``` 访问 `http://localhost:16686` 查看链路。 ### user_options OTel Mixin **无专属 user_options**，配置通过全局 `TracerProvider` + exporter 完成。 ### 代码示例 ```python from funboost import boost, BrokerEnum, fct from funboost.contrib.override_publisher_consumer_cls.funboost_otel_mixin import OtelBoosterParams init_opentelemetry("order-service") @boost(OtelBoosterParams( queue_name='otel_task_entry', broker_kind=BrokerEnum.REDIS, )) def task_entry(order_id: int): fct.logger.info(f"处理订单 {order_id}") task_process.push(order_id=order_id) # 链路上下文自动传播 return order_id @boost(OtelBoosterParams( queue_name='otel_task_process', broker_kind=BrokerEnum.REDIS, )) def task_process(order_id: int): return f"processed {order_id}" if __name__ == '__main__': task_process.consume() task_entry.consume() task_entry.push(order_id=1001) ``` ### 手动指定 Mixin ```python from funboost import boost, BoosterParams, BrokerEnum from funboost.contrib.override_publisher_consumer_cls.funboost_otel_mixin import ( AutoOtelConsumerMixin, AutoOtelPublisherMixin, ) @boost(BoosterParams( queue_name='my_otel_task', broker_kind=BrokerEnum.REDIS, consumer_override_cls=AutoOtelConsumerMixin, publisher_override_cls=AutoOtelPublisherMixin, )) def my_otel_task(x): return x + 1 ``` ### 与 logger + task_id 的关系 | 维度 | Logger + task_id | OpenTelemetry | |------|------------------|---------------| | 视角 | 一维文本日志 | 树状/甘特图 | | 跨服务 | 需手动关联 | 自动串联 | | 性能分析 | 需人工计算 | 直观看瓶颈 | 两者互补，建议生产环境同时使用。 --- ## 3. 异常告警通知（AlertNotifierConsumerMixin） ### 工作原理 - 仅做**告警通知**，不熔断、不阻塞消费 - 支持两种触发策略：`consecutive`（连续失败）和 `rate`（滑动窗口错误率） - 进入告警状态发送通知，恢复后自动发送恢复通知 - requeue、死信队列、远程 kill 不计入失败计数（避免误告警） ### user_options['alert_options'] 配置 | 参数 | 类型 | 默认值 | 说明 | |------|------|--------|------| | `strategy` | str | `'consecutive'` | `'consecutive'` 或 `'rate'` | | `failure_threshold` | int | `5` | 连续失败次数阈值（consecutive 策略） | | `errors_rate` | float | `0.5` | 错误率阈值 0.0~1.0（rate 策略） | | `period` | float | `60.0` | 统计窗口秒数（rate 策略） | | `min_calls` | int | `5` | 窗口内最少调用数（rate 策略） | | `alert_app` | str | `'wechat'` | 告警通道：`dingtalk` / `wechat` / `feishu` / `webhook` / `custom` | | `webhook_url` | str | `None` | 对应通道的 Webhook 地址（**必填**） | | `alert_interval` | int | `300` | 告警去重间隔秒数 | | `exceptions` | tuple/None | `None` | 跟踪的异常类型（None 跟踪所有） | ### 代码示例 ```python from funboost import boost, BoosterParams, BrokerEnum from funboost.contrib.override_publisher_consumer_cls.alert_notifier_mixin import ( AlertNotifierConsumerMixin, AlertNotifierBoosterParams, ) # 方式1：连续失败 5 次 → 企业微信告警（最简） @boost(AlertNotifierBoosterParams( queue_name='my_task', broker_kind=BrokerEnum.REDIS, user_options={ 'alert_options': { 'failure_threshold': 5, 'alert_app': 'wechat', 'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY', }, }, )) def my_task(x): return call_external_api(x) # 方式2：错误率策略 + 钉钉告警 @boost(BoosterParams( queue_name='my_task_rate', broker_kind=BrokerEnum.REDIS, consumer_override_cls=AlertNotifierConsumerMixin, user_options={ 'alert_options': { 'strategy': 'rate', 'errors_rate': 0.5, 'period': 60, 'min_calls': 10, 'alert_app': 'dingtalk', 'webhook_url': 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN', 'alert_interval': 600, }, }, )) def my_task_rate(x): return call_external_api(x) # 方式3：自定义告警渠道 class EmailAlertConsumer(AlertNotifierConsumerMixin): def custom_send_notification(self, message: str): send_email(to='ops@example.com', subject='任务告警', body=message) @boost(BoosterParams( queue_name='my_task_custom', broker_kind=BrokerEnum.REDIS, consumer_override_cls=EmailAlertConsumer, user_options={ 'alert_options': { 'alert_app': 'custom', 'failure_threshold': 3, }, }, )) def my_task_custom(x): return risky_operation(x) ``` --- ## 4. 周期额度限制（PeriodicQuotaConsumerMixin） ### 工作原理 - 在指定周期内限制**总执行次数**，周期结束后配额自动重置 - **周期额度 ≠ 匀速执行**：例如"每天 24 次"允许一口气用完，不要求每小时 1 次 - 可与 `qps` 参数组合：`qps` 控制执行间隔，周期额度控制总次数 - 配额用完后阻塞等待下一周期（不丢消息） ### 两种窗口模式 | 模式 | user_options | 说明 | |------|-------------|------| | 滑动窗口（默认） | `sliding_window=True` | 从程序启动时刻起算周期 |