libfluetを使ってC++プログラムからfluentdにログをとばす

完全に手前味噌な記事ですがそこはご愛嬌で。

まえがき

2000年代に主なログ収集方法だったsyslogに代わって、fluentdやLogstashなどの新しいログ収集・配送方式が広く普及するようになりました。
特にfluentdはRubyによって簡単にプラグインが作成できるため、入力・出力ともに多様なプラグインが存在します。
ただC/C++のようなネイティブプログラムからログを送信する場合、ファイル出力、DB出力などを経由する必要があり、
無駄なオーバーヘッドがかかってしまうため、直接fluentdのin_forwardにデータを送り込むことが
できるlibfluentを利用します。

既存のC++によるfluentd実装としてはfluent-logger-cppがありますが、libfluentは以下のような特徴があります。

• ネストされたメッセージ構造をサポートしています。{"a": {"b": {"c": 3}}} のようなログを送信できます。
• 配列をサポートしています。 {"a": [1, 2, 3]} のようなログを遅れます。
• バッファリング＋非同期通信をサポートしています。
• 切断時の再接続とExponential backoffを実装しています。
• ログをfluentdへの送信ではなく、ローカルのファイルに書き出すことができます

libfluentのインストール

% git clone https://github.com/m-mizutani/libfluent.git
% cd libfluent
% cmake .
% make
% sudo make install

サンプルプログラム

localhostでログを待ちま変えているfluentdにtag.httpというタグのログを送信します。

#include <fluent.hpp>

int main(int argc, char *argv[]) {
  // Loggerを生成します。
  fluent::Logger *logger = new fluent::Logger();
  // new_forwardメソッドによって新しい宛先（ホストlocalhost、ポート24224）を追加します
  logger->new_forward("localhost", 24224);

  // retain_messageメソッドによりログ（Messageインスタンス）を取得します
  // 時刻はset_ts(time_t ts)で指定できますが、初期値で現在時刻がセットされます
  fluent::Message *msg = logger->retain_message("tag.http");

  // キー "url" に対して "http://github.com" という値をセットします
  msg->set("url", "http://github.com");
  // data: {"url": "http://github.com"} となります

  // キー "port" に対して 443 という値をセットします
  msg->set("port", 443);
  // data: {"url": "http://github.com", "port": 443} となります

  // ログを送信します。Asynchronousに送信されるのでMessageインスタンスを
  // deleteする必要はありません。
  logger->emit(msg);

  // Loggerインスタンスは最後に削除します
  delete logger;
}

複雑な構造のログを送信する。

ネストされたメッセージを作るサンプルは以下のとおりです。

fluent::Message *msg = logger->retain_message("test.map");
// {}
msg->set("t1", "a");
// {"t1": "a"}
fluent::Message::Map *m1 = msg->retain_map("map1");
// {"t1": "a", "map1": {}}
m1->set("t2", "b");
// {"t2": "a", "map1": {"t2": "b"}}
fluent::Message::Map *m2 = m1->retain_map("map2");
// {"t1": "a", "map1": {"t2": "b", "map2":{}}}
m2->set("t2", "b");
// {"t2": "a", "map1": {"t2": "b", "map2":{"t2": "b"}}}

fluent::Message::Mapクラスがネストされたマップ構造のインスタンスになります。基本的にfluent::Messageと同じように set(key, value) の形で値を保存できます。現在は値としてstd::string, int, double, boolを受け付けます。

また、配列を使う場合は以下のようになります。
Arrayの中にMapを入れるなども可能です。

fluent::Message *msg = logger->retain_message("test.array");
// {}
fluent::Message::Array *arr = msg->retain_array("arr1");
// {"arr1": []}
arr->push(1);
// {"arr1": [1]}
arr->push(2);
// {"arr1": [1, 2]}
fluent::Message::Map *map = arr->retain_map("map2");
// {"arr1": [1, 2, {}]}
map->set("t", "a");
// {"arr1": [1, 2, {"t": "a"}]}

ローカルファイルに出力する

本来、fluentdにログを送信できるような状況であればそれでいいはずですが、以下の様な状況で必要になると考えて実装しておきました。

• 一時的にどのようなログが出力されるかどうかを試したいとき
• 開発・デバッグ中。特にネットワークなどの関係でfluentdにログが飛ばしにくい時

ログをローカルファイルに出力するときは、new_forward(const std::string &host, int port)の代わりに、new_dumpfile(const std::string &fname)を呼び出します。

int main(int argc, char *argv[]) {
  fluent::Logger *logger = new fluent::Logger();
  logger->new_dumpfile("./dump.msg");

ファイルに出力した内容はそのままmsgpackのフォーマットなので、

% ruby -e "require 'pp'; require 'msgpack'; MessagePack::Unpacker.new(File.open('dump.msg')).each{|m| pp m }"

として確認することもできますし、

$ cat dump.msg | nc localhost 24224

のようにすることで、あとからfluentdに流しこむこともできます。