#はじめに
とあるきっかけで、AWS kinesisからlogを取得する、fluentdのInput Plugin、
in_kinesis なるものを作ったので、その話を書きます。

ちなみに【in_kinesis】はこちらになります。
gem版はこちらになります。

#きっかけ
今回、なぜPluginを作ったかというと、
会社で新たにリアルタイムバトルを導入する事になり、そのlog周りに携わる事になったのがきっかけでした。
そこで、今まで通りのアーキテクチャで良いのかどうかの検討から行う事にしました。
その時のlogの要件はこんな感じでした。

fluentdのBigQuery pluginが有るからBigQueryに入れるのは簡単だね！
と思っていたのですが、
ここで問題となるのが「ログの数の上限値が不明」←これでした!!

こんなやり取りがあり、白羽の矢がたったのがkinesisでした。

#kinesisとはなんぞや？
まずAWSのkinesisというサービスをご存知ですか？
kinesisとはAWSが提供している、大規模なストリーミングデータをリアルタイムで処理する完全マネージド型サービスだそうです。

大量のデータを捌くパイプみたいなイメージです。

kinesisのメリットとしてはこの2点が大きいかと思います。
・リアルタイムに大量のデータが捌けて、Shardによるスケールの容易な変更が可能。
・取り込んだデータを最大7日間保持する。

#kinesisを検討してみる
そこでもう一度、問題点と照らし合わせます。

なんという事でしょう。
色んな問題が一気に解決されたではありませんか。
と、いうわけでkinesisを使う事にしました。

構成はこんな感じです。

しかし！
肝心の、kineisに対応したinput pluginが無かったのです。
そこで、pluginは自作する事にしました。

#in_kinesisの作成
自作の際に気をつけたい事は、kinesisの良さを損なわないプラグインにする事でした。
そこで、主に下記機能を備えました。

shard毎にマルチスレッドで並列処理を行う事で、kinesisのリアルタイム処理を極力損なわない様にしています。

あと、１番の目玉は「shardの増減の自動追尾」です。
ここが自動じゃないと、結局人による管理が必要になり、運用コストがかかってしまうので、kinesisの良さを損なってしまうと考えました。

そこで、shardの増減の自動追尾を可能にする為に、各shardにアクセスするshardスレッドを管理する、「supervisorスレッド」を用意して各shardの状態を監視させる事にしました。

【in_kinesisの内部構成】

supervisorスレッドはメインスレッドとjoinせずに独立して動きます。
そして、supervisorスレッドが定期的にkinesisの状態を取得し、それに合わせてshardスレッドを起動させたり終了させしりします。

それにより、shardの増減の自動追尾が可能になりました。

あとは、多様性を上げる為に、使用するshardを意図的に指定できる機能も備えました。
これはshard毎に用途を変えて使用する事を考慮して入れているのですが、この機能を使用した場合は自動追尾ができなくなってしまうので、ストリーム毎に管理する事をお勧めします。

#最後に
in_kinesisを使う事でコンシューマーを手作りする必要が無く、fluentdでいろんなoutputに吐き出せるため、意外にも欲している人は多いのでは？と思い、今回初めてオープンソースとして作ってみることにしました。

しかし、初めてのRubyという事で、つたないRubyになってしまい、そういった面でも機能面でも、
まだまだ改良の余地あると思うので、気軽にpull request等送ってください^^ノ
優しい鉞、心よりお待ちしています。