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

Issue:
use cpu based Qwen model in e2e tests
it would be good to have e2e tests without the need to use GPUs

Hints from issue comments:
@nirrozenbaum: /assign@danehans: xref https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/259

Solution patch:
diff --git a/Makefile b/Makefile
index cf053749..c3c24892 100644
--- a/Makefile
+++ b/Makefile
@@ -32,6 +32,8 @@ IMAGE_REGISTRY ?= us-central1-docker.pkg.dev/k8s-staging-images/gateway-api-infe
 IMAGE_NAME := epp
 IMAGE_REPO ?= $(IMAGE_REGISTRY)/$(IMAGE_NAME)
 IMAGE_TAG ?= $(IMAGE_REPO):$(GIT_TAG)
+ROOT_DIR:=$(shell dirname $(realpath $(firstword $(MAKEFILE_LIST))))
+E2E_MANIFEST_PATH ?= config/manifests/vllm/gpu-deployment.yaml
 
 SYNCER_IMAGE_NAME := lora-syncer
 SYNCER_IMAGE_REPO ?= $(IMAGE_REGISTRY)/$(SYNCER_IMAGE_NAME)
@@ -126,8 +128,8 @@ test-integration: manifests generate fmt vet envtest ## Run tests.
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path)" go test ./test/integration/epp/... -race -coverprofile cover.out
 
 .PHONY: test-e2e
-test-e2e: ## Run end-to-end tests against an existing Kubernetes cluster with at least 3 available GPUs.
-	go test ./test/e2e/epp -v -ginkgo.v
+test-e2e: ## Run end-to-end tests against an existing Kubernetes cluster. When using default configuration, the tests need at least 3 available GPUs.
+	MANIFEST_PATH=$(ROOT_DIR)/$(E2E_MANIFEST_PATH) go test ./test/e2e/epp/ -v -ginkgo.v
 
 .PHONY: lint
 lint: golangci-lint ## Run golangci-lint linter
@@ -236,7 +238,7 @@ bbr-image-build: ## Build the image using Docker Buildx.
 		--build-arg BUILDER_IMAGE=$(BUILDER_IMAGE) \
 		$(PUSH) \
 		$(LOAD) \
-		.
+		$(BBR_IMAGE_BUILD_EXTRA_OPTS) ./
 
 .PHONY: bbr-image-push
 bbr-image-push: PUSH=--push ## Build the image and push it to $IMAGE_REPO.
