如何在 Logstash 配置文件中输出到多个 Elasticsearch 索引？

核心思路：在 filter 阶段根据日志特征打标（如添加字段或 tag），然后在 output 的 elasticsearch 插件里用条件语句和索引名变量（如 index => "%{[@metadata][index_name]}"）把不同日志写入不同索引。

配置示例

input {
  file {
    path => "/var/log/myapp/*.log"
    start_position => "beginning"
    type => "app_log"
  }
  file {
    path => "/var/log/myapp-errors/*.log" # 本例格式为 ISO8601 ERROR: 消息，不是 Nginx 原生错误格式
    type => "error_log"
  }
}

filter {
  if [type] == "app_log" {
    grok { match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{GREEDYDATA:log_message}" } }
    mutate { add_field => { "[@metadata][index_name]" => "app-logs" } }
  } else if [type] == "error_log" {
    grok { match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} ERROR: %{GREEDYDATA:error_message}" } }
    mutate { add_field => { "[@metadata][index_name]" => "error-logs" } }
  } else {
    mutate { add_field => { "[@metadata][index_name]" => "other-logs" } }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "%{[@metadata][index_name]}-%{+yyyy.MM.dd}"
    ilm_enabled => false
    data_stream => false
  }
}

要点说明

• 用 [@metadata] 存索引名：@metadata 下的字段不会写入最终文档，只用于内部传递，比直接加字段干净。
• 按日期滚动：索引名里拼 %{+yyyy.MM.dd}，按事件 @timestamp 的 UTC 日期选索引；grok 仅提取 timestamp 字符串并不会改 @timestamp，需要 date filter 才按源时间路由。这里是普通索引模式，ILM rollover/data stream 需另配。
• 条件分流：if/else if 基于 type、tag 或字段值判断，确保每条日志落到正确的索引。

常见问题（FAQ）

Q：为什么索引名要用 @metadata 而不是普通字段？
A：@metadata 是 Logstash 的内部命名空间，output 阶段可用但不会写入 Elasticsearch，避免文档里多出一个无意义的 index_name 字段。

Q：动态索引名写错了会怎样？
A：如果引用的字段不存在，会得到形如 %{[field]} 的字面量索引名。建议给所有分支都设默认值，或在 output 前加条件防护。

Q：多索引用一个 output 还是多个 output？
A：变量法（一个 output + 动态索引名）最简洁；如果不同索引用不同的 ES 集群或不同的模板/管道，就需要写多个 output 块分别配条件。