Buffer queue size exceeds limit without BufferQueueLimitError

[hoshi-sano]

It happens particularly in the case using many kinds of tags, or in the case try_flush(enqueue_buffer) called frequently.

require 'fluent/load'
require 'logger'
logger = Logger.new(STDOUT)
$log = Fluent::Log.new(logger.instance_variable_get(:@logdev).dev, logger.level)
require 'fluent/plugin/buf_memory'

buf = Fluent::MemoryBuffer.new
buf.start

# fill the queue with chunk
buf.buffer_queue_limit.times do
  chunk = buf.new_chunk('hoge')
  buf.enqueue(chunk)
  buf.instance_variable_get(:@queue) << chunk
end

# queue size reaches limit
p buf.queue_size # => 64

# new events are emitted (with tag which is not included in @map.keys)
10000.times do |i|
  tag = i.to_s
  data = {value: i}.to_msgpack
  chain = Fluent::NullOutputChain.instance
  buf.emit(tag, data, chain)
end

# same as Output#enqueue_buffer
buf.keys.each {|key| buf.push(key) }

# queue size exceeds limit
p buf.queue_size # => 10064

I wonder why is the queue size not checked in BasicBuffer#push.
https://github.com/fluent/fluentd/blob/v0.10.49/lib/fluent/buffer.rb#L244

Or may it be necessary to check the queue size at the beginning of BasicBuffer#emit ? (not after checking storable?)
https://github.com/fluent/fluentd/blob/v0.10.49/lib/fluent/buffer.rb#L165-L183

[repeatedly]

Ah, hmm...

Or may it be necessary to check the queue size at the beginning of BasicBuffer#emit ? (not after checking storable?)

The merit of current implementation, Fluentd tries to process events as much as possible, queued chunks + chunk in @map.
I don't know which is the best but it works fine.

[hoshi-sano]

Thanks for reply.
I worried about memory exhausting by growth of @map and queue size.
(Particularly, in the case that forwarding is not working well by a high load and so on.)
However, I thought over that it may be what should/can be avoided by the tuning of the parameter.