Logstash 如何同时处理多个不同来源的输入？

核心模式：在每个 input 插件里设置 type => "名称" 打标签，然后在 filter 和 output 中用 if [type] == "名称" 条件语句对不同来源的日志分别做解析和路由。

完整配置示例

input {
  file {
    type => "web_server_logs"
    path => "/var/log/apache/access.log"
  }
  file {
    type => "application_logs"
    path => "/var/log/app/application.log"
  }
  syslog {
    type => "syslog"
    port => 514
  }
}

filter {
  if [type] == "web_server_logs" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
  else if [type] == "application_logs" {
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:msg}" }
    }
  }
  else if [type] == "syslog" {
    # syslog 消息已由插件解析，按需补充处理
  }
}

output {
  if [type] == "web_server_logs" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      ilm_enabled => false
      data_stream => false
      index => "web-logs-%{+YYYY.MM.dd}"
    }
  }
  else if [type] == "application_logs" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      ilm_enabled => false
      data_stream => false
      index => "app-logs-%{+YYYY.MM.dd}"
    }
  }
  else if [type] == "syslog" {
    elasticsearch {
      hosts => ["http://localhost:9200"]
      ilm_enabled => false
      data_stream => false
      index => "syslog-logs-%{+YYYY.MM.dd}"
    }
  }
  else { stdout { codec => rubydebug } }
}

要点说明

1. type 可作为分流标记：input 的 type 通常不会覆盖事件已有的 type，需核对输入内容；也可以用 tags（数组，支持多个）或自定义字段。
2. 每种日志独立解析：不同格式用不同的 grok 模式，互不干扰。
3. 独立路由到不同索引：便于按日志类型设置生命周期。日期取事件 @timestamp，未解析应用 timestamp 时可能是采集时间；生产还需配置 TLS、认证和写入/ILM 策略。

常见问题（FAQ）

Q：type 和 tags 有什么区别？
A：type 是单值字段，tags 是数组（一个事件可有多个 tag）。简单分流用 type，需要多维度标记（如 "nginx" + "production"）时用 tags。

Q：配置越来越长不好维护怎么办？
A：Logstash 支持多管道（pipelines.yml），把不同来源拆到不同的 pipeline 文件，可隔离队列和背压，但仍共享同一 JVM/主机资源，不能保证互不影响。

Q：条件判断性能如何？
A：事件量大时每个事件都要过一遍所有条件判断。可以用 if/else if 结构确保只命中一个分支，或拆分成多管道。