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..f7904daaf 100644 --- a/rpc/metrics.go +++ b/rpc/metrics.go @@ -38,6 +38,9 @@ type psrpcMetrics struct { streamCurrent *prometheus.GaugeVec errorTotal *prometheus.CounterVec bytesTotal *prometheus.CounterVec + requestsReceived *prometheus.CounterVec + requestsExpired *prometheus.CounterVec + claimWaitTime prometheus.ObserverVec } var ( @@ -85,6 +88,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 +131,28 @@ 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.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 +161,9 @@ 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.claimWaitTime) CurryMetricLabels(o.curryLabels) } @@ -157,6 +188,9 @@ 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), + claimWaitTime: metricsBase.claimWaitTime.MustCurryWith(metricsBase.curryLabels), }) } @@ -167,7 +201,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 +281,20 @@ 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() +} + +func (o PSRPCMetricsObserver) OnClaim(info psrpc.RPCInfo, outcome psrpc.ClaimOutcome, wait time.Duration) { + 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 new file mode 100644 index 000000000..fd9b3911f --- /dev/null +++ b/rpc/metrics_observer_test.go @@ -0,0 +1,135 @@ +// 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_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/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{}), )