Fluent::Plugin::Buffer#queued_num grows infinitely when chunks are time-sliced

[eagletmt]

• fluentd or td-agent version.
  • fluentd v1.1.2
• Environment information, e.g. OS.
  • Linux

Problem

When the buffer section has time as chunk key like https://docs.fluentd.org/v1.0/articles/buffer-section#time , Fluent::Plugin::Buffer#queued_num seems to grow infinitely.

Step to reproduce

Start fluentd with the following configuration.

% bundle exec fluentd --config fluentd.conf -v
2018-04-02 16:12:15 +0900 [info]: fluent/log.rb:322:info: parsing config file is succeeded path="fluentd.conf"
2018-04-02 16:12:15 +0900 [info]: fluent/log.rb:322:info: using configuration file: <ROOT>
  <source>
    @type forward
    @id input_forward
  </source>
  <source>
    @type debug_agent
    bind "127.0.0.1"
    port 26224
  </source>
  <match debug.echo>
    @type stdout
    <buffer tag,time>
      @type "memory"
      timekey 5s
      timekey_wait 5s
    </buffer>
  </match>
</ROOT>
2018-04-02 16:12:15 +0900 [info]: fluent/log.rb:322:info: starting fluentd-1.1.2 pid=18405 ruby="2.3.7"
2018-04-02 16:12:15 +0900 [info]: fluent/log.rb:322:info: spawn command to main:  cmdline=["/usr/share/rbenv/versions/2.3.7/bin/ruby", "-Eascii-8bit:ascii-8bit", "-rbundler/setup", "/home/eagletmt/gems/ruby/2.3.0/bin/fluentd", "--config", "fluentd.conf", "-v", "--under-supervisor"]
2018-04-02 16:12:16 +0900 [info]: fluent/log.rb:322:info: gem 'fluentd' version '1.1.2'
2018-04-02 16:12:16 +0900 [info]: fluent/log.rb:322:info: adding match pattern="debug.echo" type="stdout"
2018-04-02 16:12:16 +0900 [info]: fluent/log.rb:322:info: adding source type="forward"
2018-04-02 16:12:16 +0900 [info]: fluent/log.rb:322:info: adding source type="debug_agent"
2018-04-02 16:12:16 +0900 [info]: #0 fluent/log.rb:322:info: starting fluentd worker pid=18437 ppid=18405 worker=0
2018-04-02 16:12:16 +0900 [debug]: #0 fluent/log.rb:302:debug: buffer started instance=47459346498500 stage_size=0 queue_size=0
2018-04-02 16:12:16 +0900 [info]: #0 fluent/log.rb:322:info: listening dRuby uri="druby://127.0.0.1:26224" object="Fluent::Engine" worker=0
2018-04-02 16:12:16 +0900 [info]: #0 [input_forward] listening port port=24224 bind="0.0.0.0"
2018-04-02 16:12:16 +0900 [debug]: #0 fluent/log.rb:302:debug: enqueue_thread actually running
2018-04-02 16:12:16 +0900 [info]: #0 fluent/log.rb:322:info: fluentd worker is now running worker=0
2018-04-02 16:12:16 +0900 [debug]: #0 fluent/log.rb:302:debug: flush_thread actually running

At this moment, queued_num is empty.

% bundle exec fluent-debug -p 26224
Connected to druby://127.0.0.1:26224.
Usage:
    Fluent::Engine.match('some.tag').output  : get an output plugin instance
    Fluent::Engine.sources[i]                : get input plugin instances
    Fluent::Plugin.load_plugin(type,name)    : load plugin class (use this if you get DRb::DRbUnknown)

irb(main):001:0> Fluent::Engine.root_agent.outputs[0].buffer.queued_num
=> {}

After running echo '{"message":"hello"}' | bundle exec fluent-cat debug.echo for several times, fluentd flushes buffers and outputs debug.echo: {"message":"hello"}, but queued_num doesn't become empty.

% bundle exec fluent-debug -p 26224
Connected to druby://127.0.0.1:26224.
Usage:
    Fluent::Engine.match('some.tag').output  : get an output plugin instance
    Fluent::Engine.sources[i]                : get input plugin instances
    Fluent::Plugin.load_plugin(type,name)    : load plugin class (use this if you get DRb::DRbUnknown)

irb(main):001:0> Fluent::Engine.root_agent.outputs[0].buffer.queued_num
=> {#<struct Fluent::Plugin::Buffer::Metadata timekey=1522653270, tag="debug.echo", variables=nil>=>0, #<struct Fluent::Plugin::Buffer::Metadata timekey=1522653275, tag="debug.echo", variables=nil>=>0, #<struct Fluent::Plugin::Buffer::Metadata timekey=1522653290, tag="debug.echo", variables=nil>=>0}

It seems this causes memory-leak bug.

[repeatedly]

Release v1.1.3