p2p: add message size, read error and concurrency metrics

docs/metrics.md

@@ -99,6 +99,10 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `core_validatorapi_request_latency_seconds` | Histogram | The validatorapi request latencies in seconds by endpoint | `endpoint` |
 | `core_validatorapi_request_total` | Counter | The total number of requests per content-type and endpoint | `endpoint, content_type` |
 | `core_validatorapi_vc_user_agent` | Gauge | Gauge with label set to user agent string of requests made by VC | `user_agent` |
+| `p2p_concurrent_requests` | Histogram | Number of concurrently handled inbound messages, observed at each message arrival, by protocol and sending peer. Unlike the sampled inflight_requests gauge, this captures bursts between scrapes. | `protocol, peer` |
+| `p2p_handler_duration_seconds` | Histogram | Duration of inbound libp2p message handling from stream accept to handler completion by protocol. Explains inflight_requests: inflight equals message rate times this duration. | `protocol` |
+| `p2p_inflight_requests` | Gauge | Current number of inbound libp2p messages being handled (stream accept to handler completion) by protocol and sending peer. | `protocol, peer` |
+| `p2p_message_read_errors_total` | Counter | Total number of failures reading a libp2p message by protocol and sending peer. Includes messages exceeding the protocol read limit. | `protocol, peer` |
 | `p2p_peer_connection_total` | Counter | Total number of libp2p connections per peer. | `peer` |
 | `p2p_peer_connection_types` | Gauge | Current number of libp2p connections by peer, type (`direct` or `relay`), and protocol (`tcp`, `quic`). Note that peers may have multiple connections. | `peer, type, protocol` |
 | `p2p_peer_network_receive_bytes_total` | Counter | Total number of network bytes received from the peer by protocol and transport. Transport is based on first active connection (accurate in steady state). | `peer, protocol, transport` |
@@ -108,9 +112,11 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `p2p_ping_latency_secs` | Histogram | Ping latencies in seconds per peer | `peer` |
 | `p2p_ping_success` | Gauge | Whether the last ping was successful (1) or not (0). Can be used as proxy for connected peers | `peer` |
 | `p2p_reachability_status` | Gauge | Current libp2p reachability status of this node as detected by autonat: unknown(0), public(1) or private(2). |  |
+| `p2p_received_message_size_bytes` | Histogram | Size in bytes of received libp2p protobuf messages by protocol and sending peer. | `protocol, peer` |
 | `p2p_relay_connection_types` | Gauge | Current number of libp2p connections by relay, type (`direct` or `relay`), and protocol (`tcp`, `quic`). Note that peers may have multiple connections. | `peer, type, protocol` |
 | `p2p_relay_connections` | Gauge | Connected relays by name | `peer` |
 | `p2p_send_duration_seconds` | Histogram | Wall-clock duration of synchronous libp2p Send (one-way) and SendReceive (round-trip) calls, by peer, protocol, and topic. Topic is a sub-protocol label (e.g. qbft_pre_prepare, parsigex_proposer); empty when not set by the caller. | `peer, protocol, topic` |
+| `p2p_sent_message_size_bytes` | Histogram | Size in bytes of sent libp2p protobuf messages by protocol. Not labelled by peer since duty messages are broadcast identically to all peers. | `protocol` |
 | `relay_p2p_active_connections` | Gauge | Current number of active connections by peer and cluster | `peer, peer_cluster` |
 | `relay_p2p_connection_total` | Counter | Total number of new connections by peer and cluster | `peer, peer_cluster` |
 | `relay_p2p_network_receive_bytes_total` | Counter | Total number of network bytes received from the peer and cluster | `peer, peer_cluster` |

p2p/metrics.go

@@ -3,6 +3,7 @@
 package p2p
 
 import (
+	"sync"
 	"time"
 
 	"github.com/libp2p/go-libp2p"
@@ -12,6 +13,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"github.com/libp2p/go-libp2p/p2p/net/swarm"
 	"github.com/prometheus/client_golang/prometheus"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/obolnetwork/charon/app/promauto"
 )
@@ -99,8 +101,116 @@ var (
 		Name:      "peer_network_sent_bytes_total",
 		Help:      "Total number of network bytes sent to the peer by protocol and transport. Transport is based on first active connection (accurate in steady state).",
 	}, []string{"peer", "protocol", "transport"})
