Support nanosecond

.travis.yml

@@ -1,7 +1,6 @@
 language: ruby
 
 rvm:
-  - 1.9.3
   - 2.0.0
   - 2.1
   - 2.2

fluentd.gemspec

@@ -19,7 +19,7 @@ Gem::Specification.new do |gem|
 
   gem.required_ruby_version = '>= 1.9.3'
 
-  gem.add_runtime_dependency("msgpack", [">= 0.5.11", "< 0.6.0"])
+  gem.add_runtime_dependency("msgpack", [">= 0.7.0"])
   gem.add_runtime_dependency("json", [">= 1.4.3"])
   gem.add_runtime_dependency("yajl-ruby", ["~> 1.0"])
   gem.add_runtime_dependency("cool.io", [">= 1.4.1", "< 2.0.0"])
@@ -34,6 +34,7 @@ Gem::Specification.new do |gem|
     gem.add_runtime_dependency("win32-event", ["~> 0.6.1"])
     gem.add_runtime_dependency("windows-pr", ["~> 1.2.3"])
   end
+  gem.add_runtime_dependency("strptime", [">= 0.1.3"])
 
   gem.add_development_dependency("rake", [">= 0.9.2"])
   gem.add_development_dependency("flexmock", ["~> 1.3.3"])

lib/fluent/buffer.rb

