如何在Logstash中检查Elasticsearch查询结果并实现缺失记录告警

嘿，我完全理解你的处境——等团队开启Kibana告警功能可能要等好久，那咱们就直接用Logstash本身来实现这个“检查最近一天是否有日志，没有就告警”的需求。

核心思路是：让Elasticsearch查询总是返回一个结果（哪怕没有匹配的日志），然后在Logstash里判断结果里的文档计数是否为0，要是为0就触发告警。

第一步：修改Elasticsearch输入查询为聚合查询 #

你原来的查询如果没匹配到日志，Logstash根本不会生成事件，咱们换成聚合查询，让ES不管有没有日志都返回一个包含文档总数的结果：

input {
  elasticsearch {
    id => "es_input_plugin"
    user => "myuser"
    password => "mypassword"
    hosts => "myhost"
    ca_file => "C:\devsetup\logstash-8.6.0\rootca3.crt"
    ssl => true
    index => "log-com-ms-icm--*"
    # 改成聚合查询，统计最近24小时的文档总数
    query => '{
      "size": 0,
      "query": {
        "range": {
          "@timestamp": {
            "gte": "now-1d/d"
          }
        }
      },
      "aggs": {
        "total_docs": {
          "value_count": {
            "field": "_id"
          }
        }
      }
    }'
    schedule => "0 * * * *"
    # 告诉插件我们要获取聚合结果
    result_type => "aggregations"
  }
}

第二步：添加Filter判断是否需要告警 #

在Filter阶段提取聚合结果里的文档数，判断是否为0，标记需要告警的事件：

filter {
  ruby {
    code => '
      # 从聚合结果里取文档总数，默认0
      total_docs = event.get("[aggregations][total_docs][value]") || 0
      event.set("total_docs", total_docs)
      
      # 如果总数为0，标记告警并设置告警信息
      if total_docs == 0
        event.set("needs_alert", true)
        event.set("alert_message", "⚠️ 告警：过去24小时内log-com-ms-icm--*索引没有新记录！")
      end
    '
  }
}

第三步：配置Output发送告警 #

根据你的需求选择告警方式，比如邮件、企业微信机器人、Slack等，这里以邮件为例：

output {
  # 只处理需要告警的事件
  if [needs_alert] {
    email {
      to => "your-alert-group@company.com"
      from => "logstash-alerts@company.com"
      subject => "ELK告警：ICM服务日志缺失"
      body => "%{alert_message}"
      # 替换成你们公司的SMTP配置
      address => "smtp.yourcompany.com"
      port => 587
      user => "smtp-account"
      password => "smtp-password"
      enable_starttls_auto => true
    }
  }
  
  # 可选：把聚合结果输出到控制台，方便调试
  stdout {
    codec => rubydebug
  }
}

一些注意事项 #

• 测试的时候可以把schedule改成*/5 * * * *（每5分钟跑一次），快速验证功能是否正常
• 确保你的ES用户有查询目标索引和执行聚合的权限
• 如果不用邮件告警，可以换成HTTP输出，调用你们内部的告警接口（比如企业微信机器人的webhook）
• 可以把告警事件输出到一个专门的告警索引，方便后续回溯

备注：内容来源于stack exchange，提问作者Alex Elam