Getting "dump an error event: error_class=Fluent::Plugin::ConcatFilter::TimeoutError"

[ntanh1]

Hi team,

I'm using concat plugin v2.1.0 for my FluentD container, the config is as follow:

<source>
  type forward
  port 24224
  bind 0.0.0.0
</source>

<filter *.*>
  @type concat
  key log
  separator ""
  stream_identity_key container_id
  multiline_start_regexp /^---SL---/
  multiline_end_regexp /^---EL---&/
  flush_interval 10
</filter>

<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    log ${record["log"] != nil ? record["log"].sub('---SL---','') : ''}
  </record>
</filter>

<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    log ${record["log"] != nil ? record["log"].sub('---EL---&','') : ''}
  </record>
</filter>

further processing

................................................

So my log event will indicate its start point with ---SL--- and endpoint with ---EL---&.
There's a Java app running in another container and use fluent logging driver.

Problem is I'm getting timeout flush for some random event, e.g:

2017-09-26 07:27:16 +0000 [warn]: #0 fluent/log.rb:336:call: dump an error event: error_class=Fluent::Plugin::ConcatFilter::TimeoutError error="Timeout flush: docker.5eabe1bb5d52:5eabe1bb5d52251a978270f141ba2657c1e2ac3a5febfe081e48f0039aae7646" tag="docker.5eabe1bb5d52" time=#<Fluent::EventTime:0x007fd9c861a9c8 @sec=1506410836, @nsec=182951490> record={"container_name"=>"/gfast-sim-id-1-1", "source"=>"stdout", "log"=>"---SL---{\"date\":1506410826922,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-1-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-1 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d159, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r", "container_id"=>"5eabe1bb5d52251a978270f141ba2657c1e2ac3a5febfe081e48f0039aae7646"}

As you can see the log event is complete, we don't wait for pieces of that event and concat. So I cannot understand why the timeout happened.
It occurs quite randomly, some time with TestANV.1-1-1, some time with TestANV.1-1-2 (I have 5 such entities).

Can some one please help?

[okkez]

This is caused by race condition. I've fixed the race condition. Could you try #38 ?

[ntanh1]

thanks, I'll check and get back

[ntanh1]

still happening

2017-09-27 05:23:55 +0000 [warn]: #0 fluent/log.rb:336:call: dump an error event: error_class=Fluent::Plugin::ConcatFilter::TimeoutError error="Timeout flush: docker.d1ed37c3657a:d1ed37c3657a0c5e5f1491f8fc4bdc0710b524ef37115a2871362a11bcd42e45" tag="docker.d1ed37c3657a" time=#<Fluent::EventTime:0x007fee2028c178 @sec=1506489835, @nsec=915635486> record={"container_id"=>"d1ed37c3657a0c5e5f1491f8fc4bdc0710b524ef37115a2871362a11bcd42e45", "container_name"=>"/gfast-sim-id-1-1", "source"=>"stdout", "log"=>"---SL---{\"date\":1506489825384,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-5-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-5 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d433, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r"}
2017-09-27 05:23:55 +0000 [info]: #0 fluent/log.rb:316:call: Timeout flush: docker.d1ed37c3657a:d1ed37c3657a0c5e5f1491f8fc4bdc0710b524ef37115a2871362a11bcd42e45

bash-4.3# cat ./usr/lib/ruby/gems/2.2.0/gems/fluent-plugin-concat-2.1.0/lib/fluent/plugin/filter_concat.rb
require "fluent/plugin/filter"

