feat!: sensor rpc functionality for broadcasting txs (#708)

* feat: add `--rpc-port` flag

* feat: sensor rpc endpoint to broadcast txs

* chore: clean up

* fix: stuff

* fix: more

* fix: more

* fix: remove processing tx duplication

* docs: validate tx godoc

* fix: err shadow

* fix: use eth.TransactionsMsg

* docs: make gen-doc

* feat: --no-discovery flag

* fix: log when broadcasting txs

* fix: nits

* fix: body defer

* docs: --no-discovery

* fix: api /peers panic

* chore: refactor db creation

* feat!: remove --quick-start and use --static-nodes flag

* fix: remove leftover nodes logic

* fix: remove duplicate var

* fix: flags

* fix: more

* docs: make gen-doc

---------

Co-authored-by: Léo Vincent <[email protected]>

Author: minhd-vu

cmd/p2p/crawl/crawl.go

@@ -114,15 +114,15 @@ var CrawlCmd = &cobra.Command{
 }
 
 func init() {
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Bootnodes, "bootnodes", "b", "",
+	CrawlCmd.Flags().StringVarP(&inputCrawlParams.Bootnodes, "bootnodes", "b", "",
 		`Comma separated nodes used for bootstrapping. At least one bootnode is
 required, so other nodes in the network can discover each other.`)
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.DiscoveryDNS, "discovery-dns", "", "", `Enable EIP-1459, DNS Discovery to recover node list from given ENRTree`)
+	CrawlCmd.Flags().StringVarP(&inputCrawlParams.DiscoveryDNS, "discovery-dns", "", "", `Enable EIP-1459, DNS Discovery to recover node list from given ENRTree`)
 	CrawlCmd.MarkFlagsMutuallyExclusive("bootnodes", "discovery-dns")
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Timeout, "timeout", "t", "30m0s", "Time limit for the crawl")
-	CrawlCmd.PersistentFlags().IntVarP(&inputCrawlParams.Threads, "parallel", "p", 16, "How many parallel discoveries to attempt")
-	CrawlCmd.PersistentFlags().Uint64VarP(&inputCrawlParams.NetworkID, "network-id", "n", 0, "Filter discovered nodes by this network id")
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Database, "database", "d", "", "Node database for updating and storing client information")
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.RevalidationInterval, "revalidation-interval", "r", "10m", "Time before retrying to connect to a failed peer")
-	CrawlCmd.PersistentFlags().BoolVarP(&inputCrawlParams.OnlyURLs, "only-urls", "u", true, "Only writes the enode URLs to the output")
+	CrawlCmd.Flags().StringVarP(&inputCrawlParams.Timeout, "timeout", "t", "30m0s", "Time limit for the crawl")
+	CrawlCmd.Flags().IntVarP(&inputCrawlParams.Threads, "parallel", "p", 16, "How many parallel discoveries to attempt")
+	CrawlCmd.Flags().Uint64VarP(&inputCrawlParams.NetworkID, "network-id", "n", 0, "Filter discovered nodes by this network id")
+	CrawlCmd.Flags().StringVarP(&inputCrawlParams.Database, "database", "d", "", "Node database for updating and storing client information")
+	CrawlCmd.Flags().StringVarP(&inputCrawlParams.RevalidationInterval, "revalidation-interval", "r", "10m", "Time before retrying to connect to a failed peer")
+	CrawlCmd.Flags().BoolVarP(&inputCrawlParams.OnlyURLs, "only-urls", "u", true, "Only writes the enode URLs to the output")
 }

cmd/p2p/nodelist/nodelist.go

@@ -5,14 +5,14 @@ import (
 	"os"
 
 	"github.com/0xPolygon/polygon-cli/p2p/database"
+	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
 )
 
