feat(dot/telemetry): implement telemetry message network_state

chain/dev/genesis.json

@@ -3,7 +3,12 @@
     "id": "dev",
     "chainType": "Local",
     "bootNodes": [],
-    "telemetryEndpoints": null,
+    "telemetryEndpoints": [
+    [
+      "wss://telemetry.polkadot.io/submit/",
+      0
+    ]
+  ],
     "protocolId": "/gossamer/dev/0",
     "genesis": {
         "raw": {
@@ -32,4 +37,4 @@
     "forkBlocks": null,
     "badBlocks": null,
     "consensusEngine": ""
-}
+}

chain/gssmr/genesis.json

@@ -3,6 +3,12 @@
     "id": "gssmr",
     "chainType": "Local",
     "bootNodes": [],
+    "telemetryEndpoints": [
+    [
+      "wss://telemetry.polkadot.io/submit/",
+      0
+    ]
+  ],
     "protocolId": "/gossamer/gssmr/0",
     "genesis": {
         "raw": {
@@ -40,4 +46,4 @@
     "forkBlocks": null,
     "badBlocks": null,
     "consensusEngine": ""
-}
+}

dot/core/messages.go

@@ -57,3 +57,8 @@ func (s *Service) HandleTransactionMessage(msg *network.TransactionMessage) (boo
 
 	return len(msg.Extrinsics) > 0, nil
 }
+
+// TransactionsCount returns number for pending transactions in pool
+func (s *Service) TransactionsCount() int {
+	return len(s.transactionState.PendingInPool())
+}

dot/core/messages_test.go

@@ -21,7 +21,7 @@ import (
 	"testing"
 	"time"
 
-	. "github.com/ChainSafe/gossamer/dot/core/mocks"
+	. "github.com/ChainSafe/gossamer/dot/core/mocks" // nolint
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/state"
 	"github.com/ChainSafe/gossamer/dot/types"
@@ -38,7 +38,7 @@ import (
 
 func TestService_ProcessBlockAnnounceMessage(t *testing.T) {
 	// TODO: move to sync package
-	net := new(MockNetwork)
+	net := new(MockNetwork) // nolint
 
 	cfg := &Config{
 		Network:  net,
@@ -136,7 +136,7 @@ func TestService_HandleTransactionMessage(t *testing.T) {
 	ks := keystore.NewGlobalKeystore()
 	ks.Acco.Insert(kp)
 
-	bp := new(MockBlockProducer)
+	bp := new(MockBlockProducer) // nolint
 	blockC := make(chan types.Block)
 	bp.On("GetBlockChannel", nil).Return(blockC)
 

dot/network/service.go

@@ -19,7 +19,9 @@ package network
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
+	"math/big"
 	"os"
 	"sync"
 	"time"
@@ -304,6 +306,12 @@ func (s *Service) logPeerCount() {
 	}
 }
 
+type peerInfo struct {
+	Roles      byte   `json:"roles"`
+	BestHash   string `json:"bestHash"`
+	BestNumber uint64 `json:"bestNumber"`
+}
+
 func (s *Service) publishNetworkTelemetry(done chan interface{}) {
 	ticker := time.NewTicker(s.telemetryInterval)
 	defer ticker.Stop()
@@ -316,11 +324,36 @@ main:
 
 		case <-ticker.C:
 			o := s.host.bwc.GetBandwidthTotals()
-			err := telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-				telemetry.NewKeyValue("bandwidth_download", o.RateIn),
-				telemetry.NewKeyValue("bandwidth_upload", o.RateOut),
-				telemetry.NewKeyValue("msg", "system.interval"),
-				telemetry.NewKeyValue("peers", s.host.peerCount())))
+			err := telemetry.GetInstance().SendMessage(telemetry.NewBandwidthTM(o.RateIn, o.RateOut, s.host.peerCount()))
+
+			if err != nil {
+				logger.Debug("problem sending system.interval telemetry message", "error", err)
+			}
+			netState := make(map[string]interface{})
+			netState["peerId"] = s.host.h.ID()
+			hostAddrs := []string{}
+			for _, v := range s.host.h.Addrs() {
+				hostAddrs = append(hostAddrs, v.String())
+			}
+			netState["externalAddressess"] = hostAddrs
+			listAddrs := []string{}
+			for _, v := range s.host.h.Network().ListenAddresses() {
+				listAddrs = append(listAddrs, fmt.Sprintf("%s/p2p/%s", v, s.host.h.ID()))
+			}
+			netState["listenedAddressess"] = listAddrs
+
+			peers := make(map[string]interface{})
+			for _, v := range s.Peers() {
+				p := &peerInfo{
+					Roles:      v.Roles,
+					BestHash:   v.BestHash.String(),
+					BestNumber: v.BestNumber,
+				}
+				peers[v.PeerID] = *p
+			}
+			netState["connectedPeers"] = peers
+
+			err = telemetry.GetInstance().SendMessage(telemetry.NewNetworkStateTM(netState))
 			if err != nil {
 				logger.Debug("problem sending system.interval telemetry message", "error", err)
 			}
@@ -334,19 +367,22 @@ func (s *Service) sentBlockIntervalTelemetry() {
 		if err != nil {
 			continue
 		}
+		bestHash := best.Hash()
+
 		finalized, err := s.blockState.GetFinalizedHeader(0, 0) //nolint
 		if err != nil {
 			continue
 		}
-
-		err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-			telemetry.NewKeyValue("best", best.Hash().String()),
-			telemetry.NewKeyValue("finalized_hash", finalized.Hash().String()), //nolint
-			telemetry.NewKeyValue("finalized_height", finalized.Number),        //nolint
-			telemetry.NewKeyValue("height", best.Number),
-			telemetry.NewKeyValue("msg", "system.interval"),
-			telemetry.NewKeyValue("txcount", 0),                // todo (ed) determine where to get tx count
-			telemetry.NewKeyValue("used_state_cache_size", 0))) // todo (ed) determine where to get used_state_cache_size
+		finalizedHash := finalized.Hash()
+
+		err = telemetry.GetInstance().SendMessage(telemetry.NewBlockIntervalTM(
+			&bestHash,
+			best.Number,
+			&finalizedHash,
+			finalized.Number,
+			big.NewInt(int64(s.transactionHandler.TransactionsCount())),
+			big.NewInt(0), // todo (ed) determine where to get used_state_cache_size
+		))
 		if err != nil {
 			logger.Debug("problem sending system.interval telemetry message", "error", err)
 		}

dot/network/service_test.go

@@ -84,6 +84,7 @@ func createTestService(t *testing.T, cfg *Config) (srvc *Service) {
 	if cfg.TransactionHandler == nil {
 		mocktxhandler := &MockTransactionHandler{}
 		mocktxhandler.On("HandleTransactionMessage", mock.AnythingOfType("*TransactionMessage")).Return(nil)
+		mocktxhandler.On("TransactionsCount").Return(0)
 		cfg.TransactionHandler = mocktxhandler
 	}
 

dot/network/state.go

@@ -55,5 +55,6 @@ type Syncer interface {
 
 // TransactionHandler is the interface used by the transactions sub-protocol
 type TransactionHandler interface {
-	HandleTransactionMessage(*TransactionMessage) (bool, error)
+	HandleTransactionMessage(*TransactionMessage) error
+	TransactionsCount() int
 }

dot/network/test_helpers.go

@@ -59,6 +59,7 @@ func NewMockSyncer() *MockSyncer {
 func NewMockTransactionHandler() *MockTransactionHandler {
 	mocktxhandler := new(MockTransactionHandler)
 	mocktxhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(nil)
+	mocktxhandler.On("TransactionsCount").Return(0)
 	return mocktxhandler
 }
 

dot/network/transaction_test.go

@@ -55,7 +55,8 @@ func TestDecodeTransactionMessage(t *testing.T) {
 func TestHandleTransactionMessage(t *testing.T) {
 	basePath := utils.NewTestBasePath(t, "nodeA")
 	mockhandler := &MockTransactionHandler{}
-	mockhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(true, nil)
+	mockhandler.On("HandleTransactionMessage", mock.AnythingOfType("*network.TransactionMessage")).Return(nil)
+	mockhandler.On("TransactionsCount").Return(0)
 
 	config := &Config{
 		BasePath:           basePath,

dot/node.go

@@ -343,17 +343,16 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 	}
 
 	telemetry.GetInstance().AddConnections(gd.TelemetryEndpoints)
-
-	err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage(
-		telemetry.NewKeyValue("authority", cfg.Core.GrandpaAuthority),
-		telemetry.NewKeyValue("chain", sysSrvc.ChainName()),
-		telemetry.NewKeyValue("genesis_hash", stateSrvc.Block.GenesisHash().String()),
-		telemetry.NewKeyValue("implementation", sysSrvc.SystemName()),
-		telemetry.NewKeyValue("msg", "system.connected"),
-		telemetry.NewKeyValue("name", cfg.Global.Name),
-		telemetry.NewKeyValue("network_id", networkSrvc.NetworkState().PeerID),
-		telemetry.NewKeyValue("startup_time", strconv.FormatInt(time.Now().UnixNano(), 10)),
-		telemetry.NewKeyValue("version", sysSrvc.SystemVersion())))
+	genesisHash := stateSrvc.Block.GenesisHash()
+	err = telemetry.GetInstance().SendMessage(telemetry.NewSystemConnectedTM(
+		cfg.Core.GrandpaAuthority,
+		sysSrvc.ChainName(),
+		&genesisHash,
+		sysSrvc.SystemName(),
+		cfg.Global.Name,
+		networkSrvc.NetworkState().PeerID,
+		strconv.FormatInt(time.Now().UnixNano(), 10),
+		sysSrvc.SystemVersion()))
 	if err != nil {
 		logger.Debug("problem sending system.connected telemetry message", "err", err)
 	}

dot/sync/syncer.go

@@ -346,13 +346,13 @@ func (s *Service) handleBlock(block *types.Block) error {
 
 	logger.Debug("🔗 imported block", "number", block.Header.Number, "hash", block.Header.Hash())
 
-	err = telemetry.GetInstance().SendMessage(telemetry.NewTelemetryMessage( // nolint
-		telemetry.NewKeyValue("best", block.Header.Hash().String()),
-		telemetry.NewKeyValue("height", block.Header.Number.Uint64()),
-		telemetry.NewKeyValue("msg", "block.import"),
-		telemetry.NewKeyValue("origin", "NetworkInitialSync")))
+	blockHash := block.Header.Hash()
+	err = telemetry.GetInstance().SendMessage(telemetry.NewBlockImportTM(
+		&blockHash,
+		block.Header.Number,
+		"NetworkInitialSync"))
 	if err != nil {
-		logger.Trace("problem sending block.import telemetry message", "error", err)
+		logger.Debug("problem sending block.import telemetry message", "error", err)
 	}
 
 	return nil