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/network/service.go

@@ -301,6 +301,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()
@@ -321,6 +327,35 @@ main:
 			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, v.String())
+			}
+			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.NewTelemetryMessage(
+				telemetry.NewKeyValue("msg", "system.network_state"),
+				telemetry.NewKeyValue("state", netState)))
+			if err != nil {
+				logger.Debug("problem sending system.interval telemetry message", "error", err)
+			}
 		}
 	}
 }

dot/telemetry/telemetry.go

@@ -134,19 +134,9 @@ func (h *Handler) startListening() {
 	}
 }
 
-type response struct {
-	ID        int                    `json:"id"`
-	Payload   map[string]interface{} `json:"payload"`
-	Timestamp time.Time              `json:"ts"`
-}
-
 func msgToBytes(message Message) []byte {
-	res := response{
-		ID:        1, // todo (ed) determine how this is used
-		Payload:   message.values,
-		Timestamp: time.Now(),
-	}
-	resB, err := json.Marshal(res)
+	message.values["ts"] = time.Now()
+	resB, err := json.Marshal(message.values)
 	if err != nil {
 		return nil
 	}

dot/telemetry/telemetry_test.go

@@ -2,6 +2,7 @@ package telemetry
 
 import (
 	"bytes"
+	"fmt"
 	"log"
 	"math/big"
 	"net/http"
@@ -89,10 +90,10 @@ func TestHandler_SendMulti(t *testing.T) {
 
 	wg.Wait()
 
-	expected1 := []byte(`{"id":1,"payload":{"bandwidth_download":2,"bandwidth_upload":3,"msg":"system.interval","peers":1},"ts":`)
-	expected2 := []byte(`{"id":1,"payload":{"best":"hash","height":2,"msg":"block.import","origin":"NetworkInitialSync"},"ts":`)
-	expected3 := []byte(`{"id":1,"payload":{"authority":false,"chain":"chain","genesis_hash":"hash","implementation":"systemName","msg":"system.connected","name":"nodeName","network_id":"netID","startup_time":"startTime","version":"version"},"ts":`)
-	expected4 := []byte(`{"id":1,"payload":{"best":"0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6","finalized_hash":"0x687197c11b4cf95374159843e7f46fbcd63558db981aaef01a8bac2a44a1d6b2","finalized_height":32256,"height":32375,"msg":"system.interval","txcount":2,"used_state_cache_size":1886357},"ts":`) // nolint
+	expected1 := []byte(`{"bandwidth_download":2,"bandwidth_upload":3,"msg":"system.interval","peers":1,"ts":`)
+	expected2 := []byte(`{"best":"hash","height":2,"msg":"block.import","origin":"NetworkInitialSync","ts":`)
+	expected3 := []byte(`{"authority":false,"chain":"chain","genesis_hash":"hash","implementation":"systemName","msg":"system.connected","name":"nodeName","network_id":"netID","startup_time":"startTime","ts":`)
+	expected4 := []byte(`{"best":"0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6","finalized_hash":"0x687197c11b4cf95374159843e7f46fbcd63558db981aaef01a8bac2a44a1d6b2","finalized_height":32256,"height":32375,"msg":"system.interval","ts":`) // nolint
 
 	expected := [][]byte{expected3, expected1, expected4, expected2}
 
@@ -139,6 +140,16 @@ func TestListenerConcurrency(t *testing.T) {
 	}
 }
 
+// TestInfiniteListener starts loop that print out data received on websocket ws://localhost:8001/
+//  this can be useful to see what data is sent to telemetry server
+func TestInfiniteListener(t *testing.T) {
+	t.Skip()
+	resultCh = make(chan []byte)
+	for data := range resultCh {
+		fmt.Printf("Data %s\n", data)
+	}
+}
+
 func listen(w http.ResponseWriter, r *http.Request) {
 	c, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {