KafkaのメトリクスをFluentdに送信するプラグインを書きました

PLAZMA OSS Day: TD Tech Talk 2018 – Cooperative works for Fluentd Community の補足記事です。

背景

kafka-connect-fluentdとfluent-pluguin-kafkaの性能を比較したいと考えたときに、同じ指標で性能を比較したいと考えていました。 kafka-connect-fluentdのFluentdSourceConnectorとfluent-plugin-kafkaのoutputプラグインはどちらもKafkaに書き込むので、Kafkaのスループットを見ればよいはずです。

幸いKafkaにはメトリックスを取るためのAPIが提供されていました。

しかしKafkaに組み込みのKafkaCSVMetricsReporterは、動きませんでした¹。他にもいくつか既存のMetricsReporterを探してみましたがちょうどよいものはありませんでした。 Kafkaのメトリックスも、同じフィールドに型の異なる値が入っていることがあり、そのままでは扱いづらいものでした。

そこで、全てのメトリックスをFluentdに流せば、それ以降はFluentdのプラグインで好きなように加工できるので素早くデータを可視化するための環境を作ることができそうだと考えました。またKafkaの提供するメトリックスは大量にあるので、kafka-connect-fluentdとfluent-plugin-kafkaの性能比較に使えるものだけを選別することも簡単にできそうだと考えていました。

kafka-fluent-metrics-reporter

kafka-fluent-metrics-reporterを作りました。

KafkaのプラグインはScalaで書かれていたり、Javaで書かれていたり、Kotlinで書かれていたり様々なので、これもKotlinで実装しました。

特に工夫したところはなく、Kafkaの提供してくれるメトリックスをmapにつめてFluentdに送るだけでした。

使い方

こちらも未リリースなので自分でビルドする必要があります。

$ git clone https://github.com/okkez/kafka-fluent-metrics-reporter.git
$ cd kafka-fluent-metrics-reporter
$ ./gradlew shadowJar
$ cp build/libs/kafka-fluent-metrics-reporter-1.0-SNAPSHOT-all.jar /path/to/kafka_2.11-1.0.0/libs

以下の設定をKafka Serverの設定ファイル server.properties に追加します。

kafka.metrics.reporters=org.fluentd.kafka.metrics.KafkaFluentMetricsReporter
kafka.metrics.polling.interval.secs=5
kafka.fluent.metrics.enabled=true
kafka.fluent.metrics.host=localhost
kafka.fluent.metrics.port=24224
kafka.fluent.metrics.tagPrefix=kafka-metrics

Fluentd側は以下のように設定します。このように設定してしばらく流してみると、大体様子がわかると思います。

<source>
  @type forward
  port 24224
</source>

<match kafka-metrics.**>
  @type copy
  <store>
    @type file
    path log/${tag}
    <buffer tag>
    </buffer>
  </store>
  <store>
    @type stdout
  </store>
</match>

タグにメトリックスの名前が入ってくるので、必要なメトリックスをタグで絞り込むことができます。