module Fluent::Plugin
  class ConcatFilter < Filter
    Fluent::Plugin.register_filter("concat", self)

    helpers :timer, :event_emitter

    desc "The key for part of multiline log"
    config_param :key, :string
    desc "The separator of lines"
    config_param :separator, :string, default: "\n"
    desc "The number of lines"
    config_param :n_lines, :integer, default: nil
    desc "The regexp to match beginning of multiline"
    config_param :multiline_start_regexp, :string, default: nil
    desc "The regexp to match ending of multiline"
    config_param :multiline_end_regexp, :string, default: nil
    desc "The regexp to match continuous lines"
    config_param :continuous_line_regexp, :string, default: nil
    desc "The key to determine which stream an event belongs to"
    config_param :stream_identity_key, :string, default: nil
    desc "The interval between data flushes, 0 means disable timeout"
    config_param :flush_interval, :time, default: 60
    desc "The label name to handle timeout"
    config_param :timeout_label, :string, default: nil
    desc "Use timestamp of first record when buffer is flushed"
    config_param :use_first_timestamp, :bool, default: false

    class TimeoutError < StandardError
    end

    def initialize
      super

      @buffer = Hash.new {|h, k| h[k] = [] }
      @timeout_map = Hash.new {|h, k| h[k] = Fluent::Engine.now }
      @mutex = Thread::Mutex.new
    end

    def configure(conf)
      super

      if @n_lines && @multiline_start_regexp
        raise Fluent::ConfigError, "n_lines and multiline_start_regexp are exclusive"
      end
      if @n_lines.nil? && @multiline_start_regexp.nil?
        raise Fluent::ConfigError, "Either n_lines or multiline_start_regexp is required"
      end

      @mode = nil
      case
      when @n_lines
        @mode = :line
      when @multiline_start_regexp
        @mode = :regexp
        @multiline_start_regexp = Regexp.compile(@multiline_start_regexp[1..-2])
        if @multiline_end_regexp
          @multiline_end_regexp = Regexp.compile(@multiline_end_regexp[1..-2])
        end
        if @continuous_line_regexp
          @continuous_line_regexp = Regexp.compile(@continuous_line_regexp[1..-2])
        end
      end
    end

    def start
      super
      @finished = false
      timer_execute(:filter_concat_timer, 1, &method(:on_timer))
    end

    def shutdown
      @finished = true
      flush_remaining_buffer
      super
    end

    def filter_stream(tag, es)
      new_es = Fluent::MultiEventStream.new
      es.each do |time, record|
        if /\Afluent\.(?:trace|debug|info|warn|error|fatal)\z/ =~ tag
          new_es.add(time, record)
          next
        end
        begin
          flushed_es = process(tag, time, record)
          unless flushed_es.empty?
            flushed_es.each do |_time, new_record|
              time = _time if @use_first_timestamp
              new_es.add(time, record.merge(new_record))
            end
          end
        rescue => e
          router.emit_error_event(tag, time, record, e)
        end
      end
      new_es
    end

    private

    def on_timer
      return if @flush_interval <= 0
      return if @finished
      flush_timeout_buffer
    end

    def process(tag, time, record)
      if @stream_identity_key
        stream_identity = "#{tag}:#{record[@stream_identity_key]}"
      else
        stream_identity = "#{tag}:default"
      end
      @timeout_map[stream_identity] = Fluent::Engine.now
      case @mode
      when :line
        process_line(stream_identity, tag, time, record)
      when :regexp
        process_regexp(stream_identity, tag, time, record)
      end
    end

    def process_line(stream_identity, tag, time, record)
      new_es = Fluent::MultiEventStream.new
      @mutex.synchronize do
        @buffer[stream_identity] << [tag, time, record]
      end
      if @buffer[stream_identity].size >= @n_lines
        new_time, new_record = @mutex.synchronize do
          flush_buffer(stream_identity)
        end
        time = new_time if @use_first_timestamp
        new_es.add(time, new_record)
      end
      new_es
    end

    def process_regexp(stream_identity, tag, time, record)
      new_es = Fluent::MultiEventStream.new
      case
      when firstline?(record[@key])
        if @buffer[stream_identity].empty?
          @mutex.synchronize do
            @buffer[stream_identity] << [tag, time, record]
          end
          if lastline?(record[@key])
            new_time, new_record = @mutex.synchronize do
              flush_buffer(stream_identity)
            end
            time = new_time if @use_first_timestamp
            new_es.add(time, new_record)
          end
        else
          new_time, new_record = @mutex.synchronize do
            flush_buffer(stream_identity, [tag, time, record])
          end
          time = new_time if @use_first_timestamp
          new_es.add(time, new_record)
          if lastline?(record[@key])
            new_time, new_record = @mutex.synchronize do
              flush_buffer(stream_identity)
            end
            time = new_time if @use_first_timestamp
            new_es.add(time, new_record)
          end
          return new_es
        end
      when lastline?(record[@key])
        @mutex.synchronize do
          @buffer[stream_identity] << [tag, time, record]
          new_time, new_record = flush_buffer(stream_identity)
        end
        time = new_time if @use_first_timestamp
        new_es.add(time, new_record)
        return new_es
      else
        if @buffer[stream_identity].empty?
          new_es.add(time, record)
          return new_es
        else
          if continuous_line?(record[@key])
            # Continuation of the previous line
            @mutex.synchronize do
              @buffer[stream_identity] << [tag, time, record]
            end
          else
            new_time, new_record = @mutex.synchronize do
              flush_buffer(stream_identity)
            end
            time = new_time if @use_first_timestamp
            new_es.add(time, new_record)
            new_es.add(time, record)
          end
        end
      end
      new_es
    end

    def firstline?(text)
      @multiline_start_regexp && !!@multiline_start_regexp.match(text)
    end

    def lastline?(text)
      @multiline_end_regexp && !!@multiline_end_regexp.match(text)
    end

    def continuous_line?(text)
      if @continuous_line_regexp
        !!@continuous_line_regexp.match(text)
      else
        true
      end
    end

    def flush_buffer(stream_identity, new_element = nil)
      lines = @buffer[stream_identity].map {|_tag, _time, record| record[@key] }
      _tag, time, first_record = @buffer[stream_identity].first
      new_record = {
        @key => lines.join(@separator)
      }
      @buffer[stream_identity] = []
      @buffer[stream_identity] << new_element if new_element
      [time, first_record.merge(new_record)]
    end

    def flush_timeout_buffer
      now = Fluent::Engine.now
      timeout_stream_identities = []
      @timeout_map.each do |stream_identity, previous_timestamp|
        next if @flush_interval > (now - previous_timestamp)
        next if @buffer[stream_identity].empty?
        next if @mutex.locked?
        time, flushed_record = @mutex.synchronize do
          flush_buffer(stream_identity)
        end
        timeout_stream_identities << stream_identity
        tag = stream_identity.split(":").first
        message = "Timeout flush: #{stream_identity}"
        handle_timeout_error(tag, @use_first_timestamp ? time : now, flushed_record, message)
        log.info(message)
      end
      @timeout_map.reject! do |stream_identity, _|
        timeout_stream_identities.include?(stream_identity)
      end
    end

    def flush_remaining_buffer
      @buffer.each do |stream_identity, elements|
        next if elements.empty?

        lines = elements.map {|_tag, _time, record| record[@key] }
        new_record = {
          @key => lines.join(@separator)
        }
        tag, time, record = elements.first
        message = "Flush remaining buffer: #{stream_identity}"
        handle_timeout_error(tag, time, record.merge(new_record), message)
        log.info(message)
      end
      @buffer.clear
    end

    def handle_timeout_error(tag, time, record, message)
      if @timeout_label
        event_router = event_emitter_router(@timeout_label)
        event_router.emit(tag, time, record)
      else
        router.emit_error_event(tag, time, record, TimeoutError.new(message))
      end
    end
  end
