diff --git a/README.md b/README.md index e7f9df3..4fd862d 100644 --- a/README.md +++ b/README.md @@ -152,8 +152,11 @@ class BufferOverflowHandler def flush(messages) @buffer ||= [] - MessagePack::Unpacker.new.feed_each(messages) do |msg| - @buffer << msg + messages.each do |tag, message| + unpacker = MessagePack::Unpacker.new(StringIO.new(message)) + unpacker.each do |time, record| + @buffer << [tag, time, record] + end end end end diff --git a/lib/fluent/logger/fluent_logger.rb b/lib/fluent/logger/fluent_logger.rb index b974265..7cb108b 100644 --- a/lib/fluent/logger/fluent_logger.rb +++ b/lib/fluent/logger/fluent_logger.rb @@ -20,6 +20,8 @@ require 'monitor' require 'logger' require 'json' +require 'base64' +require 'securerandom' module Fluent module Logger @@ -131,9 +133,9 @@ 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 if @nanosecond_precision && time.is_a?(Time) - write [tag, EventTime.new(time.to_i, time.nsec), map] + write(tag, EventTime.new(time.to_i, time.nsec), map) else - write [tag, time.to_i, map] + write(tag, time.to_i, map) end end @@ -141,7 +143,9 @@ def close @mon.synchronize { if @pending begin - send_data(@pending) + @pending.each do |tag, record| + send_data([tag, record].to_msgpack) + end rescue => e set_last_error(e) @logger.error("FluentLogger: Can't send logs to #{connection_string}: #{$!}") @@ -173,7 +177,7 @@ def connection_string def pending_bytesize if @pending - @pending.bytesize + @pending.to_s.bytesize else 0 end @@ -201,20 +205,22 @@ def suppress_sec end end - def write(msg) + def write(tag, time, map) begin - data = to_msgpack(msg) + record = to_msgpack([time, map]) rescue => e set_last_error(e) + msg = [tag, time, map] @logger.error("FluentLogger: Can't convert to msgpack: #{msg.inspect}: #{$!}") return false end @mon.synchronize { if @pending - @pending << data + @pending[tag] << record else - @pending = data + @pending = Hash.new{|h, k| h[k] = "" } + @pending[tag] = record end # suppress reconnection burst @@ -225,7 +231,9 @@ def write(msg) end begin - send_data(@pending) + @pending.each do |tag, record| + send_data([tag, record].to_msgpack) + end @pending = nil true rescue => e @@ -260,6 +268,7 @@ def send_data(data) # end # data = data[n..-1] #end + true end diff --git a/spec/fluent_logger_spec.rb b/spec/fluent_logger_spec.rb index eb565f2..4098122 100644 --- a/spec/fluent_logger_spec.rb +++ b/spec/fluent_logger_spec.rb @@ -196,7 +196,7 @@ logger_io.rewind log = logger_io.read expect(log).to match /Failed to connect/ - expect(log).to match /Can't send logs to/ + expect(log).to match /Can\'t send logs to/ } it ('post limit over') do @@ -206,11 +206,11 @@ expect(fluentd.queue.last).to be_nil logger_io.rewind - expect(logger_io.read).not_to match /Can't send logs to/ + expect(logger_io.read).not_to match /Can\'t send logs to/ logger.post('tag', {'a' => ('c' * 1000)}) logger_io.rewind - expect(logger_io.read).to match /Can't send logs to/ + expect(logger_io.read).to match /Can\'t send logs to/ end it ('log connect error once') do @@ -233,8 +233,11 @@ class BufferOverflowHandler def flush(messages) @buffer ||= [] - MessagePack::Unpacker.new.feed_each(messages) do |msg| - @buffer << msg + messages.each do |tag, message| + unpacker = MessagePack::Unpacker.new(StringIO.new(message)) + unpacker.each do |time, record| + @buffer << [tag, time, record] + end end end end