fluent-plugin-beatsを作った


最近Elastic社が力を入れているBeatsと呼ばれる，各種イベントに特化したエージェントがあります．

[Beats: Data Shippers for Elasticsearch  | Elastic](https://www.elastic.co/products/beats)

現状だとシステムのメトリクスを集める`topbeats`，ネットワーク周りの情報を集める`packetbeat`，ファイルを監視してログを集める`filebeat`(fluentdのin_tailみたいなもの)があります．

これらは単体でElasticsearchに投げることが出来るんですが，フィールドを加工したりとか，様々なデータストアに投げるとか，あまり複雑なことは出来ません．
ですが，そういう時のために`lumberjack`というプロトコルを使って外部に投げることが出来るようになってます．ログを集めて加工したり転送したりといえば，軽量でロバストなFluentdというビンゴなソフトウェアがあるので，これらとやりとり出来たら便利だなぁとfluent-plugin-beatsを作りました！

[fluent-plugin-beats](https://github.com/repeatedly/fluent-plugin-beats)

ということで，以下プラグインの説明をします

## インストール

いつも通りgem経由です

```
$ gem install fluent-plugin-beats --no-document
```

## 基本設定

```aconf
<source>
  @type beats
  tag beats.event
</source>
```

複雑な設定はありません．デフォルトで`0.0.0.0:5044`で起動し，topbeat，packetbeat，filebeatすべてのイベントを受け付けます．この設定だと固定タグですが，`tag`の代わりに`metadata_as_tag`を設定すると，各イベントが持っているメタデータの設定を使うようになり，topbeatなら`topbeat`というタグが付くようになります．以下の"Elasticsearchに投げる設定例"を参照してください．

filebeatが送ってくるイベントにはログそのものが含まれているので，これを`beats`プラグインの中でパースしたい場合には，in_tailのように`format`を指定すれば処理してくれるようになります．

その他書いてない設定に関しては[READMEの設定セクション](https://github.com/repeatedly/fluent-plugin-beats#configuration)を参照してください．


## Elasticsearchに投げる設定例

各beatはlogstashのインデックスと同じ仕組みでインデックスを作るので(xxxbeat-YYYY.MM.DD)，用意されているテンプレートを使うためには，それらに合わせる必要があります．以下が設定の実例です．

```aconf
<source>
  @type beats
  metadata_as_tag
</source>

# xxxbeatのログを全部ESに投げる
<match *beat>
  @type elasticsearch_dynamic
  logstash_format true
  logstash_prefix ${tag_parts[0]}  # metadata_as_tagによりタグに各beatの名前が入っているので，それを使う
  type_name ${record['type']}      # レコードにインデックスのtypeが含まれているので，それを利用する
  flush_interval 10s
</match>
```

一つにまとめるために`elasticsearch_dynamic`を使ってますが，`filebeat`はデフォルトで常に`type_name`が`log`なので，流量が大きければ通常の`elasticsearch`にするのも手です．

もちろん，`copy`プラグインを使ってS3やTreasure Dataみたいなサービスにバックアップ用途で投げることが出来ます．topbeatのデータは長期に保存する必要性は薄いですが，filebeatの結果はログで結構膨れるので，その手のサービスに入れておくのは有りだと思います．

以下は各beatの出力をfluentd経由で保存した結果になります．beatから直接送ったのとちゃんと同じ結果になっていると思います．

![](https://i.gyazo.com/edebad53fa5a6e70eaf52198ef3011c1.png)

## 注意点

### コネクション毎にスレッドを作る

このプラグインは，logstash-input-beatsプラグインが使っているlumberjackモジュールを流用していますが，このモジュールが1コネクション毎にスレッドを作る前提の実装になっているので，たくさんのbeatからアクセスが来るとそれだけスレッドが立ち上がります．
`max_connections`という設定で最大数を制御できるので，問題になりそうな方は設定してください．一応スレッドプールとかを使ってはいますが，解決するにはCool.ioとかのモジュールで根本から修正しないと行けない．

### filebeatとの相性

lumberjackモジュールがバルクでデータを取得するというAPIを提供してくれていないので(受け取ったデータをばらして渡してくる)，実は流量が多いとそんなに効率がよくありません．これはlumberjackが1レコードずつackを送信元に返すプロトコルであるのにも起因しています．

10万行のnginxログを食わせて`flowcounter_simple`を使って手元のMBPでパフォーマンスを計測してみたところ，fluentdのin_tailが`format none`で80,000/sec，fluent-agent-hydraが100,000+/sec，filebeatが18,000/secと，同一マシン内ですら遅い結果になっています．実環境だとネットワークを跨ぐので，1レコードずつのackによる速度低下がもっと顕著になると思います([Filebeat performance when sending to Logstas](https://github.com/elastic/beats/issues/587)というissueで同じくパフォーマンス問題が指摘されている)．
パフォーマンスを気にするなら，fluentdを直接使うか，末端ではよりエコなfluent-agent-hydraを使う方が良いと思います．

### logstash-input-beatsの実装

fluent-plugin-beatsはいくつかの部分でlogstash-input-beatsの実装を参考にしました．で，色々と気づいたことがあったので，logstash-input-beatsを使う方のためにメモ代わりに書いて起きます．

- logstashの内部キューが詰まるとレスポンスを返さなくなる

logstashの内部キューがブロッキングキューなのは有名ですが，それらで問題があった時のために，コネクションをacceptするまえにCircuitBreakerで処理するかどうか判断しています．で，CircuitBreakerが働いている限りはひたすら`sleep(0.5)`をループし続けるだけの処理になっていて，だんまりしてしまいます．
設定でそのタイムアウトを調整できるようなので，気になる方は調整しましょう．

- beatsプラグイン自体もオンメモリキューを持っている

logstashはFluentdや他のログコレクタと違い，データをためる内部キューやOutputのバッファが現在オンメモリでしか持てないという非常に厳しい制限があり，かつブロッキングなのは上で述べました．で，実はこのキューはブロッキング時のタイムアウトすら設定出来ないという実装になっているらしく，上記のタイムアウトを実現するために，さらにbeatsプラグインの中で別のタイムアウト可能なオンメモリキューをもつことでタイムアウトを実現しています．
その結果，オンメモリのブロッキングキューを二つ持つことになり，処理がブロックしやすく，データの欠損がさらに起きやすくなっているという感じです．気をつけましょう．

## まとめ

個人的にBeats自体はかなり使いやすさそうだなぁと感じていて，組み込み向けエージェントならfluent-bit，通常のサーバとかならBeats，ログ集約・転送にはfluentdと，それぞれ分けて使えそうだなぁと思っています．topbeat/packetbeatを使えば，Elasticsearchを運用出来るリソースがあれば，Kibanaなどと合わせて簡易的なモニタリングシステムを構築できるので，今後この辺のユーザは増えそうですね．

### 追記

Elasticsearch勉強会#14で話したスライド: [Fluentd meets Beats](http://www.slideshare.net/repeatedly/fluentpluginbeats-at-elasticsearch-meetup-14)