+
+	receivedMsgSizeHist = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "p2p",
+		Name:      "received_message_size_bytes",
+		Help:      "Size in bytes of received libp2p protobuf messages by protocol and sending peer.",
+		Buckets:   messageSizeBuckets,
+	}, []string{"protocol", "peer"})
+
+	sentMsgSizeHist = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "p2p",
+		Name:      "sent_message_size_bytes",
+		Help:      "Size in bytes of sent libp2p protobuf messages by protocol. Not labelled by peer since duty messages are broadcast identically to all peers.",
+		Buckets:   messageSizeBuckets,
+	}, []string{"protocol"})
+
+	msgReadErrorCounter = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "p2p",
+		Name:      "message_read_errors_total",
+		Help:      "Total number of failures reading a libp2p message by protocol and sending peer. Includes messages exceeding the protocol read limit.",
+	}, []string{"protocol", "peer"})
+
+	inflightGauge = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "p2p",
+		Name:      "inflight_requests",
+		Help:      "Current number of inbound libp2p messages being handled (stream accept to handler completion) by protocol and sending peer.",
+	}, []string{"protocol", "peer"})
+
+	concurrentRequestsHist = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "p2p",
+		Name:      "concurrent_requests",
+		Help:      "Number of concurrently handled inbound messages, observed at each message arrival, by protocol and sending peer. Unlike the sampled inflight_requests gauge, this captures bursts between scrapes.",
+		Buckets:   []float64{1, 2, 4, 8, 16, 32, 64, 128, 256},
+	}, []string{"protocol", "peer"})
+
+	handlerDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: "p2p",
+		Name:      "handler_duration_seconds",
+		Help:      "Duration of inbound libp2p message handling from stream accept to handler completion by protocol. Explains inflight_requests: inflight equals message rate times this duration.",
+		Buckets:   prometheus.ExponentialBuckets(0.001, 2, 15), // 1ms .. ~16s, covers the ~10s default receive timeout.
+	}, []string{"protocol"})
 )
 
