add fluentd sink using fluent-logger.

fluentd[*1] is a open source log collector daemon written in Ruby. It supports several log transports using plugin architecture[*2] so that it can cover the main purpose of steno, 'multiple log destinations', such as mongodb, S3, cassandra, splunk, ...etc.

  [*1] http://fluentd.org/
  [*2] http://fluentd.org/plugin/

lib/steno/config.rb

@@ -52,6 +52,10 @@ def to_config_hash(hash)
         opts[:sinks] << Steno::Sink::Syslog.instance
       end
 
+      if hash[:fluentd]
+        opts[:sinks] << Steno::Sink::Fluentd.new(hash[:fluentd])
+      end
+
       if opts[:sinks].empty?
         opts[:sinks] << Steno::Sink::IO.new(STDOUT)
       end

lib/steno/sink.rb

@@ -1,3 +1,4 @@
 require "steno/sink/base"
 require "steno/sink/io"
 require "steno/sink/syslog"
+require "steno/sink/fluentd"

lib/steno/sink/fluentd.rb

@@ -0,0 +1,31 @@
+require 'fluent-logger'
+#
+# Steno sink implementation for Fluentd
+#
+#   See fluentd at http://fluentd.org/
+#   and fluent-logger at https://github.com/fluent/fluent-logger-ruby
+#
+class Steno::Sink::Fluentd < Steno::Sink::Base
+
+  # @param [Hash] opts Key :tag_prefix tag prefix of fluent logs (default: steno)
+  #                    Key :host fluentd host (default: 127.0.0.1)
+  #                    Key :port fluentd port (deafult: 24224)
+  #                    Key :buffer_limit buffer limit of fluent-logger
+  def initialize(opts = {})
+    super
+
+    @fluentd = Fluent::Logger::FluentLogger.new(opts[:tag_prefix] || "steno",
+      :host => opts[:host] || "127.0.0.1",
+      :port => opts[:port] || 24224,
+      :buffer_limit => opts[:buffer_limit] || Fluent::Logger::FluentLogger::BUFFER_LIMIT)
+    @io_lock = Mutex.new
+  end
+
+  def add_record(record)
+    @fluentd.post(record.source, record)
+  end
+
+  def flush
+    nil
+  end
+end

spec/unit/sink/fluentd_spec.rb

@@ -0,0 +1,46 @@
+require "spec_helper"
+
+describe Steno::Sink::IO do
+  let(:level) do
+    Steno::Logger.lookup_level(:info)
+  end
+
+  let(:record) do
+    Steno::Record.new("source", level, "message")
+  end
+
+  describe "#initialize" do
+    it "should initialize FluentLogger with the default option" do
+      Fluent::Logger::FluentLogger.should_receive(:new).with("steno", {
+        :host => "127.0.0.1",
+        :port => 24224,
+        :buffer_limit => Fluent::Logger::FluentLogger::BUFFER_LIMIT,
+        }).and_return()
+      sink = Steno::Sink::Fluentd.new()
+    end
+
+    it "should initialize FliuentLogger with override options" do
+      Fluent::Logger::FluentLogger.should_receive(:new).with("vcap", {
+        :host => "localhost",
+        :port => 8080,
+        :buffer_limit => 1024,
+        }).and_return()
+      sink = Steno::Sink::Fluentd.new({
+        :tag_prefix => "vcap",
+        :host => "localhost",
+        :port => 8080,
+        :buffer_limit => 1024
+        })
+    end
+  end
+
+  describe "#add_record" do
+    it "should post an record with the correct tag" do
+      fluentd = mock("fluentd")
+      Fluent::Logger::FluentLogger.should_receive(:new).and_return(fluentd)
+      fluentd.should_receive(:post).with("source", record)
+      sink = Steno::Sink::Fluentd.new()
+      sink.add_record(record)
+    end
+  end
+end

steno.gemspec

@@ -27,6 +27,7 @@ Gem::Specification.new do |gem|
 
   gem.add_dependency("grape")
   gem.add_dependency("yajl-ruby", "~> 1.0")
+  gem.add_dependency("fluent-logger")
 
   gem.add_development_dependency("ci_reporter")
   gem.add_development_dependency("rack-test")