@@ -112,7 +112,7 @@ def write_to(io)
 
     def msgpack_each(&block)
       open {|io|
-        u = MessagePack::Unpacker.new(io)
+        u = Fluent::Engine.msgpack_factory.unpacker(io)
         begin
           u.each(&block)
         rescue EOFError

lib/fluent/command/cat.rb

@@ -289,7 +289,7 @@ def abort_message(time, record)
 
 when 'msgpack'
   begin
-    u = MessagePack::Unpacker.new($stdin)
+    u = Fluent::Engine.msgpack_factory.unpacker($stdin)
     u.each {|record|
       w.write(record)
     }

lib/fluent/engine.rb

@@ -30,13 +30,17 @@ def initialize
       @log_event_queue = []
 
       @suppress_config_dump = false
+
+      @msgpack_factory = MessagePack::Factory.new
+      @msgpack_factory.register_type(Fluent::EventTime::TYPE, Fluent::EventTime)
     end
 
     MATCH_CACHE_SIZE = 1024
     LOG_EMIT_INTERVAL = 0.1
 
     attr_reader :root_agent
     attr_reader :matches, :sources
+    attr_reader :msgpack_factory
 
     def init(opts = {})
       BasicSocket.do_not_reverse_lookup = true
@@ -132,7 +136,7 @@ def flush!
 
     def now
       # TODO thread update
-      Time.now.to_i
+      Fluent::EventTime.now
     end
 
     def log_event_loop

lib/fluent/event.rb

@@ -27,12 +27,20 @@ def each(&block)
     end
 
     def to_msgpack_stream
-      out = MessagePack::Packer.new # MessagePack::Packer is fastest way to serialize events
+      out = Fluent::Engine.msgpack_factory.packer
       each {|time,record|
         out.write([time,record])
       }
       out.to_s
     end
+
+    def to_msgpack_stream_forced_integer
+      out = Fluent::Engine.msgpack_factory.packer
+      each {|time,record|
+        out.write([time.to_i,record])
+      }
+      out.to_s
+    end
   end
 
 
@@ -143,7 +151,7 @@ def repeatable?
 
     def each(&block)
       # TODO format check
-      unpacker = MessagePack::Unpacker.new
+      unpacker = Fluent::Engine.msgpack_factory.unpacker
       unpacker.feed_each(@data, &block)
       nil
     end

lib/fluent/load.rb

@@ -9,13 +9,15 @@
 require 'yajl'
 require 'uri'
 require 'msgpack'
+require 'strptime'
 begin
   require 'sigdump/setup'
 rescue
   # ignore setup error on Win or similar platform which doesn't support signal
 end
 require 'cool.io'
 
+require 'fluent/time'
 require 'fluent/env'
 require 'fluent/version'
 require 'fluent/log'

lib/fluent/mixin.rb

@@ -17,13 +17,24 @@
 module Fluent
   class TimeFormatter
     require 'fluent/timezone'
+    require 'fluent/time'
 
     def initialize(format, localtime, timezone = nil)
       @tc1 = 0
       @tc1_str = nil
       @tc2 = 0
       @tc2_str = nil
 
+      if format && format =~ /(^|[^%])(%%)*%L|(^|[^%])(%%)*%\d*N/
+        define_singleton_method(:format) {|time|
+          format_with_subsec(time)
+        }
+      else
+        define_singleton_method(:format) {|time|
+          format_without_subsec(time)
+        }
+      end
+
       if formatter = Fluent::Timezone.formatter(timezone, format)
         define_singleton_method(:format_nocache) {|time|
           formatter.call(time)
@@ -54,7 +65,7 @@ def initialize(format, localtime, timezone = nil)
       end
     end
 
-    def format(time)
+    def format_without_subsec(time)
       if @tc1 == time
         return @tc1_str
       elsif @tc2 == time
@@ -72,6 +83,28 @@ def format(time)
       end
     end
 
+    def format_with_subsec(time)
+      if Fluent::EventTime.eq?(@tc1, time)
+        return @tc1_str
+      elsif Fluent::EventTime.eq?(@tc2, time)
+        return @tc2_str
+      else
+        str = format_nocache(time)
+        if @tc1 < @tc2
+          @tc1 = time
+          @tc1_str = str
+        else
+          @tc2 = time
+          @tc2_str = str
+        end
+        return str
+      end
+    end
+
+    def format(time)
+      # will be overridden in initialize
+    end
+
     def format_nocache(time)
       # will be overridden in initialize
     end

lib/fluent/output.rb

@@ -431,13 +431,19 @@ def flush_secondary(secondary)
 
 
   class ObjectBufferedOutput < BufferedOutput
+    config_param :time_as_integer, :bool, :default => true
+
     def initialize
       super
     end
 
     def emit(tag, es, chain)
       @emit_count += 1
-      data = es.to_msgpack_stream
+      if @time_as_integer
+        data = es.to_msgpack_stream_forced_integer
+      else
+        data = es.to_msgpack_stream
+      end
       key = tag
       if @buffer.emit(key, data, chain)
         submit_flush
@@ -529,7 +535,7 @@ def configure(conf)
       else
         @flush_interval = [60, @time_slice_cache_interval].min
         @enqueue_buffer_proc = Proc.new do
-          nowslice = @time_slicer.call(Engine.now - @time_slice_wait)
+          nowslice = @time_slicer.call(Time.now - @time_slice_wait)
           @buffer.keys.each {|key|
             if key < nowslice
               @buffer.push(key)

lib/fluent/parser.rb

@@ -60,12 +60,18 @@ def initialize(time_format)
         @cache2_time = nil
         @parser =
           if time_format
-            Proc.new { |value| Time.strptime(value, time_format) }
+            begin
+              strptime = Strptime.new(time_format)
+              Proc.new { |value| Fluent::EventTime.from_time(strptime.exec(value)) }
+            rescue
+              Proc.new { |value| Fluent::EventTime.from_time(Time.strptime(value, time_format)) }
+            end
           else
-            Time.method(:parse)
+            Proc.new { |value| Fluent::EventTime.parse(value) }
           end
       end
 
+      # TODO: new cache mechanism using format string
       def parse(value)
         unless value.is_a?(String)
           raise ParserError, "value must be string: #{value}"
@@ -77,7 +83,7 @@ def parse(value)
           return @cache2_time
         else
           begin
-            time = @parser.call(value).to_i
+            time = @parser.call(value)
           rescue => e
             raise ParserError, "invalid time format: value = #{value}, error_class = #{e.class.name}, error = #{e.message}"
           end
@@ -256,7 +262,7 @@ def parse(text)
             time = @mutex.synchronize { @time_parser.parse(value) }
           else
             begin
-              time = value.to_i
+              time = Fluent::EventTime.from_time(Time.at(value.to_f))
             rescue => e
               raise ParserError, "invalid time value: value = #{value}, error_class = #{e.class.name}, error = #{e.message}"
             end

lib/fluent/plugin/buf_memory.rb

@@ -57,7 +57,7 @@ def write_to(io)
 
     # optimize
     def msgpack_each(&block)
-      u = MessagePack::Unpacker.new
+      u = Fluent::Engine.msgpack_factory.unpacker
       u.feed_each(@data, &block)
     end
   end

lib/fluent/plugin/exec_util.rb

@@ -58,7 +58,7 @@ def call(io)
 
     class MessagePackParser < Parser
       def call(io)
-        @u = MessagePack::Unpacker.new(io)
+        @u = Fluent::Engine.msgpack_factory.unpacker(io)
         begin
           @u.each(&@on_message)
         rescue EOFError

lib/fluent/plugin/filter_record_transformer.rb

@@ -88,7 +88,7 @@ def filter_stream(tag, es)
         last_record = record # for debug log
         new_record = reform(time, record, placeholders)
         if @renew_time_key && new_record.has_key?(@renew_time_key)
-          time = new_record[@renew_time_key].to_i
+          time = EventTime.from_time(Time.at(new_record[@renew_time_key].to_f))
         end
         new_es.add(time, new_record)
       end

lib/fluent/plugin/in_exec.rb

@@ -66,9 +66,15 @@ def configure(conf)
       if @time_key
         if @time_format
           f = @time_format
-          @time_parse_proc = Proc.new {|str| Time.strptime(str, f).to_i }
+          @time_parse_proc =
+            begin
+              strptime = Strptime.new(f)
+              Proc.new { |str| Fluent::EventTime.from_time(strptime.exec(str)) }
+            rescue
+              Proc.new {|str| Fluent::EventTime.from_time(Time.strptime(str, f)) }
+            end
         else
-          @time_parse_proc = Proc.new {|str| str.to_i }
+          @time_parse_proc = Proc.new {|str| Fluent::EventTime.from_time(Time.at(str.to_f)) }
         end
       end
 

lib/fluent/plugin/in_forward.rb

@@ -153,8 +153,8 @@ def on_message(msg, chunk_size, source)
         entries.each {|e|
           record = e[1]
           next if record.nil?
-          time = e[0].to_i
-          time = (now ||= Engine.now) if time == 0
+          time = e[0]
+          time = (now ||= Engine.now) if time.to_i == 0
           es.add(time, record)
         }
         router.emit_stream(tag, es)
@@ -165,7 +165,7 @@ def on_message(msg, chunk_size, source)
         record = msg[2]
         return if record.nil?
         time = msg[1]
-        time = Engine.now if time == 0
+        time = Engine.now if time.to_i == 0
         router.emit(tag, time, record)
         option = msg[3]
       end
@@ -219,7 +219,7 @@ def on_read(data)
         else
           m = method(:on_read_msgpack)
           @serializer = :to_msgpack.to_proc
-          @u = MessagePack::Unpacker.new
+          @u = Fluent::Engine.msgpack_factory.unpacker
         end
 
         (class << self; self; end).module_eval do

lib/fluent/plugin/in_http.rb

@@ -144,8 +144,8 @@ def on_request(path_info, params)
           end
         end
         time = if param_time = params['time']
-                 param_time = param_time.to_i
-                 param_time.zero? ? Engine.now : param_time
+                 param_time = param_time.to_f
+                 param_time.zero? ? Engine.now : Fluent::EventTime.from_time(Time.at(param_time))
                else
                  record_time.nil? ? Engine.now : record_time
                end
@@ -191,7 +191,7 @@ def on_request(path_info, params)
 
     def parse_params_default(params)
       record = if msgpack = params['msgpack']
-                 MessagePack.unpack(msgpack)
+                 Engine.msgpack_factory.unpacker.feed(msgpack).read
                elsif js = params['json']
                  JSON.parse(js)
                else

lib/fluent/plugin/in_stream.rb

@@ -87,8 +87,8 @@ def on_message(msg)
         entries.each {|e|
           record = e[1]
           next if record.nil?
-          time = e[0].to_i
-          time = (now ||= Engine.now) if time == 0
+          time = e[0]
+          time = (now ||= Engine.now) if time.to_i == 0
           es.add(time, record)
         }
         router.emit_stream(tag, es)
@@ -99,7 +99,7 @@ def on_message(msg)
         return if record.nil?
 
         time = msg[1]
-        time = Engine.now if time == 0
+        time = Engine.now if time.to_i == 0
         router.emit(tag, time, record)
       end
     end
@@ -130,7 +130,7 @@ def on_read(data)
           @y.on_parse_complete = @on_message
         else
           m = method(:on_read_msgpack)
-          @u = MessagePack::Unpacker.new
+          @u = Fluent::Engine.msgpack_factory.unpacker
         end
 
         (class << self; self; end).module_eval do

lib/fluent/plugin/out_exec_filter.rb

@@ -120,9 +120,15 @@ def configure(conf)
 
       if @out_time_key
         if f = @out_time_format
-          @time_parse_proc = Proc.new {|str| Time.strptime(str, f).to_i }
+          @time_parse_proc =
+            begin
+              strptime = Strptime.new(f)
+              Proc.new { |str| Fluent::EventTime.from_time(strptime.exec(str)) }
+            rescue
+              Proc.new {|str| Fluent::EventTime.from_time(Time.strptime(str, f)) }
+            end
         else
-          @time_parse_proc = Proc.new {|str| str.to_i }
+          @time_parse_proc = Proc.new {|str| Fluent::EventTime.from_time(Time.at(str.to_f)) }
         end
       elsif @out_time_format
         log.warn "out_time_format effects nothing when out_time_key is not specified: #{conf}"

lib/fluent/plugin/out_stream.rb

@@ -65,7 +65,7 @@ def write(chunk)
         chain = NullOutputChain.instance
         chunk.open {|io|
           # TODO use MessagePackIoEventStream
-          u = MessagePack::Unpacker.new(io)
+          u = Fluent::Engine.msgpack_factory.unpacker(io)
           begin
             u.each {|(tag,entries)|
               es = MultiEventStream.new