Timeout can cause fluentd keep warning

[azrle]

Hi,

I am using fluent-logger-ruby to send log to fluentd server via in_forward (tcp connection).
However, fluentd server sometimes keep warning the information like followings:

[warn]: #0 fluent/log.rb:336:call: incoming chunk is broken: host="127.0.0.1" msg=49

And there is a similar issue: fluent/fluentd#660

To make it worse, when the fluentd keep warning with this message, the performance degrades (maybe due to worker saturation?) and fluent-logger takes very long time to finish sending logs.

After some investigations, I found that

1. problem happens when I use Timeout::timeout and timeout happens;
2. fluent-logger sends duplicate data when problem happens;
3. log.warn "incoming chunk is broken:" takes time; commenting it out will improve the performance after problem occurred.

When it comes to 2), after timeout, fluent logger is going to call close and send @pending data which has been already sent. After the following patch, it gets fixed.

@@ -225,14 +225,15 @@
           end

           begin
-            send_data(@pending)
+            data_to_send = @pending
             @pending = nil
+            send_data(data_to_send)
             true
           rescue => e
             set_last_error(e)
             if pending_bytesize > @limit
               @logger.error("FluentLogger: Can't send logs to #{connection_string}: #{$!}")
-              call_buffer_overflow_handler(@pending)
+              call_buffer_overflow_handler(data_to_send)
               @pending = nil
             end
             @con.close if connect?

However even this issue can be fixed, I can still observe that the logs following the timed out log will still lead fluentd server keep warning. As a workaround, after timeout, I have to reset the connection to the fluentd server (or disable keepalive?).

The test client is as the following. You can adjust timeout to let timeout happen and large amount of warnings can be observed at fluentd server.

require 'fluent-logger'
require 'timeout'

Fluent::Logger::FluentLogger.open(nil, :host => 'localhost', :port => 24224)
timeout = 0.001
size = 256*1024
2.times do |i|
  begin
      Timeout.timeout(timeout) do
          Fluent::Logger.post("test.dummy", {"foo" => {"bar" => "1"*size}})
      end
  rescue Timeout::Error
      p "timeout #{timeout}"
      # establish a new connection
      Fluent::Logger::FluentLogger.open(nil, :host => 'localhost', :port => 24224)
      timeout = 5
  end
end

fluentd version: fluentd-0.14.21
fluentd config:

<source>
  @type forward
  bind 127.0.0.1
  port 24224
</source>

<match test.**>
  @type null
</match>

[azrle]

@repeatedly Could you please kindly check this issue?

[azrle]

cc: @frsyuki

[kdevendran-branch]

getting the error "[warn]: #0 incoming chunk is broken:" ...any fix for this?

[ganmacs]

Sorry for the delay. #83 should fix this issue.