Merge branch 'vishal/add_operator_details' of github.com:onflow/flow-go

Author: vishalchangrani

cmd/access/main.go

@@ -45,6 +45,7 @@ func main() {
 		receiptLimit                 uint
 		collectionGRPCPort           uint
 		pingEnabled                  bool
+		nodeInfoFile                 string
 		ingestEng                    *ingestion.Engine
 		requestEng                   *requester.Engine
 		followerEng                  *followereng.Engine
@@ -82,6 +83,7 @@ func main() {
 			flags.BoolVar(&logTxTimeToFinalizedExecuted, "log-tx-time-to-finalized-executed", false, "log transaction time to finalized and executed")
 			flags.BoolVar(&pingEnabled, "ping-enabled", false, "whether to enable the ping process that pings all other peers and report the connectivity to metrics")
 			flags.BoolVar(&retryEnabled, "retry-enabled", false, "whether to enable the retry mechanism at the access node level")
+			flags.StringVarP(&nodeInfoFile, "node-info-file", "", "", "full path to a json file which provides more details about nodes when reporting its reachability metrics")
 		}).
 		Module("collection node client", func(node *cmd.FlowNodeBuilder) error {
 			// collection node address is optional (if not specified, collection nodes will be chosen at random)
@@ -279,6 +281,7 @@ func main() {
 				pingMetrics,
 				pingEnabled,
 				node.Middleware,
+				nodeInfoFile,
 			)
 			if err != nil {
 				return nil, fmt.Errorf("could not create ping engine: %w", err)

engine/access/ping/engine.go

@@ -23,6 +23,7 @@ type Engine struct {
 	pingEnabled  bool
 	pingInterval time.Duration
 	middleware   *p2p.Middleware
+	nodeInfo     map[flow.Identifier]string // additional details about a node such as operator name
 }
 
 func New(
@@ -32,7 +33,9 @@ func New(
 	metrics module.PingMetrics,
 	pingEnabled bool,
 	mw *p2p.Middleware,
+	nodeInfoFile string,
 ) (*Engine, error) {
+
 	eng := &Engine{
 		unit:         engine.NewUnit(),
 		log:          log.With().Str("engine", "ping").Logger(),
@@ -44,6 +47,20 @@ func New(
 		middleware:   mw,
 	}
 
+	// if a node info file is provided, it is read and the additional node information is reported as part of the ping metric
+	if nodeInfoFile != "" {
+		nodeInfo, err := readJson(nodeInfoFile)
+		if err != nil {
+			log.Error().Err(err).Str("node_info_file", nodeInfoFile).Msg("failed to read node info file")
+		} else {
+			eng.nodeInfo = nodeInfo
+			log.Debug().Str("node_info_file", nodeInfoFile).Msg("using node info file")
+		}
+	} else {
+		// the node info file is not mandatory and should not stop the Ping engine from running
+		log.Trace().Msg("no node info file specified")
+	}
+
 	return eng, nil
 }
 
@@ -85,10 +102,12 @@ func (e *Engine) startPing() {
 	}
 }
 
-// send ping to a given node and report the reachable result to metrics
+// pingNode pings the given peer and updates the metrics with the result and the additional node information
 func (e *Engine) pingNode(peer *flow.Identity) {
-	reachable := e.pingAddress(peer.ID())
-	e.metrics.NodeReachable(peer, reachable)
+	id := peer.ID()
+	reachable := e.pingAddress(id)
+	info := e.nodeInfo[id]
+	e.metrics.NodeReachable(peer, info, reachable)
 }
 
 // pingAddress sends a ping request to the given address, and block until either receive

engine/access/ping/node_info.go

@@ -0,0 +1,51 @@
+package ping
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+
+	"github.com/onflow/flow-go/model/flow"
+)
+
+func readJson(jsonFileName string) (map[flow.Identifier]string, error) {
+
+	// read the file
+	byteValue, err := openAndReadFile(jsonFileName)
+	if err != nil {
+		return nil, err
+	}
+
+	// unmarshal json data
+	result, err := unmarshalJson(byteValue)
+	if err != nil {
+		return nil, fmt.Errorf("failed to unmarshal %s: %w", jsonFileName, err)
+	}
+	return result, nil
+}
+
+func openAndReadFile(fileName string) ([]byte, error) {
+	//open
+	jsonFile, err := os.Open(fileName)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open %s: %w", fileName, err)
+	}
+	defer jsonFile.Close()
+
+	// read
+	byteValue, err := ioutil.ReadAll(jsonFile)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read %s: %w", fileName, err)
+	}
+	return byteValue, nil
+}
+
+func unmarshalJson(jsonData []byte) (map[flow.Identifier]string, error) {
+	var result map[flow.Identifier]string
+	err := json.Unmarshal(jsonData, &result)
+	if err != nil {
+		return nil, err
+	}
+	return result, nil
+}

engine/access/ping/node_info_test.go

@@ -0,0 +1,39 @@
+package ping
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+
+	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/utils/unittest"
+
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type Suite struct {
+	suite.Suite
+}
+
+func TestHandler(t *testing.T) {
+	suite.Run(t, new(Suite))
+}
+
+func TestReadJson(t *testing.T) {
+	totalNodes := 10
+	ids := unittest.IdentifierListFixture(totalNodes)
+	testJson := make(map[flow.Identifier]string, totalNodes)
+	for i, id := range ids {
+		testJson[id] = fmt.Sprintf("Operator%d", i+1)
+	}
+	jsonAsBytes, err := json.Marshal(testJson)
+	require.NoError(t, err)
+	content, err := unmarshalJson(jsonAsBytes)
+	require.NoError(t, err)
+	require.Len(t, content, totalNodes)
+	for k, v := range testJson {
+		require.Contains(t, content, k)
+		require.Equal(t, content[k], v)
+	}
+}

module/metrics.go

@@ -316,5 +316,7 @@ type TransactionMetrics interface {
 }
 
 type PingMetrics interface {
-	NodeReachable(node *flow.Identity, reachable bool)
+	// NodeReachable tracks the node availability of the node and reports it as 1 if the node was successfully pinged, 0
+	// otherwise. The nodeInfo provides additional information about the node such as the name of the node operator
+	NodeReachable(node *flow.Identity, nodeInfo string, reachable bool)
 }

module/metrics/labels.go

@@ -8,6 +8,7 @@ const (
 	LabelMessage  = "message"
 	LabelNodeID   = "nodeid"
 	LabelNodeRole = "noderole"
+	LabelNodeInfo = "nodeinfo"
 	LabelPriority = "priority"
 )
 

module/metrics/ping.go

@@ -18,15 +18,19 @@ func NewPingCollector() *PingCollector {
 			Namespace: namespaceNetwork,
 			Subsystem: subsystemGossip,
 			Help:      "report whether a node is reachable",
-		}, []string{LabelNodeID, LabelNodeRole}),
+		}, []string{LabelNodeID, LabelNodeRole, LabelNodeInfo}),
 	}
 	return pc
 }
 
-func (pc *PingCollector) NodeReachable(node *flow.Identity, reachable bool) {
+func (pc *PingCollector) NodeReachable(node *flow.Identity, nodeInfo string, reachable bool) {
 	var val float64
 	if reachable {
 		val = 1
 	}
-	pc.reachable.With(prometheus.Labels{LabelNodeID: node.String(), LabelNodeRole: node.Role.String()}).Set(val)
+	pc.reachable.With(prometheus.Labels{
+		LabelNodeID:   node.String(),
+		LabelNodeRole: node.Role.String(),
+		LabelNodeInfo: nodeInfo}).
+		Set(val)
 }