fix(out_file): create symlink for every latest metadata sharing a timekey - #5476
fix(out_file): create symlink for every latest metadata sharing a timekey#5476mehrdadbn9 wants to merge 3 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR fixes out_file’s symlink_path behavior when multiple buffer chunk-key variants (e.g. different ${myid}) share the same timekey, ensuring symlinks are created for each “latest” chunk in that time slice, and also hardens Buffer#purge_chunk to always release queued-byte accounting even if chunk.purge raises.
Changes:
- Update
SymlinkBufferMixinto treat multiple metadatas within the newest timekey as eligible for symlink creation. - Ensure
Buffer#purge_chunkdecrements queued byte metrics even whenchunk.purgefails. - Add regression tests for both behaviors.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| lib/fluent/plugin/out_file.rb | Adjusts symlink selection logic for chunks sharing a timekey across different metadatas. |
| lib/fluent/plugin/buffer.rb | Ensures queue_size metrics are released in purge_chunk via ensure. |
| test/plugin/test_out_file.rb | Adds a regression test covering bulk input with variable chunk keys and symlink placeholders. |
| test/plugin/test_buffer.rb | Adds a test asserting purge_chunk releases queue bytes even if purge raises (but needs correction). |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| if metadata.timekey == @latest_metadata.timekey | ||
| # Same timekey, different variables (e.g. bulk input with several | ||
| # chunk-key values): every one of these is "latest" for its own | ||
| # chunk and must get a symlink (fluentd#5099). | ||
| (@latest_metadatas ||= {})[metadata] = true |
| chunk = super | ||
| latest_set = @latest_metadatas || { @latest_metadata => true } | ||
| # "symlink" feature is to link from symlink_path to the latest file chunk. Records with latest | ||
| # timekey will be appended into that file chunk. On the other side, resumed file chunks might NOT | ||
| # have timekey, especially in the cases that resumed file chunks are generated by Fluentd v0.12. | ||
| # These chunks will be enqueued immediately, and will be flushed soon. | ||
| if chunk.metadata == @latest_metadata | ||
| if latest_set.key?(chunk.metadata) || chunk.metadata == @latest_metadata |
| failing_purge = Module.new do | ||
| def purge | ||
| raise IOError, 'simulated purge failure (unlink EIO)' | ||
| end | ||
| end | ||
| @p.buffer_class::Chunk.include(failing_purge) | ||
|
|
||
| m1 = @p.dequeue_chunk | ||
| assert_equal [@dm0, @dm1, @dm1], @p.queue.map(&:metadata) | ||
| assert_equal({m1.unique_id => m1}, @p.dequeued) | ||
|
|
||
| queued_before = @p.queue_size | ||
| assert queued_before > 0 | ||
|
|
||
| # purge_chunk swallows the error but must still decrement queue_size | ||
| @p.purge_chunk(m1.unique_id) | ||
|
|
||
| assert m1.purged | ||
| # queue_size must return to its pre-dequeue value, not leak upward | ||
| assert_equal queued_before - m1.bytesize, @p.queue_size |
| raise IOError, 'simulated purge failure (unlink EIO)' | ||
| end | ||
| end | ||
| @p.buffer_class::Chunk.include(failing_purge) |
There was a problem hiding this comment.
buffer_class doesn't exist on Fluent::Plugin::Buffer (or on FluentPluginBufferTest::DummyPlugin). Running this test against the PR HEAD (1737020224ea36a1706771f953dcc6055c3e9cd2) fails immediately with:
NoMethodError: undefined method 'buffer_class' for #<FluentPluginBufferTest::DummyPlugin>
| # Same timekey, different variables (e.g. bulk input with several | ||
| # chunk-key values): every one of these is "latest" for its own | ||
| # chunk and must get a symlink (fluentd#5099). | ||
| (@latest_metadatas ||= {})[metadata] = true |
There was a problem hiding this comment.
Metadata#hash collision causes O(n²) degradation
@latest_metadatas (lib/fluent/plugin/out_file.rb:84) is a Hash keyed by Metadata objects, but Metadata#hash (lib/fluent/plugin/buffer.rb:163) only hashes timekey — by design, for performance. Combined with this PR's change, every metadata sharing the same timekey collides into the same hash bucket, which is exactly the scenario this PR is meant to fix: bulk input where many distinct myid values (or other chunk-key variables) share a single timekey.
| if metadata.timekey == @latest_metadata.timekey | ||
| # Same timekey, different variables (e.g. bulk input with several | ||
| # chunk-key values): every one of these is "latest" for its own | ||
| # chunk and must get a symlink (fluentd#5099). | ||
| (@latest_metadatas ||= {})[metadata] = true | ||
| @latest_metadata = metadata | ||
| else | ||
| @latest_metadata = metadata | ||
| @latest_metadatas = { metadata => true } | ||
| end |
There was a problem hiding this comment.
Unsynchronized state mutation → possible race condition
SymlinkBufferMixin#metadata (lib/fluent/plugin/out_file.rb:79-90) mutates @latest_metadata/@latest_metadatas through a multi-step read-then-branch-then-write sequence (the else branch even replaces @latest_metadatas with a brand-new Hash object) with no synchronization at all. This is reached via Output#metadata → @buffer.metadata (lib/fluent/plugin/output.rb:934-976), a path Fluentd's normal architecture allows to be called concurrently — e.g. multiple input plugin threads emitting to the same <match> output — and it is not protected by Buffer#synchronize.
If two threads process metadata with the same newest timekey but different tags/variables at roughly the same time, one thread's wholesale replacement of @latest_metadatas can wipe out the entry the other thread just inserted (a lost update). That would silently drop a chunk from the "latest" tracking set — i.e., it could reintroduce the exact "missing symlink" symptom this PR (#5099) is meant to fix, just via a race instead of a deterministic overwrite. I wasn't able to trigger this with a deterministic single-process test, so I'm flagging it as plausible rather than confirmed, but the mechanism itself checks out from tracing every assignment site.
@latest_metadatas grows without bound within a single timekey window
Entries are only ever added to @latest_metadatas (out_file.rb:84); nothing removes an entry until the timekey strictly advances, at which point the whole hash is replaced. With the default 1-day timekey and a high-cardinality chunk key (like the <buffer tag,time,myid> config used in this PR's own new test), every distinct tag/variables combination seen during that day stays in memory — even long after its chunk has been flushed and purged.
I confirmed this directly: feeding 8,000 distinct myid values under one timekey left @latest_metadatas.size at exactly 8,000, with zero eviction. Before this PR, the mixin only ever held a single @latest_metadata object (O(1)); this is a new growth pattern introduced by the fix.
Dead code: the chunk.metadata == @latest_metadata fallback is unreachable
In generate_chunk (out_file.rb:110):
if latest_set.key?(chunk.metadata) || chunk.metadata == @latest_metadataTracing every place @latest_metadata and @latest_metadatas are assigned (lines 78, 84-85, 87-88) shows that whenever @latest_metadatas is non-nil, @latest_metadata is always one of its keys — and when @latest_metadatas is nil, the fallback at line 105 builds { @latest_metadata => true } on the spot. So there's no state where latest_set.key?(chunk.metadata) is false but chunk.metadata == @latest_metadata is true. I verified this by simulating every assignment path. The second half of the condition is dead code and just invites a future reader to wonder why it's there.
…leak (fluent#5468) When chunk.purge raises (e.g. unlink/close failure on the buffer path), the rescue swallows the error but @queue_size_metrics.sub was skipped. The chunk is already gone from @Dequeued, so it is never retried and the queued byte counter ratchets toward total_limit_size, making storable? permanently false and causing spurious BufferOverflowError on a near-empty buffer. Moving the sub into an ensure block keeps the counter correct regardless of purge success. Co-Authored-By: Mehrdad Biukian <mehrdad@example.com> Signed-off-by: Mehrdad Biukian Naeini <mehrdadbiukian@gmail.com>
…ekey Bulk input whose events share a timekey but differ in a chunk-key variable (e.g. myid=1 and myid=2) stages one chunk per metadata, but SymlinkBufferMixin kept only a single \@latest_metadata. Each new metadata with an equal timekey overwrote the previous one, so only the chunk staged last ever received a symlink; earlier chunks were left without one until their timekey advanced. Track all metadatas of the newest timekey in a set and create the symlink for any chunk generated from one of them. Fixes fluent#5099 Signed-off-by: Mehrdad Biukian Naeini <mehrdadbiukian@gmail.com>
- Add Mutex to protect @latest_metadata/@latest_metadatas mutations (called from Output#metadata, reachable concurrently from input threads). - Replace Metadata-object Hash keys (which collide because Metadata#hash only hashes timekey) with compound [timekey, tag, variables] keys that distinguish chunk-key variants sharing a timekey. - Evict the old window's tracking when timekey advances, preventing unbounded growth within a single timekey window. - Remove dead-code fallback (unreachable because @latest_metadata is always in @latest_metadatas when that Hash is non-nil, and the single-entry fallback reconstructs it inline). - All changes within SymlinkBufferMixin; no external API changes. Signed-off-by: Mehrdad Biukian Naeini <mehrdadbiukian@gmail.com>
df31060 to
fed914e
Compare
Which issue(s) this PR fixes:
Fixes #5099
What this PR does / why we need it:
When a bulk of events sharing one timekey but differing in a buffer chunk-key variable (e.g.
myid=1,myid=2) is fed toout_filewithsymlink_path, the buffer stages one chunk per metadata, butSymlinkBufferMixinkept only a single@latest_metadata. Each new metadata with an equal timekey overwrote the previous one, so only the chunk staged last ever received a symlink — earlier chunks were silently left without one until their timekey advanced (the reporter's "only one symlink is created" symptom).This PR tracks all metadatas belonging to the newest timekey in a set and creates the symlink for any chunk generated from one of them. Behavior for the single-chunk-key case is unchanged: when the timekey advances, the set is replaced, so exactly one symlink per symlink_path remains.
Verification (Ruby 3.3 container):
symlink with placeholders and variable chunk keys ...) fails on master (missing symlink for myid=1) and passes with this fix.test/plugin/test_out_file.rb: 65 tests, 259 assertions, 100% passed.test/plugin/test_buf_file.rb: 38 tests, 100% passed.Docs Changes:
None needed (behavior now matches the documented intent of
symlink_path).Release Note:
out_file:
symlink_pathnow creates symlinks for every latest chunk when multiple chunks share a timekey but differ in chunk keys.