1 写在前面
Logstash 被称为数据处理管道，由输入插件、过滤器插件、输出插件三部分构成，在常规的配置下，一个 Logstash 实例中只有一个管道运行，从 6.0 版本后，官方逐渐推出了一些多管道的配置，能够在一个 Logstash 实例中启动多个管道执行。
2 版本说明
Logstash 6.3.0 以上
3 配置方案
3.1 单管道配置
先讲运行单个 Logstash 实例的情况，在最基本的单个管道配置下，又有以下几种方案来实现不同的输出方式。
3.1.1 单输出插件
显而易见，如果只配置一个输出插件，Logstash 便只能向一个地方输出结果。
一份简略的配置如下：
input {
file {
path => "/tmp/test.log"
}
}
filter {
json {
skip_on_invalid_json => true
source => "message"
}
}
output {
stdout {}
}
3.1.2 多输出插件
更进一步，如果配置多个输出插件，Logstash 运行正常的情况下，就能依次向多个地方输出结果。
为什么说运行正常的情况下？因为前一个输出插件如果出了意外，导致消息不能被正常处理，后面的插件就不能继续收到消息，所以这样的方式配置多个输出插件时，插件之间会存在相互影响。
一份简略的配置如下：
input {
file {
path => "/tmp/test.log"
}
}
filter {
urldecode {}
}
output {
http {
method => "get"
url => "http://127.0.0.1:8080/test"
}
stdout {}
}
3.1.3 条件输出
如果需要根据不同的情况，执行不同的输出逻辑，则可以在配置文件中使用 if…else… 语句实现。
一份简略的配置如下：
input {
file {
path => "/tmp/test.log"
}
}
filter {
json {
skip_on_invalid_json => true
source => "message"
}
}
output {
if [testParam] == "http" {
http {
method => "get"
url => "http://127.0.0.1:8080/test"
}
stdout {}
} else {
tcp {
host => "192.168.10.123"
port => 9999
}
}
}
3.2 多管道配置
在上一节单管道配置的基础上，现在可以实现单个 Logstash 实例运行多个管道，其中每个管道的配置跟上述单管道一致，这里就不再赘述，主要讲讲多管道的几种模式。
官方在 6.0 版本时推出了 multiple-pipelines 的功能，让 Logstash 能在一个实例内同时启动多个管道；之后又在 6.3 版本时推出了 pipeline-to-pipeline 的功能，让同一个 Logstash 实例内的不同管道间可以产生逻辑关系。
注意，多管道的使用必须使用 pipelines.yml 文件，这个配置文件中的配置有两点要注意，首先要指定不重复的 pipeline.id，其次要说明每个管道的插件配置，可以使用 path.config 指定插件配置文件的路径，也可以使用 config.string 直接把插件的配置写在该 pipelines.yml 文件中。而其余的配置，都跟 logstash.yml 中一样。
同时，启动 Logstash 时不再需要使用 -f 参数指定专门的配置文件，它会默认使用 pipelines.yml 的配置启动。
3.2.1 多管道相互独立
如果想在一个 Logstash 中启动多个管道，让它们各自运行，互不影响，在 6.0 版本后已经可以实现。
官方文档给的一份示例配置如下：
- pipeline.id: my-pipeline_1
path.config: "/etc/path/to/p1.config"
pipeline.workers: 3
- pipeline.id: my-other-pipeline
path.config: "/etc/different/path/p2.cfg"
queue.type: persisted
它使用插件配置文件 p1.config 启动了一个叫 my-pipeline_1 的管道，又设置这个管道的工作线程为 3 个；而同时又使用 p2.cfg 启动了一个叫 my-other-pipeline 的管道，设置这个管道的队列为持久化管道。
另外，官方还有一篇博客《如何创建易于维护和再利用的 Logstash 管道》介绍了这样的玩法：
- pipeline.id: my-pipeline_1
path.config: "<path>/{01_in,01_filter,02_filter,01_out}.cfg"
- pipeline.id: my-pipeline_2
path.config: "<path>/{02_in,02_filter,03_filter,01_out}.cfg"
简单来说，就是把各种输入插件、过滤器、输出插件拆分出来，成为不同的配置文件，再在 pipelines.yml 中使用 glob 表达式拼接它们，其中的好处是复用相同的插件配置，避免某一处问题修改还要到多个配置文件中去做。
3.2.2 多管道输出单管道
如果想要让事件（数据）在同一个 Logstash 实例上的多个管道内依次流动（处理），可以使用 pipeline-to-pipeline 功能，Logstash 提供了一个 pipeline 输出插件，可以将当前管道内的数据发送至 Logstash 实例内的虚拟服务上的指定虚拟地址，同时又提供了一个 pipeline 输入插件，根据虚拟地址，从虚拟服务中获取这些数据。
比如有多个不同来源的数据，分别经过各自的管道处理后，再流向通用的输出管道做最终处理，形成一种汇聚型的结构。
参考官方博客的一个示例配置：
- pipeline.id: beats
config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => [commonOut] } }
- pipeline.id: kafka
config.string: |
input { kafka { ... } }
output { pipeline { send_to => [commonOut] } }
- pipeline.id: partner
# This common pipeline enforces the same logic whether data comes from Kafka or Beats
config.string: |
input { pipeline { address => commonOut } }
filter {
# Always remove sensitive data from all input sources
mutate { remove_field => 'sensitive-data' }
}
output { elasticsearch { } }
它使用 beats 插件同时又使用 kafka 插件采集不同地方的数据，分别使用 pipeline 插件将所有的数据发送到 commonOut 这个虚拟地址，最后启动一个专门的管道，从该地址接收全部的数据，写入 elasticsearch。
3.2.3 单管道输出多管道
除了汇聚形式的管道搭配，还有各种分发形式的使用，从一个管道接收数据再传递到多个下游管道。
同样可以使用 if…else… 语句进行选择，究竟要发到哪个管道：
- pipeline.id: beats-server
config.string: |
input { beats { port => 5044 } }
output {
if [type] == apache {
pipeline { send_to => weblogs }
} else if [type] == system {
pipeline { send_to => syslog }
} else {
pipeline { send_to => fallback }
}
}
- pipeline.id: weblog-processing
config.string: |
input { pipeline { address => weblogs } }
filter {
# Weblog filter statements here...
}
output {
elasticsearch { hosts => [es_cluster_a_host] }
}
- pipeline.id: syslog-processing
config.string: |
input { pipeline { address => syslog } }
filter {
# Syslog filter statements here...
}
output {
elasticsearch { hosts => [es_cluster_b_host] }
}
- pipeline.id: fallback-processing
config.string: |
input { pipeline { address => fallback } }
output { elasticsearch { hosts => [es_cluster_b_host] } }
或者直接把数据发到全部下游管道：
- pipeline.id: intake
queue.type: persisted
config.string: |
input { beats { port => 5044 } }
output { pipeline { send_to => [es, http] } }
- pipeline.id: buffered-es
queue.type: persisted
config.string: |
input { pipeline { address => es } }
output { elasticsearch { } }
- pipeline.id: buffered-http
queue.type: persisted
config.string: |
input { pipeline { address => http } }
output { http { } }
3.3 多 Logstash 实例
除此之外，当然，还可以直接根据上述单管道或多管道的配置，直接启动多个 Logstash 实例。
各实例间或毫无关联，或依然存在一定的逻辑关系，这里就不再举例了。
4 最后
Logstash 的应用可以说非常灵活，尤其是官方引入了多管道配置后，可以在一个 Logstash 实例内实现多个管道的搭配运行，方便了整个 Logstash 服务的管理。