[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

pkg/ext-proc/handlers/request.go

@@ -2,7 +2,6 @@ package handlers
 
 import (
 	"encoding/json"
-	"errors"
 	"fmt"
 	"strconv"
 
@@ -12,6 +11,7 @@ import (
 	klog "k8s.io/klog/v2"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/scheduling"
+	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
@@ -27,14 +27,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.V(logutil.DEFAULT).ErrorS(err, "Error unmarshaling request body")
-		return nil, fmt.Errorf("error unmarshaling request body: %v", err)
+		return nil, errutil.Error{Code: errutil.InvalidRequest, Msg: fmt.Sprintf("error unmarshaling request body: %v", err)}
 	}
 	klogV.InfoS("Request body unmarshalled", "body", rb)
 
 	// Resolve target models.
 	model, ok := rb["model"].(string)
 	if !ok {
-		return nil, errors.New("model not found in request")
+		return nil, errutil.Error{Code: errutil.InvalidRequest, Msg: "model not found in request"}
 	}
 	klogV.InfoS("Model requested", "model", model)
 	modelName := model
@@ -44,12 +44,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, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("error finding a model object in InferenceModel for input %v", model)}
 	}
 	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, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("error getting target model name for model %v", modelObj.Name)}
 		}
 	}
 	llmReq := &scheduling.LLMRequest{
@@ -67,14 +67,14 @@ func (s *Server) HandleRequestBody(reqCtx *RequestContext, req *extProcPb.Proces
 		requestBody, err = json.Marshal(rb)
 		if err != nil {
 			klog.V(logutil.DEFAULT).ErrorS(err, "Error marshaling request body")
-			return nil, fmt.Errorf("error marshaling request body: %v", err)
+			return nil, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("error marshaling request body: %v", err)}
 		}
 		klogV.InfoS("Updated request body marshalled", "body", string(requestBody))
 	}
 
 	targetPod, err := s.scheduler.Schedule(llmReq)
 	if err != nil {
-		return nil, fmt.Errorf("failed to find target pod: %w", err)
+		return nil, errutil.Error{Code: errutil.Internal, Msg: fmt.Errorf("failed to find target pod: %w", err).Error()}
 	}
 	klogV.InfoS("Target model and pod selected", "model", llmReq.ResolvedTargetModel, "pod", targetPod)
 

pkg/ext-proc/handlers/response.go

@@ -7,6 +7,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"
 	klog "k8s.io/klog/v2"
+	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
@@ -16,6 +17,44 @@ func (s *Server) HandleResponseHeaders(reqCtx *RequestContext, req *extProcPb.Pr
 	h := req.Request.(*extProcPb.ProcessingRequest_ResponseHeaders)
 	klog.V(logutil.VERBOSE).InfoS("Headers before", "headers", 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"
+	// 		}
+	// 	  ]
+	// 	}
+	// }
+	reqCtx.ResponseStatusCode = errutil.OK
+	for _, header := range h.ResponseHeaders.Headers.GetHeaders() {
+		if header.Key == "status" {
+			code := header.RawValue[0]
+			if string(code) != "200" {
+				reqCtx.ResponseStatusCode = errutil.ModelServerError
+			}
+			break
+		}
+	}
+
 	resp := &extProcPb.ProcessingResponse{
 		Response: &extProcPb.ProcessingResponse_ResponseHeaders{
 			ResponseHeaders: &extProcPb.HeadersResponse{
@@ -71,7 +110,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, errutil.Error{Code: errutil.Internal, Msg: fmt.Sprintf("unmarshaling response body: %v", err)}
 	}
 	reqCtx.Response = res
 	reqCtx.ResponseSize = len(body.ResponseBody.Body)

pkg/ext-proc/handlers/server.go

@@ -13,6 +13,7 @@ import (
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/backend"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/scheduling"
+	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
 
@@ -105,12 +106,19 @@ func (s *Server) Process(srv extProcPb.ExternalProcessor_ProcessServer) error {
 			klog.V(logutil.DEFAULT).ErrorS(nil, "Unknown Request type", "request", v)
 			return status.Error(codes.Unknown, "unknown request type")
 		}
+
+		// This indicates error from the underlying model server.
+		if reqCtx.ResponseStatusCode != errutil.OK {
+			metrics.RecordRequestErrCounter(reqCtx.Model, reqCtx.ResolvedTargetModel, reqCtx.ResponseStatusCode)
+		}
+
 		if err != nil {
 			klog.V(logutil.DEFAULT).ErrorS(err, "Failed to process request", "request", req)
-			switch status.Code(err) {
+			metrics.RecordRequestErrCounter(reqCtx.Model, reqCtx.ResolvedTargetModel, errutil.CanonicalCode(err))
+			switch errutil.CanonicalCode(err) {
 			// This code can be returned by scheduler when there is no capacity for sheddable
 			// requests.
-			case codes.ResourceExhausted:
+			case errutil.InferencePoolResourceExhausted:
 				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 errutil.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 errutil.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        string
 }

pkg/ext-proc/metrics/README.md

@@ -41,6 +41,7 @@ spec:
 | Metric name | Metric Type  | Description | Labels | Status | 
 | ------------|--------------| ----------- | ------ | ------ |
 | inference_model_request_total | Counter      | The counter of requests broken out for each model. | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt;  | ALPHA |
+| inference_model_request_error_total | Counter      | The counter of requests errors broken out for each model. | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt;  | ALPHA |
 | inference_model_request_duration_seconds | Distribution | Distribution of response latency. | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt;  | ALPHA |
 | inference_model_request_sizes | Distribution      | Distribution of request size in bytes. | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt;  | ALPHA |
 | inference_model_response_sizes | Distribution      | Distribution of response size in bytes. | `model_name`=&lt;model-name&gt; <br> `target_model_name`=&lt;target-model-name&gt;  | ALPHA |

pkg/ext-proc/metrics/metrics.go

@@ -21,12 +21,22 @@ var (
 		&compbasemetrics.CounterOpts{
 			Subsystem:      InferenceModelComponent,
 			Name:           "request_total",
-			Help:           "Counter of inference model requests broken out for each model and target model.",
+			Help:           "Counter of inference requests broken out for each model and target model.",
 			StabilityLevel: compbasemetrics.ALPHA,
 		},
 		[]string{"model_name", "target_model_name"},
 	)
 
+	requestErrCounter = compbasemetrics.NewCounterVec(
+		&compbasemetrics.CounterOpts{
+			Subsystem:      InferenceModelComponent,
+			Name:           "request_error_total",
+			Help:           "Counter of inference requests errors broken out for each model and target model.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"model_name", "target_model_name", "error_code"},
+	)
+
 	requestLatencies = compbasemetrics.NewHistogramVec(
 		&compbasemetrics.HistogramOpts{
 			Subsystem: InferenceModelComponent,
@@ -122,6 +132,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)
@@ -138,6 +149,13 @@ func RecordRequestCounter(modelName, targetModelName string) {
 	requestCounter.WithLabelValues(modelName, targetModelName).Inc()
 }
 
+// RecordRequestErrCounter records the number of error requests.
+func RecordRequestErrCounter(modelName, targetModelName string, code string) {
+	if code != "" {
+		requestErrCounter.WithLabelValues(modelName, targetModelName, code).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

@@ -7,9 +7,11 @@ import (
 
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/component-base/metrics/testutil"
+	errutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/error"
 )
 
 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"
@@ -88,6 +90,65 @@ func TestRecordRequestCounterandSizes(t *testing.T) {
 	}
 }
 
+func TestRecordRequestErrorCounter(t *testing.T) {
+	type requests struct {
+		modelName       string
+		targetModelName string
+		error           string
+	}
+	scenarios := []struct {
+		name    string
+		reqs    []requests
+		invalid bool
+	}{{
+		name: "multiple requests",
+		reqs: []requests{
+			{
+				modelName:       "m10",
+				targetModelName: "t10",
+				error:           errutil.Internal,
+			},
+			{
+				modelName:       "m10",
+				targetModelName: "t10",
+				error:           errutil.Internal,
+			},
+			{
+				modelName:       "m10",
+				targetModelName: "t11",
+				error:           errutil.ModelServerError,
+			},
+			{
+				modelName:       "m20",
+				targetModelName: "t20",
+				error:           errutil.InferencePoolResourceExhausted,
+			},
+		},
+	},
+	}
+	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 requests errors broken out for each model and target model.
+# TYPE inference_model_request_error_total counter
+inference_model_request_error_total{error_code="Internal", model_name="m10",target_model_name="t10"} 2
+inference_model_request_error_total{error_code="ModelServerError", model_name="m10",target_model_name="t11"} 1
+inference_model_request_error_total{error_code="InferencePoolResourceExhausted", model_name="m20",target_model_name="t20"} 1

pkg/ext-proc/metrics/testdata/request_total_metric

@@ -1,4 +1,4 @@
-# HELP inference_model_request_total [ALPHA] Counter of inference model requests broken out for each model and target model.
+# HELP inference_model_request_total [ALPHA] Counter of inference requests broken out for each model and target model.
 # TYPE inference_model_request_total counter
 inference_model_request_total{model_name="m10", target_model_name="t10"} 2
 inference_model_request_total{model_name="m10", target_model_name="t11"} 1

pkg/ext-proc/scheduling/scheduler.go

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

pkg/ext-proc/util/error/error.go

@@ -0,0 +1,39 @@
+package error
+
+import (
+	"fmt"
+
+	"k8s.io/klog/v2"
+
+	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
+)
+
+// Error is an error struct for errors returned by the epp server.
+type Error struct {
+	Code string
+	Msg  string
+}
+
+const (
+	OK                             = "OK"
+	Unknown                        = "Unknown"
+	InvalidRequest                 = "InvalidRequest"
+	Internal                       = "Internal"
+	ModelServerError               = "ModelServerError"
+	InferencePoolResourceExhausted = "InferencePoolResourceExhausted"
+)
+
+// Error returns a string version of the error.
+func (e Error) Error() string {
+	return fmt.Sprintf("inference gateway: %s - %s", e.Code, e.Msg)
+}
+
+// CanonicalCode returns the error's ErrorCode.
+func CanonicalCode(err error) string {
+	e, ok := err.(Error)
+	if ok {
+		return e.Code
+	}
+	klog.V(logutil.VERBOSE).Infof("failed to convert to internal error")
+	return Unknown
+}