[bbr] Skip body re-serialization and mutation when unchanged (kubernetes-sigs#2556)

Add body mutation tracking to InferenceMessage via SetBody,
SetBodyField, RemoveBodyField, and BodyMutated so the request and
response handlers skip json.Marshal and Content-Length update when
no plugin modified the body.

SetBody replaces the entire body map for plugins that transform the
full payload (e.g., API translation). SetBodyField and RemoveBodyField
handle granular field-level mutations.

For unary mode, omit BodyMutation from the ext_proc response when
unchanged. For streaming mode, forward original bytes instead of
re-marshaling.

Add integration tests that verify body mutation tracking over real
gRPC for both unary and streaming paths.

Signed-off-by: Asaad Balum <asaad.balum@gmail.com>

Author: asaadbalum

pkg/bbr/framework/types.go

@@ -37,6 +37,7 @@ type InferenceMessage struct {
 	// mutations
 	mutatedHeaders map[string]string
 	removedHeaders sets.Set[string]
+	bodyMutated    bool
 }
 
 func (r *InferenceMessage) SetHeader(key string, value string) {
@@ -63,6 +64,27 @@ func (r *InferenceMessage) RemovedHeaders() []string {
 	return r.removedHeaders.UnsortedList()
 }
 
+func (r *InferenceMessage) SetBody(body map[string]any) {
+	r.Body = body
+	r.bodyMutated = true
+}
+
+func (r *InferenceMessage) SetBodyField(key string, value any) {
+	r.Body[key] = value
+	r.bodyMutated = true
+}
+
+func (r *InferenceMessage) RemoveBodyField(key string) {
+	if _, ok := r.Body[key]; ok {
+		delete(r.Body, key)
+		r.bodyMutated = true
+	}
+}
+
+func (r *InferenceMessage) BodyMutated() bool {
+	return r.bodyMutated
+}
+
 type InferenceRequest struct {
 	InferenceMessage
 }

pkg/bbr/framework/types_test.go

@@ -0,0 +1,103 @@
+/*
+Copyright 2026 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package framework
+
+import (
+	"testing"
+)
+
+func TestSetBodyField(t *testing.T) {
+	msg := newInferenceMessage()
+	if msg.BodyMutated() {
+		t.Error("new message should not be marked as body-mutated")
+	}
+
+	msg.SetBodyField("key", "value")
+
+	if !msg.BodyMutated() {
+		t.Error("expected BodyMutated() to return true after SetBodyField")
+	}
+	if got, ok := msg.Body["key"]; !ok || got != "value" {
+		t.Errorf("Body[\"key\"] = %v, %v; want \"value\", true", got, ok)
+	}
+}
+
+func TestSetBodyField_Overwrite(t *testing.T) {
+	msg := newInferenceMessage()
+	msg.Body["existing"] = "old"
+
+	msg.SetBodyField("existing", "new")
+
+	if !msg.BodyMutated() {
+		t.Error("expected BodyMutated() to return true after overwriting a field")
+	}
+	if got := msg.Body["existing"]; got != "new" {
+		t.Errorf("Body[\"existing\"] = %v; want \"new\"", got)
+	}
+}
+
+func TestRemoveBodyField(t *testing.T) {
+	msg := newInferenceMessage()
+	msg.Body["key"] = "value"
+
+	msg.RemoveBodyField("key")
+
+	if !msg.BodyMutated() {
+		t.Error("expected BodyMutated() to return true after RemoveBodyField")
+	}
+	if _, ok := msg.Body["key"]; ok {
+		t.Error("expected key to be removed from Body")
+	}
+}
+
+func TestRemoveBodyField_NonExistent(t *testing.T) {
+	msg := newInferenceMessage()
+
+	msg.RemoveBodyField("missing")
+
+	if msg.BodyMutated() {
+		t.Error("removing a non-existent field should not mark body as mutated")
+	}
+}
+
+func TestSetBody(t *testing.T) {
+	msg := newInferenceMessage()
+
+	msg.SetBody(map[string]any{"model": "llama", "prompt": "hello"})
+
+	if !msg.BodyMutated() {
+		t.Error("expected BodyMutated() to return true after SetBody")
+	}
+	if got, ok := msg.Body["model"]; !ok || got != "llama" {
+		t.Errorf("Body[\"model\"] = %v, %v; want \"llama\", true", got, ok)
+	}
+	if got, ok := msg.Body["prompt"]; !ok || got != "hello" {
+		t.Errorf("Body[\"prompt\"] = %v, %v; want \"hello\", true", got, ok)
+	}
+}
+
+func TestBodyMutated_FalseByDefault(t *testing.T) {
+	req := NewInferenceRequest()
+	if req.BodyMutated() {
+		t.Error("new InferenceRequest should not be marked as body-mutated")
+	}
+
+	resp := NewInferenceResponse()
+	if resp.BodyMutated() {
+		t.Error("new InferenceResponse should not be marked as body-mutated")
+	}
+}

pkg/bbr/handlers/request.go

@@ -66,12 +66,16 @@ func (s *Server) HandleRequestBody(ctx context.Context, reqCtx *RequestContext,
 	reqCtx.Request.SetHeader(BaseModelHeader, baseModel)
 	logger.Info("Base model from datastore", "baseModel", baseModel)
 
-	// TODO: check and do this only if the request body actually changed.
-	mutatedBodyBytes, err := json.Marshal(reqCtx.Request.Body)
-	if err != nil {
-		return nil, err
+	bodyMutated := reqCtx.Request.BodyMutated()
+	var mutatedBodyBytes []byte
+	if bodyMutated {
+		var err error
+		mutatedBodyBytes, err = json.Marshal(reqCtx.Request.Body)
+		if err != nil {
+			return nil, err
+		}
+		reqCtx.Request.SetHeader(contentLengthHeader, strconv.Itoa(len(mutatedBodyBytes)))
 	}
-	reqCtx.Request.SetHeader(contentLengthHeader, strconv.Itoa(len(mutatedBodyBytes)))
 
 	metrics.RecordSuccessCounter()
 
@@ -89,27 +93,35 @@ func (s *Server) HandleRequestBody(ctx context.Context, reqCtx *RequestContext,
 				},
 			},
 		})
-		ret = addStreamedBodyResponse(ret, mutatedBodyBytes)
+		if bodyMutated {
+			ret = addStreamedBodyResponse(ret, mutatedBodyBytes)
+		} else {
+			ret = addStreamedBodyResponse(ret, requestBodyBytes)
+		}
 		return ret, nil
 	}
 
+	// Necessary so that the new headers are used in the routing decision.
+	response := &eppb.CommonResponse{
+		ClearRouteCache: true,
+		HeaderMutation: &eppb.HeaderMutation{
+			SetHeaders:    envoy.GenerateHeadersMutation(reqCtx.Request.MutatedHeaders()),
+			RemoveHeaders: reqCtx.Request.RemovedHeaders(),
+		},
+	}
+	if bodyMutated {
+		response.BodyMutation = &eppb.BodyMutation{
+			Mutation: &eppb.BodyMutation_Body{
+				Body: mutatedBodyBytes,
+			},
+		}
+	}
+
 	return []*eppb.ProcessingResponse{
 		{
 			Response: &eppb.ProcessingResponse_RequestBody{
 				RequestBody: &eppb.BodyResponse{
-					Response: &eppb.CommonResponse{
-						// Necessary so that the new headers are used in the routing decision.
-						ClearRouteCache: true,
-						HeaderMutation: &eppb.HeaderMutation{
-							SetHeaders:    envoy.GenerateHeadersMutation(reqCtx.Request.MutatedHeaders()),
-							RemoveHeaders: reqCtx.Request.RemovedHeaders(),
-						},
-						BodyMutation: &eppb.BodyMutation{
-							Mutation: &eppb.BodyMutation_Body{
-								Body: mutatedBodyBytes,
-							},
-						},
-					},
+					Response: response,
 				},
 			},
 		},