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 9 commits into
Conversation
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
github-actions
Bot
removed
the
meta: awaiting author
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7661e29310
ℹ️ 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".
| let current_size: usize = events.iter() | ||
| .map(|e| e.estimated_json_encoded_size_of().get()) | ||
| .sum(); |
There was a problem hiding this comment.
When many small events target the same file, each additional event recomputes the size of every event already buffered for that path. With the default 10 MB window and no default max_events, a busy low-cardinality sink can accumulate tens or hundreds of thousands of events before the timeout/byte limit, making batch construction O(n²) and eating the syscall savings this change is meant to provide. Store the running byte count with the buffered events and update it as events are added.
Useful? React with 👍 / 👎.
| if current_size + event_size > batch_settings.size_limit | ||
| || events.len() >= batch_settings.item_limit |
There was a problem hiding this comment.
For batch.max_events = 1 (or any path that receives exactly N configured events and then goes quiet), this check only flushes before accepting the next event, so the full batch remains buffered until the timeout or another event arrives. That means users cannot use max_events to restore immediate/per-event file writes, and exact-size batches have the same delayed-flush behavior. Flush after appending when the batch reaches the item or byte limit.
Useful? React with 👍 / 👎.
|
|
||
| loop { | ||
| // Earliest flush deadline across all paths. | ||
| let flush_deadline = buffers.values().map(|(_, deadline)| *deadline).min(); |
There was a problem hiding this comment.
With high-cardinality path templates, every incoming event now scans all active path buffers just to find the earliest timeout. A workload with thousands of active partitions therefore pays O(active partitions) CPU per event before any encoding or file I/O, which undermines the high-partition performance goal of this change. Keep the next flush deadline in a delay queue/heap or update it only when buffers are inserted or removed.
Useful? React with 👍 / 👎.
| + batch_settings.timeout; | ||
| buffers.insert(path, (vec![event], deadline)); | ||
| } else { | ||
| events.push(event); |
There was a problem hiding this comment.
When batch.timeout_secs is larger than idle_timeout_secs, accepting events into an existing buffer does not reset the open-file deadline, so a file can be closed as idle while that same path is still receiving events. With truncation options such as after_close_time_secs, the later flush can reopen and truncate a file that would have stayed open under the previous per-event write path. Reset the file deadline when buffering for an already-open path, or prevent expiry while a buffer exists for that path.
Useful? React with 👍 / 👎.
| let deadline = tokio::time::Instant::now() | ||
| + batch_settings.timeout; | ||
| buffers.insert(path, (vec![event], deadline)); |
There was a problem hiding this comment.
For path templates with mostly unique rendered values, every event creates a separate buffer that is held until the timeout even though no later event will join it. At high ingest rates, or when users raise batch.timeout_secs as the new docs suggest for throughput, this can retain an arbitrary number of full Events and finalizers in memory before any writes occur. Add a global/partition cap or flush singleton partitions promptly when the active-buffer count grows.
Useful? React with 👍 / 👎.
thomasqueirozb
added
sink: file
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2604baa09c
ℹ️ 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".
| let deadline = tokio::time::Instant::now() | ||
| + batch_settings.timeout; | ||
| buffers.insert(path.clone(), vec![event]); | ||
| flush_deadlines.push(std::cmp::Reverse((deadline, path.clone()))); |
There was a problem hiding this comment.
When the input stream stays open but goes quiet after a non-full batch, this deadline is only stored in flush_deadlines; nothing in the tokio::select! awaits the next buffered deadline, and expired buffers are checked only after some other wake-up (another event, stream end, or an idle file). For the common first write to a path there is no open file yet, so the default 1-second timeout can be missed indefinitely and events remain unacknowledged/unwritten until another event arrives or the sink shuts down. Add a timer branch for the earliest flush_deadlines entry so the configured batch timeout actually wakes the loop.
Useful? React with 👍 / 👎.
| /// increases throughput at the cost of end-to-end latency. | ||
| #[configurable(derived)] | ||
| #[serde(default)] | ||
| pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>, |
There was a problem hiding this comment.
This adds a user-facing batch configuration block, but the generated file sink docs/examples were not updated; I checked website/cue/reference/components/sinks/generated/file.cue and website/generated/example-configs/sinks/file/*.yaml, and they still contain no batch entry. As a result the published configuration reference omits the new option and the generated-docs check is likely to fail until make generate-docs updates those artifacts.
AGENTS.md reference: AGENTS.md:L114-L117
Useful? React with 👍 / 👎.
| || events.len() >= batch_settings.item_limit | ||
| }); | ||
| if needs_flush { | ||
| let batch = buffers.remove(&path).unwrap(); |
There was a problem hiding this comment.
When a batch is flushed because it reached max_events/max_bytes, this removes the buffer but leaves that path's old deadline in flush_deadlines (the pre-add flush above has the same pattern). For a hot path that fills batches before timeout_secs, the heap grows with completed batches until those stale deadlines are later popped; if the same path has a new partial buffer when an old deadline is popped, the loop removes and writes that new buffer immediately, defeating the configured batching. Remove or invalidate the queued deadline whenever the buffer is drained.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 708d56d8a6
ℹ️ 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".
| events.push(event); | ||
| } | ||
| } else { | ||
| let generation = per_path_gen.entry(path.clone()).or_insert(0); |
There was a problem hiding this comment.
When path templates produce mostly unique values, every new destination path is inserted into per_path_gen, but entries are never removed when the corresponding buffer is flushed or dropped. The active buffers map is capped, but this separate map keeps a cloned Bytes key for every path ever seen, so a high-cardinality file sink can still grow memory without bound over time; remove the generation entry once no live buffer/deadline for that path remains or store the generation with an evictable structure.
Useful? React with 👍 / 👎.
|
|
||
| let next_timer_deadline = flush_deadlines.peek() | ||
| .map(|r| (r.0).0) | ||
| .copied(); |
There was a problem hiding this comment.
In every build, flush_deadlines.peek().map(|r| (r.0).0) already returns an Option<tokio::time::Instant> because the instant is copied out of the heap entry, but .copied() is only defined for Option<&T>, so this module does not compile at this line. Drop .copied() or have the map return a reference before copying.
Useful? React with 👍 / 👎.
| } | ||
| // Reset the file-handle idle deadline so the file stays open while | ||
| // events accumulate for this path. | ||
| self.files.reset_at(&path, self.deadline_at()); |
There was a problem hiding this comment.
When truncate.after_modified_time_secs is set for an already-open file, this reset happens as soon as an event is buffered and before process_batch calls should_truncate, which derives the last modification time from deadline - idle_timeout. With batch.max_events = 1 or a short batch.timeout_secs, an old file can therefore be treated as freshly modified and appended to instead of being truncated; keep the idle-deadline refresh separate from the modification timestamp or run the truncation check before this reset.
Useful? React with 👍 / 👎.
| for (finalizers, _) in succeeded { | ||
| finalizers.update_status(EventStatus::Errored); |
There was a problem hiding this comment.
If write_all returns an error after the OS has accepted a prefix of this batch, such as ENOSPC or a quota error after several records were appended, this loop marks every encoded event as errored even though the early records are already in the file. Sources using acknowledgements can then retry the whole batch and duplicate those records; write in record-sized chunks or otherwise track the completed prefix before marking finalizers.
Useful? React with 👍 / 👎.
github-actions
Bot
added
docs review on hold
…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>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 90d0c0e9ca
ℹ️ 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".
| let mut total_bytes = 0usize; | ||
| for i in 0..n { | ||
| let (buf, finalizers, event_size) = encoded.remove(0); | ||
| match file.write_all(&buf).await { |
There was a problem hiding this comment.
For the default uncompressed sink, a batch containing multiple events still invokes file.write_all once for every event. Consequently, the write-operation count remains O(events), so the new buffering adds latency without delivering the advertised reduction from per-event writes; combine the encoded records into a batch write while tracking event boundaries for partial-write acknowledgements.
Useful? React with 👍 / 👎.
| let n = encoded.len(); | ||
| let mut total_bytes = 0usize; | ||
| for i in 0..n { | ||
| let (buf, finalizers, event_size) = encoded.remove(0); |
There was a problem hiding this comment.
When a destination accumulates a large batch of small events, every remove(0) shifts all remaining entries in encoded. Repeating that for the whole default 10 MB batch makes batch draining O(n²), adding substantial CPU and memory-copy overhead precisely on the high-throughput workloads this change targets; consume the vector with an iterator or queue instead.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 26518da99e
ℹ️ 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".
| if dropped_events == n_events { | ||
| dropped_events = n_events - i; |
There was a problem hiding this comment.
When the write fails before completing the first event (written is zero or lies inside its boundary), dropped_events remains equal to n_events, so this condition is true for every subsequent errored event and repeatedly overwrites the count; after a batch of N failures it ends at 1. FileIoError uses this value for ComponentEventsDropped, causing production telemetry to undercount discarded events. Compute the count once from the number of boundaries beyond written instead.
Useful? React with 👍 / 👎.
4 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.