diff --git a/cmd/body-based-routing/main.go b/cmd/body-based-routing/main.go
index 66534389..13f841b6 100644
--- a/cmd/body-based-routing/main.go
+++ b/cmd/body-based-routing/main.go
@@ -18,18 +18,26 @@ package main
 
 import (
 	"flag"
+	"net"
+	"net/http"
 	"os"
+	"strconv"
 
 	"github.com/go-logr/logr"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	uberzap "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc"
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
+	"k8s.io/client-go/rest"
+	"k8s.io/component-base/metrics/legacyregistry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/body-based-routing/server"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
@@ -42,6 +50,8 @@ var (
 		"grpcHealthPort",
 		9005,
 		"The port used for gRPC liveness and readiness probes")
+	metricsPort = flag.Int(
+		"metricsPort", 9090, "The metrics port")
 	logVerbosity = flag.Int("v", logging.DEFAULT, "number for the log level verbosity")
 
 	setupLog = ctrl.Log.WithName("setup")
@@ -95,6 +105,11 @@ func run() error {
 		return err
 	}
 
+	// Register metrics handler.
+	if err := registerMetricsHandler(mgr, *metricsPort, cfg); err != nil {
+		return err
+	}
+
 	// Start the manager. This blocks until a signal is received.
 	setupLog.Info("Manager starting")
 	if err := mgr.Start(ctx); err != nil {
@@ -135,3 +150,58 @@ func initLogging(opts *zap.Options) {
 	logger := zap.New(zap.UseFlagOptions(opts), zap.RawZapOpts(uberzap.AddCaller()))
 	ctrl.SetLogger(logger)
 }
+
+const metricsEndpoint = "/metrics"
+
+// registerMetricsHandler adds the metrics HTTP handler as a Runnable to the given manager.
+func registerMetricsHandler(mgr manager.Manager, port int, cfg *rest.Config) error {
+	metrics.Register()
+
+	// Init HTTP server.
+	h, err := metricsHandlerWithAuthenticationAndAuthorization(cfg)
+	if err != nil {
+		return err
+	}
+
+	mux := http.NewServeMux()
+	mux.Handle(metricsEndpoint, h)
+
+	srv := &http.Server{
+		Addr:    net.JoinHostPort("", strconv.Itoa(port)),
+		Handler: mux,
+	}
+
+	if err := mgr.Add(&manager.Server{
+		Name:   "metrics",
+		Server: srv,
+	}); err != nil {
+		setupLog.Error(err, "Failed to register metrics HTTP handler")
+		return err
+	}
+	return nil
+}
+
+func metricsHandlerWithAuthenticationAndAuthorization(cfg *rest.Config) (http.Handler, error) {
+	h := promhttp.HandlerFor(
+		legacyregistry.DefaultGatherer,
+		promhttp.HandlerOpts{},
+	)
+	httpClient, err := rest.HTTPClientFor(cfg)
+	if err != nil {
+		setupLog.Error(err, "Failed to create http client for metrics auth")
+		return nil, err
+	}
+
+	filter, err := filters.WithAuthenticationAndAuthorization(cfg, httpClient)
+	if err != nil {
+		setupLog.Error(err, "Failed to create metrics filter for auth")
+		return nil, err
+	}
+	metricsLogger := ctrl.Log.WithName("metrics").WithValues("path", metricsEndpoint)
+	metricsAuthHandler, err := filter(metricsLogger, h)
+	if err != nil {
+		setupLog.Error(err, "Failed to create metrics auth handler")
+		return nil, err
+	}
+	return metricsAuthHandler, nil
+}
diff --git a/docs/proposals/004-endpoint-picker-protocol/README.md b/docs/proposals/004-endpoint-picker-protocol/README.md
index 3657a10e..5280e05c 100644
--- a/docs/proposals/004-endpoint-picker-protocol/README.md
+++ b/docs/proposals/004-endpoint-picker-protocol/README.md
@@ -9,27 +9,57 @@ This doc defines the protocol between the EPP and the proxy (e.g, Envoy).
 The EPP MUST implement the Envoy
 [external processing service](https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/ext_proc/v3/external_processor) protocol.
 
+## Endpoint Subset
+For each HTTP request, the proxy CAN communicate the subset of endpoints the EPP MUST pick from by setting an unstructured entry in the [filter metadata](https://github.com/envoyproxy/go-control-plane/blob/63a55395d7a39a8d43dcc7acc3d05e4cae7eb7a2/envoy/config/core/v3/base.pb.go#L819) field of the ext-proc request. The metadata entry for the subset list MUST be wrapped with an outer key (which represents the metadata namespace) with a default of `envoy.lb.subset_hint`.
+
+```go
+filterMetadata: {
+  "envoy.lb.subset_hint" {
+     "x-gateway-destination-endpoint-subset": [<ip:port>, <ip:port>, ...]
+  }
+}
+```
+
+If the key `x-gateway-destination-endpoint-subset` is set, the EPP MUST only select endpoints from the specified list. If none of the endpoints in the list is eligible or the list is empty, then the EPP MUST return a [ImmediateResponse](https://github.com/envoyproxy/envoy/blob/f2023ef77bdb4abaf9feef963c9a0c291f55568f/api/envoy/service/ext_proc/v3/external_processor.proto#L195) with 503 (Service Unavailable) HTTP status code. If the EPP does not select from the list, then this leads to unpredictable behavior.
+
+If the key `x-gateway-destination-endpoint-subset` is not set, then the EPP MUST select from the set defined by the `InferencePool` selector.
+
+## Destination Endpoint
 For each HTTP request, the EPP MUST communicate to the proxy the picked model server endpoint via:
 
 1. Setting the `x-gateway-destination-endpoint` HTTP header to the selected endpoint in <ip:port> format.
 
 2. Set an unstructured entry in the [dynamic_metadata](https://github.com/envoyproxy/go-control-plane/blob/c19bf63a811c90bf9e02f8e0dc1dcef94931ebb4/envoy/service/ext_proc/v3/external_processor.pb.go#L320) field of the ext-proc response. The metadata entry for the picked endpoint MUST be wrapped with an outer key (which represents the metadata namespace) with a default of `envoy.lb`.
 
-The final metadata necessary would look like: 
+The primary endpoint MUST be set using the key `x-gateway-destination-endpoint` as follows:
 ```go
 dynamicMetadata: {
   "envoy.lb": {
-    "x-gateway-destination-endpoint": <ip:port>"  
+    "x-gateway-destination-endpoint": <ip:port>
   }
 }
 ```
 
-Note:
-- If the EPP did not communicate the server endpoint via these two methods, it MUST return an error.
+Constraints:
+- If the EPP did not communicate the server endpoint via these two methods, it MUST return an error as follows:
+  -  [ImmediateResponse](https://github.com/envoyproxy/envoy/blob/f2023ef77bdb4abaf9feef963c9a0c291f55568f/api/envoy/service/ext_proc/v3/external_processor.proto#L195) with 503 (Serivce Unavailable) HTTP status code if there are no ready endpoints.
+  -  [ImmediateResponse](https://github.com/envoyproxy/envoy/blob/f2023ef77bdb4abaf9feef963c9a0c291f55568f/api/envoy/service/ext_proc/v3/external_processor.proto#L195) with 429 (Too Many Requests) HTTP status code if the request should be dropped (e.g., a Sheddable request, and the servers under heavy load).
 - The EPP MUST not set two different values in the header and the inner response metadata value. 
+- Setting different value leads to unpredictable behavior because proxies aren't guaranteed to support both paths, and so this protocol does not define what takes precedence.
+
+### Destination endpoint fallback
+A single fallback endpoint CAN be set using the key `x-gateway-destination-endpoint-fallback` in the same metadata namespace as one used for `x-gateway-destination-endpoint` as follows:
 
-## Why envoy.lb namespace as a default? 
-The `envoy.lb` namesapce is a predefined namespace used for subsetting. One common way to use the selected endpoint returned from the server, is [envoy subsets](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/upstream/load_balancing/subsets) where host metadata for subset load balancing must be placed under `envoy.lb`.
+```go
+dynamicMetadata: {
+  "envoy.lb" {
+     "x-gateway-destination-endpoint-fallback": <ip:port>
+  }
+}
+```
 
-Setting different value leads to unpredictable behavior because proxies aren't guaranteed to support both paths, and so this protocol does not define what takes precedence.
+### Why envoy.lb namespace as a default? 
+The `envoy.lb` namespace is a predefined namespace. One common way to use the selected endpoint returned from the server, is [envoy subsets](https://www.envoyproxy.io/docs/envoy/latest/intro/arch_overview/upstream/load_balancing/subsets)  where host metadata for subset load balancing must be placed under `envoy.lb`. Note that this is not related to the subsetting feature discussed above, this is an enovy implementation detail.
 
+## Matching An InferenceModel
+The model name of a request MUST match the `Spec.ModelName` parameter of one of the `InferenceModels` referencing the `InferencePool` managed by the EPP. Otherwise, the EPP MUST return a 404 status code.
diff --git a/pkg/body-based-routing/handlers/request.go b/pkg/body-based-routing/handlers/request.go
index 3c5037a9..6596e191 100644
--- a/pkg/body-based-routing/handlers/request.go
+++ b/pkg/body-based-routing/handlers/request.go
@@ -24,6 +24,7 @@ import (
 	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"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/body-based-routing/metrics"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
@@ -38,6 +39,7 @@ func (s *Server) HandleRequestBody(ctx context.Context, body *eppb.HttpBody) (*e
 
 	modelVal, ok := data["model"]
 	if !ok {
+		metrics.RecordModelNotInBodyCounter()
 		logger.V(logutil.DEFAULT).Info("Request body does not contain model parameter")
 		return &eppb.ProcessingResponse{
 			Response: &eppb.ProcessingResponse_RequestBody{
@@ -48,6 +50,7 @@ func (s *Server) HandleRequestBody(ctx context.Context, body *eppb.HttpBody) (*e
 
 	modelStr, ok := modelVal.(string)
 	if !ok {
+		metrics.RecordModelNotParsedCounter()
 		logger.V(logutil.DEFAULT).Info("Model parameter value is not a string")
 		return &eppb.ProcessingResponse{
 			Response: &eppb.ProcessingResponse_RequestBody{
@@ -56,6 +59,7 @@ func (s *Server) HandleRequestBody(ctx context.Context, body *eppb.HttpBody) (*e
 		}, fmt.Errorf("the model parameter value %v is not a string", modelVal)
 	}
 
+	metrics.RecordSuccessCounter()
 	return &eppb.ProcessingResponse{
 		Response: &eppb.ProcessingResponse_RequestBody{
 			RequestBody: &eppb.BodyResponse{
diff --git a/pkg/body-based-routing/metrics/metrics.go b/pkg/body-based-routing/metrics/metrics.go
new file mode 100644
index 00000000..fc3538fb
--- /dev/null
+++ b/pkg/body-based-routing/metrics/metrics.go
@@ -0,0 +1,103 @@
+/*
+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 metrics
+
+import (
+	"sync"
+
+	compbasemetrics "k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+const component = "bbr"
+
+var (
+	successCounter = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
+			Subsystem:      component,
+			Name:           "success_total",
+			Help:           "Count of successes pulling model name from body and injecting it in the request headers.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+	modelNotInBodyCounter = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
+			Subsystem:      component,
+			Name:           "model_not_in_body_total",
+			Help:           "Count of times the model was not present in the request body.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+	modelNotParsedCounter = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
+			Subsystem:      component,
+			Name:           "model_not_parsed_total",
+			Help:           "Count of times the model was in the request body but we could not parse it.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{},
+	)
+
+	// TODO: Uncomment and use this metrics once the core server implementation has handling to skip body parsing if header exists.
+	/*
+		modelAlreadyPresentInHeaderCounter = compbasemetrics.NewCounterVec(
+			&compbasemetrics.CounterOpts{
+				Subsystem:      component,
+				Name:           "model_already_present_in_header_total",
+				Help:           "Count of times the model was already present in request headers.",
+				StabilityLevel: compbasemetrics.ALPHA,
+			},
+			[]string{},
+		)
+	*/
+)
+
+var registerMetrics sync.Once
+
+// Register all metrics.
+func Register() {
+	registerMetrics.Do(func() {
+		legacyregistry.MustRegister(successCounter)
+		legacyregistry.MustRegister(modelNotInBodyCounter)
+		legacyregistry.MustRegister(modelNotParsedCounter)
+		// legacyregistry.MustRegister(modelAlreadyPresentInHeaderCounter)
+	})
+}
+
+// RecordSuccessCounter records the number of successful requests to inject the model name into request headers.
+func RecordSuccessCounter() {
+	successCounter.WithLabelValues().Inc()
+}
+
+// RecordModelNotInBodyCounter records the number of times the model was not found in the request body.
+func RecordModelNotInBodyCounter() {
+	modelNotInBodyCounter.WithLabelValues().Inc()
+}
+
+// RecordModelNotParsedCounter records the number of times the model was found in the body but it could not be parsed.
+func RecordModelNotParsedCounter() {
+	modelNotParsedCounter.WithLabelValues().Inc()
+}
+
+/*
+// RecordModelAlreadyInHeaderCounter records the number of times the model was already found in the request headers.
+func RecordModelAlreadyInHeaderCounter() {
+	modelAlreadyPresentInHeaderCounter.WithLabelValues().Inc()
+}
+*/
diff --git a/pkg/epp/handlers/response.go b/pkg/epp/handlers/response.go
index f9396acf..44ea6d6a 100644
--- a/pkg/epp/handlers/response.go
+++ b/pkg/epp/handlers/response.go
@@ -20,9 +20,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 
 	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"github.com/go-logr/logr"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
@@ -67,11 +69,25 @@ func (s *Server) HandleResponseHeaders(
 	// 	}
 	// }
 	for _, header := range h.ResponseHeaders.Headers.GetHeaders() {
+		var statusFound, typeFound bool
 		if header.Key == "status" {
 			code := header.RawValue[0]
 			if string(code) != "200" {
 				reqCtx.ResponseStatusCode = errutil.ModelServerError
+				statusFound = true
 			}
+		}
+		if header.Key == "content-type" {
+			contentType := header.RawValue
+			if strings.Contains(string(contentType), "text/event-stream") {
+				reqCtx.Streaming = true
+			} else {
+				reqCtx.Streaming = false
+			}
+			typeFound = true
+		}
+
+		if statusFound && typeFound {
 			break
 		}
 	}
@@ -132,22 +148,19 @@ func (s *Server) HandleResponseBody(
 ) (*extProcPb.ProcessingResponse, error) {
 	logger := log.FromContext(ctx)
 	loggerVerbose := logger.V(logutil.VERBOSE)
-	loggerVerbose.Info("Processing HandleResponseBody")
 	body := req.Request.(*extProcPb.ProcessingRequest_ResponseBody)
 
-	res := Response{}
-	if err := json.Unmarshal(body.ResponseBody.Body, &res); err != nil {
-		return nil, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("unmarshaling response body: %v", err)}
+	if reqCtx.Streaming {
+		logger.V(logutil.DEBUG).Info("Processing HandleResponseBody")
+		if err := s.HandleStreaming(ctx, reqCtx, body, loggerVerbose); err != nil {
+			return nil, err
+		}
+	} else {
+		loggerVerbose.Info("Processing HandleResponseBody")
+		if err := s.HandleNonStreaming(ctx, reqCtx, body, loggerVerbose); err != nil {
+			return nil, err
+		}
 	}
-	reqCtx.Response = res
-	reqCtx.ResponseSize = len(body.ResponseBody.Body)
-	// ResponseComplete is to indicate the response is complete. In non-streaming
-	// case, it will be set to be true once the response is processed; in
-	// streaming case, it will be set to be true once the last chunk is processed.
-	// TODO(https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/178)
-	// will add the processing for streaming case.
-	reqCtx.ResponseComplete = true
-	loggerVerbose.Info("Response generated", "response", res)
 
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_ResponseBody{
@@ -159,6 +172,76 @@ func (s *Server) HandleResponseBody(
 	return resp, nil
 }
 
+func (s *Server) HandleNonStreaming(
+	ctx context.Context,
+	reqCtx *RequestContext,
+	body *extProcPb.ProcessingRequest_ResponseBody,
+	loggerVerbose logr.Logger,
+) error {
+	loggerVerbose.Info("Processing HandleResponseBody")
+
+	res := Response{}
+	if err := json.Unmarshal(body.ResponseBody.Body, &res); err != nil {
+		return errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("unmarshaling response body: %v", err)}
+	}
+	reqCtx.Response = res
+	reqCtx.ResponseSize = len(body.ResponseBody.Body)
+	reqCtx.ResponseComplete = true
+	loggerVerbose.Info("Response generated", "response", res)
+	return nil
+}
+
+func (s *Server) HandleStreaming(
+	ctx context.Context,
+	reqCtx *RequestContext,
+	body *extProcPb.ProcessingRequest_ResponseBody,
+	loggerVerbose logr.Logger,
+) error {
+	respPrefix := "data: "
+	responseText := string(body.ResponseBody.Body)
+	// Example message if "stream_options": {"include_usage": "true"} is included in the request:
+	// data: {"id":"...","object":"text_completion","created":1739400043,"model":"tweet-summary-0","choices":[],
+	// "usage":{"prompt_tokens":7,"total_tokens":17,"completion_tokens":10}}
+	//
+	// data: [DONE]
+	//
+	// Noticed that vLLM returns two entries in one response.
+	// We need to strip the `data:` prefix and next Data: [DONE] from the message to fetch response data.
+	//
+	// If include_usage is not included in the request, `data: [DONE]` is returned separately, which
+	// indicates end of streaming.
+	if strings.Contains(responseText, "data: [DONE]") {
+		response := Response{}
+
+		lines := strings.Split(responseText, "\n")
+		for _, line := range lines {
+			if !strings.HasPrefix(line, respPrefix) {
+				continue
+			}
+			content := strings.TrimPrefix(line, respPrefix)
+			if content == "[DONE]" {
+				continue
+			}
+
+			byteSlice := []byte(content)
+			if err := json.Unmarshal(byteSlice, &response); err != nil {
+				loggerVerbose.Error(err, "unmarshaling response body")
+				continue
+			}
+		}
+		reqCtx.Response = response
+	}
+
+	if body.ResponseBody.EndOfStream {
+		loggerVerbose.Info("Streaming is completed")
+		reqCtx.ResponseComplete = true
+	} else {
+		reqCtx.ResponseSize += len(body.ResponseBody.Body)
+	}
+
+	return nil
+}
+
 type Response struct {
 	Usage Usage `json:"usage"`
 }
diff --git a/pkg/epp/handlers/server.go b/pkg/epp/handlers/server.go
index be882fc7..4f45ae82 100644
--- a/pkg/epp/handlers/server.go
+++ b/pkg/epp/handlers/server.go
@@ -117,15 +117,25 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 			resp, err = s.HandleResponseHeaders(ctx, reqCtx, req)
 			loggerVerbose.Info("Request context after HandleResponseHeaders", "context", reqCtx)
 		case *extProcPb.ProcessingRequest_ResponseBody:
-			resp, err = s.HandleResponseBody(ctx, reqCtx, req)
-			if err == nil && reqCtx.ResponseComplete {
+			// Don't send a 500 on a response error. Just let the message passthrough and log our error for debugging purposes.
+			// We assume the body is valid JSON, err messages are not guaranteed to be json, and so capturing and sending a 500 obfuscates the response message.
+			// using the standard 'err' var will send an immediate error response back to the caller.
+			var responseErr error
+			resp, responseErr = s.HandleResponseBody(ctx, reqCtx, req)
+			if responseErr != nil {
+				logger.V(logutil.DEFAULT).Error(responseErr, "Failed to process response body", "request", req)
+			} else if reqCtx.ResponseComplete {
 				reqCtx.ResponseCompleteTimestamp = time.Now()
 				metrics.RecordRequestLatencies(ctx, reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp)
 				metrics.RecordResponseSizes(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.ResponseSize)
 				metrics.RecordInputTokens(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.Response.Usage.PromptTokens)
 				metrics.RecordOutputTokens(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.Response.Usage.CompletionTokens)
 			}
-			loggerVerbose.Info("Request context after HandleResponseBody", "context", reqCtx)
+			if reqCtx.Streaming {
+				logger.V(logutil.DEBUG).Info("Request context after HandleResponseBody", "context", reqCtx)
+			} else {
+				loggerVerbose.Info("Request context after HandleResponseBody", "context", reqCtx)
+			}
 		default:
 			logger.V(logutil.DEFAULT).Error(nil, "Unknown Request type", "request", v)
 			return status.Error(codes.Unknown, "unknown request type")
@@ -139,7 +149,11 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 			}
 		}
 
-		loggerVerbose.Info("Response generated", "response", resp)
+		if !reqCtx.Streaming {
+			loggerVerbose.Info("Response generated", "response", resp)
+		} else {
+			logger.V(logutil.DEBUG).Info("Response generated", "response", resp)
+		}
 		if err := srv.Send(resp); err != nil {
 			logger.V(logutil.DEFAULT).Error(err, "Send failed")
 			return status.Errorf(codes.Unknown, "failed to send response back to Envoy: %v", err)
@@ -214,4 +228,5 @@ type RequestContext struct {
 	ResponseSize              int
 	ResponseComplete          bool
 	ResponseStatusCode        string
+	Streaming                 bool
 }
diff --git a/pkg/epp/handlers/streamingserver.go b/pkg/epp/handlers/streamingserver.go
index 2aaca7f3..adcd83ed 100644
--- a/pkg/epp/handlers/streamingserver.go
+++ b/pkg/epp/handlers/streamingserver.go
@@ -192,15 +192,21 @@ func (s *StreamingServer) Process(srv extProcPb.ExternalProcessor_ProcessServer)
 
 				// Message is buffered, we can read and decode.
 				if v.ResponseBody.EndOfStream {
-					err = decoder.Decode(&responseBody)
-					if err != nil {
-						logger.V(logutil.DEFAULT).Error(err, "Error unmarshaling request body")
+					// Don't send a 500 on a response error. Just let the message passthrough and log our error for debugging purposes.
+					// We assume the body is valid JSON, err messages are not guaranteed to be json, and so capturing and sending a 500 obfuscates the response message.
+					// using the standard 'err' var will send an immediate error response back to the caller.
+					var responseErr error
+					responseErr = decoder.Decode(&responseBody)
+					if responseErr != nil {
+						logger.V(logutil.DEFAULT).Error(responseErr, "Error unmarshaling request body")
 					}
 					// Body stream complete. Close the reader pipe.
 					reader.Close()
 
-					reqCtx, err = s.HandleResponseBody(ctx, reqCtx, responseBody)
-					if err == nil && reqCtx.ResponseComplete {
+					reqCtx, responseErr = s.HandleResponseBody(ctx, reqCtx, responseBody)
+					if responseErr != nil {
+						logger.V(logutil.DEFAULT).Error(responseErr, "Failed to process response body", "request", req)
+					} else if reqCtx.ResponseComplete {
 						reqCtx.ResponseCompleteTimestamp = time.Now()
 						metrics.RecordRequestLatencies(ctx, reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.RequestReceivedTimestamp, reqCtx.ResponseCompleteTimestamp)
 						metrics.RecordResponseSizes(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.ResponseSize)
diff --git a/site-src/guides/metrics.md b/site-src/guides/metrics.md
index f793734d..a904145d 100644
--- a/site-src/guides/metrics.md
+++ b/site-src/guides/metrics.md
@@ -4,14 +4,7 @@ This guide describes the current state of exposed metrics and how to scrape them
 
 ## Requirements
 
-Response metrics are only supported in non-streaming mode, with the follow up [issue](https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/178) to address streaming mode.
-
-Currently there are two options:
-- If requests don't use response streaming, then you can enable `Buffered` mode for response in `EnvoyExtensionPolicy`, this will buffer the response body at the proxy and forward it to the endpoint picker, which allows the endpoint picker to report response metrics.
-
-- If requests use response streaming, then it is not recommended to enable `Buffered` mode, the response body processing mode should be left empty in the `EnvoyExtensionPolicy` (default). In this case response bodies will not be forwarded to the endpoint picker, and therefore response metrics will not be reported.
-
-
+To have response metrics, set the body mode to `Buffered` or `Streamed`:
 ```
 apiVersion: gateway.envoyproxy.io/v1alpha1
 kind: EnvoyExtensionPolicy
@@ -32,6 +25,19 @@ spec:
           body: Buffered
 ```
 
+If you want to include usage metrics for vLLM model server streaming request, send the request with `include_usage`:
+
+```
+curl -i ${IP}:${PORT}/v1/completions -H 'Content-Type: application/json' -d '{
+"model": "tweet-summary",
+"prompt": "whats your fav movie?",
+"max_tokens": 10,
+"temperature": 0,
+"stream": true,
+"stream_options": {"include_usage": "true"}
+}'
+```
+
 ## Exposed metrics
 
 | **Metric name**                              | **Metric Type**  | <div style="width:200px">**Description**</div>  | <div style="width:250px">**Labels**</div>                                          | **Status**  | 



Test patch:
diff --git a/pkg/body-based-routing/handlers/request_test.go b/pkg/body-based-routing/handlers/request_test.go
index 9bdac521..76f64e0c 100644
--- a/pkg/body-based-routing/handlers/request_test.go
+++ b/pkg/body-based-routing/handlers/request_test.go
@@ -18,12 +18,16 @@ package handlers
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	basepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/protobuf/testing/protocmp"
+	"k8s.io/component-base/metrics/legacyregistry"
+	metricsutils "k8s.io/component-base/metrics/testutil"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/body-based-routing/metrics"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
@@ -48,6 +52,7 @@ const (
 )
 
 func TestHandleRequestBody(t *testing.T) {
+	metrics.Register()
 	ctx := logutil.NewTestLoggerIntoContext(context.Background())
 
 	tests := []struct {
@@ -125,4 +130,20 @@ func TestHandleRequestBody(t *testing.T) {
 			}
 		})
 	}
+
+	wantMetrics := `
+	# HELP bbr_model_not_in_body_total [ALPHA] Count of times the model was not present in the request body.
+	# TYPE bbr_model_not_in_body_total counter
+	bbr_model_not_in_body_total{} 1
+	# HELP bbr_model_not_parsed_total [ALPHA] Count of times the model was in the request body but we could not parse it.
+	# TYPE bbr_model_not_parsed_total counter
+	bbr_model_not_parsed_total{} 1
+	# HELP bbr_success_total [ALPHA] Count of successes pulling model name from body and injecting it in the request headers.
+	# TYPE bbr_success_total counter
+	bbr_success_total{} 1
+	`
+
+	if err := metricsutils.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(wantMetrics), "inference_model_request_total"); err != nil {
+		t.Error(err)
+	}
 }
diff --git a/pkg/epp/handlers/response_test.go b/pkg/epp/handlers/response_test.go
index 01f02d09..8b6f16a7 100644
--- a/pkg/epp/handlers/response_test.go
+++ b/pkg/epp/handlers/response_test.go
@@ -49,6 +49,13 @@ const (
 		}
 	}
 	`
+
+	streamingBodyWithoutUsage = `data: {"id":"cmpl-41764c93-f9d2-4f31-be08-3ba04fa25394","object":"text_completion","created":1740002445,"model":"tweet-summary-0","choices":[],"usage":null}
+	`
+
+	streamingBodyWithUsage = `data: {"id":"cmpl-41764c93-f9d2-4f31-be08-3ba04fa25394","object":"text_completion","created":1740002445,"model":"tweet-summary-0","choices":[],"usage":{"prompt_tokens":7,"total_tokens":17,"completion_tokens":10}}
+data: [DONE]
+	`
 )
 
 func TestHandleResponseBody(t *testing.T) {
@@ -57,6 +64,7 @@ func TestHandleResponseBody(t *testing.T) {
 	tests := []struct {
 		name    string
 		req     *extProcPb.ProcessingRequest_ResponseBody
+		reqCtx  *RequestContext
 		want    Response
 		wantErr bool
 	}{
@@ -84,12 +92,47 @@ func TestHandleResponseBody(t *testing.T) {
 			},
 			wantErr: true,
 		},
+		{
+			name: "streaming request without usage",
+			req: &extProcPb.ProcessingRequest_ResponseBody{
+				ResponseBody: &extProcPb.HttpBody{
+					Body: []byte(streamingBodyWithoutUsage),
+				},
+			},
+			reqCtx: &RequestContext{
+				Streaming: true,
+			},
+			wantErr: false,
+			// In the middle of streaming response, so request context response is not set yet.
+		},
+		{
+			name: "streaming request with usage",
+			req: &extProcPb.ProcessingRequest_ResponseBody{
+				ResponseBody: &extProcPb.HttpBody{
+					Body: []byte(streamingBodyWithUsage),
+				},
+			},
+			reqCtx: &RequestContext{
+				Streaming: true,
+			},
+			wantErr: false,
+			want: Response{
+				Usage: Usage{
+					PromptTokens:     7,
+					TotalTokens:      17,
+					CompletionTokens: 10,
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			server := &Server{}
-			reqCtx := &RequestContext{}
+			reqCtx := test.reqCtx
+			if reqCtx == nil {
+				reqCtx = &RequestContext{}
+			}
 			_, err := server.HandleResponseBody(ctx, reqCtx, &extProcPb.ProcessingRequest{Request: test.req})
 			if err != nil {
 				if !test.wantErr {
diff --git a/test/e2e/epp/e2e_suite_test.go b/test/e2e/epp/e2e_suite_test.go
index 01da152f..bc7dc87a 100644
--- a/test/e2e/epp/e2e_suite_test.go
+++ b/test/e2e/epp/e2e_suite_test.go
@@ -68,8 +68,6 @@ const (
 	inferExtName = "inference-gateway-ext-proc"
 	// clientManifest is the manifest for the client test resources.
 	clientManifest = "../../testdata/client.yaml"
-	// modelServerManifest is the manifest for the model server test resources.
-	modelServerManifest = "../../../config/manifests/vllm/gpu-deployment.yaml"
 	// modelServerSecretManifest is the manifest for the model server secret resource.
 	modelServerSecretManifest = "../../testdata/model-secret.yaml"
 	// inferPoolManifest is the manifest for the inference pool CRD.
@@ -80,6 +78,8 @@ const (
 	inferExtManifest = "../../../config/manifests/ext_proc.yaml"
 	// envoyManifest is the manifest for the envoy proxy test resources.
 	envoyManifest = "../../testdata/envoy.yaml"
+	// modelServerManifestFilepathEnvVar is the env var that holds absolute path to the manifest for the model server test resource.
+	modelServerManifestFilepathEnvVar = "MANIFEST_PATH"
 )
 
 var (
@@ -107,6 +107,7 @@ var _ = ginkgo.BeforeSuite(func() {
 })
 
 func setupInfra() {
+	modelServerManifest := readModelServerManifestPath()
 	crds := map[string]string{
 		"inferencepools.inference.networking.x-k8s.io":  inferPoolManifest,
 		"inferencemodels.inference.networking.x-k8s.io": inferModelManifest,
@@ -145,6 +146,7 @@ func setupSuite() {
 
 	kubeCli, err = kubernetes.NewForConfig(cfg)
 	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+	gomega.Expect(kubeCli).NotTo(gomega.BeNil())
 }
 
 func cleanupResources() {
@@ -181,6 +183,14 @@ func namespaceExists(k8sClient client.Client, ns string) {
 	}, existsTimeout, interval)
 }
 
+// readModelServerManifestPath reads from env var the absolute filepath to model server deployment for testing.
+func readModelServerManifestPath() string {
+	ginkgo.By(fmt.Sprintf("Ensuring %s environment variable is set", modelServerManifestFilepathEnvVar))
+	modelServerManifestFilepath := os.Getenv(modelServerManifestFilepathEnvVar)
+	gomega.Expect(modelServerManifestFilepath).NotTo(gomega.BeEmpty(), modelServerManifestFilepathEnvVar+" is not set")
+	return modelServerManifestFilepath
+}
+
 // createCRDs creates the Inference Extension CRDs used for testing.
 func createCRDs(k8sClient client.Client, crds map[string]string) {
 	for name, path := range crds {
@@ -215,6 +225,29 @@ func createClient(k8sClient client.Client, filePath string) {
 
 // createModelServer creates the model server resources used for testing from the given filePaths.
 func createModelServer(k8sClient client.Client, secretPath, deployPath string) {
+	ginkgo.By("Ensuring the model server manifest points to an existing file")
+	modelServerManifestArray := readYaml(deployPath)
+	gomega.Expect(modelServerManifestArray).NotTo(gomega.BeEmpty())
+	modelServerManifestYaml := modelServerManifestArray[0]
+	if strings.Contains(modelServerManifestYaml, "hf-token") {
+		createHfSecret(k8sClient, secretPath)
+	}
+
+	ginkgo.By("Creating model server resources from manifest: " + deployPath)
+	createObjsFromYaml(k8sClient, modelServerManifestArray)
+
+	// Wait for the deployment to exist.
+	deploy := &appsv1.Deployment{}
+	testutils.EventuallyExists(ctx, func() error {
+		return k8sClient.Get(ctx, types.NamespacedName{Namespace: nsName, Name: modelServerName}, deploy)
+	}, existsTimeout, interval)
+
+	// Wait for the deployment to be available.
+	testutils.DeploymentAvailable(ctx, k8sClient, deploy, modelReadyTimeout, interval)
+}
+
+// createHfSecret read HF_TOKEN from env var and creates a secret that contains the access token.
+func createHfSecret(k8sClient client.Client, secretPath string) {
 	ginkgo.By("Ensuring the HF_TOKEN environment variable is set")
 	token := os.Getenv("HF_TOKEN")
 	gomega.Expect(token).NotTo(gomega.BeEmpty(), "HF_TOKEN is not set")
@@ -226,25 +259,13 @@ func createModelServer(k8sClient client.Client, secretPath, deployPath string) {
 		outManifests = append(outManifests, strings.Replace(m, "$HF_TOKEN", token, 1))
 	}
 
-	ginkgo.By("Creating model server secret resource from manifest: " + deployPath)
+	ginkgo.By("Creating model server secret resource")
 	createObjsFromYaml(k8sClient, outManifests)
 
 	// Wait for the secret to exist before proceeding with test.
 	testutils.EventuallyExists(ctx, func() error {
 		return k8sClient.Get(ctx, types.NamespacedName{Namespace: nsName, Name: "hf-token"}, &corev1.Secret{})
 	}, existsTimeout, interval)
-
-	ginkgo.By("Creating model server resources from manifest: " + deployPath)
-	applyYAMLFile(k8sClient, deployPath)
-
-	// Wait for the deployment to exist.
-	deploy := &appsv1.Deployment{}
-	testutils.EventuallyExists(ctx, func() error {
-		return k8sClient.Get(ctx, types.NamespacedName{Namespace: nsName, Name: modelServerName}, deploy)
-	}, existsTimeout, interval)
-
-	// Wait for the deployment to be available.
-	testutils.DeploymentAvailable(ctx, k8sClient, deploy, modelReadyTimeout, interval)
 }
 
 // createEnvoy creates the envoy proxy resources used for testing from the given filePath.
diff --git a/test/e2e/epp/e2e_test.go b/test/e2e/epp/e2e_test.go
index f5cfaf24..09c8835a 100644
--- a/test/e2e/epp/e2e_test.go
+++ b/test/e2e/epp/e2e_test.go
@@ -49,11 +49,7 @@ var _ = ginkgo.Describe("InferencePool", func() {
 
 			ginkgo.By("Ensuring the InferenceModel resource exists in the namespace")
 			gomega.Eventually(func() error {
-				err := cli.Get(ctx, types.NamespacedName{Namespace: infModel.Namespace, Name: infModel.Name}, infModel)
-				if err != nil {
-					return err
-				}
-				return nil
+				return cli.Get(ctx, types.NamespacedName{Namespace: infModel.Namespace, Name: infModel.Name}, infModel)
 			}, existsTimeout, interval).Should(gomega.Succeed())
 
 			ginkgo.By("Verifying connectivity through the inference extension")