Add FluentLogger#pending_bytesize

Author: itkq

lib/fluent/logger/fluent_logger.rb

@@ -135,6 +135,14 @@ def connection_string
         @socket_path ? "#{@socket_path}" : "#{@host}:#{@port}"
       end
 
+      def pending_bytesize
+        if @pending
+          @pending.bytesize
+        else
+          0
+        end
+      end
+
       private
 
       def to_msgpack(msg)
@@ -170,7 +178,7 @@ def write(msg)
           end
 
           # suppress reconnection burst
-          if !@connect_error_history.empty? && @pending.bytesize <= @limit
+          if !@connect_error_history.empty? && pending_bytesize <= @limit
             if Time.now.to_i - @connect_error_history.last < suppress_sec
               return false
             end
@@ -182,7 +190,7 @@ def write(msg)
             true
           rescue => e
             set_last_error(e)
-            if @pending.bytesize > @limit
+            if pending_bytesize > @limit
               @logger.error("FluentLogger: Can't send logs to #{connection_string}: #{$!}")
               call_buffer_overflow_handler(@pending)
               @pending = nil

spec/fluent_logger_spec.rb

@@ -58,9 +58,11 @@
 
     context('post') do
       it ('success') {
+        expect(logger.pending_bytesize).to eq 0
         expect(logger.post('tag', {'a' => 'b'})).to be true
         fluentd.wait_transfer
         expect(fluentd.queue.last).to eq ['logger-test.tag', {'a' => 'b'}]
+        expect(logger.pending_bytesize).to eq 0
       }
 
       it ('close after post') {
@@ -72,6 +74,7 @@
         expect(logger).to be_connect
         fluentd.wait_transfer
         expect(fluentd.queue.last).to eq ['logger-test.tag', {'b' => 'c'}]
+        expect(logger.pending_bytesize).to eq 0
       }
 
       it ('large data') {
@@ -150,9 +153,11 @@
   context "not running fluentd" do
     context('fluent logger interface') do
       it ('post & close') {
+        expect(logger.pending_bytesize).to eq 0
         expect(logger.post('tag', {'a' => 'b'})).to be false
         fluentd.wait_transfer  # even if wait
         expect(fluentd.queue.last).to be_nil
+        expect(logger.pending_bytesize).to be > 0
         logger.close
         logger_io.rewind
         log = logger_io.read
@@ -204,18 +209,21 @@ def flush(messages)
       let(:buffer_overflow_handler) { Proc.new { |messages| handler.flush(messages) } }
 
       it ('post limit over') do
+        expect(logger.pending_bytesize).to eq 0
         logger.limit = 100
         event_1 = {'a' => 'b'}
         logger.post('tag', event_1)
         fluentd.wait_transfer  # even if wait
         expect(fluentd.queue.last).to be(nil)
+        expect(logger.pending_bytesize).to be > 0
 
         logger_io.rewind
         expect(logger_io.read).not_to match(/Can't send logs to/)
 
         event_2 = {'a' => ('c' * 1000)}
         logger.post('tag', event_2)
         logger_io.rewind
+        expect(logger.pending_bytesize).to eq 0
         expect(logger_io.read).to match(/Can't send logs to/)
 
         buffer = handler.buffer