end

[okkez]

still happening

Is this easy to reproduce? Could you show me minimal reproduce sequence?
I could not reproduce this in my local environment.

[ntanh1]

yes, it's reproducible all the time.

I 'create' my containers, put in the rb file with your fix, then 'up' my containers.

version: '2'
services:
    fluentd:
        image: fnms-fluent
        container_name: fnms-fluentd  
        restart: always
        ports:
            - "24224:24224"
        environment:
            - ES_IP=elasticsearch
            - ES_PORT=9200
            - KIBANA_VERSION=5.2.2
            - OPENTSDB_URL=http://opentsdb:4242
        links:
            - elasticsearch
        depends_on:
            - elasticsearch

    gfast-simulator-1:
        image: gfast-simulator-v2
        container_name: gfast-sim-id-1-1
        ports:
            - "1023:1023"
            - "9500:9500"
        environment:
            - com_nokia_anv_simulator_circuit_id_prefix=TestANV.1-1-
        stdin_open: true
        tty: true
        logging:
            driver: fluentd
            options:
               fluentd-address: "0.0.0.0:24224"
               tag: "docker.{{.ID}}"
        depends_on:
            - fluentd
            - anv
...

Logging works for some time, issue happens when there're 5 log events, there's usually 1 event missing and timeout (in this case TestANV.1-1-5):

