eth: add per-peer metrics

eth/protocols/eth/handlers.go

@@ -448,6 +448,7 @@ func handleNewPooledTransactionHashes(backend Backend, msg Decoder, peer *Peer)
 	for _, hash := range ann.Hashes {
 		peer.markTransaction(hash)
 	}
+	peer.meters.annReceived.Mark(int64(len(ann.Hashes)))
 	return backend.Handle(peer, ann)
 }
 
@@ -501,6 +502,7 @@ func handleTransactions(backend Backend, msg Decoder, peer *Peer) error {
 		}
 		peer.markTransaction(tx.Hash())
 	}
+	peer.meters.txReceived.Mark(int64(len(txs)))
 	return backend.Handle(peer, &txs)
 }
 
@@ -523,6 +525,7 @@ func handlePooledTransactions(backend Backend, msg Decoder, peer *Peer) error {
 	}
 	requestTracker.Fulfil(peer.id, peer.version, PooledTransactionsMsg, txs.RequestId)
 
+	peer.meters.pooledTxReceived.Mark(int64(len(txs.PooledTransactionsResponse)))
 	return backend.Handle(peer, &txs.PooledTransactionsResponse)
 }
 

eth/protocols/eth/peer.go

@@ -59,6 +59,8 @@ type Peer struct {
 	reqCancel   chan *cancel   // Dispatch channel to cancel pending requests and untrack them
 	resDispatch chan *response // Dispatch channel to fulfil pending requests and untrack them
 
+	meters *peerMeters // Peer-specific metrics for diagnositics and to use in decision logic
+
 	term chan struct{} // Termination channel to stop the broadcasters
 }
 
@@ -77,6 +79,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe
 		reqCancel:   make(chan *cancel),
 		resDispatch: make(chan *response),
 		txpool:      txpool,
+		meters:      newPeerMeters("peers/"+p.ID().String(), nil),
 		term:        make(chan struct{}),
 	}
 	// Start up all the broadcasters
@@ -92,6 +95,7 @@ func NewPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWriter, txpool TxPool) *Pe
 // clean it up!
 func (p *Peer) Close() {
 	close(p.term)
+	p.meters.Close()
 }
 
 // ID retrieves the peer's unique identifier.
@@ -136,6 +140,7 @@ func (p *Peer) SendTransactions(txs types.Transactions) error {
 	for _, tx := range txs {
 		p.knownTxs.Add(tx.Hash())
 	}
+	p.meters.txSent.Mark(int64(len(txs)))
 	return p2p.Send(p.rw, TransactionsMsg, txs)
 }
 
@@ -162,6 +167,7 @@ func (p *Peer) AsyncSendTransactions(hashes []common.Hash) {
 func (p *Peer) sendPooledTransactionHashes(hashes []common.Hash, types []byte, sizes []uint32) error {
 	// Mark all the transactions as known, but ensure we don't overflow our limits
 	p.knownTxs.Add(hashes...)
+	p.meters.annSent.Mark(int64(len(hashes)))
 	return p2p.Send(p.rw, NewPooledTransactionHashesMsg, NewPooledTransactionHashesPacket{Types: types, Sizes: sizes, Hashes: hashes})
 }
 
@@ -184,6 +190,7 @@ func (p *Peer) ReplyPooledTransactionsRLP(id uint64, hashes []common.Hash, txs [
 	p.knownTxs.Add(hashes...)
 
 	// Not packed into PooledTransactionsResponse to avoid RLP decoding
+	p.meters.pooledTxSent.Mark(int64(len(txs)))
 	return p2p.Send(p.rw, PooledTransactionsMsg, &PooledTransactionsRLPPacket{
 		RequestId:                     id,
 		PooledTransactionsRLPResponse: txs,

eth/protocols/eth/peer_metrics.go

@@ -0,0 +1,58 @@
+// Copyright 2025 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package eth
+
+import "github.com/ethereum/go-ethereum/metrics"
+
+type peerMeters struct {
+	base string
+	reg  metrics.Registry
+
+	txReceived       *metrics.Meter
+	txSent           *metrics.Meter
+	pooledTxSent     *metrics.Meter
+	pooledTxReceived *metrics.Meter
+	annReceived      *metrics.Meter
+	annSent          *metrics.Meter
+}
+
+// newPeerMeters registers and returns peer-level meters.
+func newPeerMeters(base string, r metrics.Registry) *peerMeters {
+	if r == nil {
+		r = metrics.DefaultRegistry
+	}
+	return &peerMeters{
+		base: base,
+		reg:  r,
+
+		txReceived:       metrics.NewRegisteredMeter(base+"/txReceived", r),
+		txSent:           metrics.NewRegisteredMeter(base+"/txSent", r),
+		pooledTxSent:     metrics.NewRegisteredMeter(base+"/pooledTxSent", r),
+		pooledTxReceived: metrics.NewRegisteredMeter(base+"/pooledTxReceived", r),
+		annReceived:      metrics.NewRegisteredMeter(base+"/annReceived", r),
+		annSent:          metrics.NewRegisteredMeter(base+"/annSent", r),
+	}
+}
+
+func (m *peerMeters) Close() {
+	m.reg.Unregister(m.base + "/txReceived")
+	m.reg.Unregister(m.base + "/txSent")
+	m.reg.Unregister(m.base + "/pooledTxSent")
+	m.reg.Unregister(m.base + "/pooledTxReceived")
+	m.reg.Unregister(m.base + "/annReceived")
+	m.reg.Unregister(m.base + "/annSent")
+}