Repository navigation
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![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: a20e2fdbe3
ℹ️ 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 usage >= limit.as_u64() as f64 { | ||
| ctx.schedule_self_msg(Duration::from_millis(100), Loop); | ||
| return Ok(()); |
There was a problem hiding this comment.
When Kinesis is used and node usage remains above the limit, returning here stops only the parent SourceActor: the per-shard consumers continue calling get_records and sending payload-bearing messages (kinesis/shard_consumer.rs:180-208) into the 1,000-slot channel (kinesis/kinesis_source.rs:118), whose receiver is drained only by emit_batches. Those queued bytes are absent from all three gauges, so sustained input during a commit/serialization pause can accumulate up to 1,000 fetch responses and OOM the node despite the new backpressure setting; propagate the pause to the shard consumers or reserve/account the channel payloads.
Useful? React with 👍 / 👎.
| | `cpu_capacity` | Advisory parameter used by the control plane. The value can expressed be in threads (e.g. `2`) or in term of millicpus (`2000m`). The control plane will attempt to schedule indexing pipelines on the different nodes proportionally to the cpu capacity advertised by the indexer. It is NOT used as a limit. All pipelines will be scheduled regardless of whether the cluster has sufficient capacity or not. The control plane does not attempt to spread the work equally when the load is well below the `cpu_capacity`. Users who need a balanced load on all of their indexer nodes can set the `cpu_capacity` to an arbitrarily low value as long as they keep it proportional to the number of threads available. | `num threads available` | | ||
| | `enable_cooperative_indexing` | Enable sharing resources more efficiently when the number of indexes actively written to is significantly higher than the number of cores but might decrease the overall indexing throughput. | `false` | | ||
|
|
||
| The `indexer.indexing_memory_limit` budget is shared across all indexing pipelines on a node. It accounts for reported Tantivy writer memory (including RAM directories) through serialization, plus raw and processed document batches queued or being handled. Both batch types are accounted for using their raw input byte sizes, not their exact heap usage; raw and processed batches can overlap during processing. When usage reaches the limit, sources stop reading and emitting new batches and retry every 100 ms. Downstream work drains normally, with existing commit timeouts and per-pipeline commit triggers unchanged. Already-admitted work can overshoot the limit. Source/network buffers, ingest queues, and parsing or indexing expansion are not fully accounted for, so this is not a strict cap on memory usage or total process RSS. It is disabled by default; set it to a positive byte size such as `2GiB` to enable it. |
There was a problem hiding this comment.
What's the rule of thumb rule to set a good value if everything is not accounted for?
| pub struct IndexerConfig { | ||
| /// Memory budget shared across all indexing pipelines on this node. Accounts for reported | ||
| /// Tantivy writer memory (including RAM directories) and raw input bytes of in-flight batches. | ||
| /// Sources stop emitting while at the limit; downstream work drains normally. Disabled by |
There was a problem hiding this comment.
What happens when the memory limit goes back below the budget? Is there any kind of backoff or source resume emitting as fast as they can?
| type Reply = (); | ||
|
|
||
| async fn handle(&mut self, _message: Loop, ctx: &SourceContext) -> Result<(), ActorExitStatus> { | ||
| if let Some(limit) = self.indexing_memory_limit { |
There was a problem hiding this comment.
I think performing the check here alone may miscount the limit.
A burst from source can admit enough future work to cause an OOM later in the indexer, even after the sources have stopped. A more robust check should be done on the indexer that decides wether to consume the next batch and eventually do an early commit.
This branch has not been deployed
2 participants
Why
Avoid OOM crashes.
How
Before emitting a batch, each source checks the combined writer memory and in-flight batch sizes across all pipelines on the node, using the existing atomic-backed gauges.
When usage reaches
indexer.indexing_memory_limit, sources stop feeding new batches into the pipeline. They check memory usage every 100 ms and resume once it drops below the limit. Downstream work drains and commits normally.The limit is disabled by default. Enable it with:
This is a soft threshold, not a process-memory cap. Batch accounting uses raw input bytes rather than exact heap usage, and already-admitted work can overshoot. Source/network buffers and ingest queues are outside the budget.
Verification
make fmtcargo nextest run -p quickwit-indexing -p quickwit-config: 369 passed, 1 skipped.