关于Logstash Elasticsearch输入插件聚合数据的实现咨询

核心结论 #

不需要使用elasticsearch过滤器的aggregation_fields参数，直接在Elasticsearch输入插件中配置完整的聚合查询即可高效获取目标聚合数据。

原因说明 #

Elasticsearch输入插件本身支持发送包含aggs的完整ES DSL查询，无需先拉取原始数据再做聚合处理——这种方式不仅减少了数据传输量，还利用ES的原生聚合能力直接返回统计结果，性能更优。

你提供的elasticsearch过滤器的aggregation_fields参数，是用于给当前流经Logstash的事件补充关联聚合数据（比如根据事件中的user.name字段，去ES查询该用户的失败登录次数并追加到事件中），并非批量获取全局聚合统计的场景，和你的需求不匹配。

正确配置示例 #

将你的聚合查询直接写入Elasticsearch输入插件的query参数中，同时设置size => 0（无需返回原始文档，只保留聚合结果）：

input {
  elasticsearch {
    hosts => ["http://your-es-server:9200"] # 替换为你的ES服务地址
    index => "your-target-index-pattern"     # 替换为要查询的索引/索引模式
    query => '{
      "query": {
        "bool": {
          "must": [
            {"match": {"event.category": "authentication"}},
            {"match": {"event.outcome": "failure"}}
          ],
          "filter": [
            {"range": {"@timestamp": {"gte": "now-1d"}}}
          ]
        }
      },
      "aggs": {
        "total_logon_per_user": {
          "terms": {"field": "user.name"}
        }
      }
    }'
    size => 0 # 仅返回聚合结果，不返回原始文档，大幅提升性能
    schedule => "* * * * *" # 可选：设置定时执行查询（示例为每分钟一次）
  }
}

output {
  # 示例：将聚合结果输出到控制台调试
  stdout { codec => rubydebug }
  # 也可以将结果写入另一个ES索引长期存储：
  # elasticsearch {
  #   hosts => ["http://your-es-server:9200"]
  #   index => "failed-logon-stats-%{+YYYY.MM.dd}"
  # }
}

聚合结果的访问方式 #

聚合结果会被Logstash解析后放在事件的[@metadata][elasticsearch][aggregations]字段中，你可以在后续的filter或output逻辑中直接引用该字段进行处理。

比如，[@metadata][elasticsearch][aggregations][total_logon_per_user][buckets]就是按user.name分组的失败登录统计数组。

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