rpc: report psrpc request delivery, expiry and claim outcomes

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

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=

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()))
+}

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
+}

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{}),
 	)