影刀RPA Prometheus监控自动化指标采集与告警联动作者林焱 | 适用人群影刀RPA新手 | 难度★★★☆☆一、什么情况用这个运维团队有了Prometheus Alertmanager按理说告警自动推送。但很多中小团队配得不完善——告警规则没调好、通知渠道不稳定、或者根本没有告警路由。用影刀RPA补位定时查询Prometheus → 拿到关键指标 → 根据自定义规则判断 → 通过企微/钉钉/邮件发出告警。不依赖Alertmanager的配置规则灵活可控。实际场景没有专职运维的中小团队、告警需要走特定审批流程的场景、需要将Prometheus数据同步到业务报表的场景。这篇文章能解决的问题影刀Python节点调用Prometheus API用PromQL查询关键指标自定义告警规则比Alertmanager更灵活将监控数据写入业务报表二、怎么做拼多多店群自动化报活动上架2.1 影刀主流程┌──────────────────────────────────────────────┐ │ ① 【执行Python代码】 │ │ 调用Prometheus APIPromeQL查询指标 │ │ ↓ │ │ ② 【条件判断】 │ │ 判断指标是否超过阈值 │ │ ↓ │ │ ③ 【发送企业微信消息】超标分支 │ │ 推送告警通知 │ │ ↓ │ │ ④ 【写入Excel】正常分支 │ │ 记录指标到日报 │ └──────────────────────────────────────────────┘2.2 核心代码步骤①PromQL查询importrequestsimportjsonfromdatetimeimportdatetime,timedelta# Prometheus配置 PROMETHEUS_URLhttp://192.168.1.100:9090defquery_prometheus(promql,timeout10):执行PromQL即时查询urlf{PROMETHEUS_URL}/api/v1/queryparams{query:promql}responserequests.get(url,paramsparams,timeouttimeout)dataresponse.json()ifdata[status]!success:print(f查询失败{data.get(error,unknown error)})return[]results[]foritemindata[data][result]:metricitem[metric]valuefloat(item[value][1])results.append({metric:metric,value:value,})returnresultsdefquery_range(promql,hours1,step5m):执行范围查询查一段时间内的数据urlf{PROMETHEUS_URL}/api/v1/query_rangeenddatetime.utcnow()startend-timedelta(hourshours)params{query:promql,start:start.isoformat()Z,end:end.isoformat()Z,step:step,}responserequests.get(url,paramsparams,timeout15)dataresponse.json()ifdata[status]!success:return[]results[]foritemindata[data][result]:metricitem[metric]values[[float(v[0]),float(v[1])]forvinitem[values]]results.append({metric:metric,values:values,})returnresults# 常用监控查询 checks[]# 1. 检查所有节点的CPU使用率cpu_resultsquery_prometheus(100 - (avg by(instance) (rate(node_cpu_seconds_total{modeidle}[5m])) * 100))forrincpu_results:instancer[metric].get(instance,unknown)cpuround(r[value],1)checks.append({type:CPU,target:instance,value:cpu,unit:%,threshold:80,alert:cpu80,})# 2. 检查内存使用率mem_resultsquery_prometheus((1 - (node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes)) * 100)forrinmem_results:instancer[metric].get(instance,unknown)memround(r[value],1)checks.append({type:内存,target:instance,value:mem,unit:%,threshold:85,alert:mem85,})# 3. 检查磁盘使用率disk_resultsquery_prometheus((1 - (node_filesystem_avail_bytes{mountpoint/} / node_filesystem_size_bytes{mountpoint/})) * 100)forrindisk_results:instancer[metric].get(instance,unknown)disk_valround(r[value],1)checks.append({type:磁盘,target:instance,value:disk_val,unit:%,threshold:85,alert:disk_val85,})# 4. 检查应用错误率假设有业务指标error_ratequery_prometheus(rate(http_requests_total{status~5..}[5m]) / rate(http_requests_total[5m]) * 100)forrinerror_rate:appr[metric].get(app,unknown)rateround(r[value],2)checks.append({type:错误率,target:app,value:rate,unit:%,threshold:1,alert:rate1,})# 输出结果 alerts[cforcinchecksifc[alert]]normal[cforcinchecksifnotc[alert]]result{check_time:datetime.now().strftime(%Y-%m-%d %H:%M:%S),total:len(checks),alerts:len(alerts),alert_details:alerts,normal_details:normal,}outputjson.dumps(result,ensure_asciiFalse)print(f共检查{len(checks)}项{len(alerts)}项超标)步骤②阈值判断在影刀中拿到outputJSON后用【条件判断】节点条件len(alert_details) 0 → 是触发告警流程 → 否写入正常日志步骤③告警推送【发送企业微信消息】 内容 监控告警 ({check_time}) 共发现 {alerts} 项指标超标  {遍历alert_details每条一行} - [{type}] {target}: {value}{unit} (阈值: {threshold}{unit})2.3 PromQL常用查询模板# PromQL速查表 # CPU使用率所有节点平均100 - avg(rate(node_cpu_seconds_total{modeidle}[5m])) * 100# 内存使用率(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100# 磁盘使用率(node_filesystem_size_bytes - node_filesystem_avail_bytes) / node_filesystem_size_bytes * 100# 网络流量入站速率单位bytes/srate(node_network_receive_bytes_total[5m])# HTTP请求QPSrate(http_requests_total[5m])# HTTP请求P99延迟histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m]))# 服务可用性最近5分钟avg_over_time(up[5m]) * 100# JVM堆内存使用率sum(jvm_memory_used_bytes{areaheap}) / sum(jvm_memory_max_bytes{areaheap}) * 100# Pod重启次数增长率rate(kube_pod_container_status_restarts_total[1h])三、有什么坑坑1PromQL语法写错API返回空而不是报错PromQL语法错误时Prometheus可能返回空结果而不是报错让你误以为指标正常。调试方法# 先在浏览器或curl中测试PromQL# http://localhost:9090/api/v1/query?query你的PromQL# 确认有数据后再写到代码里# 代码中加验证resultquery_prometheus(cpu_usage_percent 0)ifnotresult:print(⚠️ 查询返回空请检查PromQL或指标是否存在)# 可以查一下有哪些可用指标[video(video-jn5dXiit-1783620199947)(type-csdn)(url-https://live.csdn.net/v/embed/526817)(image-https://v-blog.csdnimg.cn/asset/1d3c3709da119dd8c13ab01e9b282520/cover/Cover0.jpg)(title-TEMU店群矩阵自动化运营核价报活动)]labelsquery_prometheus({__name__~.*cpu.*})print(f找到{len(labels)}个CPU相关指标)坑2瞬时值vs平均值的选择query返回的是最近一次抓取的瞬时值CPU使用率如果在峰值只持续了1秒也会触发告警——导致频繁的「抖动告警」。推荐用范围查询取平均值# 用query_range取过去5分钟的平均值cpu_avgquery_range(avg(rate(node_cpu_seconds_total{modeidle}[5m])) * 100,hours5/60,# 过去5分钟step1m,)# 取最后一个值的平均值ifcpu_avg:last_values[v[1]forvincpu_avg[0][values]]avg_cpusum(last_values)/len(last_values)# 用平均值判断告警而非瞬时值坑3API认证如果Prometheus有认证如Nginx反向代理加Basic AuthPROMETHEUS_URLhttp://192.168.1.100:9090AUTH(username,password)# Basic Authresponserequests.get(url,paramsparams,authAUTH,timeout10)坑4Prometheus数据延迟Prometheus默认15秒抓一次数据再加上rate()函数需要一定时间窗口计算。所以查到的数据比实际延后1-2分钟是正常的不要以为「数据不准」。坑5大量指标导致API响应慢如果PromQL不带label过滤可能返回成百上千个时间序列API响应慢。# ❌ 不推荐返回所有Pod的CPUsum(rate(container_cpu_usage_seconds_total[5m])) by (pod)# ✅ 推荐加上namespace过滤sum(rate(container_cpu_usage_seconds_total{namespaceproduction}[5m])) by (pod)总结影刀查Prometheus的本质就是调API PromQL。关键是PromQL要写对——先在浏览器验证再写到代码里告警用平均值而不是瞬时值减少抖大量指标注意加过滤条件。