Repository navigation
Proposal: add live tail - #6899
Proposal: add live tail #6899nico34638 wants to merge 2 commits into
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. |
|
@codex review |
![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: 91c8f0ec7a
ℹ️ 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".
| quickwit_common::spawn_named_task( | ||
| run_live_tail(context, plans, params.max_docs_per_sec, event_tx), | ||
| "live_tail", |
There was a problem hiding this comment.
When a document is processed immediately after the client receives the HTTP 200 response, this detached task may not yet have opened any indexer stream, so the hub still has no receiver and silently discards the document. Because the handler returns the ReceiverStream without any readiness handshake, clients cannot avoid this startup gap; open the initial target streams before exposing the SSE response, or provide a readiness event that clients can wait for.
Useful? React with 👍 / 👎.
| let runs_pipeline_for_index = node | ||
| .member() | ||
| .indexing_tasks | ||
| .iter() | ||
| .any(|indexing_task| indexing_task.index_uid.as_ref() == Some(&plan.index_uid)); | ||
| if runs_pipeline_for_index { |
There was a problem hiding this comment.
When an index has no currently advertised indexing task, such as a newly created ingest index before its first document, this predicate opens no subscription anywhere. The ingestion that creates the first pipeline can then be processed before its task reaches the coordinator on a later 5-second refresh, and the broadcast hub has no replay, so the first (or only) documents after opening the tail are silently missed; the added integration test avoids this by priming a pipeline before opening the tail.
Useful? React with 👍 / 👎.
| let request = OpenLiveTailRequest { | ||
| session_id: session_id.clone(), | ||
| index_uid: Some(plan.index_uid.clone()), | ||
| query_ast_json: plan.query_ast_json.clone(), | ||
| max_docs_per_sec, |
There was a problem hiding this comment.
When an index pattern resolves to multiple indexes on the same indexer, this loop sends the full requested max_docs_per_sec to every (node, index) target, and each request creates an independent DocRateLimiter. The node can therefore send up to K times the advertised per-indexer cap for K matched indexes (up to the session limit), contradicting the multi-index API's throttle and potentially overwhelming the client; the budget needs to be shared or divided per node.
Useful? React with 👍 / 👎.
| for doc_json in docs_json { | ||
| docs_ndjson.extend_from_slice(&doc_json); | ||
| docs_ndjson.push(b'\n'); | ||
|
|
||
| if docs_ndjson.len() >= TARGET_HITS_PAYLOAD_NUM_BYTES { | ||
| self.send_hits(std::mem::take(&mut docs_ndjson)).await; |
There was a problem hiding this comment.
When grpc.max_message_size is near its valid 1 MB minimum, or a single matched document is large, this fixed 1 MiB threshold is unrelated to the configured cap and is checked only after appending a whole document. The encoded LiveTailHits can therefore exceed Tonic's max encoding size, terminating every remote stream and causing the coordinator to reconnect repeatedly instead of delivering those documents; chunk sizing must account for the actual gRPC limit and oversized individual documents.
Useful? React with 👍 / 👎.
| senders | ||
| .entry(index_uid.clone()) | ||
| .or_insert_with(|| broadcast::channel(self.channel_capacity).0) |
There was a problem hiding this comment.
On nodes that create/delete many indexes or recreate an index repeatedly, every pipeline's tap() inserts a distinct IndexUid here, but the map's own Sender keeps that broadcast channel alive forever and there is no removal path after the pipeline and all subscribers disappear. This makes node memory grow with every historical index incarnation; stale entries need to be pruned once no external taps or receivers remain.
Useful? React with 👍 / 👎.
| reason: dropped.reason(), | ||
| num_dropped: dropped.num_dropped, | ||
| }, | ||
| None => continue, |
There was a problem hiding this comment.
If an indexer or a mixed-version peer yields a protobuf LiveTailMessage with its oneof unset, this branch silently treats the stream as healthy and discards the protocol violation, so the client receives neither an error nor a reconnect attempt. Return an internal/unavailable error so the forwarder reports and retries the broken stream instead of pretending it succeeded.
AGENTS.md reference: AGENTS.md:L21-L22
Useful? React with 👍 / 👎.
This branch has not been deployed
1 participant
Description
When something breaks, you want to see logs as they arrive, like tail -f. In Quickwit, a log is only searchable once its split is published, which can take from 30s to a few minutes. That's too slow to follow a deploy or debug a live issue.
This PR adds
GET /api/v1/{index_id}/tail?query=.... It streams the matching documents as Server-Sent Events as soon as the indexer has processed them, before they are committed.flowchart LR subgraph indexer["Each indexer"] DP[DocProcessor] -->|"processed docs<br/>(only while someone tails)"| HUB[LiveTailHub<br/>broadcast per index] HUB --> S[LiveTailSession<br/>filter + rate limit] end subgraph rest["Any REST node"] C[Coordinator] end S -->|gRPC stream| C C -->|SSE| CL[Client]The idea is to take the documents where they already are: in the indexing pipeline. When someone is tailing an index, the DocProcessor copies each document it processed into an in-memory broadcast channel (
LiveTailHub, one channel per index, on each indexer). When nobody is tailing, nothing is copied.On each indexer, a tail session reads from that channel. Every 250ms, it indexes the new documents into a small in-RAM Tantivy index and runs the query on it, so the query behaves exactly like a search. Matching documents are rate limited and sent back over gRPC.
The node that receives the request finds the indexers running a pipeline for the index through chitchat, and opens a gRPC stream to each of them. It refreshes this list every 5s and forwards everything to the client as SSE.
Limits are in the node config and can be overridden: