Add code for Envoy extension that support body-to-header translation

Author: rramkumar1

body-based-routing.Dockerfile

@@ -0,0 +1,30 @@
+# Dockerfile has specific requirement to put this ARG at the beginning:
+# https://docs.docker.com/engine/reference/builder/#understand-how-arg-and-from-interact
+ARG BUILDER_IMAGE=golang:1.23-alpine
+ARG BASE_IMAGE=gcr.io/distroless/base-debian10
+
+## Multistage build
+FROM ${BUILDER_IMAGE} AS builder
+ENV CGO_ENABLED=0
+ENV GOOS=linux
+ENV GOARCH=amd64
+
+# Dependencies
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+
+# Sources
+COPY cmd ./cmd
+COPY pkg ./pkg
+COPY internal ./internal
+WORKDIR /src/cmd/body-based-routing
+RUN go build -o /body-based-routing
+
+## Multistage deploy
+FROM ${BASE_IMAGE}
+
+WORKDIR /
+COPY --from=builder /body-based-routing /body-based-routing
+
+ENTRYPOINT ["/body-based-routing"]

cmd/body-based-routing/health.go

@@ -0,0 +1,40 @@
+/*
+Copyright 2025 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 main
+
+import (
+	"context"
+
+	"github.com/go-logr/logr"
+	"google.golang.org/grpc/codes"
+	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/status"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+type healthServer struct {
+	logger logr.Logger
+}
+
+func (s *healthServer) Check(ctx context.Context, in *healthPb.HealthCheckRequest) (*healthPb.HealthCheckResponse, error) {
+	s.logger.V(logutil.VERBOSE).Info("gRPC health check serving", "service", in.Service)
+	return &healthPb.HealthCheckResponse{Status: healthPb.HealthCheckResponse_SERVING}, nil
+}
+
+func (s *healthServer) Watch(in *healthPb.HealthCheckRequest, srv healthPb.Health_WatchServer) error {
+	return status.Error(codes.Unimplemented, "Watch is not implemented")
+}

cmd/body-based-routing/main.go

@@ -0,0 +1,122 @@
+/*
+Copyright 2025 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 main
+
+import (
+	"flag"
+	"os"
+
+	"github.com/go-logr/logr"
+	uberzap "go.uber.org/zap"
+	"google.golang.org/grpc"
+	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
+	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/body-based-routing/server"
+)
+
+var (
+	grpcPort = flag.Int(
+		"grpcPort",
+		runserver.DefaultGrpcPort,
+		"The gRPC port used for communicating with Envoy proxy")
+	grpcHealthPort = flag.Int(
+		"grpcHealthPort",
+		9003,
+		"The port used for gRPC liveness and readiness probes")
+
+	setupLog = ctrl.Log.WithName("setup")
+)
+
+func main() {
+	if err := run(); err != nil {
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	opts := zap.Options{Development: true}
+	opts.BindFlags(flag.CommandLine)
+	flag.Parse()
+	initLogging(&opts)
+
+	// Print all flag values
+	flags := make(map[string]any)
+	flag.VisitAll(func(f *flag.Flag) {
+		flags[f.Name] = f.Value
+	})
+	setupLog.Info("Flags processed", "flags", flags)
+
+	// Init runtime.
+	cfg, err := ctrl.GetConfig()
+	if err != nil {
+		setupLog.Error(err, "Failed to get rest config")
+		return err
+	}
+
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{})
+	if err != nil {
+		setupLog.Error(err, "Failed to create manager", "config", cfg)
+		return err
+	}
+
+	ctx := ctrl.SetupSignalHandler()
+
+	// Setup runner.
+	serverRunner := &runserver.ExtProcServerRunner{GrpcPort: *grpcPort}
+
+	// Register health server.
+	if err := registerHealthServer(mgr, ctrl.Log.WithName("health"), *grpcHealthPort); err != nil {
+		return err
+	}
+
+	// Register ext-proc server.
+	if err := mgr.Add(serverRunner.AsRunnable(ctrl.Log.WithName("ext-proc"))); err != nil {
+		setupLog.Error(err, "Failed to register ext-proc gRPC server")
+		return err
+	}
+
+	// Start the manager. This blocks until a signal is received.
+	setupLog.Info("Manager starting")
+	if err := mgr.Start(ctx); err != nil {
+		setupLog.Error(err, "Error starting manager")
+		return err
+	}
+	setupLog.Info("Manager terminated")
+	return nil
+}
+
+// registerHealthServer adds the Health gRPC server as a Runnable to the given manager.
+func registerHealthServer(mgr manager.Manager, logger logr.Logger, port int) error {
+	srv := grpc.NewServer()
+	healthPb.RegisterHealthServer(srv, &healthServer{
+		logger: logger,
+	})
+	if err := mgr.Add(
+		runnable.NoLeaderElection(runnable.GRPCServer("health", srv, port))); err != nil {
+		setupLog.Error(err, "Failed to register health server")
+		return err
+	}
+	return nil
+}
+
+func initLogging(opts *zap.Options) {
+	logger := zap.New(zap.UseFlagOptions(opts), zap.RawZapOpts(uberzap.AddCaller()))
+	ctrl.SetLogger(logger)
+}

pkg/body-based-routing/README.md

@@ -0,0 +1,14 @@
+# Body-Based Routing
+This package provides an extension that can be deployed to write the `model`
+HTTP body parameter as a header so as to enable routing capabilities on the
+model name.
+
+As per OpenAI spec, it is standard for the model name to be included in the
+body of the HTTP request. However, most implementations do not support routing
+based on the request body. This extension helps bridge the gap for clients that
+This extension works by parsing the request body. If it finds a `model` parameter in the
+request body, it will copy the value of that parameter into a request header.
+
+This extension is intended to be paired with an `ext_proc` capable Gateway. There is not
+a standard way to represent this kind of extension in Gateway API yet, so we recommend
+referring to implementation-specific documentation for how to deploy this extension.

pkg/body-based-routing/handlers/request.go

@@ -0,0 +1,97 @@
+/*
+Copyright 2025 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 handlers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	basepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	eppb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
+)
+
+// HandleRequestBody handles request bodies.
+func (s *Server) HandleRequestBody(ctx context.Context, body *eppb.HttpBody) (*eppb.ProcessingResponse, error) {
+	logger := log.FromContext(ctx)
+
+	var data map[string]any
+	if err := json.Unmarshal(body.GetBody(), &data); err != nil {
+		return nil, err
+	}
+
+	modelVal, ok := data["model"]
+	if !ok {
+		logger.V(logutil.DEFAULT).Info("Request body does not contain model parameter")
+		return &eppb.ProcessingResponse{
+			Response: &eppb.ProcessingResponse_RequestBody{
+				RequestBody: &eppb.BodyResponse{},
+			},
+		}, nil
+	}
+
+	modelStr, ok := modelVal.(string)
+	if !ok {
+		logger.V(logutil.DEFAULT).Info("Model parameter value is not a string")
+		return &eppb.ProcessingResponse{
+			Response: &eppb.ProcessingResponse_RequestBody{
+				RequestBody: &eppb.BodyResponse{},
+			},
+		}, fmt.Errorf("the model parameter value %v is not a string", modelVal)
+	}
+
+	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: []*basepb.HeaderValueOption{
+							{
+								Header: &basepb.HeaderValue{
+									Key:   "model",
+									Value: modelStr,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}, nil
+}
+
+// HandleRequestHeaders handles request headers.
+func (s *Server) HandleRequestHeaders(headers *eppb.HttpHeaders) (*eppb.ProcessingResponse, error) {
+	return &eppb.ProcessingResponse{
+		Response: &eppb.ProcessingResponse_RequestHeaders{
+			RequestHeaders: &eppb.HeadersResponse{},
+		},
+	}, nil
+}
+
+// HandleRequestTrailers handles request trailers.
+func (s *Server) HandleRequestTrailers(trailers *eppb.HttpTrailers) (*eppb.ProcessingResponse, error) {
+	return &eppb.ProcessingResponse{
+		Response: &eppb.ProcessingResponse_RequestTrailers{
+			RequestTrailers: &eppb.TrailersResponse{},
+		},
+	}, nil
+}