From f48303ae1e75159c934d5044dbc7c5ddae3c67f5 Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Wed, 29 Jul 2026 12:04:01 -0400 Subject: [PATCH] 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..a8de305ea 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-0.20260729160012-70a5d7b45265 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..96562f835 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-0.20260729160012-70a5d7b45265 h1:ca9OneKyHYinz+grAiHc7QEUaNQPQlfUpyOhNrxZ49c= +github.com/livekit/psrpc v0.7.3-0.20260729160012-70a5d7b45265/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{}), )