feat(sensor): transaction broadcast metrics (#890)

* refactor: remove global metrics, pass as dependencies

* chore: refactor metric initialization

* chore: make gen

* chore: go fmt

* feat: more flags and refactor metrics

* fix: comment

* fix: protocol.go comment

Author: minhd-vu

cmd/p2p/sensor/rpc.go

@@ -15,7 +15,6 @@ import (
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/rs/zerolog/log"
 )
 
@@ -50,11 +49,11 @@ type rpcProxy struct {
 
 // rpcParams holds shared parameters for processing JSON-RPC requests.
 type rpcParams struct {
-	conns   *p2p.Conns
-	chainID *big.Int
-	gpo     *p2p.GasPriceOracle
-	proxy   *rpcProxy
-	counter *prometheus.CounterVec
+	conns    *p2p.Conns
+	chainID  *big.Int
+	gpo      *p2p.GasPriceOracle
+	proxy    *rpcProxy
+	requests *prometheus.CounterVec
 }
 
 // handleRPC sets up the JSON-RPC server for receiving and broadcasting transactions.
@@ -67,21 +66,11 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 	chainID := new(big.Int).SetUint64(networkID)
 	gpo := p2p.NewGasPriceOracle(conns)
 
-	counter := promauto.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace: "sensor",
-			Subsystem: "rpc",
-			Name:      "calls",
-			Help:      "The number of RPC calls made",
-		},
-		[]string{"method", "proxied"},
-	)
-
 	params := &rpcParams{
-		conns:   conns,
-		chainID: chainID,
-		gpo:     gpo,
-		counter: counter,
+		conns:    conns,
+		chainID:  chainID,
+		gpo:      gpo,
+		requests: p2p.NewRPCRequestsCounter(),
 	}
 
 	if inputSensorParams.ProxyRPC {
@@ -132,12 +121,12 @@ func handleRPC(conns *p2p.Conns, networkID uint64) {
 
 		// If method not found and proxy is enabled, forward to upstream
 		if isMethodNotFound(resp) && params.proxy != nil {
-			params.counter.WithLabelValues(req.Method, "true").Inc()
+			params.requests.WithLabelValues(req.Method, "true").Inc()
 			proxyRPCRequest(w, r, body, params.proxy)
 			return
 		}
 
-		params.counter.WithLabelValues(req.Method, "false").Inc()
+		params.requests.WithLabelValues(req.Method, "false").Inc()
 
 		// Enqueue transactions for async broadcast
 		if len(txs) > 0 {
@@ -281,9 +270,9 @@ func handleBatchRequest(w http.ResponseWriter, r *http.Request, body []byte, par
 			if params.proxy != nil {
 				indices = append(indices, i)
 			}
-			params.counter.WithLabelValues(req.Method, "true").Inc()
+			params.requests.WithLabelValues(req.Method, "true").Inc()
 		} else {
-			params.counter.WithLabelValues(req.Method, "false").Inc()
+			params.requests.WithLabelValues(req.Method, "false").Inc()
 		}
 	}
 

cmd/p2p/sensor/sensor.go

@@ -22,8 +22,6 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/nat"
 	"github.com/ethereum/go-ethereum/rpc"
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
@@ -60,6 +58,8 @@ type (
 		BroadcastWorkers             int
 		TxBatchTimeout               time.Duration
 		TxBroadcastQueueSize         int
+		MaxTxPacketSize              int
+		MaxQueuedTxs                 int
 		ShouldRunPprof               bool
 		PprofPort                    uint
 		ShouldRunPrometheus          bool
@@ -200,12 +200,7 @@ var SensorCmd = &cobra.Command{
 			TD:    rpcBlock.TotalDifficulty.ToBigInt(),
 		}
 
-		peersGauge := promauto.NewGauge(prometheus.GaugeOpts{
-			Namespace: "sensor",
-			Name:      "peers",
-			Help:      "The number of peers the sensor is connected to",
-		})
-
+		peersGauge := p2p.NewPeersGauge()
 		metrics := p2p.NewBlockMetrics(head.Block)
 
 		// Create peer connection manager for broadcasting transactions
@@ -223,6 +218,8 @@ var SensorCmd = &cobra.Command{
 			BroadcastWorkers:           inputSensorParams.BroadcastWorkers,
 			TxBatchTimeout:             inputSensorParams.TxBatchTimeout,
 			TxBroadcastQueueSize:       inputSensorParams.TxBroadcastQueueSize,
+			MaxTxPacketSize:            inputSensorParams.MaxTxPacketSize,
+			MaxQueuedTxs:               inputSensorParams.MaxQueuedTxs,
 		})
 
 		opts := p2p.EthProtocolOptions{
@@ -499,7 +496,9 @@ will result in less chance of missing data but can significantly increase memory
 	f.BoolVar(&inputSensorParams.ShouldBroadcastBlockHashes, "broadcast-block-hashes", false, "broadcast block hashes to peers")
 	f.IntVar(&inputSensorParams.BroadcastWorkers, "broadcast-workers", 4, "number of concurrent broadcast workers")
 	f.DurationVar(&inputSensorParams.TxBatchTimeout, "tx-batch-timeout", 500*time.Millisecond, "timeout for batching transactions before broadcast")
-	f.IntVar(&inputSensorParams.TxBroadcastQueueSize, "tx-broadcast-queue-size", 100000, "capacity of transaction broadcast queue")
+	f.IntVar(&inputSensorParams.TxBroadcastQueueSize, "tx-broadcast-queue-size", 100_000, "capacity of transaction broadcast queue")
+	f.IntVar(&inputSensorParams.MaxTxPacketSize, "max-tx-packet-size", 100*1024, "target size in bytes for transaction broadcast packets")
+	f.IntVar(&inputSensorParams.MaxQueuedTxs, "max-queued-txs", 4096, "maximum transaction announcements to queue per peer")
 	f.BoolVar(&inputSensorParams.ShouldRunPprof, "pprof", false, "run pprof server")
 	f.UintVar(&inputSensorParams.PprofPort, "pprof-port", 6060, "port pprof runs on")
 	f.BoolVar(&inputSensorParams.ShouldRunPrometheus, "prom", true, "run Prometheus server")

cmd/pos/exitproof/cmd.go

@@ -250,7 +250,6 @@ func run(ctx context.Context) error {
 	return nil
 }
 
-
 // getBlockReceipts calls eth_getBlockReceipts for the given block number.
 func getBlockReceipts(ctx context.Context, rpc *ethrpc.Client, blockNum *big.Int) ([]*types.Receipt, error) {
 	var receipts []*types.Receipt
@@ -614,10 +613,10 @@ func fetchBlockHashesBatched(ctx context.Context, rpc *ethrpc.Client, start, end
 	hashes := make([]common.Hash, count)
 
 	type blockHeader struct {
-		Number           hexutil.Big  `json:"number"`
-		Timestamp        hexutil.Big  `json:"timestamp"`
-		TransactionsRoot common.Hash  `json:"transactionsRoot"`
-		ReceiptsRoot     common.Hash  `json:"receiptsRoot"`
+		Number           hexutil.Big `json:"number"`
+		Timestamp        hexutil.Big `json:"timestamp"`
+		TransactionsRoot common.Hash `json:"transactionsRoot"`
+		ReceiptsRoot     common.Hash `json:"receiptsRoot"`
 	}
 
 	for batchStart := uint64(0); batchStart < count; batchStart += headerFetchBatchSize {
@@ -702,7 +701,7 @@ func merkleProof(leaves []common.Hash, leafIdx uint64) []byte {
 // The Polygon contracts (ExitPayloadReader.toExitPayload) RLP-decode the payload, so it must
 // be an RLP list — not ABI-encoded. The format matches the matic-js buildReferenceTxPayload:
 // [headerNumber, blockProof, blockNumber, blockTimestamp, txRoot, receiptRoot, receipt,
-//  receiptParentNodes, branchMask, logIndex]
+// receiptParentNodes, branchMask, logIndex]
 func encodeExitPayload(
 	headerNumber *big.Int,
 	blockProof []byte,

cmd/root.go

@@ -28,9 +28,9 @@ import (
 	"github.com/0xPolygon/polygon-cli/cmd/nodekey"
 	"github.com/0xPolygon/polygon-cli/cmd/p2p"
 	"github.com/0xPolygon/polygon-cli/cmd/parsebatchl2data"
-	"github.com/0xPolygon/polygon-cli/cmd/pos"
 	"github.com/0xPolygon/polygon-cli/cmd/parseethwallet"
 	"github.com/0xPolygon/polygon-cli/cmd/plot"
+	"github.com/0xPolygon/polygon-cli/cmd/pos"
 	"github.com/0xPolygon/polygon-cli/cmd/publish"
 	"github.com/0xPolygon/polygon-cli/cmd/report"
 	"github.com/0xPolygon/polygon-cli/cmd/retest"

doc/polycli_p2p_sensor.md

@@ -149,7 +149,9 @@ polycli p2p sensor amoy-nodes.json \
       --max-known-blocks int          maximum block hashes to track per peer (0 for no limit) (default 1024)
       --max-parents int               maximum parent block hashes to track per peer (0 for no limit) (default 1024)
   -m, --max-peers int                 maximum number of peers to connect to (default 2000)
+      --max-queued-txs int            maximum transaction announcements to queue per peer (default 4096)
       --max-requests int              maximum request IDs to track per peer (0 for no limit) (default 2048)
+      --max-tx-packet-size int        target size in bytes for transaction broadcast packets (default 102400)
       --max-txs int                   maximum transactions to cache for serving to peers (0 for no limit) (default 32768)
       --nat string                    NAT port mapping mechanism (any|none|upnp|pmp|pmp:<IP>|extip:<IP>) (default "any")
   -n, --network-id uint               filter discovered nodes by this network ID

doc/polycli_p2p_sensor_metrics.md

@@ -8,6 +8,24 @@ Difference between head and oldest block numbers
 Metric Type: Gauge
 
 
+### sensor_broadcast_batch_size
+Number of transactions per broadcast batch
+
+Metric Type: Histogram
+
+
+### sensor_broadcast_queue_depth
+Number of transaction batches in broadcast queue
+
+Metric Type: Gauge
+
+
+### sensor_broadcast_send_errors
+Number of failed broadcast sends
+
+Metric Type: Counter
+
+
 ### sensor_head_block_age
 Time since head block was received (in seconds)
 
@@ -27,7 +45,7 @@ Metric Type: Gauge
 
 
 ### sensor_messages
-The number and type of messages the sensor has sent and received
+Number and type of messages the sensor has sent and received
 
 Metric Type: CounterVec
 
@@ -43,7 +61,17 @@ Metric Type: Gauge
 
 
 ### sensor_peers
-The number of peers the sensor is connected to
+Number of peers the sensor is connected to
 
 Metric Type: Gauge
 
+
+### sensor_rpc_requests
+Number of RPC requests made
+
+Metric Type: CounterVec
+
+Variable Labels:
+- method
+- proxied
+

docutil/metrics.go

@@ -15,37 +15,33 @@ import (
 type metricInfo struct {
 	Name      string
 	Namespace string
+	Subsystem string
 	Help      string
 	Type      string // Gauge, Counter, Histogram, Summary, GaugeVec, CounterVec, etc.
 	Labels    []string
 	File      string
 }
 
-// fullName returns the complete metric name with namespace.
+// fullName returns the complete metric name with namespace and subsystem.
 func (m metricInfo) fullName() string {
+	parts := []string{}
 	if m.Namespace != "" {
-		return m.Namespace + "_" + m.Name
+		parts = append(parts, m.Namespace)
 	}
-	return m.Name
+	if m.Subsystem != "" {
+		parts = append(parts, m.Subsystem)
+	}
+	parts = append(parts, m.Name)
+	return strings.Join(parts, "_")
 }
 
 // genMetricsDoc generates a METRICS.md file documenting all Prometheus metrics.
 func genMetricsDoc(outputPath string) error {
-	metrics := []metricInfo{}
-
-	// Parse p2p package metrics
-	p2pMetrics, err := parseMetricsFile("p2p/metrics.go")
+	// All sensor metrics are defined in p2p/metrics.go
+	metrics, err := parseMetricsFile("p2p/metrics.go")
 	if err != nil {
 		return fmt.Errorf("failed to parse p2p/metrics.go: %w", err)
 	}
-	metrics = append(metrics, p2pMetrics...)
-
-	// Parse sensor command metrics
-	sensorMetrics, err := parseMetricsFile("cmd/p2p/sensor/sensor.go")
-	if err != nil {
-		return fmt.Errorf("failed to parse cmd/p2p/sensor/sensor.go: %w", err)
-	}
-	metrics = append(metrics, sensorMetrics...)
 
 	// Sort metrics by full name
 	sort.Slice(metrics, func(i, j int) bool {
@@ -114,6 +110,10 @@ func parseMetricsFile(filePath string) ([]metricInfo, error) {
 						if lit, ok := kv.Value.(*ast.BasicLit); ok {
 							metric.Namespace = strings.Trim(lit.Value, `"`)
 						}
+					case "Subsystem":
+						if lit, ok := kv.Value.(*ast.BasicLit); ok {
+							metric.Subsystem = strings.Trim(lit.Value, `"`)
+						}
 					case "Help":
 						if lit, ok := kv.Value.(*ast.BasicLit); ok {
 							metric.Help = strings.Trim(lit.Value, `"`)