試験用のfluentdサーバーを作る

fluentdを使うシステムを作っていく過程で、そもそもfluentdにデータが送られていることをどうテストしようかと悩んだので色々調べてみた。


# やりたいこと

fluentdを使うシステムを作っていく過程で、fluentdにデータが送られていることを確認するテストをサクッと書きたい。

# 方針

CIを行う上で出来る限りテストの時間を短くしたいので、FluentLoggerで投げたデータをなるべくリアルタイムでoutputに連携したい。

以前は自分でソケット開いて待ち受けるコードを書いていたけど、バージョン上がってSSLが絡んで面倒になったので、なるべくpluginとかでなんとかしたい。

CIを回す際にdockerを使って周囲のミドルウェアを立てることが多いので、以前に一度Elasticsearchをミドルウェアキットに入れて試してもみたが、直接呼んでないElasticsearchをわざわざ建ててテストのためにelasticsearch-ruby入れて繋ぎ込むのもだるい。。。
特に公式イメージだとシングルノードで起きないし、バージョン変わった時に全部のマイクロサービスで追従してアップデートするのが面倒なので、出来る限り追加のミドルウェアなしでテストしてたい。


# 方式検討

## リアルタイム性

fluentdのバッファリングはいつも`flush_interval`の値をチューニングしながらやってるけど、テストだったら`flush_mode=immediate`で行けそう

* https://docs.fluentd.org/configuration/buffer-section#flushing-parameters

## CIでの使い勝手

