diff --git a/docs/metrics.md b/docs/metrics.md index 97d28d849..9b4e24299 100644 --- a/docs/metrics.md +++ b/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` | diff --git a/p2p/metrics.go b/p2p/metrics.go index 6f4a04398..96e80765d 100644 --- a/p2p/metrics.go +++ b/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) diff --git a/p2p/metrics_internal_test.go b/p2p/metrics_internal_test.go new file mode 100644 index 000000000..79a9217c2 --- /dev/null +++ b/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) +} diff --git a/p2p/receive.go b/p2p/receive.go index dc05e28c6..69b963511 100644 --- a/p2p/receive.go +++ b/p2p/receive.go @@ -53,6 +53,9 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, t0 := time.Now() name := PeerName(s.Conn().RemotePeer()) + handlerDone := observeHandlerStart(s.Protocol(), s.Conn().RemotePeer()) + defer handlerDone() + _ = s.SetReadDeadline(time.Now().Add(o.receiveTimeout)) ctx, cancel := context.WithTimeout(context.Background(), o.receiveTimeout) ctx = log.WithTopic(ctx, logTopic) @@ -80,14 +83,30 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, err := readFunc(s).ReadMsg(req) if IsRelayError(err) { + // Resets on relayed (limited) connections are benign circuit recycling, + // but a reset on a direct connection is a genuine read failure. + if !s.Conn().Stat().Limited { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) + } + return // Ignore relay errors. } else if netErr := net.Error(nil); errors.As(err, &netErr) && netErr.Timeout() { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) log.Error(ctx, "Timeout reading p2p message from peer. This may indicate network latency issues or unresponsive peer", err, z.Any("duration", time.Since(t0))) + return } else if err != nil { + incMessageReadError(s.Protocol(), s.Conn().RemotePeer()) log.Error(ctx, "Failed to read p2p request from peer. Check network connectivity and peer health", err, z.Any("duration", time.Since(t0))) + return - } else if err := protonil.Check(req); err != nil { + } + + // Observe before application-level validation so malformed-but-readable messages + // are still visible in the size histogram. + observeReceivedMessage(s.Protocol(), s.Conn().RemotePeer(), req) + + if err := protonil.Check(req); err != nil { log.Warn(ctx, "LibP2P received invalid proto", err) return } @@ -108,5 +127,7 @@ func RegisterHandler(logTopic string, p2pNode host.Host, pID protocol.ID, log.Error(ctx, "Failed to write p2p response to peer. Connection may have been closed", err) return } + + observeSentMessage(s.Protocol(), resp) }) } diff --git a/p2p/sender.go b/p2p/sender.go index 3006c03fb..c0a1a3d93 100644 --- a/p2p/sender.go +++ b/p2p/sender.go @@ -358,10 +358,20 @@ func SendReceive(ctx context.Context, p2pNode host.Host, peerID peer.ID, } } + observeSentMessage(s.Protocol(), req) + if err = reader.ReadMsg(resp); err != nil { + // Resets on relayed (limited) connections are benign circuit recycling, + // but any other failure (including a reset on a direct connection) counts. + if !IsRelayError(err) || !s.Conn().Stat().Limited { + incMessageReadError(s.Protocol(), peerID) + } + return errors.Wrap(err, "read response", z.Any("protocol", s.Protocol())) } + observeReceivedMessage(s.Protocol(), peerID, resp) + o.rttCallback(time.Since(t0)) return nil @@ -408,6 +418,8 @@ func Send(ctx context.Context, p2pNode host.Host, protoID protocol.ID, peerID pe return errors.Wrap(err, "write message", z.Any("protocol", s.Protocol())) } + observeSentMessage(s.Protocol(), msg) + return nil }