如何在 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/nginx/error.log"
    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}"
  }
}

要点说明

• 用 [@metadata] 存索引名：@metadata 下的字段不会写入最终文档，只用于内部传递，比直接加字段干净。
• 按日期滚动：索引名里拼 %{+YYYY.MM.dd}，便于按天管理生命周期。
• 条件分流：if/else if 基于 type、tag 或字段值判断，确保每条日志落到正确的索引。

观测云对照

在观测云中不需要手写这类分流配置：日志上报后可以在「日志」功能下创建多个索引，通过过滤条件把不同来源的日志划入不同索引，并为每个索引单独设置存储时长和采样策略，实现成本和 retention 的精细化管理。 如果需要在采集侧做解析和分流判断，也可以用 DataKit 的 Pipeline 完成。

常见问题（FAQ）

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

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

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