影刀RPA ELK日志分析自动化：异常日志采集与告警
影刀RPA ELK日志分析自动化：异常日志采集与告警
作者：林焱
什么情况用什么
开发团队每天要登录Kibana查日志，看有没有ERROR、有没有慢查询、有没有异常流量——这活说难不难，就是重复且占时间。而且问题往往在凌晨发生，第二天早上才看到日志时已经影响了几个小时。
用影刀RPA自动查ELK日志：定时调Kibana API → 查询过去X小时的ERROR日志 → 提取关键字段 → 格式化后推送告警。
适合：
- 按小时检查生产环境ERROR日志
- 每天汇总慢查询TOP 10
- 监控特定接口的响应时间超过阈值
- 统计异常IP访问频率
怎么做
拼多多店群自动化报活动上架！
第一步：配置Kibana访问
Kibana的搜索API比手动点页面高效得多。
影刀操作步骤：
1. 【HTTP请求】POST请求 Kibana搜索API URL: http://kibana.company.com/api/console/proxy Headers: {"kbn-xsrf": "true", "Content-Type": "application/json"} Body: {"path": "_search", "method": "POST", "body": {搜索条件}}实际的Elasticsearch搜索条件这样写：
{"query":{"bool":{"must":[{"match":{"level":"ERROR"}},{"range":{"@timestamp":{"gte":"now-1h","lte":"now"}}}]}},"size":100,"sort":[{"@timestamp":"desc"}]}注意：如果你用的是ES 8.x版本，API路径和认证方式有变化，需要带API Key或用户名密码。
第二步：解析响应
Kibana返回的JSON结果需要提取关键信息。
importjson response=json.loads(api_response)# 从影刀变量获取hits=response.get('hits',{}).get('hits',[])errors=[]forhitinhits:source=hit['_source']errors.append({'时间':source.get('@timestamp',''),'级别':source.get('level',''),'服务':source.get('service',''),'消息':source.get('message','')[:200],# 截取前200字符'主机':source.get('host',{}).get('name',''),'traceId':source.get('traceId','')})第三步：去重与分类
同一个错误可能在1小时内出现几百次，全部推送不现实。
fromcollectionsimportCounter# 按错误消息去重统计error_counter=Counter()foreinerrors:# 提取错误的关键词（去掉时间戳、ID等变量部分）key=re.sub(r'\d+','N',e['消息'])key=re.sub(r'[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}','UUID',key)error_counter[key]+=1# Top 5 高频错误top_errors=error_counter.most_common(5)第四步：推送告警
根据错误类型和频率决定告警级别。
alert_level='普通'iflen(errors)>50:# 1小时内超过50条ERRORalert_level='严重'elifany('OutOfMemory'ine['消息']or'Connection refused'ine['消息']foreinerrors):alert_level='紧急'message=f""" 【{alert_level}】日志异常告警 时间范围：{(datetime.now()-timedelta(hours=1)).strftime('%H:%M')}~{datetime.now().strftime('%H:%M')}异常总数：{len(errors)}条 TOP 5 高频错误：{chr(10).join(f'{i+1}. [{cnt}次]{err[:100]}'fori,(err,cnt)inenumerate(top_errors))}"""影刀操作步骤：
1. 【发送消息】发送到企业微信群（Webhook机器人） URL: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx 内容：告警消息 2. 如果告警级别="紧急"： 3.  - 【打电话】调用电话告警APITEMU店群矩阵自动化运营核价报活动
有什么坑
坑1：ES索引名带日期
很多ELK集群按天建索引：app-logs-2026.06.27。查询时要拼出正确的索引名，否则查出来是空的。建议用通配符：app-logs-*。
坑2：Kibana session过期
Kibana的session过期时间可能只有几小时。每次查询前先检查是否需要重新登录，或者直接改用ES API绕过Kibana：
ES API: http://es.company.com:9200/app-logs-*/_search坑3：POST请求Body嵌套
Kibana的Console API要求在body里再嵌套一层body，很容易写错。建议先在Kibana Dev Tools里验证查询语句能跑通，再搬到影刀里。
坑4：告警疲劳
如果每小时的ERROR日志都推送，很快就会被忽略。一定要做去重和频率控制——同一类错误24小时内只告警一次，除非频率突然飙升。
总结：日志监控自动化的价值在告警质量而非数量。宁可漏掉几条不重要的小错误，也不要让告警变成没人看的噪音。