+// inflightKey identifies a per-protocol, per-peer inflight request counter.
+type inflightKey struct {
+	protocol string
+	peer     string
+}
+
+// inflightCounts tracks the number of concurrently handled inbound messages per (protocol, peer).
+// It is the source of truth for both inflightGauge and concurrentRequestsHist so they never drift.
+var inflightCounts = struct {
+	sync.Mutex
+
+	counts map[inflightKey]int
+}{counts: make(map[inflightKey]int)}
+
+// observeHandlerStart records the start of inbound message handling and returns a done function
+// to be called (deferred) when handling completes.
+func observeHandlerStart(pID protocol.ID, peerID peer.ID) func() {
+	key := inflightKey{protocol: string(pID), peer: PeerName(peerID)}
+	t0 := time.Now()
+
+	inflightCounts.Lock()
+	inflightCounts.counts[key]++
+	n := inflightCounts.counts[key]
+	// Update the gauge while holding the lock so concurrent Set calls cannot be reordered.
+	inflightGauge.WithLabelValues(key.protocol, key.peer).Set(float64(n))
+	inflightCounts.Unlock()
+
+	concurrentRequestsHist.WithLabelValues(key.protocol, key.peer).Observe(float64(n))
+
+	return func() {
+		inflightCounts.Lock()
+		inflightCounts.counts[key]--
+
+		n := inflightCounts.counts[key]
+		if n <= 0 {
+			delete(inflightCounts.counts, key) // Don't grow the map with idle keys.
+		}
+
+		inflightGauge.WithLabelValues(key.protocol, key.peer).Set(float64(n))
+		inflightCounts.Unlock()
+
+		handlerDuration.WithLabelValues(key.protocol).Observe(time.Since(t0).Seconds())
+	}
+}
+
+// messageSizeBuckets covers charon p2p message sizes from small control messages up to the
+// 128MiB default read limit. The 32MiB and 128MiB edges match protocol read limits, so
+// messages that (would) exceed a limit are countable by exact bucket subtraction.
+var messageSizeBuckets = []float64{
+	256, 1 << 10, 4 << 10, 16 << 10, 64 << 10, 256 << 10, 512 << 10,
+	1 << 20, 2 << 20, 4 << 20, 8 << 20, 16 << 20, 32 << 20, 64 << 20, 128 << 20,
+}
+
+// observeReceivedMessage records the size of a successfully read libp2p protobuf message.
+func observeReceivedMessage(pID protocol.ID, peerID peer.ID, msg proto.Message) {
+	receivedMsgSizeHist.WithLabelValues(string(pID), PeerName(peerID)).Observe(float64(proto.Size(msg)))
+}
+
+// observeSentMessage records the size of a successfully written libp2p protobuf message.
+func observeSentMessage(pID protocol.ID, msg proto.Message) {
+	sentMsgSizeHist.WithLabelValues(string(pID)).Observe(float64(proto.Size(msg)))
+}
+
+// incMessageReadError increments the read error counter for the protocol and sending peer.
+func incMessageReadError(pID protocol.ID, peerID peer.ID) {
+	msgReadErrorCounter.WithLabelValues(string(pID), PeerName(peerID)).Inc()
+}
+
 func observePing(p peer.ID, d time.Duration) {
 	pingLatencies.WithLabelValues(PeerName(p)).Observe(d.Seconds())
 	pingSuccess.WithLabelValues(PeerName(p)).Set(1)

p2p/metrics_internal_test.go

@@ -0,0 +1,208 @@
+// Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1
+
+package p2p
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peerstore"
+	"github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/prometheus/client_golang/prometheus"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
+	pb "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	pbv1 "github.com/obolnetwork/charon/core/corepb/v1"
+	"github.com/obolnetwork/charon/testutil"
+)
+
+// histSample returns the sample count and sum of the histogram for the given label values.
+func histSample(t *testing.T, hist *prometheus.HistogramVec, labels ...string) (uint64, float64) {
+	t.Helper()
+
+	m := new(pb.Metric)
+
+	h, err := hist.GetMetricWithLabelValues(labels...)
+	require.NoError(t, err)
+	require.NoError(t, h.(prometheus.Histogram).Write(m))
+
+	return m.GetHistogram().GetSampleCount(), m.GetHistogram().GetSampleSum()
+}
+
+func TestMessageSizeMetrics(t *testing.T) {
+	var (
+		pID    = protocol.ID("test-msg-size")
+		ctx    = context.Background()
+		client = testutil.CreateHost(t, testutil.AvailableAddr(t))
+		server = testutil.CreateHost(t, testutil.AvailableAddr(t))
+	)
+
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	RegisterHandler("server", server, pID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			return req, true, nil
+		},
+	)
+
+	req := &pbv1.Duty{Slot: 123, Type: 4}
+	msgSize := float64(proto.Size(req))
+	require.Positive(t, msgSize)
+
+	sentCount0, sentSum0 := histSample(t, sentMsgSizeHist, string(pID))
+	recvClient0, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(server.ID()))
+	recvServer0, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(client.ID()))
+
+	resp := new(pbv1.Duty)
+	require.NoError(t, SendReceive(ctx, client, server.ID(), req, resp, pID))
+
+	// Client sent the request and server sent the identical echoed response.
+	sentCount, sentSum := histSample(t, sentMsgSizeHist, string(pID))
+	require.Equal(t, sentCount0+2, sentCount)
+	require.InDelta(t, sentSum0+2*msgSize, sentSum, 0.1)
+
+	// Server received the request from the client.
+	recvServer, recvServerSum := histSample(t, receivedMsgSizeHist, string(pID), PeerName(client.ID()))
+	require.Equal(t, recvServer0+1, recvServer)
+	require.InDelta(t, msgSize, recvServerSum, 0.1)
+
+	// Client received the response from the server.
+	recvClient, _ := histSample(t, receivedMsgSizeHist, string(pID), PeerName(server.ID()))
+	require.Equal(t, recvClient0+1, recvClient)
+}
+
+func TestMessageReadErrorMetric(t *testing.T) {
+	var (
+		pID    = protocol.ID("test-read-limit")
+		ctx    = context.Background()
+		client = testutil.CreateHost(t, testutil.AvailableAddr(t))
+		server = testutil.CreateHost(t, testutil.AvailableAddr(t))
+	)
+
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	RegisterHandler("server", server, pID,
+		func() proto.Message { return new(pbv1.ParSigExMsg) },
+		func(context.Context, peer.ID, proto.Message) (proto.Message, bool, error) {
+			require.Fail(t, "handler must not be called for oversized message")
+			return nil, false, nil
+		},
+		WithReadLimit(16),
+	)
+
+	errCounter := msgReadErrorCounter.WithLabelValues(string(pID), PeerName(client.ID()))
+	errCount0 := promtestutil.ToFloat64(errCounter)
+
+	// Message larger than the 16 byte read limit is rejected server side before the handler.
+	// Send is one-way so the local write succeeds regardless.
+	msg := &pbv1.ParSigExMsg{Duty: &pbv1.Duty{Slot: 99, Type: 2}, DataSet: &pbv1.ParSignedDataSet{
+		Set: map[string]*pbv1.ParSignedData{"0xdeadbeef": {Data: make([]byte, 1024), Signature: make([]byte, 96)}},
+	}}
+	require.NoError(t, Send(ctx, client, pID, server.ID(), msg))
+
+	require.Eventually(t, func() bool {
+		return promtestutil.ToFloat64(errCounter) >= errCount0+1
+	}, time.Second*5, time.Millisecond*10)
+}
+
+func TestInflightRequestMetrics(t *testing.T) {
+	var (
+		pID    = protocol.ID("test-inflight")
+		ctx    = context.Background()
+		client = testutil.CreateHost(t, testutil.AvailableAddr(t))
+		server = testutil.CreateHost(t, testutil.AvailableAddr(t))
+	)
+
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	RegisterHandler("server", server, pID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			return req, true, nil
+		},
+	)
+
+	clientName := PeerName(client.ID())
+
+	resp := new(pbv1.Duty)
+	require.NoError(t, SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 1}, resp, pID))
+
+	// One arrival observed in the concurrency histogram with count 1.
+	concCount, concSum := histSample(t, concurrentRequestsHist, string(pID), clientName)
+	require.Equal(t, uint64(1), concCount)
+	require.InDelta(t, 1, concSum, 0.1)
+
+	// Handler duration observed once.
+	durCount, durSum := histSample(t, handlerDuration, string(pID))
+	require.Equal(t, uint64(1), durCount)
+	require.Positive(t, durSum)
+
+	// In-flight gauge back to zero after completion.
+	gauge, err := inflightGauge.GetMetricWithLabelValues(string(pID), clientName)
+	require.NoError(t, err)
+	require.Zero(t, promtestutil.ToFloat64(gauge))
+}
+
+func TestConcurrentRequestDepths(t *testing.T) {
+	var (
+		pID    = protocol.ID("test-concurrency")
+		ctx    = context.Background()
+		client = testutil.CreateHost(t, testutil.AvailableAddr(t))
+		server = testutil.CreateHost(t, testutil.AvailableAddr(t))
+	)
+
+	client.Peerstore().AddAddrs(server.ID(), server.Addrs(), peerstore.PermanentAddrTTL)
+
+	release := make(chan struct{})
+	RegisterHandler("server", server, pID,
+		func() proto.Message { return new(pbv1.Duty) },
+		func(_ context.Context, _ peer.ID, req proto.Message) (proto.Message, bool, error) {
+			<-release
+			return req, true, nil
+		},
+	)
+
+	clientName := PeerName(client.ID())
+
+	// Issue concurrent requests against a blocked handler.
+	const n = 3
+
+	var wg sync.WaitGroup
+	for range n {
+		wg.Add(1)
+
+		go func() {
+			defer wg.Done()
+
+			resp := new(pbv1.Duty)
+			_ = SendReceive(ctx, client, server.ID(), &pbv1.Duty{Slot: 1}, resp, pID)
+		}()
+	}
+
+	gauge, err := inflightGauge.GetMetricWithLabelValues(string(pID), clientName)
+	require.NoError(t, err)
+
+	// All requests block in the handler, so the gauge reaches n.
+	require.Eventually(t, func() bool {
+		return promtestutil.ToFloat64(gauge) == n
+	}, time.Second*5, time.Millisecond*10)
+
+	close(release)
+	wg.Wait()
+
+	// Gauge returns to zero once all handlers complete.
+	require.Eventually(t, func() bool {
+		return promtestutil.ToFloat64(gauge) == 0
+	}, time.Second*5, time.Millisecond*10)
+
+	// The concurrency histogram observed depths 1, 2 and 3 (in some order), so the sum is 6.
+	concCount, concSum := histSample(t, concurrentRequestsHist, string(pID), clientName)
+	require.Equal(t, uint64(n), concCount)
+	require.InDelta(t, 6, concSum, 0.1)
+}