feat: dot/telemetry: Implement basic telemetry connection (#1497)

edwardmack · web-flow · commit fcb4159bf5cc · 2021-04-07T10:21:15.000-04:00
* implement telemetry connection server

* add telemetry for import block

* setup telemetry to use node name from Global config

* init telemetry connections based on genesis.json values

* fix error message formatting

* add tests for telemetry

* implement no-telemetry cli flag

* fix typos

* cleanup error check

* create TelemetryEndpoint struct

* created connection data struct to hold connection data

* make TelemetryEndpoints a pointer reference

* add tests for interfaceToTelemetryEndpoint

* update test to fix random name
diff --git a/cmd/gossamer/config.go b/cmd/gossamer/config.go
@@ -407,6 +407,7 @@ func setDotGlobalConfig(ctx *cli.Context, tomlCfg *ctoml.Config, cfg *dot.Global
 		cfg.MetricsPort = tomlCfg.Global.MetricsPort
 	}
 
+	// TODO: generate random name if one is not assigned (see issue #1496)
 	// check --name flag and update node configuration
 	if name := ctx.GlobalString(NameFlag.Name); name != "" {
 		cfg.Name = name
@@ -442,6 +443,8 @@ func setDotGlobalConfig(ctx *cli.Context, tomlCfg *ctoml.Config, cfg *dot.Global
 		cfg.MetricsPort = uint32(metricsPort)
 	}
 
+	cfg.NoTelemetry = ctx.Bool("no-telemetry")
+
 	logger.Debug(
 		"global configuration",
 		"name", cfg.Name,
diff --git a/cmd/gossamer/config_test.go b/cmd/gossamer/config_test.go
@@ -222,6 +222,20 @@ func TestGlobalConfigFromFlags(t *testing.T) {
 				MetricsPort:    uint32(9871),
 			},
 		},
+		{
+			"Test gossamer --no-telemetry",
+			[]string{"config", "no-telemetry", "name"},
+			[]interface{}{testCfgFile.Name(), true, testCfg.Global.Name},
+			dot.GlobalConfig{
+				Name:           testCfg.Global.Name,
+				ID:             testCfg.Global.ID,
+				BasePath:       testCfg.Global.BasePath,
+				LogLvl:         log.LvlInfo,
+				PublishMetrics: testCfg.Global.PublishMetrics,
+				MetricsPort:    testCfg.Global.MetricsPort,
+				NoTelemetry:    true,
+			},
+		},
 	}
 
 	for _, c := range testcases {
diff --git a/cmd/gossamer/flags.go b/cmd/gossamer/flags.go
@@ -98,6 +98,12 @@ var (
 		Name:  "metrics-port",
 		Usage: "Set metric listening port ",
 	}
+
+	// NoTelemetryFlag stops publishing telemetry to default defined in genesis.json
+	NoTelemetryFlag = cli.BoolFlag{
+		Name:  "no-telemetry",
+		Usage: "Disable connecting to the Substrate telemetry server",
+	}
 )
 
 // Initialization-only flags
@@ -295,6 +301,9 @@ var (
 		// metrics flag
 		PublishMetricsFlag,
 		MetricsPortFlag,
+
+		// telemetry flags
+		NoTelemetryFlag,
 	}
 )
 
diff --git a/dot/config.go b/dot/config.go
@@ -50,6 +50,7 @@ type GlobalConfig struct {
 	LogLvl         log.Lvl
 	PublishMetrics bool
 	MetricsPort    uint32
+	NoTelemetry    bool
 }
 
 // LogConfig represents the log levels for individual packages
diff --git a/dot/node.go b/dot/node.go
@@ -23,21 +23,23 @@ import (
 	"os/signal"
 	"path"
 	"runtime/debug"
+	"strconv"
 	"sync"
 	"syscall"
+	"time"
 
+	"github.com/ChainSafe/chaindb"
 	gssmrmetrics "github.com/ChainSafe/gossamer/dot/metrics"
 	"github.com/ChainSafe/gossamer/dot/network"
 	"github.com/ChainSafe/gossamer/dot/state"
+	"github.com/ChainSafe/gossamer/dot/telemetry"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/genesis"
 	"github.com/ChainSafe/gossamer/lib/keystore"
 	"github.com/ChainSafe/gossamer/lib/services"
+	log "github.com/ChainSafe/log15"
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethereum/go-ethereum/metrics/prometheus"
-
-	"github.com/ChainSafe/chaindb"
-	log "github.com/ChainSafe/log15"
 )
 
 var logger = log.New("pkg", "dot")