-const jsonIndent = "    "
-
 type (
 	nodeListParams struct {
 		ProjectID  string
+		DatabaseID string
 		OutputFile string
 		Limit      int
 	}
@@ -26,24 +26,24 @@ var NodeListCmd = &cobra.Command{
 	Use:   "nodelist [nodes.json]",
 	Short: "Generate a node list to seed a node",
 	Args:  cobra.MinimumNArgs(1),
-	PreRunE: func(cmd *cobra.Command, args []string) (err error) {
+	PreRunE: func(cmd *cobra.Command, args []string) error {
 		inputNodeListParams.OutputFile = args[0]
-		inputNodeListParams.ProjectID, err = cmd.Flags().GetString("project-id")
-		return err
+		return nil
 	},
 	RunE: func(cmd *cobra.Command, args []string) error {
 		ctx := cmd.Context()
 
 		db := database.NewDatastore(cmd.Context(), database.DatastoreOptions{
-			ProjectID: inputNodeListParams.ProjectID,
+			ProjectID:  inputNodeListParams.ProjectID,
+			DatabaseID: inputNodeListParams.DatabaseID,
 		})
 
 		nodes, err := db.NodeList(ctx, inputNodeListParams.Limit)
 		if err != nil {
 			return err
 		}
 
-		bytes, err := json.MarshalIndent(nodes, "", jsonIndent)
+		bytes, err := json.MarshalIndent(nodes, "", "    ")
 		if err != nil {
 			return err
 		}
@@ -57,6 +57,10 @@ var NodeListCmd = &cobra.Command{
 }
 
 func init() {
-	NodeListCmd.PersistentFlags().IntVarP(&inputNodeListParams.Limit, "limit", "l", 100, "Number of unique nodes to return")
-	NodeListCmd.PersistentFlags().StringVarP(&inputNodeListParams.ProjectID, "project-id", "p", "", "GCP project ID")
+	NodeListCmd.Flags().IntVarP(&inputNodeListParams.Limit, "limit", "l", 100, "Number of unique nodes to return")
+	NodeListCmd.Flags().StringVarP(&inputNodeListParams.ProjectID, "project-id", "p", "", "GCP project ID")
+	NodeListCmd.Flags().StringVarP(&inputNodeListParams.DatabaseID, "database-id", "d", "", "Datastore database ID")
+	if err := NodeListCmd.MarkFlagRequired("project-id"); err != nil {
+		log.Error().Err(err).Msg("Failed to mark project-id as required flag")
+	}
 }

cmd/p2p/ping/ping.go

@@ -127,9 +127,9 @@ can see other messages the peer sends (e.g. blocks, transactions, etc.).`,
 }
 
 func init() {
-	PingCmd.PersistentFlags().StringVarP(&inputPingParams.OutputFile, "output", "o", "", "Write ping results to output file (default stdout)")
-	PingCmd.PersistentFlags().IntVarP(&inputPingParams.Threads, "parallel", "p", 16, "How many parallel pings to attempt")
-	PingCmd.PersistentFlags().BoolVarP(&inputPingParams.Listen, "listen", "l", true,
+	PingCmd.Flags().StringVarP(&inputPingParams.OutputFile, "output", "o", "", "Write ping results to output file (default stdout)")
+	PingCmd.Flags().IntVarP(&inputPingParams.Threads, "parallel", "p", 16, "How many parallel pings to attempt")
+	PingCmd.Flags().BoolVarP(&inputPingParams.Listen, "listen", "l", true,
 		`Keep the connection open and listen to the peer. This only works if the first
 argument is an enode/enr, not a nodes file.`)
 }

cmd/p2p/query/query.go

@@ -107,6 +107,6 @@ func init() {
 	QueryCmd.Flags().Uint64VarP(&inputQueryParams.StartBlock, "start-block", "s", 0, "Block number to start querying from")
 	QueryCmd.Flags().Uint64VarP(&inputQueryParams.Amount, "amount", "a", 1, "Amount of blocks to query")
 	if err := QueryCmd.MarkFlagRequired("start-block"); err != nil {
-		log.Error().Err(err).Msg("Failed to mark start-block as required persistent flag")
+		log.Error().Err(err).Msg("Failed to mark start-block as required flag")
 	}
 }

cmd/p2p/sensor/api.go

@@ -0,0 +1,136 @@
+package sensor
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"slices"
+
+	"github.com/0xPolygon/polygon-cli/p2p"
+	"github.com/ethereum/go-ethereum/eth/protocols/eth"
+	ethp2p "github.com/ethereum/go-ethereum/p2p"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/rs/zerolog/log"
+)
+
+// nodeInfo represents information about the sensor node.
+type nodeInfo struct {
+	ENR string `json:"enr"`
+	URL string `json:"enode"`
+}
+
+// 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) {
+		if r.Method != http.MethodGet {
+			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+
+		peers := make(map[string]p2p.MessageCount)
+		for _, peer := range server.Peers() {
+			url := peer.Node().URLv4()
+			peers[url] = getPeerMessages(url, peer.Fullname(), counter)
+		}
+
+		if err := json.NewEncoder(w).Encode(peers); err != nil {
+			log.Error().Err(err).Msg("Failed to encode peers")
+		}
+	})
+
+	http.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodGet {
+			http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
+			return
+		}
+
+		info := nodeInfo{
+			ENR: server.NodeInfo().ENR,
+			URL: server.Self().URLv4(),
+		}
+
+		if err := json.NewEncoder(w).Encode(info); err != nil {
+			log.Error().Err(err).Msg("Failed to encode node info")
+		}
+	})
+
+	addr := fmt.Sprintf(":%d", inputSensorParams.APIPort)
+	if err := http.ListenAndServe(addr, nil); 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 {
+	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),
+	}
+}
+
+// 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 {
+	metric := &dto.Metric{}
+
+	err := counter.WithLabelValues(packet.Name(), url, name).Write(metric)
+	if err != nil {
+		log.Error().Err(err).Send()
+		return 0
+	}
+
+	return int64(metric.GetCounter().GetValue())
+}
+
+// removePeerMessages removes all the counters of peers that disconnected from
+// the sensor. This prevents the metrics list from infinitely growing.
+func removePeerMessages(counter *prometheus.CounterVec, urls []string) error {
+	families, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		return err
+	}
+
+	var family *dto.MetricFamily
+	for _, f := range families {
+		if f.GetName() == "sensor_messages" {
+			family = f
+			break
+		}
+	}
+
+	// During DNS-discovery or when the server is taking a while to discover
+	// peers and has yet to receive a message, the sensor_messages prometheus
+	// metric may not exist yet.
+	if family == nil {
+		log.Trace().Msg("Could not find sensor_messages metric family")
+		return nil
+	}
+
+	for _, metric := range family.GetMetric() {
+		for _, label := range metric.GetLabel() {
+			url := label.GetValue()
+			if label.GetName() != "url" || slices.Contains(urls, url) {
+				continue
+			}
+
+			counter.DeletePartialMatch(prometheus.Labels{"url": url})
+		}
+	}
+
+	return nil
+}