From 251ba25874c4d89b369e6742f8762ba2a9f31e0c Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Wed, 29 Jul 2026 12:04:01 -0400 Subject: [PATCH 1/3] rpc: report psrpc request delivery, expiry and claim outcomes PSRPCMetricsObserver implements psrpc.RequestObserver, installed via psrpc.WithServerObserver inside WithServerObservability. Reports the three server-side lifecycle events the interceptor chain cannot see, because in each case the handler is never invoked: a request read off the bus, a request dropped past its expiry, and settlement of the claim handshake. --- go.mod | 2 +- go.sum | 2 + rpc/metrics.go | 78 +++++++++++++++++++++++++++++++++++- rpc/metrics_observer_test.go | 77 +++++++++++++++++++++++++++++++++++ rpc/typed_api.go | 1 + 5 files changed, 158 insertions(+), 2 deletions(-) create mode 100644 rpc/metrics_observer_test.go diff --git a/go.mod b/go.mod index 356244140..1b2ef9dad 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/lithammer/shortuuid/v4 v4.2.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 - github.com/livekit/psrpc v0.7.2 + github.com/livekit/psrpc v0.7.3 github.com/mackerelio/go-osstat v0.2.8 github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 github.com/nyaruka/phonenumbers v1.8.1 diff --git a/go.sum b/go.sum index ee809f649..0aee18d9c 100644 --- a/go.sum +++ b/go.sum @@ -89,6 +89,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/psrpc v0.7.2 h1:6oZ+NODJ2pLyaT6VqDq1F4Qc/3TpDUSpyphj/P9MhQc= github.com/livekit/psrpc v0.7.2/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= +github.com/livekit/psrpc v0.7.3 h1:bekuZt/ZQzg8+/M8G6G5jq7bvV9fAKdPHSOZeTwrIIc= +github.com/livekit/psrpc v0.7.3/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= github.com/mackerelio/go-osstat v0.2.8 h1:I2duicTaCGWoM53XwAwA9OIe1inu0xnVs8/pqOWWVr4= github.com/mackerelio/go-osstat v0.2.8/go.mod h1:SyS3XxKdoSKJnTGTkN5Yrh6VUQVuAURACfE6y+2DN4k= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= diff --git a/rpc/metrics.go b/rpc/metrics.go index 20a7ec6ae..f42f48bee 100644 --- a/rpc/metrics.go +++ b/rpc/metrics.go @@ -23,6 +23,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "go.uber.org/atomic" + "github.com/livekit/protocol/logger" "github.com/livekit/psrpc" "github.com/livekit/psrpc/pkg/middleware" ) @@ -38,6 +39,10 @@ type psrpcMetrics struct { streamCurrent *prometheus.GaugeVec errorTotal *prometheus.CounterVec bytesTotal *prometheus.CounterVec + requestsReceived *prometheus.CounterVec + requestsExpired *prometheus.CounterVec + claimTotal *prometheus.CounterVec + claimWaitTime prometheus.ObserverVec } var ( @@ -85,6 +90,9 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { streamLabels := slices.Concat(curryLabelNames, []string{"role", "service", "method"}) errorLabels := slices.Concat(labels, []string{"error_code"}) bytesLabels := slices.Concat(labels, []string{"direction"}) + // Lifecycle metrics are server-side only, so they carry no role label. + lifecycleLabels := slices.Concat(curryLabelNames, []string{"service", "method"}) + claimLabels := slices.Concat(lifecycleLabels, []string{"outcome"}) metricsBase.requestTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: livekitNamespace, @@ -125,6 +133,34 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { ConstLabels: constLabels, }, bytesLabels) + metricsBase.requestsReceived = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: livekitNamespace, + Subsystem: "psrpc", + Name: "requests_received_total", + ConstLabels: constLabels, + }, lifecycleLabels) + metricsBase.requestsExpired = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: livekitNamespace, + Subsystem: "psrpc", + Name: "requests_expired_total", + ConstLabels: constLabels, + }, lifecycleLabels) + metricsBase.claimTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: livekitNamespace, + Subsystem: "psrpc", + Name: "claim_total", + ConstLabels: constLabels, + }, claimLabels) + metricsBase.claimWaitTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: livekitNamespace, + Subsystem: "psrpc", + Name: "claim_wait_time_ms", + ConstLabels: constLabels, + // A granted claim settles in single-digit ms; a timed-out one runs to + // the caller's selection timeout, 1s by default. + Buckets: []float64{1, 5, 10, 25, 50, 100, 250, 500, 1000, 3000}, + }, claimLabels) + metricsBase.mu.Unlock() prometheus.MustRegister(metricsBase.requestTime) @@ -133,6 +169,10 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { prometheus.MustRegister(metricsBase.streamCurrent) prometheus.MustRegister(metricsBase.errorTotal) prometheus.MustRegister(metricsBase.bytesTotal) + prometheus.MustRegister(metricsBase.requestsReceived) + prometheus.MustRegister(metricsBase.requestsExpired) + prometheus.MustRegister(metricsBase.claimTotal) + prometheus.MustRegister(metricsBase.claimWaitTime) CurryMetricLabels(o.curryLabels) } @@ -157,6 +197,10 @@ func CurryMetricLabels(labels prometheus.Labels) { streamCurrent: metricsBase.streamCurrent.MustCurryWith(metricsBase.curryLabels), errorTotal: metricsBase.errorTotal.MustCurryWith(metricsBase.curryLabels), bytesTotal: metricsBase.bytesTotal.MustCurryWith(metricsBase.curryLabels), + requestsReceived: metricsBase.requestsReceived.MustCurryWith(metricsBase.curryLabels), + requestsExpired: metricsBase.requestsExpired.MustCurryWith(metricsBase.curryLabels), + claimTotal: metricsBase.claimTotal.MustCurryWith(metricsBase.curryLabels), + claimWaitTime: metricsBase.claimWaitTime.MustCurryWith(metricsBase.curryLabels), }) } @@ -167,7 +211,10 @@ func errorCodeLabel(err error) string { return string(psrpc.Unknown) } -var _ middleware.MetricsObserver = PSRPCMetricsObserver{} +var ( + _ middleware.MetricsObserver = PSRPCMetricsObserver{} + _ psrpc.RequestObserver = PSRPCMetricsObserver{} +) type PSRPCMetricsObserver struct{} @@ -244,3 +291,32 @@ func (o UnimplementedMetricsObserver) OnStreamOpen(role middleware.MetricRole, r } func (o UnimplementedMetricsObserver) OnStreamClose(role middleware.MetricRole, rpcInfo psrpc.RPCInfo) { } + +// OnRequestReceived, OnRequestExpired and OnClaim report server-side lifecycle +// events that the interceptor chain cannot see, because in each case the +// handler is never invoked. Installed by psrpc.WithServerObserver, which is +// separate from middleware.WithServerMetrics. + +func (o PSRPCMetricsObserver) OnRequestReceived(info psrpc.RPCInfo) { + metrics.Load().requestsReceived.WithLabelValues(info.Service, info.Method).Inc() +} + +func (o PSRPCMetricsObserver) OnRequestExpired(info psrpc.RPCInfo, lateBy time.Duration) { + metrics.Load().requestsExpired.WithLabelValues(info.Service, info.Method).Inc() + logger.Warnw("psrpc request dropped: expired before dispatch", nil, + "service", info.Service, "method", info.Method, "lateBy", lateBy) +} + +func (o PSRPCMetricsObserver) OnClaim(info psrpc.RPCInfo, outcome psrpc.ClaimOutcome, wait time.Duration) { + m := metrics.Load() + m.claimTotal.WithLabelValues(info.Service, info.Method, outcome.String()).Inc() + m.claimWaitTime.WithLabelValues(info.Service, info.Method, outcome.String()).Observe(float64(wait.Milliseconds())) + + if outcome == psrpc.ClaimTimedOut { + // The caller stopped waiting for a bid before ours was accepted. It has + // already returned ErrNoResponse upstream, so without this line the + // request leaves no record on either side. + logger.Warnw("psrpc claim timed out before the caller granted it", nil, + "service", info.Service, "method", info.Method, "waited", wait) + } +} diff --git a/rpc/metrics_observer_test.go b/rpc/metrics_observer_test.go new file mode 100644 index 000000000..7802f8546 --- /dev/null +++ b/rpc/metrics_observer_test.go @@ -0,0 +1,77 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package rpc + +import ( + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + + "github.com/livekit/psrpc" +) + +// TestRequestObserverMetrics asserts the server-side lifecycle events register +// and emit. These are the only signals available for a request whose handler is +// never invoked, so a silent regression here would be invisible in production. +func TestRequestObserverMetrics(t *testing.T) { + InitPSRPCStats(prometheus.Labels{}) + o := PSRPCMetricsObserver{} + info := psrpc.RPCInfo{Service: "TestSvc", Method: "TestMethod"} + + o.OnRequestReceived(info) + o.OnRequestExpired(info, 20*time.Millisecond) + o.OnClaim(info, psrpc.ClaimGranted, 3*time.Millisecond) + o.OnClaim(info, psrpc.ClaimTimedOut, 1005*time.Millisecond) + + got := gatherPSRPCCounts(t) + require.Equal(t, 1.0, got["livekit_psrpc_requests_received_total"]) + require.Equal(t, 1.0, got["livekit_psrpc_requests_expired_total"]) + require.Equal(t, 1.0, got["livekit_psrpc_claim_total|granted"]) + require.Equal(t, 1.0, got["livekit_psrpc_claim_total|timed_out"]) + require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|timed_out"]) +} + +// gatherPSRPCCounts returns counter values and histogram sample counts for +// livekit_psrpc_* series, keyed by name and outcome label where present. +func gatherPSRPCCounts(t *testing.T) map[string]float64 { + t.Helper() + mfs, err := prometheus.DefaultGatherer.Gather() + require.NoError(t, err) + + out := map[string]float64{} + for _, mf := range mfs { + if !strings.HasPrefix(mf.GetName(), "livekit_psrpc_") { + continue + } + for _, m := range mf.GetMetric() { + key := mf.GetName() + for _, l := range m.GetLabel() { + if l.GetName() == "outcome" { + key += "|" + l.GetValue() + } + } + if c := m.GetCounter(); c != nil { + out[key] += c.GetValue() + } + if h := m.GetHistogram(); h != nil { + out[key] += float64(h.GetSampleCount()) + } + } + } + return out +} diff --git a/rpc/typed_api.go b/rpc/typed_api.go index f774a70a5..49189d928 100644 --- a/rpc/typed_api.go +++ b/rpc/typed_api.go @@ -100,6 +100,7 @@ func (p *ClientParams) Args() (psrpc.MessageBus, psrpc.ClientOption) { func WithServerObservability(logger logger.Logger) psrpc.ServerOption { return psrpc.WithServerOptions( middleware.WithServerMetrics(PSRPCMetricsObserver{}), + psrpc.WithServerObserver(PSRPCMetricsObserver{}), WithServerLogger(logger), otelpsrpc.ServerOptions(otelpsrpc.Config{}), ) From 9842cee34a1a4093fb89a0b0c9865dacf95bd5be Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Wed, 29 Jul 2026 13:27:27 -0400 Subject: [PATCH 2/3] rpc: cover the interceptor-driven psrpc metrics too The six middleware.MetricsObserver methods had no coverage. Extend the gather helper to filter by service and read gauges, so tests in this package stay independent of each other and of registry ordering. --- rpc/metrics_observer_test.go | 76 ++++++++++++++++++++++++++++++++---- 1 file changed, 68 insertions(+), 8 deletions(-) diff --git a/rpc/metrics_observer_test.go b/rpc/metrics_observer_test.go index 7802f8546..72fc93fb3 100644 --- a/rpc/metrics_observer_test.go +++ b/rpc/metrics_observer_test.go @@ -15,6 +15,7 @@ package rpc import ( + "errors" "strings" "testing" "time" @@ -23,6 +24,7 @@ import ( "github.com/stretchr/testify/require" "github.com/livekit/psrpc" + "github.com/livekit/psrpc/pkg/middleware" ) // TestRequestObserverMetrics asserts the server-side lifecycle events register @@ -31,24 +33,70 @@ import ( func TestRequestObserverMetrics(t *testing.T) { InitPSRPCStats(prometheus.Labels{}) o := PSRPCMetricsObserver{} - info := psrpc.RPCInfo{Service: "TestSvc", Method: "TestMethod"} + info := psrpc.RPCInfo{Service: "LifecycleSvc", Method: "TestMethod"} o.OnRequestReceived(info) o.OnRequestExpired(info, 20*time.Millisecond) o.OnClaim(info, psrpc.ClaimGranted, 3*time.Millisecond) o.OnClaim(info, psrpc.ClaimTimedOut, 1005*time.Millisecond) - got := gatherPSRPCCounts(t) + got := gatherPSRPCSeries(t, "LifecycleSvc") require.Equal(t, 1.0, got["livekit_psrpc_requests_received_total"]) require.Equal(t, 1.0, got["livekit_psrpc_requests_expired_total"]) require.Equal(t, 1.0, got["livekit_psrpc_claim_total|granted"]) require.Equal(t, 1.0, got["livekit_psrpc_claim_total|timed_out"]) + require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|granted"]) require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|timed_out"]) } -// gatherPSRPCCounts returns counter values and histogram sample counts for -// livekit_psrpc_* series, keyed by name and outcome label where present. -func gatherPSRPCCounts(t *testing.T) map[string]float64 { +// TestMetricsObserverMetrics covers the interceptor-driven series. Each method +// routes to a different metric depending on whether the call errored, so the +// error and success paths are asserted separately. +func TestMetricsObserverMetrics(t *testing.T) { + InitPSRPCStats(prometheus.Labels{}) + o := PSRPCMetricsObserver{} + info := psrpc.RPCInfo{Service: "ObserverSvc", Method: "TestMethod"} + boom := errors.New("boom") + + o.OnUnaryRequest(middleware.ClientRole, info, 5*time.Millisecond, nil, 10, 20) + o.OnUnaryRequest(middleware.ClientRole, info, 5*time.Millisecond, boom, 1, 2) + o.OnMultiRequest(middleware.ServerRole, info, 7*time.Millisecond, 2, 0, 30, 40) + o.OnMultiRequest(middleware.ServerRole, info, 7*time.Millisecond, 0, 1, 0, 0) + o.OnStreamSend(middleware.ClientRole, info, 3*time.Millisecond, nil, 50) + o.OnStreamRecv(middleware.ClientRole, info, nil, 60) + o.OnStreamOpen(middleware.ServerRole, info) + o.OnStreamOpen(middleware.ServerRole, info) + o.OnStreamClose(middleware.ServerRole, info) + + got := gatherPSRPCSeries(t, "ObserverSvc") + + require.Equal(t, 1.0, got["livekit_psrpc_request_time_ms|client|rpc"]) + require.Equal(t, 1.0, got["livekit_psrpc_error_total|client|rpc"]) + require.Equal(t, 1.0, got["livekit_psrpc_request_time_ms|server|multirpc"]) + require.Equal(t, 1.0, got["livekit_psrpc_error_total|server|multirpc"]) + require.Equal(t, 1.0, got["livekit_psrpc_stream_send_time_ms|client"]) + require.Equal(t, 1.0, got["livekit_psrpc_stream_receive_total|client"]) + + // stream_count is a gauge: two opens and one close leave one stream live. + require.Equal(t, 1.0, got["livekit_psrpc_stream_count|server"]) + + require.Equal(t, 11.0, got["livekit_psrpc_bytes_total|client|rpc|rx"]) + require.Equal(t, 22.0, got["livekit_psrpc_bytes_total|client|rpc|tx"]) + require.Equal(t, 30.0, got["livekit_psrpc_bytes_total|server|multirpc|rx"]) + require.Equal(t, 40.0, got["livekit_psrpc_bytes_total|server|multirpc|tx"]) + require.Equal(t, 60.0, got["livekit_psrpc_bytes_total|client|stream|rx"]) + require.Equal(t, 50.0, got["livekit_psrpc_bytes_total|client|stream|tx"]) +} + +// discriminatingLabels are appended to each key in the order listed, so a key +// reads livekit_psrpc_bytes_total|client|rpc|rx. +var discriminatingLabels = []string{"role", "kind", "direction", "outcome"} + +// gatherPSRPCSeries returns livekit_psrpc_* values for one service: counter and +// gauge values, and sample counts for histograms. Filtering on service keeps +// tests in this package independent — the registry is global and accumulates +// across them, so a shared key would make assertions order-dependent. +func gatherPSRPCSeries(t *testing.T, service string) map[string]float64 { t.Helper() mfs, err := prometheus.DefaultGatherer.Gather() require.NoError(t, err) @@ -59,15 +107,27 @@ func gatherPSRPCCounts(t *testing.T) map[string]float64 { continue } for _, m := range mf.GetMetric() { - key := mf.GetName() + labels := map[string]string{} for _, l := range m.GetLabel() { - if l.GetName() == "outcome" { - key += "|" + l.GetValue() + labels[l.GetName()] = l.GetValue() + } + if labels["service"] != service { + continue + } + + key := mf.GetName() + for _, name := range discriminatingLabels { + if v, ok := labels[name]; ok { + key += "|" + v } } + if c := m.GetCounter(); c != nil { out[key] += c.GetValue() } + if g := m.GetGauge(); g != nil { + out[key] += g.GetValue() + } if h := m.GetHistogram(); h != nil { out[key] += float64(h.GetSampleCount()) } From 2d0e9cc18107d7689b056a69c6da0a109ea4e34e Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Thu, 6 Aug 2026 11:32:29 -0400 Subject: [PATCH 3/3] rpc: drop claim_total and the lifecycle warn logs The claim histogram already carries a _count per outcome, and the metrics carry what the log lines said. --- rpc/metrics.go | 24 +----------------------- rpc/metrics_observer_test.go | 2 -- 2 files changed, 1 insertion(+), 25 deletions(-) diff --git a/rpc/metrics.go b/rpc/metrics.go index f42f48bee..f7904daaf 100644 --- a/rpc/metrics.go +++ b/rpc/metrics.go @@ -23,7 +23,6 @@ import ( "github.com/prometheus/client_golang/prometheus" "go.uber.org/atomic" - "github.com/livekit/protocol/logger" "github.com/livekit/psrpc" "github.com/livekit/psrpc/pkg/middleware" ) @@ -41,7 +40,6 @@ type psrpcMetrics struct { bytesTotal *prometheus.CounterVec requestsReceived *prometheus.CounterVec requestsExpired *prometheus.CounterVec - claimTotal *prometheus.CounterVec claimWaitTime prometheus.ObserverVec } @@ -145,12 +143,6 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { Name: "requests_expired_total", ConstLabels: constLabels, }, lifecycleLabels) - metricsBase.claimTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ - Namespace: livekitNamespace, - Subsystem: "psrpc", - Name: "claim_total", - ConstLabels: constLabels, - }, claimLabels) metricsBase.claimWaitTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: livekitNamespace, Subsystem: "psrpc", @@ -171,7 +163,6 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { prometheus.MustRegister(metricsBase.bytesTotal) prometheus.MustRegister(metricsBase.requestsReceived) prometheus.MustRegister(metricsBase.requestsExpired) - prometheus.MustRegister(metricsBase.claimTotal) prometheus.MustRegister(metricsBase.claimWaitTime) CurryMetricLabels(o.curryLabels) @@ -199,7 +190,6 @@ func CurryMetricLabels(labels prometheus.Labels) { bytesTotal: metricsBase.bytesTotal.MustCurryWith(metricsBase.curryLabels), requestsReceived: metricsBase.requestsReceived.MustCurryWith(metricsBase.curryLabels), requestsExpired: metricsBase.requestsExpired.MustCurryWith(metricsBase.curryLabels), - claimTotal: metricsBase.claimTotal.MustCurryWith(metricsBase.curryLabels), claimWaitTime: metricsBase.claimWaitTime.MustCurryWith(metricsBase.curryLabels), }) } @@ -303,20 +293,8 @@ func (o PSRPCMetricsObserver) OnRequestReceived(info psrpc.RPCInfo) { func (o PSRPCMetricsObserver) OnRequestExpired(info psrpc.RPCInfo, lateBy time.Duration) { metrics.Load().requestsExpired.WithLabelValues(info.Service, info.Method).Inc() - logger.Warnw("psrpc request dropped: expired before dispatch", nil, - "service", info.Service, "method", info.Method, "lateBy", lateBy) } func (o PSRPCMetricsObserver) OnClaim(info psrpc.RPCInfo, outcome psrpc.ClaimOutcome, wait time.Duration) { - m := metrics.Load() - m.claimTotal.WithLabelValues(info.Service, info.Method, outcome.String()).Inc() - m.claimWaitTime.WithLabelValues(info.Service, info.Method, outcome.String()).Observe(float64(wait.Milliseconds())) - - if outcome == psrpc.ClaimTimedOut { - // The caller stopped waiting for a bid before ours was accepted. It has - // already returned ErrNoResponse upstream, so without this line the - // request leaves no record on either side. - logger.Warnw("psrpc claim timed out before the caller granted it", nil, - "service", info.Service, "method", info.Method, "waited", wait) - } + metrics.Load().claimWaitTime.WithLabelValues(info.Service, info.Method, outcome.String()).Observe(float64(wait.Milliseconds())) } diff --git a/rpc/metrics_observer_test.go b/rpc/metrics_observer_test.go index 72fc93fb3..fd9b3911f 100644 --- a/rpc/metrics_observer_test.go +++ b/rpc/metrics_observer_test.go @@ -43,8 +43,6 @@ func TestRequestObserverMetrics(t *testing.T) { got := gatherPSRPCSeries(t, "LifecycleSvc") require.Equal(t, 1.0, got["livekit_psrpc_requests_received_total"]) require.Equal(t, 1.0, got["livekit_psrpc_requests_expired_total"]) - require.Equal(t, 1.0, got["livekit_psrpc_claim_total|granted"]) - require.Equal(t, 1.0, got["livekit_psrpc_claim_total|timed_out"]) require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|granted"]) require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|timed_out"]) }