如何让Celery Worker轮询Prometheus查询是否存在告警？

1. 通过Prometheus HTTP API直接查询告警 #

Prometheus内置了HTTP API，其中/api/v1/alerts端点可以返回当前所有处于pending或firing状态的告警。你可以在Celery的任务逻辑或Flask请求处理代码中调用这个API，主动获取告警状态。

实现示例（Python）

import requests
from celery import Celery

app = Celery('tasks', broker='amqp://guest@rabbitmq//')

def check_calculation_pod_alerts(prometheus_url="http://prometheus-server.monitoring.svc.cluster.local:9090"):
    try:
        response = requests.get(f"{prometheus_url}/api/v1/alerts")
        response.raise_for_status()
        alerts = response.json()['data']['alerts']
        
        # 过滤出针对Calculation Pod的资源过载告警（根据你的告警规则标签）
        firing_alerts = [
            alert for alert in alerts
            if alert['status']['state'] == 'firing'
            and alert['labels'].get('service') == 'calculation'
            and alert['labels'].get('alertname') == 'PodResourceOverload'
        ]
        
        return len(firing_alerts) > 0
    except requests.exceptions.RequestException as e:
        # 处理请求失败的情况，默认假设无告警（或根据业务逻辑调整）
        print(f"Failed to query Prometheus: {e}")
        return False

@app.task
def process_calculation_task(data):
    # 先检查是否有资源过载告警
    if check_calculation_pod_alerts():
        # 有告警，延迟重试或暂存任务
        process_calculation_task.retry(countdown=30, max_retries=5)
    
    # 无告警，调用Calculation服务处理请求
    # ... 你的业务逻辑代码 ...

2. 关键注意事项 #

• K8s内部访问配置：确保Celery Pod能访问到Prometheus的服务，通常用K8s Service的内部域名（比如prometheus-server.monitoring.svc.cluster.local），无需暴露公网。
• RBAC权限：如果Prometheus启用了认证或K8s网络策略限制访问，需要给Celery所在的ServiceAccount配置对应的访问权限。
• 查询缓存：避免频繁调用Prometheus API造成压力，可以添加简单缓存（比如用Redis或内存缓存），设置5-10秒的过期时间，减少重复查询。
• 告警规则匹配：确保你的Prometheus告警规则中给Calculation Pod的告警打上了明确的标签（比如service: calculation），方便代码过滤。

3. 更优的替代方案 #

虽然主动查询告警是可行的，但更推荐结合Kubernetes的HPA（水平Pod自动扩缩容）+ Prometheus指标来自动扩容Calculation Pod：

• 用Prometheus采集Calculation Pod的CPU/内存使用率或请求队列长度等指标
• 配置HPA基于这些指标自动增减Pod数量
• 这种方式比Celery主动判断更自动化，能从根本上解决Pod资源过载的问题

如果一定要让Celery控制请求流量，也可以直接查询Prometheus的资源使用指标（而非告警），比如通过/api/v1/query端点查询container_memory_usage_bytes或container_cpu_usage_seconds_total，计算当前Pod的资源使用率，再决定是否处理请求。

内容的提问来源于stack exchange，提问作者Jens Voorpyl