Merge pull request #63 from fluent/support-event-time

Support event time

Author: repeatedly

README.md

@@ -126,7 +126,20 @@ Fluent::Logger::ConsoleLogger.open(io)
 Fluent::Logger::NullLogger.open
 ```
 
-## Buffer overflow
+## Tips
+
+### Use nanosecond-precision time
+
+To send events with nanosecond-precision time (Fluent 0.14 and up), specify `nanosecond_precision` to `FluentLogger` constructor.
+
+```rb
+log = Fluent::Logger::FluentLogger.new(nil, :host => 'localhost', :port => 24224, :nanosecond_precision => true)
+# Use nanosecond time instead
+log.post("myapp.access", {"agent" => "foo"})
+log.post_with_time("myapp.access", {"agent" => "foo"}, Time.now) # Need Time object for post_with_time
+```
+
+### Buffer overflow
 
 You can inject your own custom proc to handle buffer overflow in the event of connection failure. This will mitigate the loss of data instead of simply throwing data away.
 

fluent-logger.gemspec

@@ -30,7 +30,7 @@ EOF
   gem.executables   = `git ls-files -- bin/*`.split("\n").map{ |f| File.basename(f) }
   gem.require_paths = ['lib']
 
-  gem.add_dependency "msgpack", ">= 0.5.6", "< 2"
+  gem.add_dependency "msgpack", ">= 1.0.0", "< 2"
   gem.add_development_dependency 'rake', '>= 0.9.2'
   gem.add_development_dependency 'rspec', '>= 3.0.0'
   gem.add_development_dependency 'rspec-its', '>= 1.1.0'

lib/fluent/logger/fluent_logger.rb

@@ -23,6 +23,30 @@
 
 module Fluent
   module Logger
+    class EventTime
+      TYPE = 0
+
+      def initialize(raw_time)
+        @time = raw_time
+      end
+
+      def to_msgpack(io = nil)
+        @time.to_i.to_msgpack(io)
+      end
+
+      def to_msgpack_ext
+        [@time.to_i, @time.nsec].pack('NN')
+      end
+
+      def self.from_msgpack_ext(data)
+        new(*data.unpack('NN'))
+      end
+
+      def to_json(*args)
+        @time.to_i
+      end
+    end
+
     class FluentLogger < LoggerBase
       BUFFER_LIMIT = 8*1024*1024
       RECONNECT_WAIT = 0.5
@@ -55,6 +79,13 @@ def initialize(tag_prefix = nil, *args)
         @host = options[:host]
         @port = options[:port]
         @socket_path = options[:socket_path]
+        @nanosecond_precision = options[:nanosecond_precision]
+
+        @factory = MessagePack::Factory.new
+        if @nanosecond_precision
+          @factory.register_type(EventTime::TYPE, EventTime)
+        end
+        @packer = @factory.packer
 
         @mon = Monitor.new
         @pending = nil
@@ -98,7 +129,11 @@ def last_error
       def post_with_time(tag, map, time)
         @logger.debug { "event: #{tag} #{map.to_json}" rescue nil } if @logger.debug?
         tag = "#{@tag_prefix}.#{tag}" if @tag_prefix
-        write [tag, time.to_i, map]
+        if @nanosecond_precision && time.is_a?(Time)
+          write [tag, EventTime.new(time), map]
+        else
+          write [tag, time.to_i, map]
+        end
       end
 
       def close
@@ -147,7 +182,11 @@ def pending_bytesize
 
       def to_msgpack(msg)
         begin
-          msg.to_msgpack
+          @mon.synchronize {
+            res = @packer.pack(msg).to_s
+            @packer.clear
+            res
+          }
         rescue NoMethodError
           JSON.parse(JSON.generate(msg)).to_msgpack
         end

spec/fluent_logger_spec.rb

@@ -12,15 +12,26 @@
     DummyFluentd.new
   }
 
-  let(:logger) {
+  let(:internal_logger) {
     @logger_io = StringIO.new
-    logger = ::Logger.new(@logger_io)
-    Fluent::Logger::FluentLogger.new('logger-test', {
+    ::Logger.new(@logger_io)
+  }
+
+  let(:logger_config) {
+    {
       :host   => 'localhost',
       :port   => fluentd.port,
-      :logger => logger,
+      :logger => internal_logger,
       :buffer_overflow_handler => buffer_overflow_handler
-    })
+    }
+  }
+
+  let(:logger) {
+    Fluent::Logger::FluentLogger.new('logger-test', logger_config)
+  }
+
+  let(:logger_with_nanosec) {
+    Fluent::Logger::FluentLogger.new('logger-test', logger_config.merge(:nanosecond_precision => true))
   }
 
   let(:buffer_overflow_handler) { nil }
@@ -65,6 +76,16 @@
         expect(logger.pending_bytesize).to eq 0
       }
 
+      if defined?(Fluent::EventTime)
+        it ('success with nanosecond') {
+          expect(logger_with_nanosec.pending_bytesize).to eq 0
+          expect(logger_with_nanosec.post('tag', {'a' => 'b'})).to be true
+          fluentd.wait_transfer
+          expect(fluentd.queue.last).to eq ['logger-test.tag', {'a' => 'b'}]
+          expect(fluentd.output.emits.first[1]).to be_a_kind_of(Fluent::EventTime)
+        }
+      end
+
       it ('close after post') {
         expect(logger).to be_connect
         logger.close