fluent / fluent/fluentd

buffer: stage_size gauge goes transiently negative when write and enqueue_chunk overlap

Open
#5,479 0 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Ruby
Stars
13.6k
Forks
1.4k
Avg merge
1d 3h
Merged PRs (30d)
20

Description

### Describe the bug

`Buffer#write` cannot update `@stage_size_metrics` while it still holds chunk locks, so it defers the add until every chunk lock has been released:

https://github.com/fluent/fluentd/blob/4d5527a4e4e1b3c413f76359d8816708d981e317/lib/fluent/plugin/buffer.rb#L410-L419

That leaves a window between [`chunk.mon_exit`](https://github.com/fluent/fluentd/blob/4d5527a4e4e1b3c413f76359d8816708d981e317/lib/fluent/plugin/buffer.rb#L386) and the deferred `add` where the bytes are already committed into the chunk and visible through `@stage[metadata]`, but have not been counted yet. If a flush thread calls `enqueue_chunk` inside that window it subtracts the *current* `chunk.bytesize`, including the bytes the writer has not added:

https://github.com/fluent/fluentd/blob/4d5527a4e4e1b3c413f76359d8816708d981e317/lib/fluent/plugin/buffer.rb#L503-L505

So `sub` runs before the matching `add` and `stage_size` drops below zero until the deferred add lands.

#5467 clamps the values at export time so exporters no longer publish negative sizes. It deliberately does not touch the gauge store, because clamping `LocalMetrics#sub`/`#dec` turns the self-correcting negative into a permanent over-count and makes `Buffer#storable?` reject every write. This issue tracks the underlying accounting window that #5467 masks.

### To Reproduce

The window is deterministically reproducible by pausing the writer on the staged chunk's `mon_exit`. This test passes on master today, i.e. it asserts the negative intermediate value:

```ruby
require_relative '../helper'
require 'fluent/plugin/buffer'
require 'fluent/plugin/buffer/memory_chunk'
require 'fluent/plugin_id'
require 'fluent/log'

module StageSizeRace
class Owner < Fluent::Plugin::Base
include Fluent::PluginId
include Fluent::PluginLoggerMixin
end

class Buf < Fluent::Plugin::Buffer
def create_metadata(timekey = nil, tag = nil, variables = nil)
Fluent::Plugin::Buffer::Metadata.new(timekey, tag, variables)
end

def resume
return {}, []
end

def generate_chunk(metadata)
Fluent::Plugin::Buffer::MemoryChunk.new(metadata)
end
end
end

class StageSizeRaceTest < ::Test::Unit::TestCase
test 'stage_size goes negative while the deferred add is pending' do
b = StageSizeRace::Buf.new
b.owner = StageSizeRace::Owner.new
b.configure(config_element('buffer', '', { 'total_limit_size' => 1024, 'chunk_limit_size' => 4096 }))
b.start

m = b.create_metadata
b.write({ m => ['a' * 400] })
chunk = b.stage[m]
assert_equal 400, b.stage_size

reached = Queue.new
resume = Queue.new
armed = false

# pause the writer right after it releases the chunk lock (buffer.rb L386)
# and before the deferred add (buffer.rb L414-L419)
chunk.define_singleton_method(:mon_exit) do
r = super()
if armed
armed = false
reached << true
resume.pop
end
r
end

armed = true
writer = Thread.new { b.write({ m => ['b' * 400] }) }
reached.pop

b.enqueue_chunk(m)
assert_equal(-400, b.stage_size) # nothing is staged, yet the gauge is negative

resume << true
writer.join

assert_equal 0, b.stage.size
assert_equal 0, b.stage_size # self-heals once the deferred add lands
end
end
```

Output at the two checkpoints:

```
mid: stage.size=0 stage_size=-400 queue_size=800
end: stage.size=0 stage_size=0 queue_size=800
```

### Expected behavior

`stage_size` should never be negative, and it should equal the total bytesize of the chunks currently in `@stage` once concurrent operations have settled.

### Your Environment

```markdown
- Fluentd version:
- Package version:
- Operating system:
- Kernel version:
```

### Your Configuration

```apache
N/A
```

### Your Error Log

```shell
N/A
```

### Additional context

_No response_

Contributor guide

Open the contributing guide

Research direction

Start in lib/fluent/plugin/buffer.rb around the write path at lines 386 and 410-419, and enqueue_chunk at lines 503-505. Run the supplied StageSizeRaceTest reproduction and inspect how stage_size_metrics is updated across the concurrent operations. Done means stage_size never becomes negative and equals the bytesize of chunks remaining in @stage after operations settle.

Written by the indexing model from the issue text.

Assessment

Tech stack
ruby
Domain
backend
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.