Logstash Multiple Pipelines

做为生产者和消费者之间数据流的一个中心组件，须要一个 Logstash 实例负责驱动多个并行事件流的状况。默认状况下，这样的使用场景的配置让人并不太开心，使用者会遭遇所谓的条件地狱(Conditional hell)。由于每一个单独的 Logstash 实例默认支持一个管道，该管道由一个输入、若干个过滤器和一个输出组成，若是要处理多个数据流，就要处处使用条件判断。

条件地狱(Conditional hell)

已知的在一个管道中实现多个独立流的方法是使用条件判断。主要方式是在输入部分经过标签标记事件，而后在过滤器中和输出阶段建立条件分支，对贴有不一样标签的事件，应用不一样的插件集。这种方式虽然能够解决问题，但在实际的使用中却很是的痛苦！下面是一个简单的 demo 片断：

input { beats { port => 3444  tag => apache } tcp { port => 4222  tag => firewall } } filter { if "apache" in [tags] { dissect { ... } } else if "firewall" in [tags] { grok { ... } } } output { if "apache" in [tags] { elasticsearch { ... } } else if "firewall" in [tags] { tcp { ... } } }

对应的 Logstash 管道配置已经被条件语句包裹的十分臃肿，而它们的惟一目的是保持流的独立性。
虽然使用条件实现独立的多个流是可行的，可是很容易看出，因为存在单个管道和处理的单个阶段，随着复杂性的增长，配置会变得很是冗长，很难管理。下图展现了包含两个流的简单管道:

不幸的是，这并非该方案的惟一缺陷。

缺少拥塞隔离

若是您熟悉 Logstash 的工做原理，就会知道管道的输出部分接收到一批事件，而且在全部事件和完成全部输出以前不会移动到下一批事件。这意味着，对于上面的管道，若是 TCP 套接字目标不可达，Logstash将不会处理其余批次的事件，这也就意味着 Elasticsearch 将不会接收事件，而且会对 TCP 输入和 Beats 输入施加反压力。

不一样的数据流须要以不一样的方式处理

若是 TCP - > Grok - > TCP 数据流处理大量的小数据，而 Beats -> Dissect -> ES 数据流中的单个数据体积大可是数量少。那么前一个数据流但愿有多个 worker 并行并其每一批次处理更多事件，第二个数据流则指望使用少许的 worker 和每批次处理少许的事件。使用单个管道，没法为单个数据流指定独立的管道配置。

经过多个 Logstash 实例解决问题

上述问题能够经过在同一台机器上运行多个 Logstash 实例来解决，而后能够独立地管理这些实例。可是即便这样的解决方案也会产生其余问题：

• 须要管理多个实例(经过 init 系统管理多个后台服务)
• 每一个 Logstash 的实例也意味着一个独立的 JVM
• 须要监视每一个 Logstash 实例

这种方式其实很糟糕！

多个管道

从 Logstash 6.0 开始，引入了 Multiple Pipelines，才完美的解决了这个问题。Multiple Pipelines 的配置很是简单：在配置文件  pipelines.yml 中添加新的 pipeline 配置并指定其配置文件就能够了。下面是一个简单的 demo 配置：

- pipeline.id: apache pipeline.batch.size: 125 queue.type: persisted path.config: "/path/to/config/apache.cfg" queue.page_capacity: 50mb - pipeline.id: test pipeline.batch.size: 2 pipeline.batch.delay: 1 queue.type: memory config.string: "input { tcp { port => 3333 } } output { stdout {} }"

这个 YAML 文件包含一个散列(或字典)列表，其中每一个散列表示一个管道，键和值为该管道设置名称。被省略的设置值返回到它们的默认值。

配置多个管道

下面来看一个真实点的例子，笔者在 Ubuntu 18.04 Server 中安装了 Logstash 6.2.4，除了在默认的配置文件目录(/etc/logstash/conf.d)中添加配置文件外，建立新的目录 /etc/logstash/myconf.d，并在 /etc/logstash/myconf.d 目录下建立 Logstash 配置文件 krtest.conf。而后在 /etc/logstash/pipelines.yml 文件中添加新的 pipeline 配置：

- pipeline.id: main
  path.config: "/etc/logstash/conf.d/*.conf"
- pipeline.id: krtest path.config: "/etc/logstash/myconf.d/krtest.conf"

其中 pipeline.id 为 main 的管道是默认的配置，咱们新添加了 id 为 krtest 的管道并指定了对应的配置文件路径。启动 Logstash，若是你安装的 X-Pack 插件就能够在 Kibana->Monitoring->Logstash 中看到新添加的名称为 krtest 的管道：

使用 Multiple Pipelines 后，咱们的 Logstash 配置文件就能够写得像下面的代码同样简练(再也不须要那么多的条件语句)了：

input { beats { port => 5064 } } filter { grok { ... } } output { elasticsearch { ... } }

参考：
Multiple Pipelines
Introducing Multiple Pipelines in Logstash