影刀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['消息'])![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/a8769e4008844b1686dab1578cd5a716.png#pic_center)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. ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/6daf7964bf544f8680e5d4c2fada9599.png#pic_center) - 【打电话】调用电话告警API

有什么坑

坑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小时内只告警一次，除非频率突然飙升。

总结：日志监控自动化的价值在告警质量而非数量。宁可漏掉几条不重要的小错误，也不要让告警变成没人看的噪音。