@@ -306,6 +308,28 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 		publishMetrics(cfg)
 	}
 
+	gd, err := stateSrvc.Storage.GetGenesisData()
+	if err != nil {
+		return nil, err
+	}
+
+	if cfg.Global.NoTelemetry {
+		return node, nil
+	}
+
+	telemetry.GetInstance().AddConnections(gd.TelemetryEndpoints)
+	data := &telemetry.ConnectionData{
+		Authority:     cfg.Core.GrandpaAuthority,
+		Chain:         sysSrvc.ChainName(),
+		GenesisHash:   stateSrvc.Block.GenesisHash().String(),
+		SystemName:    sysSrvc.SystemName(),
+		NodeName:      cfg.Global.Name,
+		SystemVersion: sysSrvc.SystemVersion(),
+		NetworkID:     networkSrvc.NetworkState().PeerID,
+		StartTime:     strconv.FormatInt(time.Now().UnixNano(), 10),
+	}
+	telemetry.GetInstance().SendConnection(data)
+
 	return node, nil
 }
 
diff --git a/dot/rpc/modules/api.go b/dot/rpc/modules/api.go
@@ -90,7 +90,7 @@ type RuntimeAPI interface {
 type SystemAPI interface {
 	SystemName() string
 	SystemVersion() string
-	NodeName() string
 	Properties() map[string]interface{}
 	ChainType() string
+	ChainName() string
 }
diff --git a/dot/rpc/modules/system.go b/dot/rpc/modules/system.go
@@ -83,7 +83,7 @@ func NewSystemModule(net NetworkAPI, sys SystemAPI, core CoreAPI,
 
 // Chain returns the runtime chain
 func (sm *SystemModule) Chain(r *http.Request, req *EmptyRequest, res *string) error {
-	*res = sm.systemAPI.NodeName()
+	*res = sm.systemAPI.ChainName()
 	return nil
 }
 
@@ -161,7 +161,7 @@ func (sm *SystemModule) NodeRoles(r *http.Request, req *EmptyRequest, res *[]int
 // AccountNextIndex Returns the next valid index (aka. nonce) for given account.
 func (sm *SystemModule) AccountNextIndex(r *http.Request, req *StringRequest, res *U64Response) error {
 	if req == nil || len(req.String) == 0 {
-		return errors.New("Account address must be valid")
+		return errors.New("account address must be valid")
 	}
 	addressPubKey := crypto.PublicAddressToByteArray(common.Address(req.String))
 
diff --git a/dot/rpc/modules/system_test.go b/dot/rpc/modules/system_test.go
@@ -210,10 +210,9 @@ func (api *mockSystemAPI) SystemVersion() string {
 	return api.info.SystemVersion
 }
 
-func (api *mockSystemAPI) NodeName() string {
+func (api *mockSystemAPI) ChainName() string {
 	return api.genData.Name
 }
-
 func (api *mockSystemAPI) Properties() map[string]interface{} {
 	return nil
 }
diff --git a/dot/sync/syncer.go b/dot/sync/syncer.go
@@ -24,12 +24,12 @@ import (
 	"os"
 
 	"github.com/ChainSafe/gossamer/dot/network"
+	"github.com/ChainSafe/gossamer/dot/telemetry"
 	"github.com/ChainSafe/gossamer/dot/types"
 	"github.com/ChainSafe/gossamer/lib/blocktree"
 	"github.com/ChainSafe/gossamer/lib/common"
 	"github.com/ChainSafe/gossamer/lib/runtime"
 	rtstorage "github.com/ChainSafe/gossamer/lib/runtime/storage"
-
 	log "github.com/ChainSafe/log15"
 )
 
@@ -339,6 +339,7 @@ func (s *Service) handleBlock(block *types.Block) error {
 		}
 	} else {
 		logger.Debug("🔗 imported block", "number", block.Header.Number, "hash", block.Header.Hash())
+		telemetry.GetInstance().SendBlockImport(block.Header.Hash().String(), block.Header.Number)
 	}
 
 	// handle consensus digest for authority changes
diff --git a/dot/system/service.go b/dot/system/service.go
@@ -50,8 +50,8 @@ func (s *Service) SystemVersion() string {
 	return s.systemInfo.SystemVersion
 }
 
-// NodeName returns the nodeName (chain name)
-func (s *Service) NodeName() string {
+// ChainName returns the chain name defined in genesis.json
+func (s *Service) ChainName() string {
 	return s.genesisData.Name
 }
 
diff --git a/dot/system/service_test.go b/dot/system/service_test.go
@@ -24,10 +24,10 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestService_NodeName(t *testing.T) {
+func TestService_ChainName(t *testing.T) {
 	svc := newTestService()
 
-	name := svc.NodeName()
+	name := svc.ChainName()
 	require.Equal(t, "gssmr", name)
 }
 
diff --git a/dot/telemetry/telemetry.go b/dot/telemetry/telemetry.go
@@ -0,0 +1,123 @@
+// Copyright 2021 ChainSafe Systems (ON) Corp.
+// This file is part of gossamer.
+//
+// The gossamer library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The gossamer library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the gossamer library. If not, see <http://www.gnu.org/licenses/>.
+package telemetry
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"math/big"
+	"sync"
+	"time"
+
+	"github.com/ChainSafe/gossamer/lib/genesis"
+	"github.com/gorilla/websocket"
+	log "github.com/sirupsen/logrus"
+)
+
+// Handler struct for holding telemetry related things
+type Handler struct {
+	buf             bytes.Buffer
+	wsConn          []*websocket.Conn
+	telemetryLogger *log.Entry
+}
+
+// MyJSONFormatter struct for defining JSON Formatter
+type MyJSONFormatter struct {
+}
+
+// Format function for handling JSON formatting, this overrides default logging formatter to remove
+//  log level, line number and timestamp
+func (f *MyJSONFormatter) Format(entry *log.Entry) ([]byte, error) {
+	serialized, err := json.Marshal(entry.Data)
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal fields to JSON, %w", err)
+	}
+	return append(serialized, '\n'), nil
+}
+
+var (
+	once            sync.Once
+	handlerInstance *Handler
+)
+
+// GetInstance singleton pattern to for accessing TelemetryHandler
+func GetInstance() *Handler {
+	if handlerInstance == nil {
+		once.Do(
+			func() {
+				handlerInstance = &Handler{
+					buf: bytes.Buffer{},
+				}
+				log.SetOutput(&handlerInstance.buf)
+				log.SetFormatter(new(MyJSONFormatter))
+			})
+	}
+	return handlerInstance
+}
+
+// AddConnections adds connections to telemetry sever
+func (h *Handler) AddConnections(conns []*genesis.TelemetryEndpoint) {
+	for _, v := range conns {
+		c, _, err := websocket.DefaultDialer.Dial(v.Endpoint, nil)
+		if err != nil {
+			fmt.Printf("Error %v\n", err)
+			return
+		}
+		h.wsConn = append(h.wsConn, c)
+	}
+}
+
+// ConnectionData struct to hold connection data
+type ConnectionData struct {
+	Authority     bool
+	Chain         string
+	GenesisHash   string
+	SystemName    string
+	NodeName      string
+	SystemVersion string
+	NetworkID     string
+	StartTime     string
+}
+
+// SendConnection sends connection request message to telemetry connection
+func (h *Handler) SendConnection(data *ConnectionData) {
+	payload := log.Fields{"authority": data.Authority, "chain": data.Chain, "config": "", "genesis_hash": data.GenesisHash,
+		"implementation": data.SystemName, "msg": "system.connected", "name": data.NodeName, "network_id": data.NetworkID, "startup_time": data.StartTime,
+		"version": data.SystemVersion}
+	h.telemetryLogger = log.WithFields(log.Fields{"id": 1, "payload": payload, "ts": time.Now()})
+	h.telemetryLogger.Print()
+	h.sendTelemtry()
+}
+
+// SendBlockImport sends block imported message to telemetry connection
+func (h *Handler) SendBlockImport(bestHash string, height *big.Int) {
+	payload := log.Fields{"best": bestHash, "height": height.Int64(), "msg": "block.import", "origin": "NetworkInitialSync"}
+	h.telemetryLogger = log.WithFields(log.Fields{"id": 1, "payload": payload, "ts": time.Now()})
+	h.telemetryLogger.Print()
+	h.sendTelemtry()
+}
+
+func (h *Handler) sendTelemtry() {
+	for _, c := range h.wsConn {
+		err := c.WriteMessage(websocket.TextMessage, h.buf.Bytes())
+		if err != nil {
+			// TODO (ed) determine how to handle this error
+			fmt.Printf("ERROR connecting to telemetry %v\n", err)
+		}
+	}
+	h.buf.Reset()
+}
diff --git a/dot/telemetry/telemetry_test.go b/dot/telemetry/telemetry_test.go
diff --git a/go.mod b/go.mod
diff --git a/go.sum b/go.sum
diff --git a/lib/genesis/genesis.go b/lib/genesis/genesis.go
diff --git a/lib/genesis/genesis_test.go b/lib/genesis/genesis_test.go
diff --git a/lib/genesis/test_utils.go b/lib/genesis/test_utils.go

Original file line number	Diff line number	Diff line change
`@@ -407,6 +407,7 @@ func setDotGlobalConfig(ctx cli.Context, tomlCfg ctoml.Config, cfg *dot.Global`
`407`	`407`	`cfg.MetricsPort = tomlCfg.Global.MetricsPort`
`408`	`408`	`}`
`409`	`409`
	`410`	`+ // TODO: generate random name if one is not assigned (see issue #1496)`
`410`	`411`	`// check --name flag and update node configuration`
`411`	`412`	`if name := ctx.GlobalString(NameFlag.Name); name != "" {`
`412`	`413`	`cfg.Name = name`
`@@ -442,6 +443,8 @@ func setDotGlobalConfig(ctx cli.Context, tomlCfg ctoml.Config, cfg *dot.Global`
`442`	`443`	`cfg.MetricsPort = uint32(metricsPort)`
`443`	`444`	`}`
`444`	`445`
	`446`	`+ cfg.NoTelemetry = ctx.Bool("no-telemetry")`
	`447`	`+`
`445`	`448`	`logger.Debug(`
`446`	`449`	`"global configuration",`
`447`	`450`	`"name", cfg.Name,`
Original file line number	Diff line number	Diff line change
`@@ -98,6 +98,12 @@ var (`
`98`	`98`	`Name: "metrics-port",`
`99`	`99`	`Usage: "Set metric listening port ",`
`100`	`100`	`}`
	`101`	`+`
	`102`	`+ // NoTelemetryFlag stops publishing telemetry to default defined in genesis.json`
	`103`	`+ NoTelemetryFlag = cli.BoolFlag{`
	`104`	`+ Name: "no-telemetry",`
	`105`	`+ Usage: "Disable connecting to the Substrate telemetry server",`
	`106`	`+ }`
`101`	`107`	`)`
`102`	`108`
`103`	`109`	`// Initialization-only flags`
`@@ -295,6 +301,9 @@ var (`
`295`	`301`	`// metrics flag`
`296`	`302`	`PublishMetricsFlag,`
`297`	`303`	`MetricsPortFlag,`
	`304`	`+`
	`305`	`+ // telemetry flags`
	`306`	`+ NoTelemetryFlag,`
`298`	`307`	`}`
`299`	`308`	`)`
`300`	`309`
Original file line number	Diff line number	Diff line change
`@@ -50,6 +50,7 @@ type GlobalConfig struct {`
`50`	`50`	`LogLvl log.Lvl`
`51`	`51`	`PublishMetrics bool`
`52`	`52`	`MetricsPort uint32`
	`53`	`+ NoTelemetry bool`
`53`	`54`	`}`
`54`	`55`
`55`	`56`	`// LogConfig represents the log levels for individual packages`
Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ type RuntimeAPI interface {`
`90`	`90`	`type SystemAPI interface {`
`91`	`91`	`SystemName() string`
`92`	`92`	`SystemVersion() string`
`93`		`- NodeName() string`
`94`	`93`	`Properties() map[string]interface{}`
`95`	`94`	`ChainType() string`
	`95`	`+ ChainName() string`
`96`	`96`	`}`
Original file line number	Diff line number	Diff line change
`@@ -83,7 +83,7 @@ func NewSystemModule(net NetworkAPI, sys SystemAPI, core CoreAPI,`
`83`	`83`
`84`	`84`	`// Chain returns the runtime chain`
`85`	`85`	`func (sm SystemModule) Chain(r http.Request, req EmptyRequest, res string) error {`
`86`		`- *res = sm.systemAPI.NodeName()`
	`86`	`+ *res = sm.systemAPI.ChainName()`
`87`	`87`	`return nil`
`88`	`88`	`}`
`89`	`89`
`@@ -161,7 +161,7 @@ func (sm SystemModule) NodeRoles(r http.Request, req EmptyRequest, res []int`
`161`	`161`	`// AccountNextIndex Returns the next valid index (aka. nonce) for given account.`
`162`	`162`	`func (sm SystemModule) AccountNextIndex(r http.Request, req StringRequest, res U64Response) error {`
`163`	`163`	`if req == nil \|\| len(req.String) == 0 {`
`164`		`- return errors.New("Account address must be valid")`
	`164`	`+ return errors.New("account address must be valid")`
`165`	`165`	`}`
`166`	`166`	`addressPubKey := crypto.PublicAddressToByteArray(common.Address(req.String))`
`167`	`167`
Original file line number	Diff line number	Diff line change
`@@ -210,10 +210,9 @@ func (api *mockSystemAPI) SystemVersion() string {`
`210`	`210`	`return api.info.SystemVersion`
`211`	`211`	`}`
`212`	`212`
`213`		`-func (api *mockSystemAPI) NodeName() string {`
	`213`	`+func (api *mockSystemAPI) ChainName() string {`
`214`	`214`	`return api.genData.Name`
`215`	`215`	`}`
`216`		`-`
`217`	`216`	`func (api *mockSystemAPI) Properties() map[string]interface{} {`
`218`	`217`	`return nil`
`219`	`218`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,12 +24,12 @@ import (`
`24`	`24`	`"os"`
`25`	`25`
`26`	`26`	`"github.com/ChainSafe/gossamer/dot/network"`
	`27`	`+ "github.com/ChainSafe/gossamer/dot/telemetry"`
`27`	`28`	`"github.com/ChainSafe/gossamer/dot/types"`
`28`	`29`	`"github.com/ChainSafe/gossamer/lib/blocktree"`
`29`	`30`	`"github.com/ChainSafe/gossamer/lib/common"`
`30`	`31`	`"github.com/ChainSafe/gossamer/lib/runtime"`
`31`	`32`	`rtstorage "github.com/ChainSafe/gossamer/lib/runtime/storage"`
`32`		`-`
`33`	`33`	`log "github.com/ChainSafe/log15"`
`34`	`34`	`)`
`35`	`35`
`@@ -339,6 +339,7 @@ func (s Service) handleBlock(block types.Block) error {`
`339`	`339`	`}`
`340`	`340`	`} else {`
`341`	`341`	`logger.Debug("🔗 imported block", "number", block.Header.Number, "hash", block.Header.Hash())`
	`342`	`+ telemetry.GetInstance().SendBlockImport(block.Header.Hash().String(), block.Header.Number)`
`342`	`343`	`}`
`343`	`344`
`344`	`345`	`// handle consensus digest for authority changes`
Original file line number	Diff line number	Diff line change
`@@ -50,8 +50,8 @@ func (s *Service) SystemVersion() string {`
`50`	`50`	`return s.systemInfo.SystemVersion`
`51`	`51`	`}`
`52`	`52`
`53`		`-// NodeName returns the nodeName (chain name)`
`54`		`-func (s *Service) NodeName() string {`
	`53`	`+// ChainName returns the chain name defined in genesis.json`
	`54`	`+func (s *Service) ChainName() string {`
`55`	`55`	`return s.genesisData.Name`
`56`	`56`	`}`
`57`	`57`