TestANV.1-1-1 says hello with : NetconfClientInfo{username='TLS-CLIENT', sessionId=432, m_remoteHost='anv', m_remotePort='6524'}

TestANV.1-1-3 says hello with : NetconfClientInfo{username='TLS-CLIENT', sessionId=431, m_remoteHost='anv', m_remotePort='6524'}

TestANV.1-1-2 says hello with : NetconfClientInfo{username='TLS-CLIENT', sessionId=429, m_remoteHost='anv', m_remotePort='6524'}

TestANV.1-1-4 says hello with : NetconfClientInfo{username='TLS-CLIENT', sessionId=430, m_remoteHost='anv', m_remotePort='6524'}

[okkez]

Hmm, I could not reproduce with docker-compose and cat log file.

log file is like following:

---SL---
1aaaa
bbbb
cccc
dddd
eeee
ffff
---EL---&
(snip)
---SL---
5aaaa
bbbb
cccc
dddd
eeee
ffff
---EL---&

BTW, I noticed that your error message contains ---EL---&\r this ends with \r. This is strange behavior, I think.
I tested both LF log file and CRLF log file, but fluentd does not show \r at all.
I believe \n follows after ---EL---&\r.

Could you take sigdump when timeout is occured?
https://docs.fluentd.org/v0.12/articles/trouble-shooting#dump-fluentd-internal-information

And one more question, do logging keep running after timeout? Do logging stop after timeout?

[ntanh1]

That log event is just same with others, I don't know if the others have \r, let me know if you need any further info.

Please find below sigdumps:
sigdump-12.log
sigdump-25.log

PID   USER     TIME   COMMAND
    1 root       0:00 /bin/sh -c ./startFluentd.sh
   11 root       0:00 {startFluentd.sh} /bin/bash ./startFluentd.sh
   12 root       0:01 {fluentd} /usr/bin/ruby /usr/bin/fluentd -c /fluentd/etc/fluent.conf -p /fluentd/plugins -v
   25 root       0:06 /usr/bin/ruby -Eascii-8bit:ascii-8bit /usr/bin/fluentd -c /fluentd/etc/fluent.conf -p /fluentd/plugins -v --under-supervisor
  151 root       0:00 bash
  158 root       0:00 ps
bash-4.3# kill -CONT 12;kill -CONT 25

---

I can see logging has stopped after timeout.

[okkez]

I want step by step reproducible sequence and full configuration to reproduce the issue.
I want to debug in my local environment.

I can see logging has stopped after timeout.

Next, I want to know that logs sent from docker containers reach Fluentd in_forward or not.
You can use Wireshark or tcpdump for this purpose.

Another way, use filter_stdout like following.
Insert filter_stdout before filter_concat.

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

<filter *.*>
  @type stdout
</filter>

<filter *.*>
  @type concat
  (snip)
</filter>
(snip)

[okkez]

Could you tell me following info?

• Docker version
• docker-compose version
• About your docker images

[ntanh1]

• Docker version: Docker version 17.05.0-ce, build 89658be
• Docker-compose: docker-compose version 1.8.0, build f3628c7
• My docker image contains a Java application, it uses log4j with custom JsonLayout to produce logs as below:

public class JsonLayout extends Layout {
    private static final String START_LOG_TOKEN = "---SL---";
    private static final String END_LOG_TOKEN = "---EL---&";
...
    @Override
    public String format(LoggingEvent event) {
        Map<String, Object> jsonMapping = new HashMap<String, Object>();
        Map<String, Object> properties = parseLoggingProperties(event);
        jsonMapping.putAll(properties);
        jsonMapping.put("date", event.getTimeStamp());
        jsonMapping.put("level", event.getLevel().toString());
        jsonMapping.put("thread", event.getThreadName());
        jsonMapping.put("category", event.getLoggerName());
        jsonMapping.put("message", event.getMessage());
        jsonMapping.put("throwable", formatThrowable(event));
        return START_LOG_TOKEN + m_gson.toJson(jsonMapping) + END_LOG_TOKEN + "\n";
    }
...
}

