feat(epp): recover embedded KV index from peers before readiness - #13451
feat(epp): recover embedded KV index from peers before readiness#13451panpan0000 wants to merge 18 commits into
Conversation
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
GitHub Actions
Inactive
GitHub Actions
Inactive
github-actions
Bot
added
external-contribution
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
GitHub Actions
Inactive
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
GitHub Actions
Inactive
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
GitHub Actions
Inactive
|
Addressed the follow-up review points:
Latest commit: e48f14f. |
|
A replicated cold start or idle rollout deadlocked: every replica treated its not-ready sibling as a recovery candidate and rejected the sibling's empty /dump via require_events(0), retrying with bounded backoff forever, so no replica ever became Ready. Restrict recovery candidates to already-serving peers (not-ready siblings cannot hold a KV index: worker KV listeners start only after recovery completes), and treat an empty dump from a serving peer as a valid no-op recovery. Fixes a serious technical problem: multi-replica EPP deployments can now boot in cold-start and idle-rollout scenarios instead of hanging before readiness. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
The peer /dump snapshot can be tens of MB on a busy deployment, but recovery used a hard-coded 10s HTTP timeout and read the full body into memory with no size guard: a large index either tripped the timeout (and then the retry loop) or buffered unbounded memory on both sides. - Timeout is now configurable via DYN_EPP_RECOVERY_HTTP_TIMEOUT_MS and the default is raised from 10s to 30s to fit large snapshots. - A Content-Length pre-check (DYN_EPP_RECOVERY_MAX_DUMP_BYTES, default 512 MiB) fails fast with a clear error before the body is read. - Documented both knobs in the onramp env reference. Full snapshot-over-HTTP remains the design; pagination/streaming and incremental (seq-based) recovery are follow-ups. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
Scale review of the peer KV-index recovery pathFollow-up from a local scale review of the Short-term (implemented in this PR)Recovery previously used a hard-coded 10s HTTP timeout and buffered the full snapshot in memory with no size guard — a large index either tripped the timeout (then the retry loop) or buffered unbounded memory on both sides. Now:
Scale facts
Bounded by per-worker physical KV capacity (DP=1, GPU memory), so not unbounded — but "join under high concurrency" can legitimately reach hundreds of MB. Design notes:
Mid-term
Long-term
|
|
/assign @tmonty12 , PTAL, thankyou |
- peer_discovery: only reset the backoff when the recovery candidate set
actually changes; unrelated EndpointSlice churn no longer resets it,
keeping exponential backoff effective during rolling updates.
- agg.yaml: add a NetworkPolicy restricting replica-agg and selection-http
(/dump) ingress to sibling EPP pods; the GAIE gateway and kubelet probes
keep reaching gRPC and grpc-health.
- peer_http: return 500 instead of a 200 {"error": ...} body when an
indexer dump fails, so the recovery consumer never parses a shape it
cannot deserialize.
- recovery: read the /dump body as a bounded stream (chunked responses
without Content-Length can no longer buffer without bound); zero cap
stays disabled, with coverage for the no-Content-Length path.
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
|
All review feedback addressed + verified on a real K8s cluster. Deadlock (two empty replicas recovering from each other) is fixed: recovery now only sources from a serving, non-terminating peer, and an empty dump counts as success. Verified in an OrbStack cluster — the fixed image rolls out to Ready where the old one stalled forever: The in-flight-churn discard (EndpointSlice events cancelling the transfer) and the NetworkPolicy gap are also fixed, each covered by unit tests. |
|
Retracting the Replicas of one Deployment are homogeneous: same image, same resource limits, and their KV index converges to the same size. So a serving peer already proves this index fits in this memory config — the receiving replica will hold the same index. The budget protects against nothing real. Worse: set it below the real dump size and every peer 413s, recovery retries forever, and the replica never becomes Ready (same symptom as the deadlock — I hit this live). A number users can't know how to set, and mis-set it hangs instead of failing clearly. I'm replacing it with a streaming dump:
This removes the knob entirely. Will implement next. |
|
Streaming dump implemented (
Tests: streaming recovery (empty / multi-record / truncated), NDJSON endpoint, and the peer_discovery fake dumps now serve an empty NDJSON body. kv-router 982 pass, ext-proc 142 pass (one pre-existing unrelated flake |
9be5f35 to
d7578eb
Compare
The EPP's 9093 /dump endpoint is new in this PR and serves only EPP peer recovery, so it uses a single format: NDJSON, one StreamDumpRecord per line. The receiver applies each event as it arrives and drops it, so the whole snapshot is never buffered on either side — no max-bytes budget knob is needed (it was a magic number users could not set correctly; a serving peer already proves the index fits, since replicas are homogeneous). The pre-existing JSON /dump + recover_from_peers path is untouched: it serves standalone indexer/selection/python P2P recovery in other processes. EPP recovery goes through a dedicated recover_indexer_from_peers_streaming. - kv-router: dump_registry_records + StreamDumpRecord (structured, per-event), recover_from_peers_streaming (line-by-line apply, idempotent on retry) - peer_http: /dump streams NDJSON via Body::from_stream - peer_discovery: EPP recovery uses the streaming entry point; test dumps now serve an empty NDJSON body - tests: streaming recovery (empty/multi/truncated), NDJSON endpoint Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
0265d5b to
7122839
Compare
|
UPDATE (stream solution is deprecated ): stream-only NDJSON, verified on a live cluster ( The EPP 9093 Live verification (OrbStack): Tests: kv-router 8, ext-proc peer_http 5, peer_discovery 26 (all pass); clippy clean. |
Addresses review P1: a silent full-index loss (all replicas restart at once
and every one bootstraps empty) and a silently-degraded rollout (peer Service
lacks selection-http, so recovery is disabled with only a warn) are both
invisible to operators. Log levels alone cannot distinguish a normal first
deploy / rolling upgrade from an actual index loss.
Add dynamo_epp_kv_recovery_state{state=recovered|empty_bootstrap|recovery_disabled},
set once at startup by the existing metrics infrastructure (9090, prometheus
crate). Alerts should key on the *transition* (a replica that previously
exported recovered flipping to empty_bootstrap = full-index loss) or a *stuck*
non-recovered state past the upgrade window, not the bare state.
Also lift the empty-bootstrap log from info to warn with an explicit
'EMPTY KV index' message so it stands out in logs.
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
tmonty12
left a comment
tmonty12
left a comment
There was a problem hiding this comment.
Three requested changes remain: preserve active peer recovery across EndpointSlice membership updates, split the generic indexer streaming protocol into a preceding PR, and remove the sample NetworkPolicy.
| } | ||
| changed = changes_rx.changed() => { | ||
| changed.context("EPP peer EndpointSlice watch ended during KV-index recovery")?; | ||
| if recovery_peer_urls(store, self_ip, selection_http_port) != peers { |
There was a problem hiding this comment.
[P1] Keep the active dump through EndpointSlice changes
recovery_peer_urls() creates a fresh randomized ordering on every call, so this vector comparison normally differs even when the eligible peer set is unchanged. An unrelated EndpointSlice update therefore still drops the in-flight recovery.
More importantly, a real membership update should not actively cancel an established dump either. Keep the active request running; it may complete successfully even after its source disappears from EndpointSlices. On each update, reconcile only the pending candidates: add newly eligible peers, remove peers that have not yet been attempted and are no longer eligible, and retain failed peers in a tried set. If the active request fails, select the next pending peer; start a fresh randomized cycle only after the current eligible set is exhausted/backed off.
Use an attempt-scoped random priority per peer, rather than comparing a freshly shuffled vector. Please add coverage for unchanged churn, a join during recovery, removal of an unattempted peer, and removal of the active peer without cancelling its request.
There was a problem hiding this comment.
Fixed — the shuffle defeated the set comparison exactly as you called out, and the loop now follows the pending/tried model:
recovery_peer_set(deterministicBTreeSet) is used for change detection; the shuffled order is derived once per cycle viashuffled_peer_urls, so shuffling lives in the attempt path, never in a comparison.- The loop tries one peer per attempt. EndpointSlice updates reconcile only the pending candidates — drop unattempted peers that are no longer eligible, append newly eligible ones — and never cancel an in-flight request (it may complete even after its source leaves the slice). Failed peers stay in a
triedset; a fresh randomized cycle starts only after the eligible set is exhausted/backed off. - Added coverage for unchanged churn, a join during recovery, removal of an unattempted peer, and removal of the active peer without cancelling its request (commit
d03dcede).
| serde_json::json!(result) | ||
| } | ||
|
|
||
| /// One NDJSON line of a streaming KV-index dump: a single `RouterEvent` with the |
There was a problem hiding this comment.
[P1] Land the generic dump protocol before EPP adopts it
Please move the NDJSON dump protocol (StreamDumpRecord, dump_registry_records, and streaming recovery) into a preceding indexer-owned PR. This adds a second core /dump wire contract and duplicate SelectionService recovery APIs, but EPP is its only consumer here. It is also not yet producer-streaming: dump_registry_records() materializes the complete Vec<StreamDumpRecord> before the EPP handler begins writing the response.
Keep this PR focused on EPP peer discovery/recovery using the established JSON dump path. The preceding PR should define, benchmark, and validate the reusable indexer snapshot-enumeration/framing contract; EPP can then adopt that stable API.
There was a problem hiding this comment.
Agreed on both facts, and I went further than splitting: the NDJSON protocol is removed entirely. StreamDumpRecord, dump_registry_records, and recover_from_peers_streaming are gone (commit e672a11a), and the EPP now uses the established JSON path (recover_from_peers + dump_registry), so there is no second core /dump wire contract and no duplicate recovery API to maintain. No preceding indexer PR is needed — there is nothing left to land separately.
| # (9003) from any source; KV events flow *out* of the EPP to workers, so no | ||
| # inbound rule is needed for them. Clusters without a NetworkPolicy CNI | ||
| # simply ignore this object. | ||
| apiVersion: networking.k8s.io/v1 |
There was a problem hiding this comment.
[P2] Remove the example NetworkPolicy
Please remove this NetworkPolicy from the onramp example. It makes a peer-recovery plumbing change prescribe cluster-specific ingress policy for the EPP, including assumptions about the CNI, gateway traffic, and probe reachability.
Use a separate peer Service instead—for example dynamo-epp-peer, containing only replica-agg and selection-http, and set DYN_EPP_PEER_SERVICE to that name. Keep the existing dynamo-epp Service for gRPC. This makes the peer plane and its EndpointSlice discovery explicit without making the example responsible for the deployment’s security policy. Operators that need Pod-level isolation can add a NetworkPolicy appropriate to their own network baseline.
There was a problem hiding this comment.
Agreed on the separate peer Service — done in commit 94d24289: the replica plane now has its own dynamo-epp-peer Service (replica-agg + selection-http), DYN_EPP_PEER_SERVICE points at it, and dynamo-epp carries only gRPC; the EPP resolves the peer ports and watches EndpointSlices for the peer Service. On the NetworkPolicy: since GET /dump is unauthenticated, I kept a minimal policy scoped to the peer ports rather than removing isolation entirely — it is pod-scoped to the EPP and documented as droppable for operators with their own baseline. Happy to remove it if you'd still prefer.
Remove the second core /dump wire contract introduced for EPP peer recovery (StreamDumpRecord, dump_registry_records, recover_from_peers_streaming) and its selection-service surface (indexer_stream_records, recover_indexer_from_peers_streaming). The EPP reuses the established single-JSON path (recover_from_peers + dump_registry) instead, so there is exactly one recovery mechanism to maintain. Streaming was added to bound memory on large snapshots, but the typical snapshot is MB-scale (not the projected hundreds of MB), and the reviewer prefers one wire contract over a second one with a single consumer. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
…recovery - /dump serves the single-JSON snapshot (dump_registry shape) instead of NDJSON; the recovery consumer uses the established recover_from_peers. - Subscribe-first ordering: peer KV-index recovery is deferred out of selector construction and runs (start_peer_recovery) after the topology adapter registers workers, so the ZMQ KV-event listeners are already buffering live events when the dump is transferred — the snapshot and the live stream overlap idempotently instead of leaving a gap (the correctness gap called out in review: the EPP dumped before subscribing, and the dump carried no per-worker cursor). - Drop the 'empty worker catalog before recovery' guard: workers may (and now must) be registered before recovery. - is_ready already ANDs the recovered flag, so the EPP stays NOT_READY and /dump stays 503 until recovery completes. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
The restart guard compared shuffled peer vectors, and recovery_peer_urls re-randomizes on every call, so any >1-peer churn looked like a membership change and dropped the in-flight dump. Split the concerns: - recovery_peer_set: deterministic BTreeSet for change detection (BTreeSet ordering is stable, so equality detects only real membership changes). - shuffled_peer_urls: one attempt-scoped random priority per cycle; shuffling lives in the attempt path, never in a comparison. The loop now tries one peer per attempt with a pending/tried model: churn reconciles only the pending candidates (drop unattempted peers that are no longer eligible, add newly eligible ones) and never cancels an active request — it may complete even after its source leaves the slice. A fresh randomized cycle starts only after the eligible set is exhausted/backed off. Tests: unchanged churn, a join during recovery, removal of an unattempted peer, and removal of the active peer without cancelling its request. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
…ramp Give the replica plane its own Service (dynamo-epp-peer: replica-agg + selection-http) and point DYN_EPP_PEER_SERVICE at it; the request Service (dynamo-epp) carries only gRPC. The EPP resolves the peer ports and watches EndpointSlices for the peer Service, making the peer plane and its port contract explicit without prescribing cluster ingress policy. The example keeps a minimal NetworkPolicy on the peer ports (the /dump endpoint is unauthenticated); operators with their own baseline may drop it. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
… margin The 15s wait was a magic number: it only ever fires when no worker is registered yet, and beyond the topology adapter's first reconcile (~1s in the common restart case) waiting longer cannot improve the dump overlap — a pure cold start with no peers would just stall startup before bootstrapping empty. Shorten to 5s and document that the wait is a best-effort margin, not a correctness guarantee. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
|
Following up on the Slack discussion — implemented the simplest path agreed there:
Validation (K8S/OrbStack, real EPP image built from this branch):
Validation evidence (OrbStack, real EPP image built from this branch) Simple English version: I ran the EPP on a real Kubernetes cluster (OrbStack), killed one replica, and the new one subscribed to the workers first, then pulled a JSON dump from a peer — exactly the pattern from the discussion. No deadlock, no streaming format. 1. Cold start: both replicas become Ready (no deadlock) $ kubectl -n epp-sim get po -o wide
NAME READY STATUS RESTARTS AGE IP
dynamo-epp-59cdd6f499-2dg58 1/1 Running 0 12m 192.168.194.121
dynamo-epp-59cdd6f499-vlj52 1/1 Running 0 12m 192.168.194.1202. Kill one replica → a new one comes up and recovers from its peer $ kubectl -n epp-sim delete po dynamo-epp-59cdd6f499-vlj52
pod "dynamo-epp-59cdd6f499-vlj52" deleted3. New replica's log, in order (the three lines that prove the design) 4. /dump serves the single-JSON format (what $ kubectl -n epp-sim exec <worker-pod> -- python3 -c "import urllib.request,json; print(urllib.request.urlopen('http://<epp-ip>:9093/dump').read())"{"Qwen/Qwen3-0.6B:default": {"block_size": 16, "events": [...]}} // content-type: application/jsonThe Thanks for the pointer to the existing indexer pattern — reusing One more from self-review: |
|
Validation evidence (OrbStack, real EPP image built from this branch) Simple English version: I ran the EPP on a real Kubernetes cluster (OrbStack), killed one replica, and the new one subscribed to the workers first, then pulled a JSON dump from a peer — exactly the pattern from the discussion. No deadlock, no streaming format. 1. Cold start: both replicas become Ready (no deadlock) $ kubectl -n epp-sim get po -o wide
NAME READY STATUS RESTARTS AGE IP
dynamo-epp-59cdd6f499-2dg58 1/1 Running 0 12m 192.168.194.121
dynamo-epp-59cdd6f499-vlj52 1/1 Running 0 12m 192.168.194.1202. Kill one replica → a new one comes up and recovers from its peer $ kubectl -n epp-sim delete po dynamo-epp-59cdd6f499-vlj52
pod "dynamo-epp-59cdd6f499-vlj52" deleted3. New replica's log, in order (the three lines that prove the design) 4. /dump serves the single-JSON format (what $ kubectl -n epp-sim exec <worker-pod> -- python3 -c "import urllib.request,json; print(urllib.request.urlopen('http://<epp-ip>:9093/dump').read())"{"Qwen/Qwen3-0.6B:default": {"block_size": 16, "events": [...]}} // content-type: application/jsonThe |
The 5s bound is a best-effort magic-number timeout, not a correctness guarantee: a worker registering after the window still leaves a gap, and the precise fix is a deterministic first-reconcile signal rather than a wall clock. Document that instead of implying the wait closes the gap. Signed-off-by: Peter Pan <Peter.Pan@daocloud.io>
2 participants
Summary
selection-http/dumpendpoint backed by the embedded selection service KV-index snapshot.The recovery gate intentionally does not claim an atomic snapshot/live-event handoff; that remains a follow-up design boundary.
Closes #13403
Validation
cargo fmt -p dynamo-ext-proc -- --checkcargo build -p dynamo-ext-proccargo clippy -p dynamo-ext-proc --no-deps --all-targets -- -D warningsgit diff --checkpassed============
Kubernetes Validation Summary
Tested on a real Kubernetes cluster (OrbStack, k8s v1.34, GAIE CRDs v1.2.1 + agentgateway v1.0.0 gateway) ,
using a GPU-less mock vLLM worker that publishes synthetic ZMQ KV events: https://github.com/panpan0000/simple-kv-event-mocker :-)
Step 1: cold start with 2 EPP and 2 mocker to generate kv-events
Step2 : delete one replica
Step3 : wait new replica
Step 4: validate kv-indexer syncing 🔥
the most important :
Step 5: check amount of events
and two new enhancement when doing the development
#13537
#13534
Summary by CodeRabbit
selection-httpendpoint.