fluentdとKinesisで雑にログを集める

aws-fluent-plugin-kinesis v0.3.0リリースされました。待望のPutRecords対応。

ただし、入力プラグインはまだマージされてません。風の噂によると近いリリースでマージされるとか、されないとか…

https://github.com/awslabs/aws-fluent-plugin-kinesis/pull/10

本家にマージはされていないのですが、フォークしたリポジトリから直接インストールすれば、Kinesisの入力プラグインが使えます。

git clone https://github.com/winebarrel/aws-fluent-plugin-kinesis.git
cd aws-fluent-plugin-kinesis
git checkout add_in_plugin
bundle install
bundle exec rake install

以下のような設定でfluentdを起動すると、ストリーム「my-stream」から最新のレコードを適用に読み込みます。

<source>
  type kinesis
  stream_name my-stream
  tag debug.kinesis
  shard_iterator_type LATEST
  db /tmp/kinesis.db
  json_data true
</source>

<match debug.**>
  type stdout
</match>

json_dataがtrueの場合、レコードデータをJSONとしてパースしてfluetndに取り込みます。 駆け込みでプルリクエストに追加したので、本家にマージされるかは微妙ですが…

で、雑にデータを突っ込みたかったのでkinesis_catというcliを作りました。gemでインストールできます。

gem install kinesis_cat

fluent-catのKinesis版で、標準入力からJSONデータを読み込んでKinsisに投げるツールです。 以下のようにJSONのデータをKinesisに入力すると…

echo '{"key":"val"}' |  kinesis-cat --stream-name my-stream

fluentdでデータを集約することができます。

2015-01-12 23:26:31 +0900 debug.kinesis:
{
  "sequence_number":"49546909991764836347635699150894029849815117811545014274",
  "partition_key":"1421072787",
  "shard_id":"shardId-000000000000",
  "key":"val"
}
(整形済み)

で？

えーっと何がよいかというと、データを送る側にfluentdが不要な点です。あとネットワークが別でもログが入れられるとかもありますね。

まあ、fluentdをインストールして起動しておくのもたいした手間ではないんですが台数が増えると結構めんどくさいですし、stream名を指定してKinesisに投げておくだけで、よきに計らってくれるのは結構便利かな…と。

2015/01/15 追記

kinesis-catのgolang版を書きました。