Full config:

<source>
  type forward
  port 24224
  bind 0.0.0.0
</source>

<filter *.*>
  @type concat
  key log
  separator ""
  stream_identity_key container_id
  multiline_start_regexp /^---SL---/
  multiline_end_regexp /^---EL---&/
  flush_interval 10
</filter>


<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    log ${record["log"] != nil ? record["log"].sub('---SL---','') : ''}
  </record>
</filter>

<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    log ${record["log"] != nil ? record["log"].sub('---EL---&','') : ''}
  </record>
</filter>

#this filter handles log events in Json format. If log format is not in Json (mostly from echo, see FNMS-7280), this logs a warning but still
# moving to the next filters
<filter *.*>
  @type parser
  format json
  key_name log
  reserve_data true
  suppress_parse_error_log true
</filter>

#dedicated for Syncope containers, we configure this as a logging tag in yml file
<filter syncope.*>
  @type record_transformer
  enable_ruby true
  auto_typecast true #to keep timeMillis as float instead of String
  <record>
    date ${record["timeMillis"]} #add new field date, value copied from timeMillis
    category ${record["loggerName"]} #add new field category, value copied from loggerName
    throwable -${record["thrown"]}
  </record>
  remove_keys timeMillis,endOfBatch,loggerFqcn,threadId,threadPriority,loggerName,thrown #remove uninteresting stuffs
</filter>


<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    # nil might be log from fluent itself
    container_name ${record["container_name"] != nil ? record["container_name"].sub('/','') : 'fluentd'}
  </record>
</filter>

<filter *.*>
  @type record_transformer

  #this allows ruby syntax in the below conversion
  enable_ruby true
  <record>
    #convert date from milli seconds to %Y-%m-%dT%H:%M:%S.%3N%z format. If 'date' is not available in log event (mostly from echo, see FNMS-7280),
    # then take from Fluentd local time
    date ${record["date"] != nil ? Time.at(record["date"]/1000.0).strftime('%Y-%m-%dT%H:%M:%S.%3N%z') : Time.now.strftime('%Y-%m-%dT%H:%M:%S.%3N%z')}

    #if 'message' is not available in log event (mostly from echo, see FNMS-7280), then take from the 'log' field.
    message ${record["message"] != nil ? record["message"] : record["log"]}

    #if 'category' is not available in log event (mostly from echo, see FNMS-7280), then take from the container name.
    category ${record["category"] != nil ? record["category"] : record["container_name"]}
  </record>
  remove_keys log
</filter>

<match *.*>
    @type rewrite_tag_filter
    capitalize_regex_backreference yes
    rewriterule1 category !^(com\.nokia\.anv\.sm\.)(api\.MetricManager|consumer\.MetricAnnotationHandler)$ logstash-normal
    rewriterule2 category .+                                                                               logstash-metric
</match>

<match logstash-metric>
    @type opentsdb_metrics
    url "#{ENV['OPENTSDB_URL']}"
</match>

<match logstash-*>
    @type forest
    subtype elasticsearch
    <template>
        reload_on_failure true
        reconnect_on_error true
        logstash_format true
        type_name fluentd
        host "#{ENV['ES_IP']}"
        port "#{ENV['ES_PORT']}"
        buffer_type file
        buffer_chunk_limit 16m
        buffer_queue_limit 4096
        buffer_path /opt/fluentd/buffer/
        retry_wait 10s
        flush_interval 5s
        time_key date
        time_key_exclude_timestamp true
    </template>
</match>

ok, will try with your config and get back.

[ntanh1]

Got following result:

