diff --git a/go.mod b/go.mod index 7aad1df98..798e3016a 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.20260727202133-fe4559ce1ddf github.com/mackerelio/go-osstat v0.2.7 github.com/maxbrunsfeld/counterfeiter/v6 v6.12.2 github.com/nyaruka/phonenumbers v1.8.0 diff --git a/go.sum b/go.sum index 23714658a..49e12a435 100644 --- a/go.sum +++ b/go.sum @@ -91,6 +91,8 @@ github.com/livekit/psrpc v0.7.2-0.20260604225640-4bab4033deca h1:d6itWQ1sp5wkUPs 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.20260727202133-fe4559ce1ddf h1:reVLyqpMHLnyGEkm8lEExZt81/XCRr2yzLIJH5Q2BXk= +github.com/livekit/psrpc v0.7.3-0.20260727202133-fe4559ce1ddf/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..b4bbeb996 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; an abandoned 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. psrpc wires these automatically for any observer +// passed to 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.ClaimAbandoned { + // 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 abandoned by caller", 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..02fbd0b59 --- /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.ClaimAbandoned, 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|abandoned"]) + require.Equal(t, 1.0, got["livekit_psrpc_claim_wait_time_ms|abandoned"]) +} + +// 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 +}