From f72bb8a4c30f5e6aa3d9f89619516635a6c7ff45 Mon Sep 17 00:00:00 2001 From: Sangkyoon Nam Date: Fri, 2 Oct 2026 11:37:51 +0900 Subject: [PATCH] feat(mcp): record metrics for backend notification streams Signed-off-by: Sangkyoon Nam --- internal/mcpproxy/session.go | 110 ++++++ internal/mcpproxy/session_test.go | 530 +++++++++++++++++++++++++++ internal/metrics/mcp_metrics.go | 133 +++++++ internal/metrics/mcp_metrics_test.go | 105 ++++++ internal/metrics/register.go | 9 + 5 files changed, 887 insertions(+) diff --git a/internal/mcpproxy/session.go b/internal/mcpproxy/session.go index c2d32297b3..d077c16602 100644 --- a/internal/mcpproxy/session.go +++ b/internal/mcpproxy/session.go @@ -29,6 +29,7 @@ import ( "github.com/envoyproxy/ai-gateway/internal/filterapi" "github.com/envoyproxy/ai-gateway/internal/json" + "github.com/envoyproxy/ai-gateway/internal/metrics" "github.com/envoyproxy/ai-gateway/internal/tracing/tracingapi" ) @@ -404,9 +405,16 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b } s.reqCtx.l.Debug("sending MCP request", args...) } + // Only the legacy GET notification stream records stream metrics; other methods leave this nil. + var streamRec *notificationStreamRecorder + if httpMethod == http.MethodGet { + streamRec = ¬ificationStreamRecorder{metrics: s.reqCtx.metrics.WithBackend(backend.Name)} + streamRec.attempt(ctx) + } startAt := time.Now() httpResp, err := s.reqCtx.client.Do(req) if err != nil { + streamRec.failedToOpen(ctx, err) if errors.Is(err, context.Canceled) { return nil } @@ -425,6 +433,12 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b case "gzip": gr, gzErr := gzip.NewReader(httpResp.Body) if gzErr != nil { + // A known non-200 status (e.g. 204 with an empty gzip body) keeps its status outcome. + if outcome, ok := notificationStreamStatusOutcome(httpResp.StatusCode); ok { + streamRec.outcome(ctx, outcome) + } else { + streamRec.failedToOpen(ctx, gzErr) + } return fmt.Errorf("failed to create gzip decompressor: %w", gzErr) } defer gr.Close() @@ -436,9 +450,12 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b switch httpResp.StatusCode { case http.StatusNoContent, http.StatusMethodNotAllowed, http.StatusAccepted: // No notifications. + streamRec.outcome(ctx, metrics.MCPNotificationStreamOutcomeUnsupported) return nil case http.StatusOK: default: + outcome, _ := notificationStreamStatusOutcome(httpResp.StatusCode) + streamRec.outcome(ctx, outcome) body, _ := io.ReadAll(bodyReader) return fmt.Errorf("MCP GET request failed with status code %d, body=%s", httpResp.StatusCode, string(body)) } @@ -448,10 +465,13 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b var respBody []byte respBody, err = io.ReadAll(bodyReader) if err != nil { + streamRec.failedToOpen(ctx, err) return fmt.Errorf("failed to read MCP response body: %w", err) } msg, ok := tryDecodeJSONRPCMessage(respBody) if ok { + // A single JSON-RPC message answers the GET without opening a stream. + streamRec.outcome(ctx, metrics.MCPNotificationStreamOutcomeUnsupported) eventChan <- &backendEvent{ sseEvent: &sseEvent{ backend: backend.Name, @@ -472,6 +492,7 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b // io.Copy won't flush until the end, which doesn't happen for streaming responses. // So we need to read the body in chunks and flush after each chunk. parser := newSSEEventParser(bodyReader, backend.Name) + streamRec.opened(ctx) for { var event *sseEvent event, err = parser.next() @@ -491,8 +512,10 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b // or the upstream closes the connection. Either way, the error is not recoverable or not worth // the logging. errors.Is(err, io.ErrUnexpectedEOF) { + streamRec.ended(ctx, err) break } + streamRec.ended(ctx, err) _ = httpResp.Body.Close() return fmt.Errorf("failed to read MCP GET response body: %w", err) } @@ -500,6 +523,93 @@ func (s *session) sendRequestPerBackend(ctx context.Context, eventChan chan<- *b return nil } +// notificationStreamRecorder records the lifecycle of a legacy GET notification stream to a single backend. +// All methods are no-ops on a nil receiver so that non-GET requests record nothing. +type notificationStreamRecorder struct { + metrics metrics.MCPMetrics + openedAt time.Time +} + +func (r *notificationStreamRecorder) attempt(ctx context.Context) { + if r == nil { + return + } + r.metrics.RecordNotificationStreamOpenAttempt(ctx) +} + +func (r *notificationStreamRecorder) outcome(ctx context.Context, outcome metrics.MCPNotificationStreamOutcome) { + if r == nil { + return + } + r.metrics.RecordNotificationStreamOpenOutcome(ctx, outcome) +} + +// failedToOpen records the outcome of an error that happened before the stream opened. +func (r *notificationStreamRecorder) failedToOpen(ctx context.Context, err error) { + if errors.Is(err, context.Canceled) || ctx.Err() != nil { + r.outcome(ctx, metrics.MCPNotificationStreamOutcomeCancelled) + return + } + r.outcome(ctx, metrics.MCPNotificationStreamOutcomeTransportError) +} + +func (r *notificationStreamRecorder) opened(ctx context.Context) { + if r == nil { + return + } + r.openedAt = time.Now() + r.metrics.RecordNotificationStreamOpenOutcome(ctx, metrics.MCPNotificationStreamOutcomeOpened) +} + +// ended records the end of an opened stream given the error that terminated the read loop. +func (r *notificationStreamRecorder) ended(ctx context.Context, err error) { + if r == nil { + return + } + reason := metrics.MCPNotificationStreamEndReasonError + switch { + case errors.Is(err, context.Canceled) || ctx.Err() != nil: + reason = metrics.MCPNotificationStreamEndReasonCancelled + case isCleanEOF(err): + reason = metrics.MCPNotificationStreamEndReasonEOF + } + r.metrics.RecordNotificationStreamEnd(ctx, r.openedAt, reason) +} + +// isCleanEOF reports whether err is io.EOF, possibly wrapped, without any other error joined to it. +// The SSE parser joins io.EOF with the parse error of a malformed trailing event, which is not a normal close. +func isCleanEOF(err error) bool { + if joined, ok := err.(interface{ Unwrap() []error }); ok { + for _, e := range joined.Unwrap() { + if !isCleanEOF(e) { + return false + } + } + return true + } + if wrapped := errors.Unwrap(err); wrapped != nil { + return isCleanEOF(wrapped) + } + return err == io.EOF //nolint:errorlint // wrappers are unwrapped above. +} + +// notificationStreamStatusOutcome maps the HTTP status code of a notification stream request to its outcome. +// It returns false for 200 OK, whose outcome depends on the response body. +func notificationStreamStatusOutcome(statusCode int) (metrics.MCPNotificationStreamOutcome, bool) { + switch { + case statusCode == http.StatusOK: + return "", false + case statusCode == http.StatusNoContent, statusCode == http.StatusMethodNotAllowed, statusCode == http.StatusAccepted: + return metrics.MCPNotificationStreamOutcomeUnsupported, true + case statusCode >= 400 && statusCode < 500: + return metrics.MCPNotificationStreamOutcomeHTTP4xx, true + case statusCode >= 500 && statusCode < 600: + return metrics.MCPNotificationStreamOutcomeHTTP5xx, true + default: + return metrics.MCPNotificationStreamOutcomeHTTPOther, true + } +} + type ( // clientToGatewaySessionID is the ID of a session in MCP in the client<>Gateway direction. // diff --git a/internal/mcpproxy/session_test.go b/internal/mcpproxy/session_test.go index bb91269955..b2448e3c79 100644 --- a/internal/mcpproxy/session_test.go +++ b/internal/mcpproxy/session_test.go @@ -11,11 +11,13 @@ import ( "context" "encoding/base64" "errors" + "fmt" "io" "log/slog" "net/http" "net/http/httptest" "strings" + "sync" "sync/atomic" "testing" "time" @@ -24,10 +26,14 @@ import ( "github.com/modelcontextprotocol/go-sdk/jsonrpc" mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp" "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/envoyproxy/ai-gateway/internal/filterapi" "github.com/envoyproxy/ai-gateway/internal/internalapi" "github.com/envoyproxy/ai-gateway/internal/metrics" + "github.com/envoyproxy/ai-gateway/internal/testing/testotel" ) // stubMetrics implements metrics.MCPMetrics with no-ops. @@ -49,6 +55,13 @@ func (stubMetrics) RecordServerCapabilities(context.Context, *mcpsdk.ServerCapab } func (stubMetrics) RecordProgress(context.Context, mcpsdk.Params) {} +func (stubMetrics) RecordNotificationStreamOpenAttempt(context.Context) {} +func (stubMetrics) RecordNotificationStreamOpenOutcome(context.Context, metrics.MCPNotificationStreamOutcome) { +} + +func (stubMetrics) RecordNotificationStreamEnd(context.Context, time.Time, metrics.MCPNotificationStreamEndReason) { +} + func TestEncodeCapabilityFlags(t *testing.T) { t.Parallel() tests := []struct { @@ -1037,3 +1050,520 @@ func TestGetHeartbeatInterval(t *testing.T) { }) } } + +const ( + streamAttemptsMetric = "mcp.notification_stream.open.attempts" + streamOutcomesMetric = "mcp.notification_stream.open.outcomes" + streamActiveMetric = "mcp.notification_stream.active" + streamDurationMetric = "mcp.notification_stream.duration" +) + +func backendAttrs(backend string) attribute.Set { + return attribute.NewSet(attribute.String("mcp.backend", backend)) +} + +func outcomeAttrs(backend string, outcome metrics.MCPNotificationStreamOutcome) attribute.Set { + return attribute.NewSet( + attribute.String("mcp.backend", backend), + attribute.String("mcp.notification_stream.outcome", string(outcome)), + ) +} + +func endReasonAttrs(backend string, reason metrics.MCPNotificationStreamEndReason) attribute.Set { + return attribute.NewSet( + attribute.String("mcp.backend", backend), + attribute.String("mcp.notification_stream.end_reason", string(reason)), + ) +} + +// lookupStreamMetric returns the value of a sum metric, or the count of a histogram, for the given +// attributes. Unlike the testotel helpers it does not fail when the data point does not exist yet, +// so it can be used for polling and for asserting absence. +func lookupStreamMetric(t *testing.T, mr *sdkmetric.ManualReader, name string, attrs attribute.Set) (float64, bool) { + t.Helper() + var data metricdata.ResourceMetrics + require.NoError(t, mr.Collect(t.Context(), &data)) + for _, sm := range data.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != name { + continue + } + switch d := m.Data.(type) { + case metricdata.Sum[float64]: + for _, dp := range d.DataPoints { + if dp.Attributes.Equals(&attrs) { + return dp.Value, true + } + } + case metricdata.Histogram[float64]: + for _, dp := range d.DataPoints { + if dp.Attributes.Equals(&attrs) { + return float64(dp.Count), true + } + } + } + } + } + return 0, false +} + +// totalStreamOutcomes returns the number of open outcomes recorded for the backend across all outcome values. +func totalStreamOutcomes(t *testing.T, mr *sdkmetric.ManualReader, backend string) float64 { + t.Helper() + var data metricdata.ResourceMetrics + require.NoError(t, mr.Collect(t.Context(), &data)) + var total float64 + for _, sm := range data.ScopeMetrics { + for _, m := range sm.Metrics { + if m.Name != streamOutcomesMetric { + continue + } + for _, dp := range m.Data.(metricdata.Sum[float64]).DataPoints { + if v, ok := dp.Attributes.Value("mcp.backend"); ok && v.AsString() == backend { + total += dp.Value + } + } + } + } + return total +} + +func requireStreamMetricEventually(t *testing.T, mr *sdkmetric.ManualReader, name string, attrs attribute.Set, want float64) { + t.Helper() + require.Eventually(t, func() bool { + v, ok := lookupStreamMetric(t, mr, name, attrs) + return ok && v == want + }, 5*time.Second, 10*time.Millisecond, "%s %v never reached %v", name, attrs.ToSlice(), want) +} + +// newNotificationStreamTestProxy returns a proxy that records metrics into the returned reader +// and sends backend requests to a test server running handler. The request context seen by +// handler is also cancelled at test cleanup, so handlers that hold the response until the +// request is cancelled cannot block server shutdown even if the client never cancels. +func newNotificationStreamTestProxy(t *testing.T, handler http.HandlerFunc) (*mcpRequestContext, *sdkmetric.ManualReader) { + release := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithCancel(r.Context()) + defer cancel() + go func() { + select { + case <-release: + cancel() + case <-ctx.Done(): + } + }() + handler(w, r.WithContext(ctx)) + })) + t.Cleanup(func() { + close(release) + srv.CloseClientConnections() + srv.Close() + }) + mr := sdkmetric.NewManualReader() + proxy := newTestMCPProxyWithOTEL(mr, noopTracer) + proxy.backendListenerAddr = srv.URL + return proxy, mr +} + +// openNotificationStream opens the legacy GET notification stream to backend1. +func openNotificationStream(ctx context.Context, proxy *mcpRequestContext) error { + s := &session{reqCtx: proxy} + return s.sendRequestPerBackend(ctx, make(chan *backendEvent, 10), "route1", filterapi.MCPBackend{Name: "backend1"}, + &compositeSessionEntry{sessionID: "sess1"}, http.MethodGet, nil, nil) +} + +// runInBackground runs fn with a cancellable context. The returned wait function returns fn's +// result, failing the test if fn does not return within 5s. Only an actual completion is cached, +// so cleanup still cancels and then joins the worker with a bounded wait after an earlier timeout. +func runInBackground(t *testing.T, fn func(context.Context) error) (cancel context.CancelFunc, wait func() error) { + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- fn(ctx) }() + var ( + finished bool + result error + ) + wait = func() error { + if finished { + return result + } + select { + case result = <-done: + finished = true + case <-time.After(5 * time.Second): + t.Error("background worker did not return within 5s") + } + return result + } + t.Cleanup(func() { + cancel() + _ = wait() + }) + return cancel, wait +} + +func sseTestEvent(t *testing.T) string { + id, _ := jsonrpc.MakeID("1") + msg, err := jsonrpc.EncodeMessage(&jsonrpc.Request{Method: "a1", ID: id}) + require.NoError(t, err) + return "event: a1\ndata: " + string(msg) + "\n\n" +} + +func TestStreamNotifications_Metrics_HealthyAndRejectedBackends(t *testing.T) { + originalHeartbeatInterval := heartbeatInterval + heartbeatInterval = 20 * time.Millisecond + t.Cleanup(func() { heartbeatInterval = originalHeartbeatInterval }) + + event := sseTestEvent(t) + proxy, mr := newNotificationStreamTestProxy(t, func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get(internalapi.MCPBackendHeader) == "backend2" { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + // backend1 keeps a healthy stream open until the client goes away. + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(event)) + w.(http.Flusher).Flush() + <-r.Context().Done() + }) + s := &session{ + reqCtx: proxy, + perBackendSessions: map[filterapi.MCPBackendName]*compositeSessionEntry{ + "backend1": {backendName: "backend1", sessionID: "s1"}, + "backend2": {backendName: "backend2", sessionID: "s2"}, + }, + route: "test-route", + } + cancel, wait := runInBackground(t, func(ctx context.Context) error { + return s.streamNotifications(ctx, httptest.NewRecorder(), proxy.toolChangeSignaler) + }) + + requireStreamMetricEventually(t, mr, streamActiveMetric, backendAttrs("backend1"), 1) + requireStreamMetricEventually(t, mr, streamOutcomesMetric, outcomeAttrs("backend2", metrics.MCPNotificationStreamOutcomeUnsupported), 1) + for _, b := range []string{"backend1", "backend2"} { + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamAttemptsMetric, backendAttrs(b))) + require.Equal(t, float64(1), totalStreamOutcomes(t, mr, b)) + } + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs("backend1", metrics.MCPNotificationStreamOutcomeOpened))) + // The rejected backend never opened a stream, so it has no active streams. + _, ok := lookupStreamMetric(t, mr, streamActiveMetric, backendAttrs("backend2")) + require.False(t, ok) + + cancel() + require.ErrorIs(t, wait(), context.Canceled) + requireStreamMetricEventually(t, mr, streamDurationMetric, endReasonAttrs("backend1", metrics.MCPNotificationStreamEndReasonCancelled), 1) + require.Equal(t, float64(0), testotel.GetCounterValue(t, mr, streamActiveMetric, backendAttrs("backend1"))) +} + +func TestStreamNotifications_Metrics_AllBackendStreamsEndWhileHeartbeatsContinue(t *testing.T) { + originalHeartbeatInterval := heartbeatInterval + heartbeatInterval = 20 * time.Millisecond + t.Cleanup(func() { heartbeatInterval = originalHeartbeatInterval }) + + event := sseTestEvent(t) + // Every backend sends one event and closes its stream normally. + proxy, mr := newNotificationStreamTestProxy(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(event)) + }) + s := &session{ + reqCtx: proxy, + perBackendSessions: map[filterapi.MCPBackendName]*compositeSessionEntry{ + "backend1": {backendName: "backend1", sessionID: "s1"}, + "backend2": {backendName: "backend2", sessionID: "s2"}, + }, + route: "test-route", + } + rr := &lockedRecorder{rec: httptest.NewRecorder()} + cancel, wait := runInBackground(t, func(ctx context.Context) error { + return s.streamNotifications(ctx, rr, proxy.toolChangeSignaler) + }) + + backends := []string{"backend1", "backend2"} + for _, b := range backends { + requireStreamMetricEventually(t, mr, streamDurationMetric, endReasonAttrs(b, metrics.MCPNotificationStreamEndReasonEOF), 1) + } + // All backend streams have ended, but the client stream stays open and keeps receiving heartbeats. + heartbeatsAfterEnd := rr.count(`"method":"ping"`) + require.Eventually(t, func() bool { + return rr.count(`"method":"ping"`) > heartbeatsAfterEnd + }, 5*time.Second, 10*time.Millisecond) + + for _, b := range backends { + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamAttemptsMetric, backendAttrs(b))) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs(b, metrics.MCPNotificationStreamOutcomeOpened))) + require.Equal(t, float64(0), testotel.GetCounterValue(t, mr, streamActiveMetric, backendAttrs(b))) + } + + cancel() + require.ErrorIs(t, wait(), context.Canceled, "streamNotifications must only return on cancellation") +} + +// lockedRecorder is an http.ResponseWriter that can be read while streamNotifications writes to it. +type lockedRecorder struct { + mu sync.Mutex + rec *httptest.ResponseRecorder +} + +func (l *lockedRecorder) Header() http.Header { + l.mu.Lock() + defer l.mu.Unlock() + return l.rec.Header() +} + +func (l *lockedRecorder) Write(b []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.rec.Write(b) +} + +func (l *lockedRecorder) WriteHeader(code int) { + l.mu.Lock() + defer l.mu.Unlock() + l.rec.WriteHeader(code) +} + +func (l *lockedRecorder) count(substr string) int { + l.mu.Lock() + defer l.mu.Unlock() + return strings.Count(l.rec.Body.String(), substr) +} + +func TestSendRequestPerBackend_NotificationStreamOpenOutcomes(t *testing.T) { + id, _ := jsonrpc.MakeID("1") + ping, _ := jsonrpc.EncodeMessage(&jsonrpc.Request{Method: "ping", ID: id}) + tests := []struct { + name string + status int + contentType, encoding string + contentLength, body string + want metrics.MCPNotificationStreamOutcome + }{ + {name: "202", status: http.StatusAccepted, want: metrics.MCPNotificationStreamOutcomeUnsupported}, + {name: "204", status: http.StatusNoContent, want: metrics.MCPNotificationStreamOutcomeUnsupported}, + {name: "405", status: http.StatusMethodNotAllowed, want: metrics.MCPNotificationStreamOutcomeUnsupported}, + {name: "404", status: http.StatusNotFound, want: metrics.MCPNotificationStreamOutcomeHTTP4xx}, + {name: "500", status: http.StatusInternalServerError, want: metrics.MCPNotificationStreamOutcomeHTTP5xx}, + {name: "503", status: http.StatusServiceUnavailable, want: metrics.MCPNotificationStreamOutcomeHTTP5xx}, + {name: "201", status: http.StatusCreated, want: metrics.MCPNotificationStreamOutcomeHTTPOther}, + { + name: "single JSON-RPC message instead of a stream", status: http.StatusOK, + contentType: "application/json", body: string(ping), want: metrics.MCPNotificationStreamOutcomeUnsupported, + }, + { + name: "truncated JSON body", status: http.StatusOK, contentType: "application/json", + contentLength: "100", body: `{"jsonrpc":`, want: metrics.MCPNotificationStreamOutcomeTransportError, + }, + { + name: "204 with gzip encoding and empty body", status: http.StatusNoContent, + encoding: "gzip", want: metrics.MCPNotificationStreamOutcomeUnsupported, + }, + { + name: "503 with invalid gzip body", status: http.StatusServiceUnavailable, + encoding: "gzip", body: "not gzip", want: metrics.MCPNotificationStreamOutcomeHTTP5xx, + }, + { + name: "200 with invalid gzip body", status: http.StatusOK, contentType: "text/event-stream", + encoding: "gzip", body: "not gzip", want: metrics.MCPNotificationStreamOutcomeTransportError, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + proxy, mr := newNotificationStreamTestProxy(t, func(w http.ResponseWriter, _ *http.Request) { + for k, v := range map[string]string{"Content-Type": tc.contentType, "Content-Encoding": tc.encoding, "Content-Length": tc.contentLength} { + if v != "" { + w.Header().Set(k, v) + } + } + w.WriteHeader(tc.status) + _, _ = w.Write([]byte(tc.body)) + }) + _ = openNotificationStream(t.Context(), proxy) + + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamAttemptsMetric, backendAttrs("backend1"))) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, outcomeAttrs("backend1", tc.want))) + require.Equal(t, float64(1), totalStreamOutcomes(t, mr, "backend1"), "exactly one outcome per attempt") + _, ok := lookupStreamMetric(t, mr, streamActiveMetric, backendAttrs("backend1")) + require.False(t, ok, "a stream that never opened must not be counted as active") + }) + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func TestSendRequestPerBackend_NotificationStreamTransportError(t *testing.T) { + mr := sdkmetric.NewManualReader() + proxy := newTestMCPProxyWithOTEL(mr, noopTracer) + proxy.client = http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("connection refused") + })} + + require.Error(t, openNotificationStream(t.Context(), proxy)) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs("backend1", metrics.MCPNotificationStreamOutcomeTransportError))) +} + +func TestSendRequestPerBackend_NotificationStreamEndReasons(t *testing.T) { + event := sseTestEvent(t) + tests := []struct { + name string + handler http.HandlerFunc + wantErr bool + want metrics.MCPNotificationStreamEndReason + }{ + { + name: "valid final event without trailing blank line", + handler: func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(strings.TrimSuffix(event, "\n\n"))) + }, + want: metrics.MCPNotificationStreamEndReasonEOF, + }, + { + name: "malformed final event without trailing blank line", + handler: func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(event + "data: {invalid json}")) + }, + want: metrics.MCPNotificationStreamEndReasonError, + }, + { + name: "malformed complete event", + handler: func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {invalid json}\n\n")) + }, + wantErr: true, + want: metrics.MCPNotificationStreamEndReasonError, + }, + { + name: "connection cut mid-chunk", + handler: func(w http.ResponseWriter, _ *http.Request) { + conn, buf, err := w.(http.Hijacker).Hijack() + if err != nil { + return + } + defer conn.Close() + _, _ = buf.WriteString("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n") + _, _ = buf.WriteString("20\r\nevent: a1\n") + _ = buf.Flush() + }, + want: metrics.MCPNotificationStreamEndReasonError, + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + proxy, mr := newNotificationStreamTestProxy(t, tc.handler) + err := openNotificationStream(t.Context(), proxy) + if tc.wantErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs("backend1", metrics.MCPNotificationStreamOutcomeOpened))) + count, _ := testotel.GetHistogramValues(t, mr, streamDurationMetric, endReasonAttrs("backend1", tc.want)) + require.Equal(t, uint64(1), count) + require.Equal(t, float64(0), testotel.GetCounterValue(t, mr, streamActiveMetric, backendAttrs("backend1"))) + }) + } +} + +func TestSendRequestPerBackend_NotificationStreamCancellation(t *testing.T) { + t.Run("before open", func(t *testing.T) { + entered := make(chan struct{}) + proxy, mr := newNotificationStreamTestProxy(t, func(_ http.ResponseWriter, r *http.Request) { + close(entered) + <-r.Context().Done() // Withhold the response, so the stream never opens. + }) + cancel, wait := runInBackground(t, func(ctx context.Context) error { return openNotificationStream(ctx, proxy) }) + + select { + case <-entered: + case <-time.After(5 * time.Second): + require.FailNow(t, "backend request was not received") + } + // The attempt is counted while the response is still withheld. + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamAttemptsMetric, backendAttrs("backend1"))) + require.Equal(t, float64(0), totalStreamOutcomes(t, mr, "backend1")) + + cancel() + require.NoError(t, wait()) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs("backend1", metrics.MCPNotificationStreamOutcomeCancelled))) + _, ok := lookupStreamMetric(t, mr, streamActiveMetric, backendAttrs("backend1")) + require.False(t, ok) + }) + + t.Run("after open", func(t *testing.T) { + proxy, mr := newNotificationStreamTestProxy(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + w.(http.Flusher).Flush() + <-r.Context().Done() + }) + cancel, wait := runInBackground(t, func(ctx context.Context) error { return openNotificationStream(ctx, proxy) }) + + requireStreamMetricEventually(t, mr, streamActiveMetric, backendAttrs("backend1"), 1) + const minLifetime = 50 * time.Millisecond + time.Sleep(minLifetime) + cancel() + require.NoError(t, wait()) + + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, streamOutcomesMetric, + outcomeAttrs("backend1", metrics.MCPNotificationStreamOutcomeOpened))) + count, sum := testotel.GetHistogramValues(t, mr, streamDurationMetric, + endReasonAttrs("backend1", metrics.MCPNotificationStreamEndReasonCancelled)) + require.Equal(t, uint64(1), count) + // The lifetime is measured from when the stream opened. + require.GreaterOrEqual(t, sum, minLifetime.Seconds()) + require.Less(t, sum, 60.0) + require.Equal(t, float64(0), testotel.GetCounterValue(t, mr, streamActiveMetric, backendAttrs("backend1"))) + }) +} + +func TestIsCleanEOF(t *testing.T) { + bad := errors.New("bad frame") + for _, tc := range []struct { + name string + err error + want bool + }{ + {name: "EOF", err: io.EOF, want: true}, + {name: "wrapped EOF", err: fmt.Errorf("read: %w", io.EOF), want: true}, + {name: "EOF joined with nil", err: errors.Join(io.EOF, nil), want: true}, + {name: "EOF joined with parse error", err: errors.Join(io.EOF, bad), want: false}, + {name: "wrapped join with parse error", err: fmt.Errorf("read: %w", errors.Join(io.EOF, bad)), want: false}, + {name: "unexpected EOF", err: io.ErrUnexpectedEOF, want: false}, + {name: "other error", err: bad, want: false}, + } { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.want, isCleanEOF(tc.err)) + }) + } +} + +func TestSendRequestPerBackend_POSTDoesNotRecordNotificationStreamMetrics(t *testing.T) { + proxy, mr := newNotificationStreamTestProxy(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{}}\n\n")) + }) + s := &session{reqCtx: proxy} + id, _ := jsonrpc.MakeID("1") + err := s.sendRequestPerBackend(t.Context(), make(chan *backendEvent, 1), "route1", filterapi.MCPBackend{Name: "backend1"}, + &compositeSessionEntry{sessionID: "sess1"}, http.MethodPost, &jsonrpc.Request{Method: "tools/list", ID: id}, nil) + require.NoError(t, err) + + var data metricdata.ResourceMetrics + require.NoError(t, mr.Collect(t.Context(), &data)) + for _, sm := range data.ScopeMetrics { + for _, m := range sm.Metrics { + require.NotContains(t, m.Name, "mcp.notification_stream.") + } + } +} diff --git a/internal/metrics/mcp_metrics.go b/internal/metrics/mcp_metrics.go index cec53342d7..8d063b2785 100644 --- a/internal/metrics/mcp_metrics.go +++ b/internal/metrics/mcp_metrics.go @@ -42,6 +42,33 @@ const ( mcpCapabilitiesNegotiated = "mcp.capabilities.negotiated" // MCP Progress Notifications is a counter metric that records the total number of MCP progress notifications sent. mpcProgressNotifications = "mcp.progress.notifications" + // MCP Notification Stream Open Attempts is a counter metric that records the total number of legacy GET + // notification streams the gateway tried to open to an upstream MCP backend. It is recorded when the request starts. + // + // Dimensions: + // - mcp.backend + mcpNotificationStreamOpenAttempts = "mcp.notification_stream.open.attempts" + // MCP Notification Stream Open Outcomes is a counter metric that records how each legacy GET notification + // stream open attempt ended. See MCPNotificationStreamOutcome for all outcomes. + // + // Dimensions: + // - mcp.backend + // - mcp.notification_stream.outcome + mcpNotificationStreamOpenOutcomes = "mcp.notification_stream.open.outcomes" + // MCP Notification Stream Active is an up/down counter metric that records the number of legacy GET + // notification streams that are currently open to an upstream MCP backend. + // + // Dimensions: + // - mcp.backend + mcpNotificationStreamActive = "mcp.notification_stream.active" + // MCP Notification Stream Duration is a histogram metric that records the lifetime in seconds of an opened + // legacy GET notification stream, from when it opened until it ended. See MCPNotificationStreamEndReason + // for all end reasons. + // + // Dimensions: + // - mcp.backend + // - mcp.notification_stream.end_reason + mcpNotificationStreamDuration = "mcp.notification_stream.duration" // MCP JSON-RPC method name attribute. mcpAttributeMethodName = "mcp.method.name" // MCP status attribute, which is either "success" or "error". See mcpStatusType for all statuses. @@ -54,6 +81,10 @@ const ( mcpAttributeCapabilitySide = "capability.side" // MCP backend attribute, which identifies the upstream MCP backend that handled the request. mcpAttributeBackend = "mcp.backend" + // MCP notification stream open outcome attribute. See MCPNotificationStreamOutcome for all outcomes. + mcpAttributeStreamOutcome = "mcp.notification_stream.outcome" + // MCP notification stream end reason attribute. See MCPNotificationStreamEndReason for all reasons. + mcpAttributeStreamEndReason = "mcp.notification_stream.end_reason" ) // MCPErrorType defines the type of error that occurred during an MCP request. @@ -85,6 +116,44 @@ const ( MCPStatusError MCPStatusType = "error" ) +// MCPNotificationStreamOutcome defines the outcome of an attempt to open a legacy GET notification stream +// to an upstream MCP backend. +type MCPNotificationStreamOutcome string + +const ( + // MCPNotificationStreamOutcomeOpened indicates that the backend accepted the request and the stream is open. + MCPNotificationStreamOutcomeOpened MCPNotificationStreamOutcome = "opened" + // MCPNotificationStreamOutcomeUnsupported indicates that the backend does not offer a notification stream, + // i.e. it returned 202, 204 or 405, or answered with a single JSON-RPC message instead of a stream. + MCPNotificationStreamOutcomeUnsupported MCPNotificationStreamOutcome = "unsupported" + // MCPNotificationStreamOutcomeHTTP4xx indicates that the backend rejected the request with a 4xx status code + // other than 405. + MCPNotificationStreamOutcomeHTTP4xx MCPNotificationStreamOutcome = "http_4xx" + // MCPNotificationStreamOutcomeHTTP5xx indicates that the backend failed the request with a 5xx status code. + MCPNotificationStreamOutcomeHTTP5xx MCPNotificationStreamOutcome = "http_5xx" + // MCPNotificationStreamOutcomeHTTPOther indicates that the backend returned an unexpected status code + // outside the 4xx and 5xx classes, e.g. 201 or 3xx. + MCPNotificationStreamOutcomeHTTPOther MCPNotificationStreamOutcome = "http_other" + // MCPNotificationStreamOutcomeTransportError indicates that the request failed before a usable response + // was received, e.g. a connection error or an undecodable response body. + MCPNotificationStreamOutcomeTransportError MCPNotificationStreamOutcome = "transport_error" + // MCPNotificationStreamOutcomeCancelled indicates that the request was cancelled before the stream opened. + MCPNotificationStreamOutcomeCancelled MCPNotificationStreamOutcome = "cancelled" +) + +// MCPNotificationStreamEndReason defines why an opened legacy GET notification stream ended. +type MCPNotificationStreamEndReason string + +const ( + // MCPNotificationStreamEndReasonEOF indicates that the backend closed the stream normally. + MCPNotificationStreamEndReasonEOF MCPNotificationStreamEndReason = "eof" + // MCPNotificationStreamEndReasonCancelled indicates that the stream was cancelled, e.g. the client disconnected. + MCPNotificationStreamEndReasonCancelled MCPNotificationStreamEndReason = "cancelled" + // MCPNotificationStreamEndReasonError indicates that the stream ended with an error, including an + // unexpected EOF when the connection was cut mid-event. + MCPNotificationStreamEndReasonError MCPNotificationStreamEndReason = "error" +) + // mcpCapabilityType defines the type of capability that is negotiated between client and server. type mcpCapabilityType string @@ -131,6 +200,14 @@ type MCPMetrics interface { RecordServerCapabilities(ctx context.Context, capabilities *mcpsdk.ServerCapabilities, meta mcpsdk.Params) // RecordProgress records a progress notification sent/received. RecordProgress(ctx context.Context, meta mcpsdk.Params) + // RecordNotificationStreamOpenAttempt records the start of an attempt to open a legacy GET notification stream. + RecordNotificationStreamOpenAttempt(ctx context.Context) + // RecordNotificationStreamOpenOutcome records the outcome of an attempt to open a legacy GET notification stream. + // The opened outcome also increments the number of active streams. + RecordNotificationStreamOpenOutcome(ctx context.Context, outcome MCPNotificationStreamOutcome) + // RecordNotificationStreamEnd records the end of a legacy GET notification stream that opened at openedAt. + // It decrements the number of active streams and records the stream lifetime. + RecordNotificationStreamEnd(ctx context.Context, openedAt time.Time, reason MCPNotificationStreamEndReason) } type mcp struct { @@ -139,6 +216,10 @@ type mcp struct { initializationDuration metric.Float64Histogram capabilitiesNegotiated metric.Float64Counter progressNotifications metric.Float64Counter + streamOpenAttempts metric.Float64Counter + streamOpenOutcomes metric.Float64Counter + streamActive metric.Float64UpDownCounter + streamDuration metric.Float64Histogram requestHeaderAttributeMapping map[string]string // maps HTTP headers to metric attribute names. defaultAttributes []attribute.KeyValue } @@ -172,6 +253,27 @@ func NewMCP(meter metric.Meter, requestHeaderAttributeMapping map[string]string) mpcProgressNotifications, metric.WithDescription("Total number of MCP progress notifications sent"), ), + streamOpenAttempts: mustRegisterCounter( + meter, + mcpNotificationStreamOpenAttempts, + metric.WithDescription("Total number of attempts to open an MCP backend notification stream"), + ), + streamOpenOutcomes: mustRegisterCounter( + meter, + mcpNotificationStreamOpenOutcomes, + metric.WithDescription("Total number of MCP backend notification stream open attempts by outcome"), + ), + streamActive: mustRegisterUpDownCounter( + meter, + mcpNotificationStreamActive, + metric.WithDescription("Number of currently open MCP backend notification streams"), + ), + streamDuration: mustRegisterHistogram(meter, + mcpNotificationStreamDuration, + metric.WithDescription("Lifetime of MCP backend notification streams"), + metric.WithUnit("s"), + metric.WithExplicitBucketBoundaries(0.1, 1, 5, 15, 30, 60, 300, 900, 1800, 3600, 7200), + ), } } @@ -183,6 +285,10 @@ func (m *mcp) WithBackend(backend string) MCPMetrics { initializationDuration: m.initializationDuration, capabilitiesNegotiated: m.capabilitiesNegotiated, progressNotifications: m.progressNotifications, + streamOpenAttempts: m.streamOpenAttempts, + streamOpenOutcomes: m.streamOpenOutcomes, + streamActive: m.streamActive, + streamDuration: m.streamDuration, requestHeaderAttributeMapping: m.requestHeaderAttributeMapping, defaultAttributes: append( slices.Clone(m.defaultAttributes), @@ -201,6 +307,10 @@ func (m *mcp) WithRequestAttributes(req *http.Request) MCPMetrics { initializationDuration: m.initializationDuration, capabilitiesNegotiated: m.capabilitiesNegotiated, progressNotifications: m.progressNotifications, + streamOpenAttempts: m.streamOpenAttempts, + streamOpenOutcomes: m.streamOpenOutcomes, + streamActive: m.streamActive, + streamDuration: m.streamDuration, requestHeaderAttributeMapping: m.requestHeaderAttributeMapping, } @@ -265,6 +375,29 @@ func (m *mcp) RecordProgress(ctx context.Context, params mcpsdk.Params) { m.progressNotifications.Add(ctx, 1, m.withDefaultAttributes(params)) } +// RecordNotificationStreamOpenAttempt implements [MCPMetrics.RecordNotificationStreamOpenAttempt]. +func (m *mcp) RecordNotificationStreamOpenAttempt(ctx context.Context) { + m.streamOpenAttempts.Add(ctx, 1, m.withDefaultAttributes(nil)) +} + +// RecordNotificationStreamOpenOutcome implements [MCPMetrics.RecordNotificationStreamOpenOutcome]. +func (m *mcp) RecordNotificationStreamOpenOutcome(ctx context.Context, outcome MCPNotificationStreamOutcome) { + m.streamOpenOutcomes.Add(ctx, 1, m.withDefaultAttributes(nil, + attribute.Key(mcpAttributeStreamOutcome).String(string(outcome)), + )) + if outcome == MCPNotificationStreamOutcomeOpened { + m.streamActive.Add(ctx, 1, m.withDefaultAttributes(nil)) + } +} + +// RecordNotificationStreamEnd implements [MCPMetrics.RecordNotificationStreamEnd]. +func (m *mcp) RecordNotificationStreamEnd(ctx context.Context, openedAt time.Time, reason MCPNotificationStreamEndReason) { + m.streamActive.Add(ctx, -1, m.withDefaultAttributes(nil)) + m.streamDuration.Record(ctx, time.Since(openedAt).Seconds(), m.withDefaultAttributes(nil, + attribute.Key(mcpAttributeStreamEndReason).String(string(reason)), + )) +} + // RecordClientCapabilities implements [MCPMetrics.RecordClientCapabilities]. func (m *mcp) RecordClientCapabilities(ctx context.Context, capabilities *mcpsdk.ClientCapabilities, params mcpsdk.Params) { if capabilities == nil { diff --git a/internal/metrics/mcp_metrics_test.go b/internal/metrics/mcp_metrics_test.go index c7a1b5d036..057b40c30c 100644 --- a/internal/metrics/mcp_metrics_test.go +++ b/internal/metrics/mcp_metrics_test.go @@ -14,6 +14,7 @@ import ( "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "github.com/envoyproxy/ai-gateway/internal/testing/testotel" ) @@ -205,6 +206,110 @@ func TestRecordProgressNotifications(t *testing.T) { require.Equal(t, float64(2), val) } +func TestRecordNotificationStreamLifecycle(t *testing.T) { + mr := metric.NewManualReader() + meter := metric.NewMeterProvider(metric.WithReader(mr)).Meter("test") + + m := NewMCP(meter, nil).WithBackend("backend1") + backendAttrs := attribute.NewSet(attribute.String(mcpAttributeBackend, "backend1")) + + m.RecordNotificationStreamOpenAttempt(t.Context()) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, mcpNotificationStreamOpenAttempts, backendAttrs)) + + m.RecordNotificationStreamOpenOutcome(t.Context(), MCPNotificationStreamOutcomeOpened) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, mcpNotificationStreamOpenOutcomes, attribute.NewSet( + attribute.String(mcpAttributeBackend, "backend1"), + attribute.String(mcpAttributeStreamOutcome, string(MCPNotificationStreamOutcomeOpened)), + ))) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, mcpNotificationStreamActive, backendAttrs)) + + m.RecordNotificationStreamEnd(t.Context(), time.Now().Add(-30*time.Second), MCPNotificationStreamEndReasonEOF) + require.Equal(t, float64(0), testotel.GetCounterValue(t, mr, mcpNotificationStreamActive, backendAttrs)) + count, sum := testotel.GetHistogramValues(t, mr, mcpNotificationStreamDuration, attribute.NewSet( + attribute.String(mcpAttributeBackend, "backend1"), + attribute.String(mcpAttributeStreamEndReason, string(MCPNotificationStreamEndReasonEOF)), + )) + require.Equal(t, uint64(1), count) + // The lower bound is exact; the upper bound only tolerates scheduler pauses. + require.GreaterOrEqual(t, sum, 30.0) + require.Less(t, sum, 40.0) + + // The lifetime histogram is in seconds with buckets that cover short failures and long-lived streams. + var data metricdata.ResourceMetrics + require.NoError(t, mr.Collect(t.Context(), &data)) + var found bool + for _, sm := range data.ScopeMetrics { + for _, md := range sm.Metrics { + if md.Name != mcpNotificationStreamDuration { + continue + } + found = true + require.Equal(t, "s", md.Unit) + dps := md.Data.(metricdata.Histogram[float64]).DataPoints + require.Len(t, dps, 1) + require.Equal(t, []float64{0.1, 1, 5, 15, 30, 60, 300, 900, 1800, 3600, 7200}, dps[0].Bounds) + } + } + require.True(t, found) +} + +func TestRecordNotificationStreamOpenOutcome(t *testing.T) { + mr := metric.NewManualReader() + meter := metric.NewMeterProvider(metric.WithReader(mr)).Meter("test") + + m := NewMCP(meter, nil).WithBackend("backend1") + backendAttrs := attribute.NewSet(attribute.String(mcpAttributeBackend, "backend1")) + // One open stream so that the active gauge exists and can be checked for changes. + m.RecordNotificationStreamOpenOutcome(t.Context(), MCPNotificationStreamOutcomeOpened) + + for _, outcome := range []MCPNotificationStreamOutcome{ + MCPNotificationStreamOutcomeUnsupported, + MCPNotificationStreamOutcomeHTTP4xx, + MCPNotificationStreamOutcomeHTTP5xx, + MCPNotificationStreamOutcomeHTTPOther, + MCPNotificationStreamOutcomeTransportError, + MCPNotificationStreamOutcomeCancelled, + } { + m.RecordNotificationStreamOpenOutcome(t.Context(), outcome) + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, mcpNotificationStreamOpenOutcomes, attribute.NewSet( + attribute.String(mcpAttributeBackend, "backend1"), + attribute.String(mcpAttributeStreamOutcome, string(outcome)), + )), "outcome %s", outcome) + // Only the opened outcome changes the number of active streams. + require.Equal(t, float64(1), testotel.GetCounterValue(t, mr, mcpNotificationStreamActive, backendAttrs), "outcome %s", outcome) + } +} + +func TestRecordNotificationStreamEnd(t *testing.T) { + mr := metric.NewManualReader() + meter := metric.NewMeterProvider(metric.WithReader(mr)).Meter("test") + + m := NewMCP(meter, nil).WithBackend("backend1") + backendAttrs := attribute.NewSet(attribute.String(mcpAttributeBackend, "backend1")) + + reasons := []MCPNotificationStreamEndReason{ + MCPNotificationStreamEndReasonEOF, + MCPNotificationStreamEndReasonCancelled, + MCPNotificationStreamEndReasonError, + } + for range reasons { + m.RecordNotificationStreamOpenOutcome(t.Context(), MCPNotificationStreamOutcomeOpened) + } + require.Equal(t, float64(len(reasons)), testotel.GetCounterValue(t, mr, mcpNotificationStreamActive, backendAttrs)) + + for i, reason := range reasons { + m.RecordNotificationStreamEnd(t.Context(), time.Now().Add(-10*time.Second), reason) + count, sum := testotel.GetHistogramValues(t, mr, mcpNotificationStreamDuration, attribute.NewSet( + attribute.String(mcpAttributeBackend, "backend1"), + attribute.String(mcpAttributeStreamEndReason, string(reason)), + )) + require.Equal(t, uint64(1), count, "reason %s", reason) + require.GreaterOrEqual(t, sum, 10.0, "reason %s", reason) + require.Less(t, sum, 20.0, "reason %s", reason) + require.Equal(t, float64(len(reasons)-i-1), testotel.GetCounterValue(t, mr, mcpNotificationStreamActive, backendAttrs)) + } +} + func TestWithBackend(t *testing.T) { mr := metric.NewManualReader() meter := metric.NewMeterProvider(metric.WithReader(mr)).Meter("test") diff --git a/internal/metrics/register.go b/internal/metrics/register.go index cfa9c3e079..648bc88692 100644 --- a/internal/metrics/register.go +++ b/internal/metrics/register.go @@ -16,6 +16,15 @@ func mustRegisterCounter(meter metric.Meter, name string, options ...metric.Floa return h } +// mustRegisterUpDownCounter registers an up/down counter with the meter and panics if it fails. +func mustRegisterUpDownCounter(meter metric.Meter, name string, options ...metric.Float64UpDownCounterOption) metric.Float64UpDownCounter { + h, err := meter.Float64UpDownCounter(name, options...) + if err != nil { + panic(err) + } + return h +} + // mustRegisterHistogram registers a histogram with the meter and panics if it fails. func mustRegisterHistogram(meter metric.Meter, name string, options ...metric.Float64HistogramOption) metric.Float64Histogram { h, err := meter.Float64Histogram(name, options...)