feat: skeleton of AiProtocolManager filter with external buffer#45899
Open
penguingao wants to merge 29 commits into
Open
feat: skeleton of AiProtocolManager filter with external buffer#45899penguingao wants to merge 29 commits into
penguingao wants to merge 29 commits into
Conversation
|
CC @envoyproxy/api-shepherds: Your approval is needed for changes made to |
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 |
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>
Fill the untested code paths in the AI Protocol Manager filter: - buffer_manager: add a configurable FailingExternalBuffer and tests for the write/read error paths (onExternalBufferError) and that errors after onDestroy() are inert. - filter_chain_bridge: new unit test covering both the decoder and the (previously entirely uncovered) encoder bridge -- accessors, data injection, pause/resume, watermark register/forward/unregister, and the unrecoverable-error local-reply path. - config: new unit test for the factory (stream filter from proto and empty proto, and registry registration). Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
/retest |
Contributor
Author
|
/assign @botengyao |
Contributor
Author
|
/assign @wbpcode |
botengyao
reviewed
Jul 2, 2026
botengyao
left a comment
botengyao
left a comment
Member
There was a problem hiding this comment.
this is great, lgtm module the code namespace question.
/wait
| // - Replay: re-injected data is paced against the chain's back-pressure, | ||
| // delivered through ReplayWatermarkHandler. We pause issuing reads/injects | ||
| // while the chain is backed up. | ||
| class BufferManager : public ReplayWatermarkHandler, Logger::Loggable<Logger::Id::filter> { |
Member
There was a problem hiding this comment.
I wonder if we want to make this class to be out of the specific filter's scope and use in a common namespace.
Contributor
Author
There was a problem hiding this comment.
SG - moved to source/common/external_buffer and made the namespace Envoy::ExternalBuffer.
| // call returns) or asynchronously on a later iteration -- whichever fits its | ||
| // store -- and the BufferManager tolerates both. Destroying the buffer cancels | ||
| // any pending callbacks; they will not fire afterwards. | ||
| class ExternalBuffer { |
Member
There was a problem hiding this comment.
similarly it will make sense we move this interface to a common namespace, wdyt
Contributor
Author
There was a problem hiding this comment.
Done. Moved to source/common/external_buffer.
Member
There was a problem hiding this comment.
sry, actually this interface file will make sense to move to the folder /envoy/buffer, where almost of interface .h are hosted.
Contributor
Author
There was a problem hiding this comment.
I think it makes more sense to not mix-up with envoy/buffer - it is where most people should look at for vast majority of the in-envoy, in-memory operations. the external buffer is a different concept despite it also has "buffer" in its name. it does not interact with rest of the system the same way regular buffer does.
WDYT?
…ce/common/external_buffer directory Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
Thanks @botengyao for the quick turn around. |
…ce/common/external_buffer directory for real... Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
/retest |
botengyao
reviewed
Jul 2, 2026
botengyao
left a comment
botengyao
left a comment
Member
There was a problem hiding this comment.
here is another pass.
/wait
| // call returns) or asynchronously on a later iteration -- whichever fits its | ||
| // store -- and the BufferManager tolerates both. Destroying the buffer cancels | ||
| // any pending callbacks; they will not fire afterwards. | ||
| class ExternalBuffer { |
Member
There was a problem hiding this comment.
sry, actually this interface file will make sense to move to the folder /envoy/buffer, where almost of interface .h are hosted.
| replay_high_watermark_count_); | ||
| if (replay_high_watermark_count_ == 0) { | ||
| // Drained: resume replay where we paused. | ||
| maybeReadNextChunk(); |
Member
There was a problem hiding this comment.
In the call, it will inject data to the filter chain, and then the next filter could local reply or reset the request to trigger the onDestroy() of the filter and then the buffer manager, which it will unregister replay watermarks while FilterManager::callUpstreamLowWatermarkCallbacks() is still iterating, all these can happen synchronously. Should we do not inject or continue decoding directly from onReplayBelowLowWatermark(). Schedule replay_cb_ next-iteration and let replay resume off the watermark callback stack?
Contributor
Author
There was a problem hiding this comment.
Great catch. Fixed.
It also fixes the case where onExternalBufferError() gets called during watermark and thus causing router cleanup and watermark re-entrant problem.
Signed-off-by: Peng Gao <pengg@google.com>
Merge envoyproxy#45932 to fix build failure. Signed-off-by: Peng Gao <pengg@google.com>
Contributor
Author
|
/wait |
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 |
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
4 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