Skip to content

Commit 7ebc2d8

Browse files
noyitznoyitz
committed
feat: add response body mode framework for streaming support
Add ResponseBodyMode type and ResponseBodyRequirement interface that allows plugins to declare their response body needs (BodyNotNeeded, BodyChunked, or BodyFull). The framework pre-computes per-profile whether response body buffering is needed at startup. When NeedsResponseBuffering is false for the selected profile, each response body chunk is acked immediately and forwarded to the client, enabling real-time streaming. Plugins that don't implement the interface default to BodyFull (backward compatible). Closes llm-d#168 Signed-off-by: Noy Itzikowitz <nitzikow@redhat.com>
1 parent 1a22645 commit 7ebc2d8

9 files changed

Lines changed: 532 additions & 10 deletions

File tree

‎docs/proposals/044-plugin-body-mode-capabilities/README.md‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
# Plugin Response Body Mode Capabilities
2+
3+
## Proposal Status
4+
***Implementing***
5+
6+
## Summary
7+
8+
This proposal introduces a **response body mode capability declaration** for IPP plugins. Each response plugin declares what level of access it needs to the response body, enabling the framework to pre-compute per-profile whether response body buffering is needed. When no plugin in the selected profile needs the full response body, the framework passes each chunk through to the client immediately — enabling real-time streaming.
9+
10+
## Problem Statement
11+
12+
The ext_proc `response_body_mode` is set to `FULL_DUPLEX_STREAMED` in deployment. The IPP framework accumulates all response body chunks in memory before processing ([server.go](../../../pkg/handlers/server.go)), which means:
13+
14+
1. **Streaming is broken** — clients see all response chunks arrive at once after the full response is collected
15+
2. **Memory is unbounded** — each concurrent request accumulates the full response body with no upper limit
16+
3. **All plugins pay the cost** — even profiles where no plugin needs the response body force full buffering
17+
18+
## Design
19+
20+
### Response Body Mode Declaration
21+
22+
Plugins optionally implement `ResponseBodyRequirement` to declare their needs:
23+
24+
```go
25+
type ResponseBodyMode int
26+
27+
const (
28+
BodyNotNeeded ResponseBodyMode = iota // headers-only plugin
29+
BodyChunked // can process individual chunks (future: ChunkProcessor)
30+
BodyFull // needs complete body in memory
31+
)
32+
33+
type ResponseBodyRequirement interface {
34+
ResponseBodyMode() ResponseBodyMode
35+
}
36+
```
37+
38+
Plugins that don't implement `ResponseBodyRequirement` default to `BodyFull` (backward compatible).
39+
40+
### Pre-Computation
41+
42+
After profiles are built in the config loader, the framework iterates each profile's `ResponsePlugins`:
43+
- If **any** plugin returns `BodyFull` (or doesn't implement the interface) → `profile.NeedsResponseBuffering = true`
44+
- Otherwise → `profile.NeedsResponseBuffering = false`
45+
46+
This is computed once at startup — no per-request overhead.
47+
48+
### Conditional Buffering in server.go
49+
50+
At runtime, after the `ProfilePicker` selects a profile for the request:
51+
52+
**When `NeedsResponseBuffering = true`** (current behavior):
53+
- Accumulate all chunks: `responseBody = append(responseBody, chunk...)`
54+
- On EndOfStream: call `HandleResponseBody(ctx, reqCtx, responseBody)`
55+
- Response plugins get the full parsed body
56+
57+
**When `NeedsResponseBuffering = false`** (new streaming path):
58+
- Each chunk is acked immediately via `ackResponseBodyChunk()`
59+
- Envoy forwards the chunk to the client in real-time
60+
- No accumulation, no `HandleResponseBody` call
61+
- Event notification still fires on EndOfStream for metrics/observability
62+
63+
### Startup Logging
64+
65+
The framework logs per-profile buffering decisions at startup:
66+
```
67+
INFO Profile response buffering computed profile=default needsResponseBuffering=true bufferingPlugins=[api-translation]
68+
INFO Profile response buffering computed profile=headers-only needsResponseBuffering=false bufferingPlugins=[]
69+
```
70+
71+
Plugins that don't implement `ResponseBodyRequirement` get a warning:
72+
```
73+
INFO Response plugin does not declare ResponseBodyRequirement, defaulting to BodyFull profile=default plugin=legacy-plugin/legacy
74+
```
75+
76+
## Implementation Phases
77+
78+
| Phase | Scope | Status |
79+
|-------|-------|--------|
80+
| 1 — Framework | `ResponseBodyMode`, `ResponseBodyRequirement` interface, pre-computation, conditional buffering, tests | **This PR** |
81+
| 2 — ChunkProcessor | `ChunkProcessor` interface for `BodyChunked` plugins, per-chunk dispatch in server loop | Next PR |
82+
| 3 — Plugin declarations | Existing plugins implement `ResponseBodyRequirement` (ODH repo) | Separate PR |
83+
84+
## Migration
85+
86+
Existing plugins that don't implement `ResponseBodyRequirement` default to `BodyFull`. No breaking changes — everything works exactly as before. Profiles with only legacy plugins will continue to buffer.
87+
88+
## Future: Dynamic Mode Override
89+
90+
Envoy ext_proc supports `mode_override` in responses. A future enhancement could override `response_body_mode` to `NONE` per-request when no response plugin in the selected profile needs the body at all. This is out of scope for the initial implementation.

‎pkg/config/loader/configloader.go‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,8 @@ func LoadConfiguration(configBytes []byte, handle plugin.Handle, processor datas
9696
return nil, err
9797
}
9898

99+
computeResponseBuffering(profiles, logger)
100+
99101
return &config.Config{
100102
ProfilePicker: profilePicker,
101103
Profiles: profiles,
@@ -307,6 +309,36 @@ func buildPostProcessors(rawConfig *configapi.PluginRefList, handle plugin.Handl
307309
return postProcessors, nil
308310
}
309311

312+
// computeResponseBuffering pre-computes NeedsResponseBuffering for each profile based on the
313+
// ResponseBodyMode declared by each response plugin. If any response plugin returns BodyFull
314+
// or doesn't implement ResponseBodyRequirement, the profile needs buffering.
315+
func computeResponseBuffering(profiles map[string]*requesthandling.Profile, logger logr.Logger) {
316+
for name, profile := range profiles {
317+
needsBuffering := false
318+
var bufferingPlugins []string
319+
320+
for _, rp := range profile.ResponsePlugins {
321+
mode := requesthandling.BodyFull
322+
if req, ok := rp.(requesthandling.ResponseBodyRequirement); ok {
323+
mode = req.ResponseBodyMode()
324+
} else {
325+
logger.Info("Response plugin does not declare ResponseBodyRequirement, defaulting to BodyFull",
326+
"profile", name, "plugin", rp.TypedName())
327+
}
328+
if mode == requesthandling.BodyFull {
329+
needsBuffering = true
330+
bufferingPlugins = append(bufferingPlugins, rp.TypedName().Name)
331+
}
332+
}
333+
334+
profile.NeedsResponseBuffering = needsBuffering
335+
logger.Info("Profile response buffering computed",
336+
"profile", name,
337+
"needsResponseBuffering", needsBuffering,
338+
"bufferingPlugins", bufferingPlugins)
339+
}
340+
}
341+
310342
// buildModelSelector iterates all built profiles and, for each model-selector plugin found in
311343
// RequestPlugins, calls AddPlugins with the profile's ModelSelectorPlugins. If no Picker was
312344
// configured, MaxScorePicker is used as the default.

‎pkg/config/loader/response_buffering_test.go‎

Lines changed: 171 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,171 @@
1+
/*
2+
Copyright 2026 The llm-d Authors.
3+
4+
Licensed under the Apache License, Version 2.0 (the "License");
5+
you may not use this file except in compliance with the License.
6+
You may obtain a copy of the License at
7+
8+
http://www.apache.org/licenses/LICENSE-2.0
9+
10+
Unless required by applicable law or agreed to in writing, software
11+
distributed under the License is distributed on an "AS IS" BASIS,
12+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
See the License for the specific language governing permissions and
14+
limitations under the License.
15+
*/
16+
17+
package loader
18+
19+
import (
20+
"context"
21+
"testing"
22+
23+
"sigs.k8s.io/controller-runtime/pkg/log"
24+
25+
logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging"
26+
"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/plugin"
27+
"github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/requesthandling"
28+
)
29+
30+
type fakeResponsePlugin struct {
31+
name string
32+
mode *requesthandling.ResponseBodyMode
33+
}
34+
35+
func (p *fakeResponsePlugin) TypedName() plugin.TypedName {
36+
return plugin.TypedName{Type: "fake", Name: p.name}
37+
}
38+
39+
func (p *fakeResponsePlugin) ProcessResponse(_ context.Context, _ *plugin.CycleState, _ *requesthandling.InferenceResponse) error {
40+
return nil
41+
}
42+
43+
func (p *fakeResponsePlugin) ResponseBodyMode() requesthandling.ResponseBodyMode {
44+
if p.mode != nil {
45+
return *p.mode
46+
}
47+
return requesthandling.BodyFull
48+
}
49+
50+
var (
51+
_ requesthandling.ResponseProcessor = &fakeResponsePlugin{}
52+
_ requesthandling.ResponseBodyRequirement = &fakeResponsePlugin{}
53+
)
54+
55+
type legacyResponsePlugin struct {
56+
name string
57+
}
58+
59+
func (p *legacyResponsePlugin) TypedName() plugin.TypedName {
60+
return plugin.TypedName{Type: "legacy", Name: p.name}
61+
}
62+
63+
func (p *legacyResponsePlugin) ProcessResponse(_ context.Context, _ *plugin.CycleState, _ *requesthandling.InferenceResponse) error {
64+
return nil
65+
}
66+
67+
var _ requesthandling.ResponseProcessor = &legacyResponsePlugin{}
68+
69+
func modePtr(m requesthandling.ResponseBodyMode) *requesthandling.ResponseBodyMode { return &m }
70+
71+
func TestComputeResponseBuffering(t *testing.T) {
72+
logger := log.FromContext(logutil.NewTestLoggerIntoContext(context.Background()))
73+
74+
tests := []struct {
75+
name string
76+
plugins []requesthandling.ResponseProcessor
77+
wantBuffering bool
78+
}{
79+
{
80+
name: "no response plugins",
81+
plugins: nil,
82+
wantBuffering: false,
83+
},
84+
{
85+
name: "all BodyNotNeeded",
86+
plugins: []requesthandling.ResponseProcessor{
87+
&fakeResponsePlugin{name: "a", mode: modePtr(requesthandling.BodyNotNeeded)},
88+
&fakeResponsePlugin{name: "b", mode: modePtr(requesthandling.BodyNotNeeded)},
89+
},
90+
wantBuffering: false,
91+
},
92+
{
93+
name: "all BodyChunked",
94+
plugins: []requesthandling.ResponseProcessor{
95+
&fakeResponsePlugin{name: "a", mode: modePtr(requesthandling.BodyChunked)},
96+
},
97+
wantBuffering: false,
98+
},
99+
{
100+
name: "one BodyFull forces buffering",
101+
plugins: []requesthandling.ResponseProcessor{
102+
&fakeResponsePlugin{name: "a", mode: modePtr(requesthandling.BodyNotNeeded)},
103+
&fakeResponsePlugin{name: "b", mode: modePtr(requesthandling.BodyFull)},
104+
},
105+
wantBuffering: true,
106+
},
107+
{
108+
name: "legacy plugin without ResponseBodyRequirement defaults to BodyFull",
109+
plugins: []requesthandling.ResponseProcessor{
110+
&legacyResponsePlugin{name: "old-plugin"},
111+
},
112+
wantBuffering: true,
113+
},
114+
{
115+
name: "mixed: BodyChunked + legacy forces buffering",
116+
plugins: []requesthandling.ResponseProcessor{
117+
&fakeResponsePlugin{name: "a", mode: modePtr(requesthandling.BodyChunked)},
118+
&legacyResponsePlugin{name: "legacy"},
119+
},
120+
wantBuffering: true,
121+
},
122+
{
123+
name: "mixed: BodyNotNeeded + BodyChunked — no buffering",
124+
plugins: []requesthandling.ResponseProcessor{
125+
&fakeResponsePlugin{name: "a", mode: modePtr(requesthandling.BodyNotNeeded)},
126+
&fakeResponsePlugin{name: "b", mode: modePtr(requesthandling.BodyChunked)},
127+
},
128+
wantBuffering: false,
129+
},
130+
}
131+
132+
for _, tc := range tests {
133+
t.Run(tc.name, func(t *testing.T) {
134+
profiles := map[string]*requesthandling.Profile{
135+
"test": {
136+
ResponsePlugins: tc.plugins,
137+
},
138+
}
139+
computeResponseBuffering(profiles, logger)
140+
if profiles["test"].NeedsResponseBuffering != tc.wantBuffering {
141+
t.Errorf("NeedsResponseBuffering = %v, want %v", profiles["test"].NeedsResponseBuffering, tc.wantBuffering)
142+
}
143+
})
144+
}
145+
}
146+
147+
func TestComputeResponseBuffering_MultipleProfiles(t *testing.T) {
148+
logger := log.FromContext(logutil.NewTestLoggerIntoContext(context.Background()))
149+
150+
profiles := map[string]*requesthandling.Profile{
151+
"streaming": {
152+
ResponsePlugins: []requesthandling.ResponseProcessor{
153+
&fakeResponsePlugin{name: "headers-only", mode: modePtr(requesthandling.BodyNotNeeded)},
154+
},
155+
},
156+
"full-body": {
157+
ResponsePlugins: []requesthandling.ResponseProcessor{
158+
&fakeResponsePlugin{name: "translator", mode: modePtr(requesthandling.BodyFull)},
159+
},
160+
},
161+
}
162+
163+
computeResponseBuffering(profiles, logger)
164+
165+
if profiles["streaming"].NeedsResponseBuffering {
166+
t.Error("streaming profile should not need buffering")
167+
}
168+
if !profiles["full-body"].NeedsResponseBuffering {
169+
t.Error("full-body profile should need buffering")
170+
}
171+
}

‎pkg/framework/interface/requesthandling/plugins.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,3 +56,36 @@ type PostProcessor interface {
5656
// PostProcess is invoked to post-process requests after the response plugins of the selected profile run.
5757
PostProcess(ctx context.Context, cycleState *plugin.CycleState, response *InferenceResponse) error
5858
}
59+
60+
type ResponseBodyMode int
61+
62+
const (
63+
// BodyNotNeeded indicates the plugin does not need the response body at all (headers-only plugin).
64+
BodyNotNeeded ResponseBodyMode = iota
65+
// BodyChunked indicates the plugin can process individual response body chunks as they stream through.
66+
// Plugins declaring BodyChunked MUST also implement ChunkProcessor (validated at startup).
67+
BodyChunked
68+
// BodyFull indicates the plugin needs the complete response body buffered in memory before processing.
69+
// This is the default for plugins that do not implement ResponseBodyRequirement (backward compatible).
70+
BodyFull
71+
)
72+
73+
func (m ResponseBodyMode) String() string {
74+
switch m {
75+
case BodyNotNeeded:
76+
return "BodyNotNeeded"
77+
case BodyChunked:
78+
return "BodyChunked"
79+
case BodyFull:
80+
return "BodyFull"
81+
default:
82+
return "Unknown"
83+
}
84+
}
85+
86+
// ResponseBodyRequirement allows response plugins to declare what level of access they need
87+
// to the response body. Plugins that don't implement this interface default to BodyFull
88+
// (backward compatible — the framework buffers the full response before calling ProcessResponse).
89+
type ResponseBodyRequirement interface {
90+
ResponseBodyMode() ResponseBodyMode
91+
}

‎pkg/framework/interface/requesthandling/types.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,4 +120,10 @@ type Profile struct {
120120
// ModelSelectorPlugins are the Filter, Scorer (including WeightedScorer), and Picker plugin
121121
// instances to be wired into any model-selector plugin present in RequestPlugins.
122122
ModelSelectorPlugins []plugin.Plugin
123+
124+
// NeedsResponseBuffering is pre-computed at startup from the ResponsePlugins' ResponseBodyMode
125+
// declarations. When true, the server accumulates all response body chunks and calls
126+
// HandleResponseBody with the full body on EndOfStream. When false, each chunk is acked
127+
// immediately and forwarded to the client without buffering.
128+
NeedsResponseBuffering bool
123129
}

‎pkg/handlers/response.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,29 @@ func (s *Server) generateEmptyResponseBodyResponse(responseBodyBytes []byte) []*
131131
return responses
132132
}
133133

134+
// ackResponseBodyChunk returns an immediate ack for a response body chunk, allowing Envoy
135+
// to forward it to the client without waiting for the full body to be accumulated.
136+
func (s *Server) ackResponseBodyChunk(body *eppb.HttpBody) []*eppb.ProcessingResponse {
137+
return []*eppb.ProcessingResponse{
138+
{
139+
Response: &eppb.ProcessingResponse_ResponseBody{
140+
ResponseBody: &eppb.BodyResponse{
141+
Response: &eppb.CommonResponse{
142+
BodyMutation: &eppb.BodyMutation{
143+
Mutation: &eppb.BodyMutation_StreamedResponse{
144+
StreamedResponse: &eppb.StreamedBodyResponse{
145+
Body: body.Body,
146+
EndOfStream: body.EndOfStream,
147+
},
148+
},
149+
},
150+
},
151+
},
152+
},
153+
},
154+
}
155+
}
156+
134157
// HandleResponseTrailers handles response trailers.
135158
func (s *Server) HandleResponseTrailers(trailers *eppb.HttpTrailers) ([]*eppb.ProcessingResponse, error) {
136159
return []*eppb.ProcessingResponse{

0 commit comments

Comments
 (0)