今回はActiveJob等でも使ってるRedisをうまく活用できないか考えたところ、[公式のプラグインリスト](https://www.fluentd.org/plugins/all)に良さげなプラグインがあったので試してみる。

* https://github.com/pokehanai/fluent-plugin-redis-store

MySQLのもあったけど、FluentLoggerから送るデータ構造が割とバラバラなので、どっちかというとそのまま入れてくれるredisのプラグインの方がマッチしてた。


# 実際にやってみた

方式検討を踏まえつつ必要なGemのインストールとfluentdの設定を進めていく。

## 0. 準備

gemは本体の他にfluent-plugin-redis-storeのプラグインがあればいいので、まとめて入れてしまう

```sh:command
gem install fluentd fluent-plugin-redis-store --no-document
```

Redisのインストールについては割愛するが、brewなりdockerなりを使ってredisは建てておく。

fluentdはforwardから受け取り、redis_storeに出していくので、fluent.confはこんな感じになる。

```xml:fluent.conf
<source>
  @type forward
  @id input1
  @label @mainstream
  port 24224
</source>

<label @mainstream>
  <filter **>
    @type record_transformer
    <record>
      tag ${tag}
    </record>
  </filter>
  <match **>
    @type copy
    <store>
      @type stdout
    </store>
    <store>
      @type redis_store
      key_path tag
      <buffer>
        flush_mode immediate
      </buffer>
    </store>
  </match>
</label>
```

やってることはこんな感じ。

* sourceとしてforwardから受け取ったものに`@mainstream`のラベルをつける
* filterとして`@mainstream`の全てのデータに対してtagというキーでタグの値をセットする
* matchとして`@mainstream`の全てのデータに対してstdoutとredis_storeの２つへアウトプットする
* redis_storeの設定として、Redisのキーとして使う値に上記のfilterでセットしたタグを設定する
* redis_storeのバッファ設定として、`flush_mode immediate`としてバッファに入った後すぐにアウトプットする

stdoutに出すのはおまけだけど、出しておくと後々デバックしやすい。

これを`./fluent.conf`に書き込み、fluentdを起動する

```sh
fluentd -c fluent.conf
```

これで準備が整ったので、データを投げ込んでみたいところだが、Redisをモニタリングしておいた方が動きがわかりやすいので、別窓を開いてモニタリングしておく

```shell
redis-cli monitor
```


## 1. fluentdにデータを投げる

fluentdにデータを送るところは、実際のアプリケーションは [fluent-logger-ruby](https://github.com/fluent/fluent-logger-ruby)を使ってデータを送るが、今回は`fluent-cat`を使って代用してみる


```shell
echo '{ "key" : "sample" }' | bundle exe fluent-cat debug.test
```

これは`{ "key" : "sample" }`というデータに `debug.test` というタグと現在時刻のタイムスタンプを付与してデータを送信してくれるコマンド。設定しない場合はlocalhost宛になるので、上記で建てたfluentdサーバーに対してデータが送信される。

そうするとfluentd側の標準出力に下記のような出力が出る。

```text:fluentdの実行ウィンドウ
2020-11-03 09:37:28.016591000 +0900 debug.test: {"key":"sample","tag":"debug.test"}
```

これは`fluent.conf`に設定した`@type stdout`が出力しているもので、実際にデータを受け取った後にfilterがタグを付加してこの形になっていることが確認できる。

さらにRedisをモニタリングしている窓を見ると下記のように表示されている。

```text:redis-cliの実行ウインドウ
1604363848.025602 [0 172.28.0.1:40998] "zadd" "debug.test" "1604363848.016591" "{\"key\":\"sample\",\"tag\":\"debug.test\"}"
```

pluginのreadmeに書いてあるとおりzaddされている。keyは`key_path tag`の設定が効いていて、データに含まれているtagの値である`debug.test`がセットされている。


## 2. 投げたデータの確認

Redisからデータを取得してみる。

```shell
redis-cli zrange debug.test 0 -1 withscores
```

```
1) "{\"key\":\"sample\",\"tag\":\"debug.test\"}"
2) "1604363848.016591"
```

タイムスタンプがスコアになってるので大量に入れてもいい感じにソートされる。

ちなみにrubyでかくとこんな感じ。

```ruby
require 'redis'

Redis.new.zrange 'debug.test', 0, -1, withscores: true
```
```ruby:output
 => [["{\"key\":\"sample\",\"tag\":\"debug.test\"}", 1604363848.016591]]
```

とりあえずシンプルに値は取れるので、割と使い勝手は良さそう。

# おまけ

## 並列テストでの使い勝手を考えてみる

`parallel_tests`や`test-queue`を使ってテストの並列実行をした際、どのテストプロセスが入れたログなのか判定する必要が出てくる。その時には送るデータにプロセスIDを含めておいて、それを使ってキーを設定する形にしてしまえばいい。

プロセスIDや時間、リクエストIDはRailsのログファイルにも入れておきたいものなので、特に害にはならないと考える。

その場合送るデータは下記のようになる

```json
{ "key" : "sample", "pid" : 123 }
```

これに伴って`fluent.conf`も下記のように変更する。

```diff
  <label @mainstream>
    <filter **>
      @type record_transformer
      <record>
        tag ${tag}
+       tag_with_pid '${tag}.${record["pid"]}'
      </record>
    </filter>
    <match **>
      @type copy
      <store>
        @type stdout
      </store>
      <store>
        @type redis_store
-       key_path tag
+       key_path tag_with_pid
        <buffer>
          flush_mode immediate
        </buffer>
      </store>
    </match>
  </label>
```

これでRedisに登録される時のキーにプロセスIDが入るので、あとはそのプロセスIDを使って自プロセスのログを特定し、それを評価することで実際にデータが送られたのかどうか判定することができるようになる。


## Redisの宛先を変える

READMEに書いてあるように下記の形で設定を変えられるらしい


```diff
  <label @mainstream>
    <filter **>
      @type record_transformer
      <record>
        tag ${tag}
      </record>
    </filter>
    <match **>
      @type copy
      <store>
        @type stdout
      </store>
      <store>
        @type redis_store
        key_path tag
+       host 10.0.0.1
+       db   11
        <buffer>
          flush_mode immediate
        </buffer>
      </store>
    </match>
  </label>
```


## RSpecやCucumberで使う時のヘルパー

実際にテストで使う時にはこんな感じのヘルパー作れば楽に使えそう

```ruby
# frozen_string_literal: true

require 'redis'

module FluentdLogHelper

  def fetch_fluentd_log_by(tag:, pid: nil)
    redis_key = pid ? "#{tag}.#{pid}" : tag
    redis.zrange redis_key, 0, -1
  end

  def redis(options = {})
    options[:db] ||= 0
    @redis ||= Redis.new(options)
  end

end
```

```ruby:シンプルな呼び方
fetch_fluentd_log_by tag: 'debug.test'
```

```ruby:プロセスIDを伴う呼び方
fetch_fluentd_log_by tag: 'debug.test', pid: Process.pid
```