feat(sensor): track metrics direction label

Author: minhd-vu

cmd/p2p/sensor/api.go

@@ -20,9 +20,17 @@ type nodeInfo struct {
 	URL string `json:"enode"`
 }
 
+// peerMessages represents the messages sent and received by a peer.
+type peerMessages struct {
+	Received        p2p.MessageCount `json:"received"`
+	Sent            p2p.MessageCount `json:"sent"`
+	PacketsReceived p2p.MessageCount `json:"packets_received"`
+	PacketsSent     p2p.MessageCount `json:"packets_sent"`
+}
+
 // handleAPI sets up the API for interacting with the sensor. The `/peers`
 // endpoint returns a list of all peers connected to the sensor, including the
-// types and counts of eth packets sent by each peer.
+// types and counts of eth packets sent and received by each peer.
 func handleAPI(server *ethp2p.Server, counter *prometheus.CounterVec) {
 	http.HandleFunc("/peers", func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
@@ -33,10 +41,15 @@ func handleAPI(server *ethp2p.Server, counter *prometheus.CounterVec) {
 		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 
-		peers := make(map[string]p2p.MessageCount)
+		peers := make(map[string]peerMessages)
 		for _, peer := range server.Peers() {
 			url := peer.Node().URLv4()
-			peers[url] = getPeerMessages(url, peer.Fullname(), counter)
+			peers[url] = peerMessages{
+				Received:        getPeerMessages(counter, url, peer.Fullname(), p2p.MsgReceived, false),
+				Sent:            getPeerMessages(counter, url, peer.Fullname(), p2p.MsgSent, false),
+				PacketsReceived: getPeerMessages(counter, url, peer.Fullname(), p2p.MsgReceived, true),
+				PacketsSent:     getPeerMessages(counter, url, peer.Fullname(), p2p.MsgSent, true),
+			}
 		}
 
 		if err := json.NewEncoder(w).Encode(peers); err != nil {
@@ -68,27 +81,32 @@ func handleAPI(server *ethp2p.Server, counter *prometheus.CounterVec) {
 
 // getPeerMessages retrieves the count of various types of eth packets sent by a
 // peer.
-func getPeerMessages(url, name string, counter *prometheus.CounterVec) p2p.MessageCount {
+func getPeerMessages(counter *prometheus.CounterVec, url, name string, direction p2p.Direction, isPacket bool) p2p.MessageCount {
 	return p2p.MessageCount{
-		BlockHeaders:        getCounterValue(new(eth.BlockHeadersPacket), url, name, counter),
-		BlockBodies:         getCounterValue(new(eth.BlockBodiesPacket), url, name, counter),
-		Blocks:              getCounterValue(new(eth.NewBlockPacket), url, name, counter),
-		BlockHashes:         getCounterValue(new(eth.NewBlockHashesPacket), url, name, counter),
-		BlockHeaderRequests: getCounterValue(new(eth.GetBlockHeadersPacket), url, name, counter),
-		BlockBodiesRequests: getCounterValue(new(eth.GetBlockBodiesPacket), url, name, counter),
-		Transactions: getCounterValue(new(eth.TransactionsPacket), url, name, counter) +
-			getCounterValue(new(eth.PooledTransactionsPacket), url, name, counter),
-		TransactionHashes:   getCounterValue(new(eth.NewPooledTransactionHashesPacket), url, name, counter),
-		TransactionRequests: getCounterValue(new(eth.GetPooledTransactionsRequest), url, name, counter),
+		BlockHeaders:        getCounterValue(new(eth.BlockHeadersPacket), counter, url, name, direction, isPacket),
+		BlockBodies:         getCounterValue(new(eth.BlockBodiesPacket), counter, url, name, direction, isPacket),
+		Blocks:              getCounterValue(new(eth.NewBlockPacket), counter, url, name, direction, isPacket),
+		BlockHashes:         getCounterValue(new(eth.NewBlockHashesPacket), counter, url, name, direction, isPacket),
+		BlockHeaderRequests: getCounterValue(new(eth.GetBlockHeadersPacket), counter, url, name, direction, isPacket),
+		BlockBodiesRequests: getCounterValue(new(eth.GetBlockBodiesPacket), counter, url, name, direction, isPacket),
+		Transactions: getCounterValue(new(eth.TransactionsPacket), counter, url, name, direction, isPacket) +
+			getCounterValue(new(eth.PooledTransactionsPacket), counter, url, name, direction, isPacket),
+		TransactionHashes:   getCounterValue(new(eth.NewPooledTransactionHashesPacket), counter, url, name, direction, isPacket),
+		TransactionRequests: getCounterValue(new(eth.GetPooledTransactionsRequest), counter, url, name, direction, isPacket),
 	}
 }
 
 // getCounterValue retrieves the count of packets for a specific type from the
 // Prometheus counter.
-func getCounterValue(packet eth.Packet, url, name string, counter *prometheus.CounterVec) int64 {
+func getCounterValue(packet eth.Packet, counter *prometheus.CounterVec, url, name string, direction p2p.Direction, isPacket bool) int64 {
 	metric := &dto.Metric{}
 
-	err := counter.WithLabelValues(packet.Name(), url, name).Write(metric)
+	messageName := packet.Name()
+	if isPacket {
+		messageName += p2p.PacketSuffix
+	}
+
+	err := counter.WithLabelValues(messageName, url, name, string(direction)).Write(metric)
 	if err != nil {
 		log.Error().Err(err).Send()
 		return 0

cmd/p2p/sensor/sensor.go

@@ -186,8 +186,8 @@ var SensorCmd = &cobra.Command{
 		msgCounter := promauto.NewCounterVec(prometheus.CounterOpts{
 			Namespace: "sensor",
 			Name:      "messages",
-			Help:      "The number and type of messages the sensor has received",
-		}, []string{"message", "url", "name"})
+			Help:      "The number and type of messages the sensor has sent and received",
+		}, []string{"message", "url", "name", "direction"})
 
 		// Create peer connection manager for broadcasting transactions
 		conns := p2p.NewConns()

p2p/log.go

@@ -4,6 +4,19 @@ import (
 	"sync/atomic"
 )
 
+// Direction represents the direction of a message (sent or received).
+type Direction string
+
+const (
+	// MsgReceived represents messages received from peers.
+	MsgReceived Direction = "received"
+	// MsgSent represents messages sent to peers.
+	MsgSent Direction = "sent"
+
+	// PacketSuffix is appended to message names to create packet count metrics.
+	PacketSuffix = "Packet"
+)
+
 // MessageCount is used to help the outer goroutine to receive summary of the
 // number and type of messages that were sent. This is used for distributed
 // logging. It can be used to count the different types of messages received

p2p/protocol.go

@@ -183,6 +183,7 @@ func (c *conn) statusExchange(packet *eth.StatusPacket) error {
 	errc := make(chan error, 2)
 
 	go func() {
+		c.countMsgSent((&eth.StatusPacket{}).Name(), 1)
 		errc <- ethp2p.Send(c.rw, eth.StatusMsg, &packet)
 	}()
 
@@ -207,9 +208,21 @@ func (c *conn) statusExchange(packet *eth.StatusPacket) error {
 	return nil
 }
 
-// AddCount increments the prometheus counter for this connection with the given message name and count.
-func (c *conn) AddCount(messageName string, count float64) {
-	c.counter.WithLabelValues(messageName, c.node.URLv4(), c.peer.Fullname()).Add(count)
+// countMsg increments the prometheus counter for this connection with the given direction, message name, and count.
+func (c *conn) countMsg(direction Direction, messageName string, count float64) {
+	c.counter.WithLabelValues(messageName, c.node.URLv4(), c.peer.Fullname(), string(direction)).Add(count)
+}
+
+// countMsgReceived increments the prometheus counter for received messages.
+func (c *conn) countMsgReceived(messageName string, count float64) {
+	c.countMsg(MsgReceived, messageName, count)
+	c.countMsg(MsgReceived, messageName+PacketSuffix, 1)
+}
+
+// countMsgSent increments the prometheus counter for sent messages.
+func (c *conn) countMsgSent(messageName string, count float64) {
+	c.countMsg(MsgSent, messageName, count)
+	c.countMsg(MsgSent, messageName+PacketSuffix, 1)
 }
 
 func (c *conn) readStatus(packet *eth.StatusPacket) error {
@@ -260,6 +273,7 @@ func (c *conn) getBlockData(hash common.Hash) error {
 		},
 	}
 
+	c.countMsgSent(headersRequest.Name(), 1)
 	if err := ethp2p.Send(c.rw, eth.GetBlockHeadersMsg, headersRequest); err != nil {
 		return err
 	}
@@ -284,6 +298,7 @@ func (c *conn) getBlockData(hash common.Hash) error {
 		GetBlockBodiesRequest: []common.Hash{hash},
 	}
 
+	c.countMsgSent(bodiesRequest.Name(), 1)
 	return ethp2p.Send(c.rw, eth.GetBlockBodiesMsg, bodiesRequest)
 }
 
@@ -320,7 +335,7 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error {
 
 	tfs := time.Now()
 
-	c.AddCount(packet.Name(), float64(len(packet)))
+	c.countMsgReceived(packet.Name(), float64(len(packet)))
 
 	// Collect unique hashes for database write.
 	uniqueHashes := make([]common.Hash, 0, len(packet))
@@ -389,7 +404,7 @@ func (c *conn) handleTransactions(ctx context.Context, msg ethp2p.Msg) error {
 
 	tfs := time.Now()
 
-	c.AddCount(txs.Name(), float64(len(txs)))
+	c.countMsgReceived(txs.Name(), float64(len(txs)))
 
 	c.db.WriteTransactions(ctx, c.node, txs, tfs)
 
@@ -402,13 +417,11 @@ func (c *conn) handleGetBlockHeaders(msg ethp2p.Msg) error {
 		return err
 	}
 
-	c.AddCount(request.Name(), 1)
+	c.countMsgReceived(request.Name(), 1)
 
-	return ethp2p.Send(
-		c.rw,
-		eth.BlockHeadersMsg,
-		&eth.BlockHeadersPacket{RequestId: request.RequestId},
-	)
+	response := &eth.BlockHeadersPacket{RequestId: request.RequestId}
+	c.countMsgSent(response.Name(), 0)
+	return ethp2p.Send(c.rw, eth.BlockHeadersMsg, response)
 }
 
 func (c *conn) handleBlockHeaders(ctx context.Context, msg ethp2p.Msg) error {
@@ -420,7 +433,7 @@ func (c *conn) handleBlockHeaders(ctx context.Context, msg ethp2p.Msg) error {
 	tfs := time.Now()
 
 	headers := packet.BlockHeadersRequest
-	c.AddCount(packet.Name(), float64(len(headers)))
+	c.countMsgReceived(packet.Name(), float64(len(headers)))
 
 	for _, header := range headers {
 		if err := c.getParentBlock(ctx, header); err != nil {
@@ -438,13 +451,11 @@ func (c *conn) handleGetBlockBodies(msg ethp2p.Msg) error {
 		return err
 	}
 
-	c.AddCount(request.Name(), float64(len(request.GetBlockBodiesRequest)))
+	c.countMsgReceived(request.Name(), float64(len(request.GetBlockBodiesRequest)))
 
-	return ethp2p.Send(
-		c.rw,
-		eth.BlockBodiesMsg,
-		&eth.BlockBodiesPacket{RequestId: request.RequestId},
-	)
+	response := &eth.BlockBodiesPacket{RequestId: request.RequestId}
+	c.countMsgSent(response.Name(), 0)
+	return ethp2p.Send(c.rw, eth.BlockBodiesMsg, response)
 }
 
 func (c *conn) handleBlockBodies(ctx context.Context, msg ethp2p.Msg) error {
@@ -459,7 +470,7 @@ func (c *conn) handleBlockBodies(ctx context.Context, msg ethp2p.Msg) error {
 		return nil
 	}
 
-	c.AddCount(packet.Name(), float64(len(packet.BlockBodiesResponse)))
+	c.countMsgReceived(packet.Name(), float64(len(packet.BlockBodiesResponse)))
 
 	var hash *common.Hash
 	for e := c.requests.Front(); e != nil; e = e.Next() {
@@ -490,7 +501,7 @@ func (c *conn) handleNewBlock(ctx context.Context, msg ethp2p.Msg) error {
 
 	tfs := time.Now()
 
-	c.AddCount(block.Name(), 1)
+	c.countMsgReceived(block.Name(), 1)
 
 	// Set the head block if newer.
 	c.headMutex.Lock()
@@ -520,12 +531,11 @@ func (c *conn) handleGetPooledTransactions(msg ethp2p.Msg) error {
 		return err
 	}
 
-	c.AddCount(request.Name(), float64(len(request.GetPooledTransactionsRequest)))
+	c.countMsgReceived(request.Name(), float64(len(request.GetPooledTransactionsRequest)))
 
-	return ethp2p.Send(
-		c.rw,
-		eth.PooledTransactionsMsg,
-		&eth.PooledTransactionsPacket{RequestId: request.RequestId})
+	response := &eth.PooledTransactionsPacket{RequestId: request.RequestId}
+	c.countMsgSent(response.Name(), 0)
+	return ethp2p.Send(c.rw, eth.PooledTransactionsMsg, response)
 }
 
 func (c *conn) handleNewPooledTransactionHashes(version uint, msg ethp2p.Msg) error {
@@ -544,17 +554,15 @@ func (c *conn) handleNewPooledTransactionHashes(version uint, msg ethp2p.Msg) er
 		return errors.New("protocol version not found")
 	}
 
-	c.AddCount(name, float64(len(hashes)))
+	c.countMsgReceived(name, float64(len(hashes)))
 
 	if !c.db.ShouldWriteTransactions() || !c.db.ShouldWriteTransactionEvents() {
 		return nil
 	}
 
-	return ethp2p.Send(
-		c.rw,
-		eth.GetPooledTransactionsMsg,
-		&eth.GetPooledTransactionsPacket{GetPooledTransactionsRequest: hashes},
-	)
+	request := &eth.GetPooledTransactionsPacket{GetPooledTransactionsRequest: hashes}
+	c.countMsgSent(request.Name(), float64(len(hashes)))
+	return ethp2p.Send(c.rw, eth.GetPooledTransactionsMsg, request)
 }
 
 func (c *conn) handlePooledTransactions(ctx context.Context, msg ethp2p.Msg) error {
@@ -565,7 +573,7 @@ func (c *conn) handlePooledTransactions(ctx context.Context, msg ethp2p.Msg) err
 
 	tfs := time.Now()
 
-	c.AddCount(packet.Name(), float64(len(packet.PooledTransactionsResponse)))
+	c.countMsgReceived(packet.Name(), float64(len(packet.PooledTransactionsResponse)))
 
 	c.db.WriteTransactions(ctx, c.node, packet.PooledTransactionsResponse, tfs)
 
@@ -577,9 +585,10 @@ func (c *conn) handleGetReceipts(msg ethp2p.Msg) error {
 	if err := msg.Decode(&request); err != nil {
 		return err
 	}
-	return ethp2p.Send(
-		c.rw,
-		eth.ReceiptsMsg,
-		&eth.ReceiptsPacket{RequestId: request.RequestId},
-	)
+
+	c.countMsgReceived(request.Name(), float64(len(request.GetReceiptsRequest)))
+
+	response := &eth.ReceiptsPacket{RequestId: request.RequestId}
+	c.countMsgSent(response.Name(), 0)
+	return ethp2p.Send(c.rw, eth.ReceiptsMsg, response)
 }