脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - LogstashのPipeline-to-Pipelineで複数のbeatからの入力を受け付ける

はじめに

どうも！生産技術部で製品の検査工程を担当しているエンジニアです。脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - Elastic Stackとはの続きになります。

対象読者

この記事では、Elastic Stackをそもそも知らない方、これから試そうと思っている方を対象としています。

この記事の内容

Logstashの特定ポート（５０４４）で受け付け、複数のbeatからの入力を別のパイプラインに流しました。

GitLabに設定ファイル一式を置いておきましたので参考にしてください。
リポジトリはこちら -> elastic-stack

Pipeline-to-Pipelineの設定

beats-serverパイプラインを用意し、5044番ポートで受けます。[source]がfilebeatかmetricbeatで分岐し、それぞれのパイプラインに流します。公式のドキュメントはこちらです。sourceで判断していますが、公式通りにtypeで判断すれば良いと思います。

logstash/config/pipelines.yml

- pipeline.id: beats-server
  config.string: |
    input { beats { port => 5044 } }
    output {
        if [source] == 'filebeat' {
          pipeline { send_to => filebeatlog }
        } else if [source] == 'metricbeat' {
          pipeline { send_to => metricbeatlog }
        }
    }

- pipeline.id: filebeat-processing
  path.config: "/usr/share/logstash/pipeline/{input/filebeat_in,filter/filebeat_filter,output/filebeat_out}.cfg"
  pipeline.batch.size: 50
  pipeline.batch.delay: 50

- pipeline.id: metricbeat-processing
  path.config: "/usr/share/logstash/pipeline/{input/metricbeat_in,filter/metricbeat_filter,output/metricbeat_out}.cfg"
  pipeline.batch.size: 50
  pipeline.batch.delay: 50

inputの設定

5044番ポートで受けていたものを、pipelineのアドレスで受けるように変更します。

logstash/pipeline/input/filebeat_in.cfg

input {
#  beats {
#    port => 5044
#  }

  pipeline {
    address => filebeatlog
  }
}

Beatsの設定（公式のやり方とは違う）

logstashでどこから来たのか識別するために、sourceフィールドを設定します。（公式に従ったtypeに設定が間違いないと思います。）fields_under_rootを設定することで、filebeatから出力する時に、トップレベルのフィールドにストアすることができます。この設定をしないと正しく動作しません。

beats/filebeat/config/filebeat.yml

  fields:
    source: 'filebeat'
  fields_under_root: true

最後に

これで、複雑なパイプラインも組むことができるようになりました。
今後は、metricbeatなどについて紹介したいと思います。