こんにちは。Fluentdチームの藤田です。
今回はfluent-logger-rubyという、Rubyプログラムから手軽にログをFluentdに送信するためのライブラリを取り上げます。
Rubyには並列処理を手軽に扱えるparallelというライブラリがあり、マルチプロセスまたはマルチスレッドで処理が行えるようになります。
これらを組み合わせて、並列的にfluent-logger-rubyでログを送信したときに問題なくログが送信できるか調べてみました。 実装次第では受信データが破損するケースがあるため、注意が必要です。
並列処理でfluent-logger-rubyを使う
受信したログが壊れるケース
以下のようにParallel.eachの外でFluent::Logger::FluentLoggerのオブジェクトを作成すると、期待通りに動作しない可能性があります。
require "fluent-logger"
require "parallel"
HOST = "127.0.0.1"
PORT = 24224
DATA_LENGTH = 1024 * 1024 * 10
logger = Fluent::Logger::FluentLogger.new(nil, host: HOST, port: PORT)
Parallel.each(("1".."100"), in_processes: 4) do |str|
logger.post('test', { message: str * DATA_LENGTH })
end
このコードでは、親プロセスで作成されたFluent::Logger::FluentLoggerオブジェクトを子プロセスでも利用しようとしています。
Parallel.eachをin_processesのオプション付きで使用すると、マルチプロセス環境になり、
Fluent::Logger::FluentLoggerオブジェクトが保有しているTCPソケットをプロセス間で共有することになります。
巨大なデータをTCPソケットで送信する場合、OSのレイヤーでデータが複数に分割して送信されることがあります。 1つのソケットに対して複数のプロセスから書き込みを行っていると、データを分割送信している際に混ざることがあり、受信側でデータが壊れる可能性があります。
正しくログが転送されるケース
以下のようにログを送信すると、Fluentdはデータの破損なしに受信することができます。
require "fluent-logger"
require "parallel"
HOST = "127.0.0.1"
PORT = 24224
DATA_LENGTH = 1024 * 1024 * 10
Parallel.each(("A".."Z"), in_processes: 4) do |str|
logger = Fluent::Logger::FluentLogger.new(nil, host: HOST, port: PORT)
logger.post('test', { message: str * DATA_LENGTH })
end
この方法では、それぞれのプロセス内でFluent::Logger::FluentLoggerのオブジェクトが作成され、独立してログを送信するため期待通りに動作します。
まとめ
並列処理でfluent-logger-rubyを正しく動作させるためには、各プロセス内でFluent::Logger::FluentLoggerオブジェクトを作成することが重要です。
また、並列処理の際にはFluentdへの接続数やリソースの使用状況も考慮する必要があるため、環境に応じたチューニングを行うことをおすすめします。
以上、fluent-logger-rubyを並列処理で使用する際の注意点と正しい使い方について紹介しました。 正しく実装すれば、並列処理を活用して効率的にログをFluentdへ送信できるので、ぜひ試してみてください！
また、クリアコードはFluentdのサポートサービスを行っています。 詳しくはFluentdのサポートサービスをご覧いただき、お問い合わせフォームよりお気軽にお問い合わせください。
