From 661e964c1ec06bac1874e637a9f3bf2e3fb4457d 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 | 4 +- go.sum | 6 +-- rpc/metrics.go | 78 +++++++++++++++++++++++++++++++++++- rpc/metrics_observer_test.go | 77 +++++++++++++++++++++++++++++++++++ rpc/typed_api.go | 1 + 5 files changed, 159 insertions(+), 7 deletions(-) create mode 100644 rpc/metrics_observer_test.go diff --git a/go.mod b/go.mod index 7aad1df98..1ccefd1d4 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.7 github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 github.com/nyaruka/phonenumbers v1.8.0 @@ -24,6 +24,7 @@ require ( github.com/pion/sdp/v3 v3.0.18 github.com/pion/webrtc/v4 v4.2.11 github.com/prometheus/client_golang v1.23.2 + github.com/prometheus/client_model v0.6.2 github.com/prometheus/procfs v0.20.1 github.com/puzpuzpuz/xsync/v4 v4.5.0 github.com/redis/go-redis/v9 v9.20.0 @@ -82,7 +83,6 @@ require ( github.com/pion/transport/v4 v4.0.2 // indirect github.com/pion/turn/v5 v5.0.8 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect - github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.68.1 // indirect github.com/wlynxg/anet v0.0.5 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect diff --git a/go.sum b/go.sum index 23714658a..26c9e9a54 100644 --- a/go.sum +++ b/go.sum @@ -87,10 +87,8 @@ github.com/lithammer/shortuuid/v4 v4.2.0 h1:LMFOzVB3996a7b8aBuEXxqOBflbfPQAiVzkI github.com/lithammer/shortuuid/v4 v4.2.0/go.mod h1:D5noHZ2oFw/YaKCfGy0YxyE7M0wMbezmMjPdhyEFe6Y= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5ATTo469PQPkqzdoU7be46ryiCDO3boc= github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/psrpc v0.7.2-0.20260604225640-4bab4033deca h1:d6itWQ1sp5wkUPsgo7k2hNHbZyDp18NtSJg1B6lNJKA= -github.com/livekit/psrpc v0.7.2-0.20260604225640-4bab4033deca/go.mod h1:rAI+m2+/cb4x9RXhLRtUx5ZwdfjjXOl4zi46IjEetaw= -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.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94= github.com/mackerelio/go-osstat v0.2.7/go.mod h1:dwpYh5pIPmvk+IEwBKNIWRFMB92mrC08CmXOhDC7nQk= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= diff --git a/rpc/metrics.go b/rpc/metrics.go index c1b966fd4..21c7084ec 100644 --- a/rpc/metrics.go +++ b/rpc/metrics.go @@ -23,6 +23,7 @@ import ( "go.uber.org/atomic" "golang.org/x/exp/maps" + "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 ( @@ -84,6 +89,9 @@ func InitPSRPCStats(constLabels prometheus.Labels, opts ...PSRPCMetricsOption) { labels := append(curryLabelNames, "role", "kind", "service", "method") streamLabels := append(curryLabelNames, "role", "service", "method") bytesLabels := append(labels, "direction") + // Lifecycle metrics are server-side only, so they carry no role label. + lifecycleLabels := append(curryLabelNames, "service", "method") + claimLabels := append(lifecycleLabels, "outcome") metricsBase.requestTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Namespace: livekitNamespace, @@ -124,6 +132,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) @@ -132,6 +168,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) } @@ -156,10 +196,17 @@ 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), }) } -var _ middleware.MetricsObserver = PSRPCMetricsObserver{} +var ( + _ middleware.MetricsObserver = PSRPCMetricsObserver{} + _ psrpc.RequestObserver = PSRPCMetricsObserver{} +) type PSRPCMetricsObserver struct{} @@ -235,3 +282,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 5510b6e25f7699a3dea4893cb7a7332ec1e82062 Mon Sep 17 00:00:00 2001 From: github-actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:23:20 +0000 Subject: [PATCH 2/3] generated protobuf --- rpc/agent.psrpc.go | 2 +- rpc/agent_dispatch.psrpc.go | 2 +- rpc/egress.psrpc.go | 2 +- rpc/ingress.psrpc.go | 2 +- rpc/io.psrpc.go | 2 +- rpc/keepalive.psrpc.go | 2 +- rpc/participant.psrpc.go | 2 +- rpc/room.psrpc.go | 2 +- rpc/roommanager.psrpc.go | 2 +- rpc/signal.psrpc.go | 2 +- rpc/sip.psrpc.go | 2 +- rpc/whip_signal.psrpc.go | 2 +- 12 files changed, 12 insertions(+), 12 deletions(-) diff --git a/rpc/agent.psrpc.go b/rpc/agent.psrpc.go index 434d66530..40f7489e5 100644 --- a/rpc/agent.psrpc.go +++ b/rpc/agent.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/agent.proto package rpc diff --git a/rpc/agent_dispatch.psrpc.go b/rpc/agent_dispatch.psrpc.go index 29d546646..2715d016a 100644 --- a/rpc/agent_dispatch.psrpc.go +++ b/rpc/agent_dispatch.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/agent_dispatch.proto package rpc diff --git a/rpc/egress.psrpc.go b/rpc/egress.psrpc.go index 2cd187b02..f41b6491d 100644 --- a/rpc/egress.psrpc.go +++ b/rpc/egress.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/egress.proto package rpc diff --git a/rpc/ingress.psrpc.go b/rpc/ingress.psrpc.go index 25faf9c1d..3c433b129 100644 --- a/rpc/ingress.psrpc.go +++ b/rpc/ingress.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/ingress.proto package rpc diff --git a/rpc/io.psrpc.go b/rpc/io.psrpc.go index 6f225e1b5..6a00fcc92 100644 --- a/rpc/io.psrpc.go +++ b/rpc/io.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/io.proto package rpc diff --git a/rpc/keepalive.psrpc.go b/rpc/keepalive.psrpc.go index 4735f8d58..107bb8fa8 100644 --- a/rpc/keepalive.psrpc.go +++ b/rpc/keepalive.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/keepalive.proto package rpc diff --git a/rpc/participant.psrpc.go b/rpc/participant.psrpc.go index 5a3fd5e7a..b8d1293d2 100644 --- a/rpc/participant.psrpc.go +++ b/rpc/participant.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/participant.proto package rpc diff --git a/rpc/room.psrpc.go b/rpc/room.psrpc.go index 0b1a8b622..83818a5e4 100644 --- a/rpc/room.psrpc.go +++ b/rpc/room.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/room.proto package rpc diff --git a/rpc/roommanager.psrpc.go b/rpc/roommanager.psrpc.go index 9dbc22974..5df0da239 100644 --- a/rpc/roommanager.psrpc.go +++ b/rpc/roommanager.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/roommanager.proto package rpc diff --git a/rpc/signal.psrpc.go b/rpc/signal.psrpc.go index acd68bb05..5ac766af3 100644 --- a/rpc/signal.psrpc.go +++ b/rpc/signal.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/signal.proto package rpc diff --git a/rpc/sip.psrpc.go b/rpc/sip.psrpc.go index b1012975c..26979c257 100644 --- a/rpc/sip.psrpc.go +++ b/rpc/sip.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/sip.proto package rpc diff --git a/rpc/whip_signal.psrpc.go b/rpc/whip_signal.psrpc.go index 29707b064..0f4f4ddf7 100644 --- a/rpc/whip_signal.psrpc.go +++ b/rpc/whip_signal.psrpc.go @@ -1,4 +1,4 @@ -// Code generated by protoc-gen-psrpc v0.7.2, DO NOT EDIT. +// Code generated by protoc-gen-psrpc v0.7.3, DO NOT EDIT. // source: rpc/whip_signal.proto package rpc From 489cc78faa006d7adc7c8b43ee7233c46a23ed17 Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Wed, 29 Jul 2026 13:27:27 -0400 Subject: [PATCH 3/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()) }