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}" }
    }
    mutate { convert => { "response_time" => "float" } }
  }
  else if [type] == "syslog" {
    # syslog 消息已由插件解析，按需补充处理
  }
}

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

要点说明

1. type 是最常用的分流标记：也可以用 tags（数组，支持多个）或自定义字段。
2. 每种日志独立解析：不同格式用不同的 grok 模式，互不干扰。
3. 独立路由到不同索引：便于按日志类型设置生命周期。

观测云对照

观测云 DataKit 原生支持多源采集：日志文件、syslog、容器 stdout、journald 等可以在同一个 DataKit 中并行采集，每类来源独立的采集配置；到达平台后可用 Pipeline 按来源分流处理，比 Logstash 的单一管道更易管理。

常见问题（FAQ）

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

Q：配置越来越长不好维护怎么办？
A：Logstash 支持多管道（pipelines.yml），把不同来源拆到不同的 pipeline 文件，各自独立运行互不影响。

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