feat(sensor): track metrics direction label (#737)

* feat(sensor): track metrics direction label

* feat: connected time metrics; join API endpoints

* fix: docs

* fix: invert IsZero

Author: minhd-vu

cmd/p2p/sensor/api.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"slices"
+	"time"
 
 	"github.com/0xPolygon/polygon-cli/p2p"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
@@ -14,17 +15,32 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-// nodeInfo represents information about the sensor node.
-type nodeInfo struct {
-	ENR string `json:"enr"`
-	URL string `json:"enode"`
+// peerData represents the metrics and connection information for a peer.
+// It includes both message counts (items sent/received) and packet counts
+// (number of p2p messages), along with connection timing information.
+type peerData struct {
+	Received        p2p.MessageCount `json:"received"`
+	Sent            p2p.MessageCount `json:"sent"`
+	PacketsReceived p2p.MessageCount `json:"packets_received"`
+	PacketsSent     p2p.MessageCount `json:"packets_sent"`
+	ConnectedAt     string           `json:"connected_at"`
+	DurationSeconds float64          `json:"duration_seconds"`
 }
 
-// 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.
-func handleAPI(server *ethp2p.Server, counter *prometheus.CounterVec) {
-	http.HandleFunc("/peers", func(w http.ResponseWriter, r *http.Request) {
+// apiData represents all sensor information including node info and peer data.
+type apiData struct {
+	ENR       string              `json:"enr"`
+	URL       string              `json:"enode"`
+	PeerCount int                 `json:"peer_count"`
+	Peers     map[string]peerData `json:"peers"`
+}
+
+// handleAPI sets up the API for interacting with the sensor. All endpoints
+// return information about the sensor node and all connected peers, including
+// the types and counts of eth packets sent and received by each peer.
+func handleAPI(server *ethp2p.Server, counter *prometheus.CounterVec, conns *p2p.Conns) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodGet {
 			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 			return
@@ -33,62 +49,74 @@ 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]peerData)
 		for _, peer := range server.Peers() {
 			url := peer.Node().URLv4()
-			peers[url] = getPeerMessages(url, peer.Fullname(), counter)
-		}
+			peerID := peer.Node().ID().String()
+			name := peer.Fullname()
+			connectedAt := conns.GetPeerConnectedAt(peerID)
+			if connectedAt.IsZero() {
+				continue
+			}
 
-		if err := json.NewEncoder(w).Encode(peers); err != nil {
-			log.Error().Err(err).Msg("Failed to encode peers")
-		}
-	})
+			msgs := peerData{
+				Received:        getPeerMessages(counter, url, name, p2p.MsgReceived, false),
+				Sent:            getPeerMessages(counter, url, name, p2p.MsgSent, false),
+				PacketsReceived: getPeerMessages(counter, url, name, p2p.MsgReceived, true),
+				PacketsSent:     getPeerMessages(counter, url, name, p2p.MsgSent, true),
+				ConnectedAt:     connectedAt.UTC().Format(time.RFC3339),
+				DurationSeconds: time.Since(connectedAt).Seconds(),
+			}
 
-	http.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != http.MethodGet {
-			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
-			return
+			peers[url] = msgs
 		}
 
-		info := nodeInfo{
-			ENR: server.NodeInfo().ENR,
-			URL: server.Self().URLv4(),
+		data := apiData{
+			ENR:       server.NodeInfo().ENR,
+			URL:       server.Self().URLv4(),
+			PeerCount: len(peers),
+			Peers:     peers,
 		}
 
-		if err := json.NewEncoder(w).Encode(info); err != nil {
-			log.Error().Err(err).Msg("Failed to encode node info")
+		if err := json.NewEncoder(w).Encode(data); err != nil {
+			log.Error().Err(err).Msg("Failed to encode sensor data")
 		}
 	})
 
 	addr := fmt.Sprintf(":%d", inputSensorParams.APIPort)
-	if err := http.ListenAndServe(addr, nil); err != nil {
+	if err := http.ListenAndServe(addr, mux); err != nil {
 		log.Error().Err(err).Msg("Failed to start API handler")
 	}
 }
 
 // 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/rpc.go

@@ -45,7 +45,8 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 	// Use network ID as chain ID for signature validation
 	chainID := new(big.Int).SetUint64(networkID)
 
-	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		if r.Method != http.MethodPost {
 			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
 			return
@@ -85,7 +86,7 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 
 	addr := fmt.Sprintf(":%d", inputSensorParams.RPCPort)
 	log.Info().Str("addr", addr).Msg("Starting JSON-RPC server")
-	if err := http.ListenAndServe(addr, nil); err != nil {
+	if err := http.ListenAndServe(addr, mux); err != nil {
 		log.Error().Err(err).Msg("Failed to start RPC server")
 	}
 }

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()
@@ -258,7 +258,7 @@ var SensorCmd = &cobra.Command{
 			go handlePrometheus()
 		}
 
-		go handleAPI(&server, msgCounter)
+		go handleAPI(&server, msgCounter, conns)
 
 		// Start the RPC server for receiving transactions
 		go handleRPC(conns, inputSensorParams.NetworkID)

p2p/conns.go

@@ -2,6 +2,7 @@ package p2p
 
 import (
 	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
@@ -77,3 +78,16 @@ func (c *Conns) Nodes() []*enode.Node {
 
 	return nodes
 }
+
+// GetPeerConnectedAt returns the connection time for a peer by their ID.
+// Returns zero time if the peer is not found.
+func (c *Conns) GetPeerConnectedAt(peerID string) time.Time {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	if cn, ok := c.conns[peerID]; ok {
+		return cn.connectedAt
+	}
+
+	return time.Time{}
+}

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