enhancement(file sink): batch writes per partition to reduce syscall … - #26081
enhancement(file sink): batch writes per partition to reduce syscall …#26081scMarkus wants to merge 1 commit into
Conversation
…overhead Events sharing the same rendered path are now accumulated into a single buffer and flushed with one write_all syscall per batch, rather than one syscall per event. This eliminates the O(events) syscall cost that caused throughput to degrade as partition count grew. Adds a `batch` config block (max_bytes, timeout_secs) using the standard BatchConfig infrastructure. Defaults (10 MiB / 1 s) match other sinks. Benchmark (10 000 events × 200 B, direct sink, no topology overhead): single file: 124 K → 1.2 M elem/s (+875%) 4 partitions: 208 K → 494 K elem/s (+138%) 32 partitions: 60 K → 72 K elem/s (+20%) 64 partitions: 32 K → 36 K elem/s (+13%) A follow-up could enable concurrent writes across partitions by lifting the file-handle map out of &mut self. Closes: vectordotdev#20394 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
github-actions
Bot
added
the
domain: sinks
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: aefa907a55
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
|
|
||
| fn partition(&self, event: &Self::Item) -> Self::Key { | ||
| match self.path.render(event) { | ||
| Ok(bytes) => Some(bytes), |
There was a problem hiding this comment.
With confinement enabled, this new partitioner returns the raw rendered path, so process_batch calls should_truncate/open_file before PathConfinement::confine has lexically rejected .. components. open_file runs create_dirs_nofollow before verify_parent, so an event field such as ../../../tmp/vector-escape/x in a /base/{{ field }}/out.log template can create /tmp/vector-escape outside base_dir before the batch is dropped. Preserve the old ordering by confining/normalizing the rendered path before any filesystem call.
Useful? React with 👍 / 👎.
| // for event in events { | ||
| // event.metadata().update_status(EventStatus::Errored); | ||
| // } | ||
| return; |
There was a problem hiding this comment.
When open_file fails, for example due to permission denied while creating/opening the destination, this branch logs the batch as dropped and immediately returns with all event finalizers still at their default status. When those events are dropped the notifier records Dropped, which does not change the batch status, so sources with acknowledgements can receive a delivered ack for data that was never written; the confinement-error branch below has the same early return. Update every event's status to Errored before returning.
Useful? React with 👍 / 👎.
| let mut batched = | ||
| input.batched_partitioned(partitioner, batch_settings.timeout, move |_| { | ||
| batch_settings.as_item_size_config(FileBatchSizer) | ||
| }); |
There was a problem hiding this comment.
Using batched_partitioned here can reorder writes within one destination when a non-empty batch is followed by an event whose estimated size is at least batch.max_bytes: the partitioned batcher queues the previous batch, then immediately closes the oversized event's batch and drains closed batches with pop(), so the oversized record is yielded and written before earlier records for the same path. The file sink previously appended events in input order per file; flush closed batches FIFO or handle oversized items without reversing them.
Useful? React with 👍 / 👎.
| let mut batched = | ||
| input.batched_partitioned(partitioner, batch_settings.timeout, move |_| { | ||
| batch_settings.as_item_size_config(FileBatchSizer) | ||
| }); |
There was a problem hiding this comment.
The partitioned batcher used here resets an existing partition's expiration every time another event arrives for that path, so batch.timeout_secs is an inactivity timer rather than a maximum batch age. With the default 10 MB batch and a steady stream that stays below that size limit, a file can receive no writes for minutes or hours until traffic pauses or the batch fills, whereas this sink previously appended each event immediately and the new config describes a 1-second flush timeout.
Useful? React with 👍 / 👎.
pront
added
the
meta: awaiting author
3 participants
…overhead
Events sharing the same rendered path are now accumulated into a single buffer and flushed with one write_all syscall per batch, rather than one syscall per event. This eliminates the O(events) syscall cost that caused throughput to degrade as partition count grew.
Adds a
batchconfig block (max_bytes, timeout_secs) using the standard BatchConfig infrastructure. Defaults (10 MiB / 1 s) match other sinks.Benchmark (10 000 events × 200 B, direct sink, no topology overhead):
single file: 124 K → 1.2 M elem/s (+875%)
4 partitions: 208 K → 494 K elem/s (+138%)
32 partitions: 60 K → 72 K elem/s (+20%)
64 partitions: 32 K → 36 K elem/s (+13%)
A follow-up could enable concurrent writes across partitions by lifting the file-handle map out of &mut self.
Closes: #20394
Summary
Vector configuration
How did you test this PR?
Is this a breaking change?
Does this PR include user facing changes?
no-changeloglabel to this PR.References
Notes
@vectordotdev/vectorto reach out to us regarding this PR.pre-pushhook, please see this template.make fmtmake check-clippy(if there are failures it's possible some of them can be fixed withmake clippy-fix)make testgit merge origin masterandgit push.Cargo.lock), pleaserun
make build-licensesto regenerate the license inventory and commit the changes (if any). More details on the dd-rust-license-tool.