2017-09-28 03:22:05.000000000 +0000 docker.18c8019ac54f: {"container_name":"/gfast-sim-id-1-1","source":"stdout","log":"---SL---{\"date\":1506568925190,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-2-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-2 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d184, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r","container_id":"18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214"}
2017-09-28 03:22:05.000000000 +0000 docker.18c8019ac54f: {"container_name":"/gfast-sim-id-1-1","source":"stdout","log":"---SL---{\"date\":1506568925233,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-1-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-1 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d185, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r","container_id":"18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214"}
2017-09-28 03:22:15 +0000 [warn]: #0 fluent/log.rb:336:call: dump an error event: error_class=Fluent::Plugin::ConcatFilter::TimeoutError error="Timeout flush: docker.18c8019ac54f:18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214" tag="docker.18c8019ac54f" time=#<Fluent::EventTime:0x007f6862939168 @sec=1506568935, @nsec=472123088> record={"container_name"=>"/gfast-sim-id-1-1", "source"=>"stdout", "log"=>"---SL---{\"date\":1506568925233,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-1-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-1 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d185, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r", "container_id"=>"18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214"}
2017-09-28 03:22:15 +0000 [info]: #0 fluent/log.rb:316:call: Timeout flush: docker.18c8019ac54f:18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214
2017-09-28 03:22:15.472280020 +0000 fluent.warn: {"error":"#<Fluent::Plugin::ConcatFilter::TimeoutError: Timeout flush: docker.18c8019ac54f:18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214>","tag":"docker.18c8019ac54f","time":1506568935,"record":{"container_name":"/gfast-sim-id-1-1","source":"stdout","log":"---SL---{\"date\":1506568925233,\"level\":\"INFO\",\"thread\":\"TestANV.1-1-1-Thread-1\",\"category\":\"com.alcatel.netconf.simulator.fwk.DeviceServer\",\"message\":\"TestANV.1-1-1 says hello with : NetconfClientInfo{username\\u003d\\u0027TLS-CLIENT\\u0027, sessionId\\u003d185, m_remoteHost\\u003d\\u0027anv\\u0027, m_remotePort\\u003d\\u00276524\\u0027}\"}---EL---&\r","container_id":"18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214"},"message":"dump an error event: error_class=Fluent::Plugin::ConcatFilter::TimeoutError error=\"Timeout flush: docker.18c8019ac54f:18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214\" tag=\"docker.18c8019ac54f\" time=#<Fluent::EventTime:0x007f6862939168 @sec=1506568935, @nsec=472123088> record={\"container_name\"=>\"/gfast-sim-id-1-1\", \"source\"=>\"stdout\", \"log\"=>\"---SL---{\\\"date\\\":1506568925233,\\\"level\\\":\\\"INFO\\\",\\\"thread\\\":\\\"TestANV.1-1-1-Thread-1\\\",\\\"category\\\":\\\"com.alcatel.netconf.simulator.fwk.DeviceServer\\\",\\\"message\\\":\\\"TestANV.1-1-1 says hello with : NetconfClientInfo{username\\\\u003d\\\\u0027TLS-CLIENT\\\\u0027, sessionId\\\\u003d185, m_remoteHost\\\\u003d\\\\u0027anv\\\\u0027, m_remotePort\\\\u003d\\\\u00276524\\\\u0027}\\\"}---EL---&\\r\", \"container_id\"=>\"18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214\"}"}
2017-09-28 03:22:15.472687269 +0000 fluent.info: {"message":"Timeout flush: docker.18c8019ac54f:18c8019ac54ff62ca0d2bf6c47beb17720c5a8c5c5a4f42c1f9a4e1ee2c19214"}

timeout for TestANV.1-1-1 but not TestANV.1-1-2

[okkez]

Thank you for your info.

Got following result:

This seems that filter_stdout does not exist.
I want to test that log event from Docker logging driver reach Fluentd.
filter_stdout before filter_concat can display raw log (multiple lines).

If filter_stdout does not display raw log after timeout occured. I think fluent-plugin-concat does not have problem.

BTW, Gson.toJson can generate single line JSON string. So you can stop using fluent-plugin-concat, If you can generate single line JSON string by your JsonLayout.format.

[okkez]

Could you try very simple configuration like following?
If reproduce problem, there are no problem in fluent-plugin-concat side.

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

<match *.*>
  @type stdout # or something
</match>

And, could you try without filter_concat?
If reproduce problem, there are no problem in fluent-plugin-concat.