[Metrics] Add request error metrics

This change defines some general errors, the list might grow in the future if more finer error types are needed.

Author: JeffLuoo

go.mod

@@ -29,6 +29,7 @@ require (
 	k8s.io/utils v0.0.0-20241210054802-24370beab758
 	sigs.k8s.io/controller-runtime v0.20.1
 	sigs.k8s.io/structured-merge-diff/v4 v4.5.0
+	sigs.k8s.io/yaml v1.4.0
 )
 
 require (
@@ -104,7 +105,6 @@ require (
 	github.com/spf13/cobra v1.8.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/stoewer/go-strcase v1.3.0 // indirect
-	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.53.0 // indirect
 	go.opentelemetry.io/otel v1.32.0 // indirect
@@ -140,5 +140,4 @@ require (
 	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.31.0 // indirect
 	sigs.k8s.io/controller-tools v0.14.0 // indirect
 	sigs.k8s.io/json v0.0.0-20241010143419-9aa6b5e7a4b3 // indirect
-	sigs.k8s.io/yaml v1.4.0 // indirect
 )

go.sum

@@ -215,8 +215,6 @@ github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8w
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
-github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
-github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=

pkg/ext-proc/handlers/request.go

@@ -2,14 +2,14 @@ package handlers
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strconv"
 
 	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/scheduling"
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 	klog "k8s.io/klog/v2"
 )
@@ -25,14 +25,14 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 	var rb map[string]interface{}
 	if err := json.Unmarshal(v.RequestBody.Body, &rb); err != nil {
 		klog.Errorf("Error unmarshaling request body: %v", err)
-		return nil, fmt.Errorf("error unmarshaling request body: %v", err)
+		return nil, infextprocerror.Error{Code: infextprocerror.InvalidRequest, Msg: fmt.Sprintf("error unmarshaling request body: %v", err)}
 	}
 	klog.V(logutil.VERBOSE).Infof("Request body: %v", rb)
 
 	// Resolve target models.
 	model, ok := rb["model"].(string)
 	if !ok {
-		return nil, errors.New("model not found in request")
+		return nil, infextprocerror.Error{Code: infextprocerror.InvalidRequest, Msg: "model not found in request"}
 	}
 	klog.V(logutil.VERBOSE).Infof("Model requested: %v", model)
 	modelName := model
@@ -42,12 +42,12 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 	// are able to be requested by using their distinct name.
 	modelObj := s.datastore.FetchModelData(model)
 	if modelObj == nil {
-		return nil, fmt.Errorf("error finding a model object in InferenceModel for input %v", model)
+		return nil, infextprocerror.Error{Code: infextprocerror.Internal, Msg: fmt.Sprintf("error finding a model object in InferenceModel for input %v", modelObj.Name)}
 	}
 	if len(modelObj.Spec.TargetModels) > 0 {
 		modelName = backend.RandomWeightedDraw(modelObj, 0)
 		if modelName == "" {
-			return nil, fmt.Errorf("error getting target model name for model %v", modelObj.Name)
+			return nil, infextprocerror.Error{Code: infextprocerror.Internal, Msg: fmt.Sprintf("error getting target model name for model %v", modelObj.Name)}
 		}
 	}
 	llmReq := &scheduling.LLMRequest{
@@ -65,14 +65,14 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 		requestBody, err = json.Marshal(rb)
 		if err != nil {
 			klog.Errorf("Error marshaling request body: %v", err)
-			return nil, fmt.Errorf("error marshaling request body: %v", err)
+			return nil, infextprocerror.Error{Code: infextprocerror.Internal, Msg: fmt.Sprintf("error marshaling request body: %v", err)}
 		}
 		klog.V(logutil.VERBOSE).Infof("Updated body: %v", string(requestBody))
 	}
 
 	targetPod, err := s.scheduler.Schedule(llmReq)
 	if err != nil {
-		return nil, fmt.Errorf("failed to find target pod: %w", err)
+		return nil, infextprocerror.Error{Code: infextprocerror.Internal, Msg: fmt.Errorf("failed to find target pod: %w", err).Error()}
 	}
 	klog.V(logutil.VERBOSE).Infof("Selected target model %v in target pod: %v\n", llmReq.ResolvedTargetModel, targetPod)
 

pkg/ext-proc/handlers/response.go

@@ -6,6 +6,7 @@ import (
 
 	configPb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 	klog "k8s.io/klog/v2"
 )
@@ -16,6 +17,43 @@ func (s *Server) HandleResponseHeaders(reqCtx *RequestContext, req *extProcPb.Pr
 	h := req.Request.(*extProcPb.ProcessingRequest_ResponseHeaders)
 	klog.V(logutil.VERBOSE).Infof("Headers before: %+v\n", h)
 
+	// Example header
+	// {
+	// 	"ResponseHeaders": {
+	// 	  "headers": [
+	// 		{
+	// 		  "key": ":status",
+	// 		  "raw_value": "200"
+	// 		},
+	// 		{
+	// 		  "key": "date",
+	// 		  "raw_value": "Thu, 30 Jan 2025 18:50:48 GMT"
+	// 		},
+	// 		{
+	// 		  "key": "server",
+	// 		  "raw_value": "uvicorn"
+	// 		},
+	// 		{
+	// 		  "key": "content-type",
+	// 		  "raw_value": "text/event-stream; charset=utf-8"
+	// 		},
+	// 		{
+	// 		  "key": "transfer-encoding",
+	// 		  "raw_value": "chunked"
+	// 		}
+	// 	  ]
+	// 	}
+	// }
+	for _, header := range h.ResponseHeaders.Headers.GetHeaders() {
+		if header.Key == "status" {
+			code := header.RawValue[0]
+			if string(code) != "200" {
+				reqCtx.ResponseStatusCode = infextprocerror.ModelServerError
+			}
+			break
+		}
+	}
+
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_ResponseHeaders{
 			ResponseHeaders: &extProcPb.HeadersResponse{
@@ -71,7 +109,7 @@ func (s *Server) HandleResponseBody(reqCtx *RequestContext, req *extProcPb.Proce
 
 	res := Response{}
 	if err := json.Unmarshal(body.ResponseBody.Body, &res); err != nil {
-		return nil, fmt.Errorf("unmarshaling response body: %v", err)
+		return nil, infextprocerror.Error{Code: infextprocerror.Internal, Msg: fmt.Sprintf("unmarshaling response body: %v", err)}
 	}
 	reqCtx.Response = res
 	reqCtx.ResponseSize = len(body.ResponseBody.Body)

pkg/ext-proc/handlers/server.go

@@ -12,6 +12,7 @@ import (
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/metrics"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/scheduling"
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 	klog "k8s.io/klog/v2"
 )
@@ -105,12 +106,19 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 			klog.Errorf("Unknown Request type %+v", v)
 			return status.Error(codes.Unknown, "unknown request type")
 		}
+
+		// This indicates error from the underlying model server.
+		if reqCtx.ResponseStatusCode == infextprocerror.ModelServerError {
+			metrics.RecordRequestErrCounter(reqCtx.Model, reqCtx.ResolvedTargetModel, infextprocerror.ModelServerError)
+		}
+
 		if err != nil {
 			klog.Errorf("failed to process request: %v", err)
-			switch status.Code(err) {
+			metrics.RecordRequestErrCounter(reqCtx.Model, reqCtx.ResolvedTargetModel, infextprocerror.CanonicalCode(err))
+			switch infextprocerror.CanonicalCode(err) {
 			// This code can be returned by scheduler when there is no capacity for sheddable
 			// requests.
-			case codes.ResourceExhausted:
+			case infextprocerror.ResourceExhausted:
 				resp = &extProcPb.ProcessingResponse{
 					Response: &extProcPb.ProcessingResponse_ImmediateResponse{
 						ImmediateResponse: &extProcPb.ImmediateResponse{
@@ -120,6 +128,28 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 						},
 					},
 				}
+			// This code can be returned by when EPP processes the request and run into server-side errors.
+			case infextprocerror.Internal:
+				resp = &extProcPb.ProcessingResponse{
+					Response: &extProcPb.ProcessingResponse_ImmediateResponse{
+						ImmediateResponse: &extProcPb.ImmediateResponse{
+							Status: &envoyTypePb.HttpStatus{
+								Code: envoyTypePb.StatusCode_InternalServerError,
+							},
+						},
+					},
+				}
+			// This code can be returned when users provide invalid json request.
+			case infextprocerror.InvalidRequest:
+				resp = &extProcPb.ProcessingResponse{
+					Response: &extProcPb.ProcessingResponse_ImmediateResponse{
+						ImmediateResponse: &extProcPb.ImmediateResponse{
+							Status: &envoyTypePb.HttpStatus{
+								Code: envoyTypePb.StatusCode_BadRequest,
+							},
+						},
+					},
+				}
 			default:
 				return status.Errorf(status.Code(err), "failed to handle request: %v", err)
 			}
@@ -144,4 +174,5 @@ type RequestContext struct {
 	Response                  Response
 	ResponseSize              int
 	ResponseComplete          bool
+	ResponseStatusCode        infextprocerror.ErrorCode
 }

pkg/ext-proc/metrics/metrics.go

@@ -4,6 +4,7 @@ import (
 	"sync"
 	"time"
 
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	compbasemetrics "k8s.io/component-base/metrics"
 	"k8s.io/component-base/metrics/legacyregistry"
 	klog "k8s.io/klog/v2"
@@ -24,6 +25,16 @@ var (
 		[]string{"model_name", "target_model_name"},
 	)
 
+	requestErrCounter = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
+			Subsystem:      InferenceModelComponent,
+			Name:           "request_error_total",
+			Help:           "Counter of inference model requests errors broken out for each model and target model.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"model_name", "target_model_name", "type"},
+	)
+
 	requestLatencies = compbasemetrics.NewHistogramVec(
 		&compbasemetrics.HistogramOpts{
 			Subsystem: InferenceModelComponent,
@@ -96,6 +107,7 @@ var registerMetrics sync.Once
 func Register() {
 	registerMetrics.Do(func() {
 		legacyregistry.MustRegister(requestCounter)
+		legacyregistry.MustRegister(requestErrCounter)
 		legacyregistry.MustRegister(requestLatencies)
 		legacyregistry.MustRegister(requestSizes)
 		legacyregistry.MustRegister(responseSizes)
@@ -109,6 +121,11 @@ func RecordRequestCounter(modelName, targetModelName string) {
 	requestCounter.WithLabelValues(modelName, targetModelName).Inc()
 }
 
+// RecordRequestErrCounter records the number of error requests.
+func RecordRequestErrCounter(modelName, targetModelName string, code infextprocerror.ErrorCode) {
+	requestErrCounter.WithLabelValues(modelName, targetModelName, code.String()).Inc()
+}
+
 // RecordRequestSizes records the request sizes.
 func RecordRequestSizes(modelName, targetModelName string, reqSize int) {
 	requestSizes.WithLabelValues(modelName, targetModelName).Observe(float64(reqSize))

pkg/ext-proc/metrics/metrics_test.go

@@ -5,11 +5,13 @@ import (
 	"testing"
 	"time"
 
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/component-base/metrics/testutil"
 )
 
 const RequestTotalMetric = InferenceModelComponent + "_request_total"
+const RequestErrorTotalMetric = InferenceModelComponent + "_request_error_total"
 const RequestLatenciesMetric = InferenceModelComponent + "_request_duration_seconds"
 const RequestSizesMetric = InferenceModelComponent + "_request_sizes"
 const ResponseSizesMetric = InferenceModelComponent + "_response_sizes"
@@ -86,6 +88,65 @@ func TestRecordRequestCounterandSizes(t *testing.T) {
 	}
 }
 
+func TestRecordRequestErrorCounter(t *testing.T) {
+	type requests struct {
+		modelName       string
+		targetModelName string
+		error           infextprocerror.ErrorCode
+	}
+	scenarios := []struct {
+		name    string
+		reqs    []requests
+		invalid bool
+	}{{
+		name: "multiple requests",
+		reqs: []requests{
+			{
+				modelName:       "m10",
+				targetModelName: "t10",
+				error:           infextprocerror.Internal,
+			},
+			{
+				modelName:       "m10",
+				targetModelName: "t10",
+				error:           infextprocerror.Internal,
+			},
+			{
+				modelName:       "m10",
+				targetModelName: "t11",
+				error:           infextprocerror.ModelServerError,
+			},
+			{
+				modelName:       "m20",
+				targetModelName: "t20",
+				error:           infextprocerror.ResourceExhausted,
+			},
+		},
+	},
+	}
+	Register()
+	for _, scenario := range scenarios {
+		t.Run(scenario.name, func(t *testing.T) {
+			for _, req := range scenario.reqs {
+				RecordRequestErrCounter(req.modelName, req.targetModelName, req.error)
+			}
+
+			wantRequestErrorCounter, err := os.Open("testdata/request_error_total_metric")
+			defer func() {
+				if err := wantRequestErrorCounter.Close(); err != nil {
+					t.Error(err)
+				}
+			}()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := testutil.GatherAndCompare(legacyregistry.DefaultGatherer, wantRequestErrorCounter, RequestErrorTotalMetric); err != nil {
+				t.Error(err)
+			}
+		})
+	}
+}
+
 func TestRecordRequestLatencies(t *testing.T) {
 	timeBaseline := time.Now()
 	type requests struct {

pkg/ext-proc/metrics/testdata/request_error_total_metric

@@ -0,0 +1,5 @@
+# HELP inference_model_request_error_total [ALPHA] Counter of inference model requests errors broken out for each model and target model.
+# TYPE inference_model_request_error_total counter
+inference_model_request_error_total{model_name="m10",target_model_name="t10",type="Internal"} 2
+inference_model_request_error_total{model_name="m10",target_model_name="t11",type="ModelServer"} 1
+inference_model_request_error_total{model_name="m20",target_model_name="t20",type="ResourceExhausted"} 1

pkg/ext-proc/scheduling/scheduler.go

@@ -5,9 +5,8 @@ import (
 	"fmt"
 	"math/rand"
 
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 	"inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
+	infextprocerror "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "inference.networking.x-k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 	klog "k8s.io/klog/v2"
 )
@@ -84,8 +83,8 @@ var (
 			name: "drop request",
 			filter: func(req *LLMRequest, pods []*backend.PodMetrics) ([]*backend.PodMetrics, error) {
 				klog.Infof("Dropping request %v", req)
-				return []*backend.PodMetrics{}, status.Errorf(
-					codes.ResourceExhausted, "dropping request due to limited backend resources")
+				return []*backend.PodMetrics{}, infextprocerror.Error{
+					Code: infextprocerror.ResourceExhausted, Msg: "dropping request due to limited backend resources"}
 			},
 		},
 	}