feat: skeleton of AiProtocolManager filter with external buffer#45899
Merged
Conversation
|
CC @envoyproxy/api-shepherds: Your approval is needed for changes made to |
penguingao
marked this pull request as draft
June 30, 2026 16:41
Contributor
Author
penguingao
changed the title
Contributor
Author
|
cc @cdmetcalf |
Contributor
Author
|
cc @abeyad |
Contributor
Author
|
/retest |
Contributor
Author
|
format failure is pre-existing. Fix is in #45907 |
penguingao
marked this pull request as ready for review
June 30, 2026 23:56
Contributor
Author
|
/retest |
Introduce a new HTTP filter `envoy.filters.http.ai_protocol_manager` with an empty config message. The filter is currently a no-op pass-through and is marked alpha / work-in-progress. Signed-off-by: Peng Gao <pengg@google.com>
…control
Add the first real feature to the AiProtocolManager HTTP filter: stream the
request body into an external buffer as it arrives and replay it back into the
filter chain once the stream ends, so the full payload can later be parsed and
validated (for routing/admission decisions) without pinning it in the
connection manager's buffers.
External buffer:
- external_buffer.h defines an async, append-only, random-access byte store
(append/read-by-offset/length) with watermark-based flow control, plus a
factory. All data ops are asynchronous since a real backing store does I/O.
- external_buffer_impl.{h,cc} is a pure in-memory reference implementation that
posts completions on the dispatcher and cancels pending callbacks on
destruction.
Filter:
- On decodeData, append each chunk to the external buffer and return
StopIterationNoBuffer (the filter owns buffering/continuation). On end_stream,
read the payload back in bounded chunks and inject it into the chain.
- Ingest flow control: the external buffer's in-flight watermark pushes back on
the request source via onDecoderFilterAboveWriteBufferHighWatermark.
Upstream-request flow control (new core plumbing):
- DownstreamWatermarkCallbacks only fire on downstream (client) backup, so a
filter replaying request data had no way to observe upstream back-pressure;
the router's onDecoderFilterAboveWriteBufferHighWatermark signal terminated at
the HCM by read-disabling the downstream codec, which does nothing for a
filter producing its own data.
- Add Http::UpstreamWatermarkCallbacks and
add/removeUpstreamWatermarkCallbacks on StreamDecoderFilterCallbacks. The HCM
ActiveStream now fans the aggregate upstream back-pressure out to subscribers
in onDecoderFilter{Above,Below}WriteBuffer*Watermark, with the FilterManager
tracking outstanding-high depth so a mid-stream subscriber is brought current.
- The filter subscribes and pauses/resumes replay accordingly.
Signed-off-by: Peng Gao <pengg@google.com>
Extract the offload+replay pipeline and its bidirectional flow control out of AiProtocolManagerFilter into a reusable, path-agnostic BufferManager that reaches the filter chain only through an abstract FilterChainBridge. Provide DecoderFilterChainBridge (decode path, wired) and EncoderFilterChainBridge (encode path, provided but not yet wired), mirroring the ext_proc ProcessorState precedent. This lets the response path reuse the exact same logic by constructing a second BufferManager with the encoder bridge. The filter becomes a thin delegator. Decode behavior is preserved: the existing filter_test passes unchanged and now doubles as the decode integration test. New buffer_manager_test unit-tests the agnostic logic against a hand-written fake bridge. Signed-off-by: Peng Gao <pengg@google.com>
…inated streams When a request body is terminated by trailers (the last DATA frame carries end_stream=false and the trailers carry END_STREAM), decodeData never observed end_stream, so replay never started: the offloaded body sat in the external buffer forever and the request hung. The filter also did not override decodeTrailers, so the held trailers could never advance past the stuck body. BufferManager now exposes onTrailers(): it marks the stream complete, drives replay (final body frame injected with end_stream=false), and releases the held trailers via a new path-agnostic FilterChainBridge::continueIteration() hook (continueDecoding/continueEncoding) once the body has been replayed, preserving body-then-trailers ordering even under replay back-pressure. With no buffered body it returns Continue so trailers flow normally. The filter overrides decodeTrailers to delegate to the manager. Adds unit coverage (trailer round-trip, large multi-chunk, empty body, no body) and a decode-path integration test asserting continueDecoding() fires after the body. Signed-off-by: Peng Gao <pengg@google.com>
Routing and admission decisions for AI traffic depend on the parsed request payload, but the rest of the filter chain would otherwise act on the headers before that payload is available. decodeHeaders() now returns StopIteration whenever a body follows, pinning the headers at this filter while decodeData() keeps offloading the body. The held headers are released to downstream filters only when replay begins -- the first injectDecodedDataToFilterChain() call flushes them ahead of the replayed body. Headers-only requests carry no payload and are passed through (Continue) to avoid deadlock. Signed-off-by: Peng Gao <pengg@google.com>
Exercise the filter end-to-end over the full protocol matrix (HTTP/1, HTTP/2, HTTP/3 downstream x upstream), asserting the offload/replay path is transparent: the upstream observes headers, the complete body across a range of sizes, and trailers unchanged. Covers header-only, header+body (0B..1MiB, single- and multi-frame), header+body+trailers, and header+trailers-with-no-body. Signed-off-by: Peng Gao <pengg@google.com>
ExternalBuffer::append() previously took a borrowed Buffer::Instance& that each implementation had to drain synchronously to keep the bytes alive across its asynchronous I/O. That ownership grab is storage-agnostic, so every backend (disk, remote, ...) would have repeated it identically. Change append() to receive an already-owned Buffer::InstancePtr and do the hand-off once in BufferManager::onData(), before dispatching to the backend. The in-memory implementation drops its staging step and just moves the owned buffer into its completion callback. The filter-facing onData(Buffer::Instance&, ...) signature is unchanged. Signed-off-by: Peng Gao <pengg@google.com>
…a common practice. also make sure we don't busy poll on the current epoll iteration via post Signed-off-by: Peng Gao <pengg@google.com>
Replace the ExternalBuffer append() API, which left the buffer responsible
for an internal write queue and watermark-based flow control, with a
single-writer write(): the caller issues at most one write at a time. This
lets the buffer drop setWatermarks() and ExternalBufferWatermarkCallbacks,
shrinking the interface to write()/read()/length().
The write queue and ingest flow control move into BufferManager, which now:
- serializes writes (one outstanding) and queues the backlog in pending_;
- tracks not-yet-durable bytes (queued plus in-flight) against the
configured buffer limit and drives source pause/resume via the bridge;
- batches small frames into chunk-sized writes (WriteFlushThreshold) so a
store that keeps up does not get a stream of tiny writes, flushing early
at end_stream or while the source is paused (the latter also prevents a
stall when the buffer limit is below the threshold).
An empty terminal frame issues no write, so replay is scheduled on a fresh
event-loop iteration rather than started re-entrantly from the data callback.
Signed-off-by: Peng Gao <pengg@google.com>
…ream
Turn replay() into replay(offset, length, done): the caller streams back the
ranges it wants (chaining sub-ranges from the done callback, one in flight at a
time) and learns when each range has been injected. Add length() so the caller
can size ranges, covering durable, queued, and in-flight bytes.
With ranges caller-driven, end-of-stream handling moves out of the BufferManager
to the filter, where it belongs:
- injectData() drops its end_stream flag; replay only injects data frames.
- trailers_pending_, the end_stream-on-last-frame logic, the empty end_stream
marker, and FilterChainBridge::continueIteration() are gone.
- The filter terminates the stream from the done callback: an empty end_stream
frame for a body-terminated request, continueDecoding() to release the held
trailers for a trailer-terminated one.
endStream() loses its trailers_pending argument and now only flushes the batched
backlog. streamBackToFilterChain() folds into maybeStartReplay(), and the replay
read loop no longer carries end-stream/trailer special cases.
Signed-off-by: Peng Gao <pengg@google.com>
…comment cleanups
Defer the replay start in replay() with scheduleCallbackCurrentIteration() instead
of next-iteration: it still runs off the caller's filter-callback stack (so no
re-entrant injection) but later in the same event-loop pass, matching the
write-completion path (which rides dispatcher.post(), itself current-iteration)
and avoiding an extra iteration of latency. The per-chunk replay yield keeps using
next-iteration.
Also:
- split updateIngestBackpressure() into two guarded early-return conditions;
- treat a zero-length offload as empty();
- note why onReadComplete() finishes a range directly rather than via
maybeReadNextChunk() (the back-pressure pause check precedes the end check);
- tighten the ExternalBuffer and write() contract comments and refresh the
AiProtocolManagerFilter overview (streaming parse/validate-while-offloading,
replay-when-valid, future AI extension chain).
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
I fixed a few more potential issues that are related to re-entering filter-chain synchronously and coming back from filter-chain synchronously. PTAL. While we're at it, what's the current thought on co-routine in Envoy? I would like to attempt modeling the forth coming AI filter chain using co-routine - I tend to think it might make filters easier to write and reason about, but that's a later point we can discuss. |
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
/retest |
botengyao
previously approved these changes
Jul 6, 2026
botengyao
left a comment
botengyao
left a comment
Member
There was a problem hiding this comment.
lgtm, thanks for the great contribution.
Contributor
Author
|
/cc @VTWoods |
…tern Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
/retest |
wbpcode
requested changes
Jul 7, 2026
wbpcode
left a comment
wbpcode
left a comment
Member
There was a problem hiding this comment.
Thanks for this great contribution. I take a first quick pass. And I guess for now this PR only contains buffer offloading. It's fine to me except the code position.
Member
|
/lgtm api |
Member
|
I do be curious to follow up PRs 😄 |
… to ai_protocol_manager Signed-off-by: Peng Gao <pengg@google.com>
pull in main format fixes. Signed-off-by: Peng Gao <pengg@google.com>
botengyao
pushed a commit
that referenced
this pull request
Jul 7, 2026
<!-- !!!ATTENTION!!! If you are fixing **any** crash or **any** potential security issue, **do not open a pull request**. Instead, please [open a GitHub Security Advisory](https://github.com/envoyproxy/envoy/security/advisories/new) (preferred). Alternatively, you may email envoy-security@googlegroups.com. Thank you in advance for helping to keep Envoy secure. !!!ATTENTION!!! For an explanation of how to fill out the fields, please see the relevant section in [PULL_REQUESTS.md](https://github.com/envoyproxy/envoy/blob/main/PULL_REQUESTS.md) !!!ATTENTION!!! Please check the [use of generative AI policy](https://github.com/envoyproxy/envoy/blob/main/CONTRIBUTING.md?plain=1#L41). You may use generative AI only if you fully understand the code. You need to disclose this usage in the PR description to ensure transparency. --> Commit Message: Introduce UpstreamWatermarkCallbacks (envoy/http/codec.h) and the (add|remove)UpstreamWatermarkCallbacks subscription API on StreamDecoderFilterCallbacks (envoy/http/filter.h), implemented by the FilterManager and fanned out from the connection manager's onDecoderFilter{Above,Below}WriteBuffer{High,Low}Watermark hooks. Useful when a filter uses injectDecodedDataToFilterChain and wants to avoid overrunning the upstream buffer. This is a spawn-off of #45899 as per review. Also adds stub overrides in the async client and tcp_proxy StreamDecoder FilterCallbacks implementations, a mock, and connection-manager coverage. This change is a spawn-off of #45899, which I used AI to generate. I have read and understand the code. Additional Description: Risk Level: Medium - new callbacks to HCM, but not yet used Testing: unit tests Docs Changes: n/a Release Notes: n/a Platform Specific Features: n/a [Optional Runtime guard:] [Optional Fixes #Issue] [Optional Fixes commit #PR or SHA] [Optional Deprecated:] [Optional [API Considerations](https://github.com/envoyproxy/envoy/blob/main/api/review_checklist.md):] Signed-off-by: Peng Gao <pengg@google.com>
Merge main to get the HCM upstream watermark PR. Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
@wbpcode - PTAL and let me know if anything else is needed before we merge. |
tyxia
approved these changes
Jul 8, 2026
tyxia
left a comment
tyxia
left a comment
Member
There was a problem hiding this comment.
Impressive work and start for such a complex feature, Thanks!
| } | ||
| // Not-yet-durable bytes: queued backlog plus the in-flight write. | ||
| const uint64_t unacked = pending_.length() + in_flight_write_size_; | ||
| if (!source_paused_ && unacked > high_watermark_) { |
Member
There was a problem hiding this comment.
For InMemoryExternalBuffer store, the write/ingest part is fast as there is no I/O etc and ack after one loop from dispatcher post. Thus, in practice unacked will unlikely to get larger than high_watermark_ for inMemoryStore? With that said, the max cap (in another comment) for the buffer in inMemory store is important here as it could grow unboundedly without back pressue.
Contributor
Author
There was a problem hiding this comment.
Yes, this doesn't trigger as-is. My follow-up change will add a file system based implementation, which makes this more meaningful.
Member
There was a problem hiding this comment.
One thing worth discussing, in future PR: on the ingest/write path, back-pressuring the sender may not always be the right option when we need to parse the whole body before streaming out any data to upstream (i.e., no drain in the store).
It could result in a deadlock state: we can't make progress without more data, but we've stalled the sender that would provide it
Contributor
Author
There was a problem hiding this comment.
Thanks for bringing this up. Though, I don't think there is the risk of deadlock. To make the point, we need to clear two points:
unackedcounts only bytes that hasn't been fully handed to external buffer- the size of the external buffer is separately tracked
Flow control only cares unacked to make sure we maintain a bound memory footprint here. Bytes in external buffer, even if it's in-memory, it doesn't take part in flow control. This means, as long as we can write to the buffer, we don't deadlock.
A size limit indeed is useful, but it is to make sure we only access and process reasonable requests and responses. It is not mainly to control memory footprint (it may as well be used that way when only a in-memory buffer is used, but it doesn't necessarily have anything to do with memory footprint when the buffer is truly external).
wbpcode
approved these changes
Jul 8, 2026
This was referenced Jul 25, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Commit Message: This is part of #44681. This PR adds a skeletal AiProtocolManager HTTP filter and marks it as alpha. This PR only focuses on payload offloading to external buffer. The goal is for the Envoy to be able to parse large HTTP json payload that can be seen in the AI protocols (MCP, A2A, OpenAi completion and response API). It is necessary for the Envoy to see the whole payload so that it can avoid field / model / tool smuggling by sending json playload with duplicated keys.
This PR provides a BufferManager, interface of ExternalBuffer, and a simple implementation of in-memory only buffer. A follow-up will add storage based external buffer. The BufferManager and ExternalBuffer is defined such that it is reasonably simple to implement an external buffer be it local file system backed, network attached storage, or a remote server. This is achieved by putting much of the asynchronous I/O coordination in the buffer manager itself for it to be re-usable.
It finishes some of the unfinished wiring of upstream back pressure callbacks.
I used AI to generate part of the PR and I have fully reviewed the changes before sending the PR
Additional Description:
Risk Level: Medium - new filter
Testing: unit test and integration tests
Docs Changes: added place holder
Release Notes: n/a - WIP filter
Platform Specific Features: n/a