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/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/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..72fc93fb3 --- /dev/null +++ b/rpc/metrics_observer_test.go @@ -0,0 +1,137 @@ +// 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 ( + "errors" + "strings" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + + "github.com/livekit/psrpc" + "github.com/livekit/psrpc/pkg/middleware" +) + +// 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: "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 := 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"]) +} + +// 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) + + out := map[string]float64{} + for _, mf := range mfs { + if !strings.HasPrefix(mf.GetName(), "livekit_psrpc_") { + continue + } + for _, m := range mf.GetMetric() { + labels := map[string]string{} + for _, l := range m.GetLabel() { + 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()) + } + } + } + return out +} 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/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{}), ) 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