Merge pull request #59 from jcalvert/unix_socket_support

Add UNIX socket support

Author: repeatedly

README.md

@@ -19,6 +19,19 @@ end
 # output: myapp.access {"agent":"foo"}
 ```
 
+### UNIX socket
+
+```ruby
+require 'fluent-logger'
+
+log = Fluent::Logger::FluentLogger.new(nil, :socket_path  => "/tmp/fluent.sock")
+unless log.post("myapp.access", {"agent"=>"foo"})
+  p log.last_error # You can get last error object via last_error method
+end
+
+# output: myapp.access {"agent":"foo"}
+```
+
 ### Singleton
 ```ruby
 require 'fluent-logger'

lib/fluent/logger/fluent_logger.rb

@@ -54,6 +54,7 @@ def initialize(tag_prefix = nil, *args)
         @tag_prefix = tag_prefix
         @host = options[:host]
         @port = options[:port]
+        @socket_path = options[:socket_path]
 
         @mon = Monitor.new
         @pending = nil
@@ -63,7 +64,6 @@ def initialize(tag_prefix = nil, *args)
         @log_reconnect_error_threshold = options[:log_reconnect_error_threshold] ||  RECONNECT_WAIT_MAX_COUNT
 
         @buffer_overflow_handler = options[:buffer_overflow_handler]
-
         if logger = options[:logger]
           @logger = logger
         else
@@ -108,7 +108,7 @@ def close
               send_data(@pending)
             rescue => e
               set_last_error(e)
-              @logger.error("FluentLogger: Can't send logs to #{@host}:#{@port}: #{$!}")
+              @logger.error("FluentLogger: Can't send logs to #{connection_string}: #{$!}")
               call_buffer_overflow_handler(@pending)
             end
           end
@@ -123,6 +123,18 @@ def connect?
         @con && !@con.closed?
       end
 
+      def create_socket!
+        if @socket_path
+          @con = UNIXSocket.new(@socket_path)
+        else
+          @con = TCPSocket.new(@host, @port)
+        end
+      end
+
+      def connection_string
+        @socket_path ? "#{@socket_path}" : "#{@host}:#{@port}"
+      end
+
       private
       def to_msgpack(msg)
         begin
@@ -170,7 +182,7 @@ def write(msg)
           rescue => e
             set_last_error(e)
             if @pending.bytesize > @limit
-              @logger.error("FluentLogger: Can't send logs to #{@host}:#{@port}: #{$!}")
+              @logger.error("FluentLogger: Can't send logs to #{connection_string}: #{$!}")
               call_buffer_overflow_handler(@pending)
               @pending = nil
             end
@@ -203,7 +215,7 @@ def send_data(data)
       end
 
       def connect!
-        @con = TCPSocket.new(@host, @port)
+        create_socket!
         @con.sync = true
         @connect_error_history.clear
         @logged_reconnect_error = false
@@ -230,7 +242,7 @@ def call_buffer_overflow_handler(pending)
       end
 
       def log_reconnect_error
-        @logger.error("FluentLogger: Can't connect to #{@host}:#{@port}(#{@connect_error_history.size} retried): #{$!}")
+        @logger.error("FluentLogger: Can't connect to #{connection_string}(#{@connect_error_history.size} retried): #{$!}")
       end
 
       def set_last_error(e)

spec/fluent_logger_spec.rb

@@ -230,4 +230,44 @@ def flush(messages)
       end
     end
   end
+
+  context "using socket_path" do
+
+    let(:socket_logger) {
+      @logger_io = StringIO.new
+      logger = ::Logger.new(@logger_io)
+      Fluent::Logger::FluentLogger.new('logger-test', {
+        :socket_path   => fluentd.socket_path,
+        :logger => logger,
+        :buffer_overflow_handler => buffer_overflow_handler
+      })
+    }
+
+    context "running fluentd" do
+      before(:all) do
+        @serverengine = DummyServerengine.new
+        @serverengine.startup
+      end
+
+      before(:each) do
+        fluentd.socket_startup
+      end
+
+      after(:each) do
+        fluentd.shutdown
+      end
+
+      after(:all) do
+        @serverengine.shutdown
+      end
+
+      context('post') do
+        it ('success') {
+          expect(socket_logger.post('tag', {'b' => 'a'})).to be true
+          fluentd.wait_transfer
+          expect(fluentd.queue.last).to eq ['logger-test.tag', {'b' => 'a'}]
+        }
+      end
+    end
+  end
 end

spec/support/dummy_fluentd.rb

@@ -10,6 +10,7 @@ def initialize
   end
 
   WAIT = ENV['WAIT'] ? ENV['WAIT'].to_f : 0.3
+  SOCKET_PATH = ENV['SOCKET_PATH'] ? ENV['SOCKET_PATH'] : "/tmp/dummy_fluent.sock"
 
   def wait_transfer
     sleep WAIT
@@ -29,6 +30,10 @@ def port
     @port
   end
 
+  def socket_path
+    SOCKET_PATH
+  end
+
   def output
     sleep 0.0001 # next tick
     if Fluent::Engine.respond_to?(:match)
@@ -66,6 +71,26 @@ def startup
     wait_transfer
   end
 
+  def socket_startup
+    config = Fluent::Config.parse(<<EOF, '(logger-spec)', '(logger-spec-dir)', true)
+<source>
+  type unix
+  path #{socket_path}
+</source>
+<match logger-test.**>
+  type test
+</match>
+EOF
+    Fluent::Test.setup
+    Fluent::Engine.run_configure(config)
+    @coolio_default_loop = nil
+    @thread = Thread.new {
+      @coolio_default_loop = Coolio::Loop.default
+      Fluent::Engine.run
+    }
+    wait_transfer
+  end
+
   def shutdown
     @coolio_default_loop.stop rescue nil
     begin