Slack Real Time Messaging APIのfluent pluginを作ってみた話

Embulkのfilter pluginを作ってみようと思ったんですが
その前にfluentdのpluginを作ってみました。
こちらです。
一応動くかと・・・

Slack RTM(Real Time Messaging)のメッセージをfluentdで取得してくれるpluginです。
なんとなく作ってみただけなので、
テストとかも無いし、Gemfileにもしていません。
そのうちする予定です。
Configから取得するところのテストくらいは書きたいです。

※SlackRTMとは
Legacy: Real Time Messaging API | Slack

SlackAPIから取得する部分の処理は
以下のライブラリを利用させてもらったので、
特に苦労しませんでしたｗ

github.com

書いたコード

書いたコードは以下の程度です。

configure(conf)
start
shutdown

がoverrideしたメソッドです。
Rubyのコード全然書いたこと無いので、{}を書いたりして大変でした←

/path/to/repo/lib/fluent/plugin/in_slackrtm.rb

require 'slack-rtmapi'

class SlackRTMInput < Fluent::Input
  Fluent::Plugin.register_input('slackrtm', self)

  def initialize
    super
  end

  def configure(conf)
    super
    @tag = conf['tag'].nil? ? '*' : conf['tag']
    @token = conf['token']
  end

  def start
    super
    @messages_thread = Thread.new do
      url = SlackRTM.get_url token: @token
      client = SlackRTM::Client.new websocket_url: url
      client.on(:message) do |data|
        emit(data)
      end
      client.main_loop
    end
    @messages_thread.abort_on_exception = true
  end

  def shutdown
    Thread.kill(@messages_thread)
  end

  def emit(data)
    time = Time.now.to_i
    Fluent::Engine.emit(@tag, time, data)
  end

end

使い方

下記のようにに指定すれば使えます。

/path/to/tmpdir/fluent.conf

<source>
    type slackrtm
    token xxxx-yyy...
</source>

<match *>
    type stdout
</match>

使うときはpオプションでファイルがあるパスを指定して使います。
-p /path/to/repo/lib/fluent/plugin という感じです。

$fluentd -c plugin_tester/slack_rtm_test.conf -p ./fluent-plugin-slack/lib/fluent/plugin/
2015-05-09 20:35:33 +0900 [info]: reading config file path="plugin_tester/slack_rtm_test.conf"
2015-05-09 20:35:33 +0900 [info]: starting fluentd-0.12.8
2015-05-09 20:35:34 +0900 [info]: gem 'fluentd' version '0.12.8'
2015-05-09 20:35:34 +0900 [info]: gem 'fluentd' version '0.12.7'
2015-05-09 20:35:34 +0900 [info]: using configuration file: <ROOT>
  <source>
    type slackrtm
    token xxxx-yyy...
  </source>
  <match *>
    type stdout
  </match>
</ROOT>
2015-05-09 20:35:34 +0900 [info]: adding match pattern="*" type="stdout"
2015-05-09 20:35:34 +0900 [info]: adding source type="slackrtm"
2015-05-09 20:35:39 +0900 *: {"type":"hello"}
...

# ctrl + c
2015-05-09 20:41:14 +0900 [info]: shutting down fluentd
2015-05-09 20:41:14 +0900 [info]: process finished code=0