fix (dot/telemetry): NoTelemetry flag stops telemetry

dot/node.go

@@ -340,9 +340,7 @@ func NewNode(cfg *Config, ks *keystore.GlobalKeystore, stopFunc func()) (*Node,
 		return nil, err
 	}
 
-	if cfg.Global.NoTelemetry {
-		return node, nil
-	}
+	telemetry.Enabled = !cfg.Global.NoTelemetry
 
 	telemetry.GetInstance().AddConnections(gd.TelemetryEndpoints)
 

dot/telemetry/telemetry.go

@@ -46,6 +46,13 @@ type Handler struct {
 	sendMessageTimeout time.Duration
 }
 
+// Instance functions that telemetry handler instance needs to implement
+type Instance interface {
+	AddConnections(conns []*genesis.TelemetryEndpoint)
+	SendMessage(msg *Message) error
+	startListening()
+}
+
 // KeyValue object to hold key value pairs used in telemetry messages
 type KeyValue struct {
 	key   string
@@ -54,13 +61,16 @@ type KeyValue struct {
 
 var (
 	once            sync.Once
-	handlerInstance *Handler
+	handlerInstance Instance
+
+	// Enabled flag to determine if telemetry is enabled
+	Enabled = true // enabled by default
 )
 
 const defaultMessageTimeout = time.Second
 
 // GetInstance singleton pattern to for accessing TelemetryHandler
-func GetInstance() *Handler { //nolint
+func GetInstance() Instance {
 	if handlerInstance == nil {
 		once.Do(
 			func() {
@@ -72,6 +82,10 @@ func GetInstance() *Handler { //nolint
 				go handlerInstance.startListening()
 			})
 	}
+	if !Enabled {
+		return &NoopHandler{}
+	}
+
 	return handlerInstance
 }
 
@@ -158,3 +172,17 @@ func msgToBytes(message Message) []byte {
 	}
 	return resB
 }
+
+// NoopHandler struct no op handling (ignoring) telemetry messages
+type NoopHandler struct {
+}
+
+func (h *NoopHandler) startListening() {}
+
+// SendMessage no op for telemetry send message function
+func (h *NoopHandler) SendMessage(msg *Message) error {
+	return nil
+}
+
+// AddConnections no op for telemetry add connections function
+func (h *NoopHandler) AddConnections(conns []*genesis.TelemetryEndpoint) {}

dot/telemetry/telemetry_test.go

@@ -139,6 +139,42 @@ func TestListenerConcurrency(t *testing.T) {
 	}
 }
 
+func TestKillInstance(t *testing.T) {
+	const qty = 1000
+	var wg sync.WaitGroup
+	wg.Add(qty)
+
+	resultCh = make(chan []byte)
+	for i := 0; i < qty; i++ {
+		if i == qty/2 {
+			Enabled = false
+		}
+		go func() {
+			GetInstance().SendMessage(NewTelemetryMessage(
+				NewKeyValue("best", "hash"),
+				NewKeyValue("height", big.NewInt(2)),
+				NewKeyValue("msg", "block.import"),
+				NewKeyValue("origin", "NetworkInitialSync")))
+			wg.Done()
+		}()
+	}
+	wg.Wait()
+	counter := 0
+	tk := time.NewTicker(time.Second * 2)
+main:
+	for {
+		select {
+		case <-tk.C:
+			break main
+		case <-resultCh:
+			counter++
+		}
+	}
+	tk.Stop()
+
+	require.LessOrEqual(t, counter, qty/2)
+}
+
 func listen(w http.ResponseWriter, r *http.Request) {
 	c, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {