o11y: Add request error counters (#132)

tao12345666333 · web-flow · commit 88aa203966d7 · 2025-09-14T11:26:11.000-04:00
Signed-off-by: Jintao Zhang &lt;zhangjintao9020@gmail.com&gt;
diff --git a/src/semantic-router/pkg/extproc/error_metrics_test.go b/src/semantic-router/pkg/extproc/error_metrics_test.go
@@ -0,0 +1,120 @@
+package extproc
+
+import (
+	"testing"
+
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	ext_proc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+)
+
+// getCounterValue returns the sum of a counter across metrics matching the given labels
+func getCounterValue(metricName string, want map[string]string) float64 {
+	var sum float64
+	mfs, _ := prometheus.DefaultGatherer.Gather()
+	for _, fam := range mfs {
+		if fam.GetName() != metricName || fam.GetType() != dto.MetricType_COUNTER {
+			continue
+		}
+		for _, m := range fam.GetMetric() {
+			labels := m.GetLabel()
+			match := true
+			for k, v := range want {
+				found := false
+				for _, l := range labels {
+					if l.GetName() == k && l.GetValue() == v {
+						found = true
+						break
+					}
+				}
+				if !found {
+					match = false
+					break
+				}
+			}
+			if match && m.GetCounter() != nil {
+				sum += m.GetCounter().GetValue()
+			}
+		}
+	}
+	return sum
+}
+
+func TestRequestParseErrorIncrementsErrorCounter(t *testing.T) {
+	r := &OpenAIRouter{}
+	r.InitializeForTesting()
+
+	ctx := &RequestContext{}
+	// Invalid JSON triggers parse error
+	badBody := []byte("not-json")
+	v := &ext_proc.ProcessingRequest_RequestBody{
+		RequestBody: &ext_proc.HttpBody{Body: badBody},
+	}
+
+	before := getCounterValue("llm_request_errors_total", map[string]string{"reason": "parse_error", "model": "unknown"})
+
+	// Use test helper wrapper to access unexported method
+	_, _ = r.HandleRequestBody(v, ctx)
+
+	after := getCounterValue("llm_request_errors_total", map[string]string{"reason": "parse_error", "model": "unknown"})
+	if !(after > before) {
+		t.Fatalf("expected llm_request_errors_total(parse_error,unknown) to increase: before=%v after=%v", before, after)
+	}
+}
+
+func TestResponseParseErrorIncrementsErrorCounter(t *testing.T) {
+	r := &OpenAIRouter{}
+	r.InitializeForTesting()
+
+	ctx := &RequestContext{RequestModel: "model-a"}
+	// Invalid JSON triggers parse error in response body handler
+	badJSON := []byte("{invalid}")
+	v := &ext_proc.ProcessingRequest_ResponseBody{
+		ResponseBody: &ext_proc.HttpBody{Body: badJSON},
+	}
+
+	before := getCounterValue("llm_request_errors_total", map[string]string{"reason": "parse_error", "model": "model-a"})
+
+	_, _ = r.HandleResponseBody(v, ctx)
+
+	after := getCounterValue("llm_request_errors_total", map[string]string{"reason": "parse_error", "model": "model-a"})
+	if !(after > before) {
+		t.Fatalf("expected llm_request_errors_total(parse_error,model-a) to increase: before=%v after=%v", before, after)
+	}
+}
+
+func TestUpstreamStatusIncrements4xx5xxCounters(t *testing.T) {
+	r := &OpenAIRouter{}
+	r.InitializeForTesting()
+
+	ctx := &RequestContext{RequestModel: "m"}
+
+	// 503 -> upstream_5xx
+	hdrs5xx := &ext_proc.ProcessingRequest_ResponseHeaders{
+		ResponseHeaders: &ext_proc.HttpHeaders{
+			Headers: &core.HeaderMap{Headers: []*core.HeaderValue{{Key: ":status", Value: "503"}}},
+		},
+	}
+
+	before5xx := getCounterValue("llm_request_errors_total", map[string]string{"reason": "upstream_5xx", "model": "m"})
+	_, _ = r.HandleResponseHeaders(hdrs5xx, ctx)
+	after5xx := getCounterValue("llm_request_errors_total", map[string]string{"reason": "upstream_5xx", "model": "m"})
+	if !(after5xx > before5xx) {
+		t.Fatalf("expected upstream_5xx to increase for model m: before=%v after=%v", before5xx, after5xx)
+	}
+
+	// 404 -> upstream_4xx
+	hdrs4xx := &ext_proc.ProcessingRequest_ResponseHeaders{
+		ResponseHeaders: &ext_proc.HttpHeaders{
+			Headers: &core.HeaderMap{Headers: []*core.HeaderValue{{Key: ":status", Value: "404"}}},
+		},
+	}
+
+	before4xx := getCounterValue("llm_request_errors_total", map[string]string{"reason": "upstream_4xx", "model": "m"})
+	_, _ = r.HandleResponseHeaders(hdrs4xx, ctx)
+	after4xx := getCounterValue("llm_request_errors_total", map[string]string{"reason": "upstream_4xx", "model": "m"})
+	if !(after4xx > before4xx) {
+		t.Fatalf("expected upstream_4xx to increase for model m: before=%v after=%v", before4xx, after4xx)
+	}
+}
diff --git a/src/semantic-router/pkg/extproc/processor.go b/src/semantic-router/pkg/extproc/processor.go
@@ -7,6 +7,7 @@ import (
 	"log"
 
 	ext_proc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
+	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/metrics"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -32,15 +33,26 @@ func (r *OpenAIRouter) Process(stream ext_proc.ExternalProcessor_ProcessServer)
 			// Handle gRPC status-based cancellations/timeouts
 			if s, ok := status.FromError(err); ok {
 				switch s.Code() {
-				case codes.Canceled, codes.DeadlineExceeded:
+				case codes.Canceled:
 					log.Println("Stream canceled gracefully")
+					metrics.RecordRequestError(ctx.RequestModel, "cancellation")
+					return nil
+				case codes.DeadlineExceeded:
+					log.Println("Stream deadline exceeded")
+					metrics.RecordRequestError(ctx.RequestModel, "timeout")
 					return nil
 				}
 			}
 
 			// Handle context cancellation from the server-side context
-			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			if errors.Is(err, context.Canceled) {
 				log.Println("Stream canceled gracefully")
+				metrics.RecordRequestError(ctx.RequestModel, "cancellation")
+				return nil
+			}
+			if errors.Is(err, context.DeadlineExceeded) {
+				log.Println("Stream deadline exceeded")
+				metrics.RecordRequestError(ctx.RequestModel, "timeout")
 				return nil
 			}
 
diff --git a/src/semantic-router/pkg/extproc/request_handler.go b/src/semantic-router/pkg/extproc/request_handler.go
@@ -164,15 +164,23 @@ func (r *OpenAIRouter) handleRequestBody(v *ext_proc.ProcessingRequest_RequestBo
 	openAIRequest, err := parseOpenAIRequest(ctx.OriginalRequestBody)
 	if err != nil {
 		log.Printf("Error parsing OpenAI request: %v", err)
+		// Attempt to determine model for labeling (may be unknown here)
+		metrics.RecordRequestError(ctx.RequestModel, "parse_error")
+		// Count this request as well, with unknown model if necessary
+		metrics.RecordModelRequest(ctx.RequestModel)
 		return nil, status.Errorf(codes.InvalidArgument, "invalid request body: %v", err)
 	}
 
 	// Store the original model
 	originalModel := string(openAIRequest.Model)
 	log.Printf("Original model: %s", originalModel)
 
-	// Record the initial request to this model
+	// Record the initial request to this model (count all requests)
 	metrics.RecordModelRequest(originalModel)
+	// Also set the model on context early so error metrics can label it
+	if ctx.RequestModel == "" {
+		ctx.RequestModel = originalModel
+	}
 
 	// Get content from messages
 	userContent, nonUserMessages := extractUserAndNonUserContent(openAIRequest)
@@ -202,6 +210,7 @@ func (r *OpenAIRouter) performSecurityChecks(ctx *RequestContext, userContent st
 		if err != nil {
 			log.Printf("Error performing jailbreak analysis: %v", err)
 			// Continue processing despite jailbreak analysis error
+			metrics.RecordRequestError(ctx.RequestModel, "classification_failed")
 		} else if hasJailbreak {
 			// Find the first jailbreak detection for response
 			var jailbreakType string
@@ -224,6 +233,8 @@ func (r *OpenAIRouter) performSecurityChecks(ctx *RequestContext, userContent st
 				"confidence":     confidence,
 				"request_id":     ctx.RequestID,
 			})
+			// Count this as a blocked request
+			metrics.RecordRequestError(ctx.RequestModel, "jailbreak_block")
 			jailbreakResponse := http.CreateJailbreakViolationResponse(jailbreakType, confidence)
 			return jailbreakResponse, true
 		} else {
@@ -347,6 +358,7 @@ func (r *OpenAIRouter) handleModelRouting(openAIRequest *openai.ChatCompletionNe
 										"model":       matchedModel,
 										"denied_pii":  defaultDeniedPII,
 									})
+									metrics.RecordRequestError(matchedModel, "pii_policy_denied")
 									piiResponse := http.CreatePIIViolationResponse(matchedModel, defaultDeniedPII)
 									return piiResponse, nil
 								}
@@ -359,6 +371,7 @@ func (r *OpenAIRouter) handleModelRouting(openAIRequest *openai.ChatCompletionNe
 								"model":       matchedModel,
 								"denied_pii":  deniedPII,
 							})
+							metrics.RecordRequestError(matchedModel, "pii_policy_denied")
 							piiResponse := http.CreatePIIViolationResponse(matchedModel, deniedPII)
 							return piiResponse, nil
 						}
@@ -397,12 +410,14 @@ func (r *OpenAIRouter) handleModelRouting(openAIRequest *openai.ChatCompletionNe
 				modifiedBody, err := serializeOpenAIRequest(openAIRequest)
 				if err != nil {
 					log.Printf("Error serializing modified request: %v", err)
+					metrics.RecordRequestError(actualModel, "serialization_error")
 					return nil, status.Errorf(codes.Internal, "error serializing modified request: %v", err)
 				}
 
 				modifiedBody, err = r.setReasoningModeToRequestBody(modifiedBody, useReasoning, categoryName)
 				if err != nil {
 					log.Printf("Error setting reasoning mode %v to request: %v", useReasoning, err)
+					metrics.RecordRequestError(actualModel, "serialization_error")
 					return nil, status.Errorf(codes.Internal, "error setting reasoning mode: %v", err)
 				}
 
@@ -489,6 +504,7 @@ func (r *OpenAIRouter) handleModelRouting(openAIRequest *openai.ChatCompletionNe
 				"model":       originalModel,
 				"denied_pii":  deniedPII,
 			})
+			metrics.RecordRequestError(originalModel, "pii_policy_denied")
 			piiResponse := http.CreatePIIViolationResponse(originalModel, deniedPII)
 			return piiResponse, nil
 		}
@@ -595,6 +611,7 @@ func (r *OpenAIRouter) handleToolSelection(openAIRequest *openai.ChatCompletionN
 			openAIRequest.Tools = nil
 			return r.updateRequestWithTools(openAIRequest, response, ctx)
 		}
+		metrics.RecordRequestError(getModelFromCtx(ctx), "classification_failed")
 		return err
 	}
 
@@ -613,6 +630,7 @@ func (r *OpenAIRouter) handleToolSelection(openAIRequest *openai.ChatCompletionN
 			// Convert the tool to OpenAI SDK format
 			toolBytes, err := json.Marshal(tool)
 			if err != nil {
+				metrics.RecordRequestError(getModelFromCtx(ctx), "serialization_error")
 				return err
 			}
 			var sdkTool openai.ChatCompletionToolParam
diff --git a/src/semantic-router/pkg/extproc/response_handler.go b/src/semantic-router/pkg/extproc/response_handler.go
@@ -3,8 +3,10 @@ package extproc
 import (
 	"encoding/json"
 	"log"
+	"strconv"
 	"time"
 
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	ext_proc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 
 	"github.com/openai/openai-go"
@@ -13,7 +15,18 @@ import (
 )
 
 // handleResponseHeaders processes the response headers
-func (r *OpenAIRouter) handleResponseHeaders(_ *ext_proc.ProcessingRequest_ResponseHeaders, ctx *RequestContext) (*ext_proc.ProcessingResponse, error) {
+func (r *OpenAIRouter) handleResponseHeaders(v *ext_proc.ProcessingRequest_ResponseHeaders, ctx *RequestContext) (*ext_proc.ProcessingResponse, error) {
+	// Detect upstream HTTP status and record non-2xx as errors
+	if v != nil && v.ResponseHeaders != nil && v.ResponseHeaders.Headers != nil {
+		if statusCode := getStatusFromHeaders(v.ResponseHeaders.Headers); statusCode != 0 {
+			if statusCode >= 500 {
+				metrics.RecordRequestError(getModelFromCtx(ctx), "upstream_5xx")
+			} else if statusCode >= 400 {
+				metrics.RecordRequestError(getModelFromCtx(ctx), "upstream_4xx")
+			}
+		}
+	}
+
 	// Best-effort TTFT measurement: record on first response headers if we have a start time and model
 	if ctx != nil && !ctx.TTFTRecorded && !ctx.ProcessingStartTime.IsZero() && ctx.RequestModel != "" {
 		ttft := time.Since(ctx.ProcessingStartTime).Seconds()
@@ -38,6 +51,35 @@ func (r *OpenAIRouter) handleResponseHeaders(_ *ext_proc.ProcessingRequest_Respo
 	return response, nil
 }
 
+// getStatusFromHeaders extracts :status pseudo-header value as integer
+func getStatusFromHeaders(headerMap *core.HeaderMap) int {
+	if headerMap == nil {
+		return 0
+	}
+	for _, hv := range headerMap.Headers {
+		if hv.Key == ":status" {
+			if hv.Value != "" {
+				if code, err := strconv.Atoi(hv.Value); err == nil {
+					return code
+				}
+			}
+			if len(hv.RawValue) > 0 {
+				if code, err := strconv.Atoi(string(hv.RawValue)); err == nil {
+					return code
+				}
+			}
+		}
+	}
+	return 0
+}
+
+func getModelFromCtx(ctx *RequestContext) string {
+	if ctx == nil || ctx.RequestModel == "" {
+		return "unknown"
+	}
+	return ctx.RequestModel
+}
+
 // handleResponseBody processes the response body
 func (r *OpenAIRouter) handleResponseBody(v *ext_proc.ProcessingRequest_ResponseBody, ctx *RequestContext) (*ext_proc.ProcessingResponse, error) {
 	completionLatency := time.Since(ctx.StartTime)
@@ -49,6 +91,7 @@ func (r *OpenAIRouter) handleResponseBody(v *ext_proc.ProcessingRequest_Response
 	var parsed openai.ChatCompletion
 	if err := json.Unmarshal(responseBody, &parsed); err != nil {
 		log.Printf("Error parsing tokens from response: %v", err)
+		metrics.RecordRequestError(ctx.RequestModel, "parse_error")
 	}
 	promptTokens := int(parsed.Usage.PromptTokens)
 	completionTokens := int(parsed.Usage.CompletionTokens)
diff --git a/src/semantic-router/pkg/metrics/metrics.go b/src/semantic-router/pkg/metrics/metrics.go
@@ -102,6 +102,15 @@ var (
 		[]string{"model"},
 	)
 
+	// RequestErrorsTotal tracks request errors categorized by reason
+	RequestErrorsTotal = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "llm_request_errors_total",
+			Help: "The total number of request errors categorized by reason (e.g., timeout, upstream_5xx, pii_policy_denied, jailbreak_block, parse_error, serialization_error, cancellation)",
+		},
+		[]string{"model", "reason"},
+	)
+
 	// ModelCost tracks the total cost attributed to each model by currency
 	ModelCost = promauto.NewCounterVec(
 		prometheus.CounterOpts{
@@ -353,9 +362,32 @@ var (
 
 // RecordModelRequest increments the counter for requests to a specific model
 func RecordModelRequest(model string) {
+	if model == "" {
+		model = "unknown"
+	}
 	ModelRequests.WithLabelValues(model).Inc()
 }
 
+// RecordRequestError increments request error counters labeled by model and normalized reason
+func RecordRequestError(model, reason string) {
+	if model == "" {
+		model = "unknown"
+	}
+	if reason == "" {
+		reason = "unknown"
+	}
+	// Normalize a few common variants to canonical reasons
+	switch reason {
+	case "deadline_exceeded":
+		reason = "timeout"
+	case "upstream_500", "upstream_502", "upstream_503", "upstream_504":
+		reason = "upstream_5xx"
+	case "upstream_400", "upstream_401", "upstream_403", "upstream_404", "upstream_429":
+		reason = "upstream_4xx"
+	}
+	RequestErrorsTotal.WithLabelValues(model, reason).Inc()
+}
+
 // RecordModelRouting records that a request was routed from one model to another
 func RecordModelRouting(sourceModel, targetModel string) {
 	if sourceModel != targetModel {
diff --git a/src/semantic-router/pkg/utils/http/response.go b/src/semantic-router/pkg/utils/http/response.go
@@ -6,10 +6,11 @@ import (
 	"log"
 	"time"
 
+	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/metrics"
+
 	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	ext_proc "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	typev3 "github.com/envoyproxy/go-control-plane/envoy/type/v3"
-	"github.com/vllm-project/semantic-router/src/semantic-router/pkg/metrics"
 )
 
 // CreatePIIViolationResponse creates an HTTP response for PII policy violations
