feat(indexer): drain indexing pipelines on planned teardowns - #6768
feat(indexer): drain indexing pipelines on planned teardowns#6768Totodore wants to merge 21 commits into
Conversation
Drain indexing pipelines on planned teardowns instead of killing them: the source stops emitting, flushes the in-flight batches with a forced commit, and the pipeline exits on its own once everything is published and settled. Sources opt in via source_needs_drain; others keep the plain kill semantics. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Drain every pipeline on teardown (sources without acknowledgment state exit as soon as their flush batch is pushed), share the drain state machine between both pipeline flavors, and fix the drain accounting: DrainAllPipelines now also waits for pipelines a plan change already detached, merge pipelines survive their draining indexing pipelines, and a paused draining pipeline is no longer reaped as exited. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Bound the node-shutdown drain wait with a configurable indexer shutdown_drain_timeout (QW_INDEXER_SHUTDOWN_DRAIN_TIMEOUT, 300s by default) so a drain can no longer outlive the deployment's grace period. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![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: b615e9fee0
ℹ️ 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".
|
Thanks for submitting this. I'll have a look early next week. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a23df941fc
ℹ️ 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".
| // sends a last empty batch with force commit to wait for the entire pipeline flush. | ||
| let flush_batch = RawDocBatch::new(Vec::new(), SourceCheckpointDelta::default(), true); | ||
| self.source_sink | ||
| .send_raw_doc_batch(flush_batch, ctx) | ||
| .await?; |
There was a problem hiding this comment.
For the acknowledgment-only sources this API is designed to support, the final batch can have an empty checkpoint delta. If all in-flight documents are rejected by the doc processor, the empty force-commit reaches an indexer workbench with neither indexed splits nor checkpoint progress; the indexer emits nothing, so the publisher never sends SuggestTruncate and the source has no way to learn that those messages were processed. is_drained() therefore remains false until the drain deadline, after which the poison messages are redelivered indefinitely. The drain barrier needs to generate a downstream settlement notification even when publication produces no split or checkpoint update.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Might be a follow up if required (a custom signal rather than suggest_truncate, which also has a bad naming for its usage).
…lled synchronously to match previous semantics
There was a problem hiding this comment.
💡 Codex Review
https://github.com/quickwit-oss/quickwit/blob/1574496e815a225249a5fa1a7d75fc7a56955b99/quickwit-indexing/src/actors/indexing_service.rs#L1179
During node shutdown, this loop awaits each pipeline's DrainPipeline handler sequentially while the caller wraps the entire DrainAllPipelines request in the same indexer_shutdown_drain_timeout. If an early pipeline is busy spawning or slow to terminate, it can consume most or all of that shared timeout before later pipelines are even told to stop emitting; universe.quit() then kills those later pipelines without any drain attempt. Initiate the global drain requests concurrently (there are no replacement pipelines during global shutdown) or give initiation a separate budget before waiting for completion.
https://github.com/quickwit-oss/quickwit/blob/1574496e815a225249a5fa1a7d75fc7a56955b99/quickwit-indexing/src/actors/indexing_pipeline.rs#L278
When any child actor fails or becomes unhealthy after draining starts, this branch logs that the drain failed but terminates the supervisor with ActorExitStatus::Success. IndexingService::handle_supervise consequently increments num_successful_pipelines, and the shutdown exit statuses conceal that the final publication or acknowledgment may have failed; the metrics pipeline mirrors the same behavior. Keep the no-respawn behavior, but exit with a failure status so the unsuccessful drain remains observable.
AGENTS.md reference: AGENTS.md:L21-L22
ℹ️ 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".
| | `enable_otlp_endpoint` | If true, enables the OpenTelemetry exporter endpoint to ingest logs and traces via the OpenTelemetry Protocol (OTLP). | `false` | | ||
| | `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` | | ||
| | `shutdown_drain_timeout` | Time budget granted to each indexing pipeline to drain gracefully before its remaining actors are killed, on node shutdown and when the control plane tears down a pipeline. Set it above the largest `commit_timeout_secs` plus the time to upload and publish the final splits. Beware that on shutdown, draining only starts after the ingester and compactor decommissions complete, so the deployment's shutdown grace period (e.g. `terminationGracePeriodSeconds` on Kubernetes) must exceed the largest decommission timeout *plus* this value — about 600 seconds with the default timeouts. Can be overridden with the `QW_INDEXER_SHUTDOWN_DRAIN_TIMEOUT` environment variable. | `300s` | |
There was a problem hiding this comment.
The fresh evidence is the still-unchanged guidance in docs/configuration/index-config.md:608, which tells indexer operators to size the infrastructure grace period only for the longest commit timeout. The new shutdown path can first consume the ingester/compactor decommission budget and then the drain timeout documented here, so readers following that existing guidance can still configure a grace period that ends before draining completes. Update that paragraph to describe or link to the combined budget in this row.
AGENTS.md reference: AGENTS.md:L23-L24
Useful? React with 👍 / 👎.
…lled synchronously to match previous semantics
2 participants
What
Adds a generic pipeline drain mechanism so planned teardowns (plan-driven pipeline shutdowns, node shutdown) publish and settle in-flight batches instead of dropping them.
Why
Some sources dont support partitionned consumers or dont support checkpoints-based consumers. These sources need a way to acknowledge the incoming messages once they are indexed even when shutting-down a pipeline.
Without a graceful source drain, these ack-only sources are subjects to at-leat-once delivery whereas they could be in the exactly-once delivery scenario (except in case in of crashes).
Our specific nats case
We want to implement a Nats source. Nats does not support partitionned consumers, so there is two ways to make it work:
Checkpointmechanism. However this does not allow for pipeline scaling (because without named partitions we cant key checkpoints per-pipelines). Scaling is required for our nats implementation.suggest_truncate, with this solution it is possible to scale the number of pipelines as Nats dynamically spread load on every consumer instance (each pipelines). The tweak is that we must be able to keep the exactly-once delivery semantic in case of pipeline teardown, hence the drain mechanism.Other source could benefit from that, the GCP pubsub source has a lot of todo related to these issues.
Design
Mirrors the
FinishPendingMergesAndShutdownPipelinepattern: everything stays inside the actor system, no detached tasks, no caller-side timeouts.Source::should_be_drained()(defaultfalse), captured once at spawn. Pipelines whose source does not opt in are torn down by theDrainPipelinehandler exactly as a kill would (terminate the child actors, exit). sources other than the future acknowledgment-based ones are guaranteed to keep their pre-existing teardown semantics.Drain(SourceActor): stopsemit_batches, pushes an empty force-commit batch so the indexer flushes its workbench, and the source exits with success on its own once everything it delivered is settled (Source::is_drained).DrainPipeline(IndexingPipeline / MetricsPipeline): fire-and-forget initiation. The sharedDrainStaterecords the pipeline's drain deadline (own commit timeout + 30s grace) and guards against respawn; the pipeline's existing supervise loop enforces the deadline, and the source's success exit cascades until the pipeline exits with success by itself.IndexingService:shutdown_pipelinessendsDrainPipelinewith an associated user-configuredshutdown_drain_timeoutto every detached pipeline without awaiting; the supervise loop reaps the handles by actor state. Merge pipelines survive their draining indexing pipelines so the final force-committed splits still reach the merge planner (immature-split pickup is node-scoped).DrainAllPipelines(node shutdown, beforeuniverse.quit()): deferred reply completed by the supervise loop once every draining pipeline has exited. Ordered after the ingester decommission, which needs pipelines still consuming, and bounded by the newshutdown_drain_timeoutconfig.Indexer: an empty batch with no open workbench is dropped instead of creating one just to commit it empty.Tests
test_source_actor_drain: a source without ack state exits immediately on drain, after emitting the flush batch.indexing_service+indexing_pipelinesuites green (non-opt-in sources keep plain kill semantics).