如何在Kibana中针对全新异常日志触发告警？

核心思路 #

通过Elasticsearch的查询规则结合告警功能（Watcher或Elastic Alerting），对比当前异常日志与近一周的历史异常消息库，自动识别全新出现的异常并触发告警，替代手动排查的低效方式。

具体实现步骤 #

1. 明确异常过滤规则 #

先锁定需要监控的异常范围，比如基于log_level: ERROR字段过滤，或结合业务特定字段（如exception_type）缩小目标，确保只关注核心异常日志。

2. 构建历史异常消息数据集 #

通过Elasticsearch聚合查询，定期获取近一周内所有出现过的异常消息：

GET /your_log_index/_search
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        {"term": {"log_level": "ERROR"}},
        {"range": {"@timestamp": {"gte": "now-7d/d"}}}
      ]
    }
  },
  "aggs": {
    "unique_exceptions": {
      "terms": {
        "field": "exception_message.keyword",
        "size": 10000
      }
    }
  }
}

该查询会返回近一周所有已出现的异常消息列表，作为判断“新异常”的基准。

3. 配置自动告警规则 #

在Kibana中配置Watcher（或Elastic Alerting），设置定时执行任务（比如每分钟一次），执行以下逻辑：

• 查询最近5分钟内的新异常日志
• 对比这些异常的exception_message是否不在历史异常列表中
• 若存在全新异常，触发指定渠道告警（邮件、Slack、Webhook等）

示例Watcher核心逻辑片段：

"input": {
  "search": {
    "request": {
      "indices": ["your_log_index"],
      "body": {
        "query": {
          "bool": {
            "filter": [
              {"term": {"log_level": "ERROR"}},
              {"range": {"@timestamp": {"gte": "now-5m/m"}}},
              {"bool": {
                "must_not": {
                  "terms": {
                    "exception_message.keyword": ["历史异常消息1", "历史异常消息2"]
                  }
                }
              }
            ]
          }
        }
      }
    }
  }
}

实际配置中可通过脚本动态拉取历史异常列表，无需手动维护。

4. 优化与验证 #

• 调整查询时间窗口，平衡告警及时性与资源消耗
• 对异常消息做归一化处理（比如移除动态ID、时间戳等变量），避免误判新异常
• 设置告警阈值，比如同一新异常出现3次再触发，减少误报

兼容旧版本的替代方案 #

如果使用低版本ELK，可通过Logstash的ruby插件结合Redis存储历史异常消息，在过滤阶段判断是否为新异常并标记，再通过Watcher触发告警。

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