Repository: kubernetes-sigs/gateway-api-inference-extension

Issue:
Support long context requests
**What happened**:

When sending long context requests, longer than 128kb, envoy is currently failing with a 500 internal error

**What you expected to happen**:

Not to fail. 

**How to reproduce it (as minimally and precisely as possible)**:

**Anything else we need to know?**:

Currently EPP buffers the whole request body and [sends it back in one go to the proxy](https://github.com/kubernetes-sigs/gateway-api-inference-extension/blob/e8834c311ed599e2a99f85328cf2e0ae143402c3/pkg/epp/handlers/server.go#L344). We probably need to send it in chunks instead.

**Environment**: GKE


Hints from issue comments:
@nirrozenbaum: @ahg-g @kfswain have we done some stress test after the fix to understand what are the limits of the system in terms of long context request?
I assume we verified that 128kb is working, but do we know if there's a new, higher limit? 

Solution patch:
diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go
index 7731ded..4b849c8 100644
--- a/pkg/epp/handlers/server.go
+++ b/pkg/epp/handlers/server.go
@@ -37,6 +37,11 @@ import (
 	requtil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/request"
 )
 
+const (
+	// Certain envoy implementations set a max limit of 64Kb per streamed chunk, intentionally setting this lower for a safe margin.
+	bodyByteLimit = 62000
+)
+
 func NewStreamingServer(destinationEndpointHintMetadataNamespace, destinationEndpointHintKey string, datastore Datastore, director Director) *StreamingServer {
 	return &StreamingServer{
 		destinationEndpointHintMetadataNamespace: destinationEndpointHintMetadataNamespace,
@@ -460,3 +465,32 @@ func BuildErrResponse(err error) (*extProcPb.ProcessingResponse, error) {
 
 	return resp, nil
 }
+
+func buildCommonResponses(bodyBytes []byte, byteLimit int) []*extProcPb.CommonResponse {
+	responses := []*extProcPb.CommonResponse{}
+	startingIndex := 0
+	bodyLen := len(bodyBytes)
+
+	for startingIndex < bodyLen {
+		eos := false
+		len := min(bodyLen-startingIndex, byteLimit)
+		chunk := bodyBytes[startingIndex : len+startingIndex]
+		if len+startingIndex == bodyLen {
+			eos = true
+		}
+
+		commonResp := &extProcPb.CommonResponse{
+			BodyMutation: &extProcPb.BodyMutation{
+				Mutation: &extProcPb.BodyMutation_StreamedResponse{
+					StreamedResponse: &extProcPb.StreamedBodyResponse{
+						Body:        chunk,
+						EndOfStream: eos,
+					},
+				},
+			},
+		}
+		responses = append(responses, commonResp)
+		startingIndex += len
+	}
+	return responses
+}



Test patch:
diff --git a/pkg/epp/handlers/server_test.go b/pkg/epp/handlers/server_test.go
new file mode 100644
index 0000000..cc99a51
--- /dev/null
+++ b/pkg/epp/handlers/server_test.go
@@ -0,0 +1,64 @@
+package handlers
+
+import (
+	"crypto/rand"
+	"testing"
+)
+
+func TestBuildCommonResponses(t *testing.T) {
+	tests := []struct {
+		name                 string
+		count                int
+		expectedMessageCount int
+	}{
+		{
+			name:                 "below limit",
+			count:                bodyByteLimit - 1000,
+			expectedMessageCount: 1,
+		},
+		{
+			name:                 "at limit",
+			count:                bodyByteLimit,
+			expectedMessageCount: 1,
+		},
+		{
+			name:                 "off by one error?",
+			count:                bodyByteLimit + 1,
+			expectedMessageCount: 2,
+		},
+		{
+			name:                 "above limit",
+			count:                bodyByteLimit + 1000,
+			expectedMessageCount: 2,
+		},
+		{
+			name:                 "above limit",
+			count:                (bodyByteLimit * 2) + 1000,
+			expectedMessageCount: 3,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			arr := generateBytes(test.count)
+			responses := buildCommonResponses(arr, bodyByteLimit)
+			for i, response := range responses {
+				eos := response.BodyMutation.GetStreamedResponse().GetEndOfStream()
+				if eos == true && i+1 != len(responses) {
+					t.Fatalf("EoS should not be set")
+				}
+				if eos == false && i+1 == len(responses) {
+					t.Fatalf("EoS should be set")
+				}
+			}
+			if len(responses) != test.expectedMessageCount {
+				t.Fatalf("Expected: %v, Got %v", test.expectedMessageCount, len(responses))
+			}
+		})
+	}
+}
+
+func generateBytes(count int) []byte {
+	arr := make([]byte, count)
+	_, _ = rand.Read(arr)
+	return arr
+}