fluentdに画像データを送ってみる

fluentdの使い方として適切でないような気がするのですが、画像データをfluentdに送って処理してもらうことを試しました。

ちなみにfluentdのインストール方法は、以下を参照してください。
centosにfluentdをインストールしてrubyでデータを投げてみた


imageデータを受け取って、指定したフォルダに保存してくれるpluginもある気配です。
https://github.com/bash0C7/fluent-plugin-imagefile

でも今回はデータを受け取ったら、プログラムを起動させて、そのプログラムに処理させる設定をしてみようと思います。


まずは試しに簡単なプログラムを起動させてみます。

起動させるプログラムやデータを保存する場所を準備します。
mkdir -p /var/td-agent/bin
mkdir -p /var/td-agent/tmp
mkdir -p /var/td-agent/data/test

fluentdから起動されるプログラムを作ります。
fluentdはプログラムを呼び出す際に、受け取ったデータをファイルにしてプログラムに渡すようです。

vi /var/td-agent/bin/out_test.rb

open("/var/td-agent/data/test/hoge.txt", "w") {|f|
  f.write ARGV
  f.write File.new(ARGV[0]).read
}

所有者情報を変更しておきます。
chown -R td-agent:td-agent /var/td-agent

fluetndには、以下のような設定を追加します。
vi /etc/td-agent/td-agent.conf
---------------------------
<match exec_test.**>
  type exec
  command /usr/local/bin/ruby /var/td-agent/bin/out_test.rb
  time_key got_at
  format json
  flush_interval 5s
  buffer_path /var/td-agent/tmp/buf
</match>
---------------------------

プログラムを実行させるにはtypeをexecにします。
実行させるプログラムはcommandで指定して、データを受け渡す際に受け取った時間をtime_kleyで指定したkeyに設定することができるようです。
そして、受け渡すファイルの形式をformatで選択します。jsonを選択した場合は、1行が一つのjsonデータになるようで、一つのファイルが一つのjsonになるわけではないようです。
flush_intervalで実行するプログラムの間隔を指定することになり、buffer_pathにプログラムに受け渡すファイルが一時的に格納されるようです。

設定後に設定を反映させます。
/etc/init.d/td-agent reload

テストで以下を実行してみます。
curl -X POST -d 'json={"json":"message"}' http://localhost:8888/exec_test.test

するとjsonデータが
/var/td-agent/data/test/hoge.txt
に保存されていることを確認できます。


ここで画像を受け取って保存できるようにプログラムを修正してみます。
vi /var/td-agent/bin/out_test.rb

require 'json'
require 'base64'

inf = open(ARGV[0])
inf.each do |l|
  data = JSON.parse l

  file_name=data['got_at'] + ".jpg"
  File.open("/var/td-agent/data/test/#{file_name}", 'wb') {|f|
    f.write Base64.decode64(data["image_data"])
  }
end

画像データを投げるプログラムは以下のような感じです。
vi image_send.rb

require 'fluent-logger'
require 'base64'

filename='aaa.jpg'
image=File.new(filename).read
image_data=Base64.encode64(image)
log = Fluent::Logger::FluentLogger.new(nil, :host=>'localhost', :port=>24224)
unless log.post("exec_test.image", {"file_name"=>filename,"image_data"=>image_data})
  p log.last_error # You can get last error object via last_error method
end

aaa.jpgを用意してimage_send.rbを実行すると
/var/td-agent/data/test/xxxxxxx.jp
としてaaa.jpgと同じものが保存されます。