telemetry when GRANDPA receieve commit or vote messages

Send `afg.received_commit` when grandpa receives a commit message.
Send `afg.received_precommit` or `afg.received_prevote` when grandpa
receives a vote message

Closes #1840
Closes #1839
Closes #1838

dot/telemetry/afg_received.go

@@ -0,0 +1,82 @@
+// 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 "github.com/ChainSafe/gossamer/lib/common"
+
+// AfG ("Al's Finality Gadget") is synonymous with GRANDPA.
+
+type afgReceivedTM struct {
+	TargetHash   common.Hash `json:"target_hash"`
+	TargetNumber string      `json:"target_number"`
+	Voter        string      `json:"voter"`
+}
+
+// afgReceivedPrecommitTM holds `afg.received_precommit` telemetry message which is
+// supposed to be sent when grandpa client receives a precommit.
+type afgReceivedPrecommitTM afgReceivedTM
+
+// NewAfgReceivedPrecommitTM gets a new afgReceivedPrecommitTM struct.
+func NewAfgReceivedPrecommitTM(targetHash common.Hash, targetNumber, voter string) Message {
+	return &afgReceivedPrecommitTM{
+		TargetHash:   targetHash,
+		TargetNumber: targetNumber,
+		Voter:        voter,
+	}
+}
+
+func (afgReceivedPrecommitTM) messageType() string {
+	return afgReceivedPrecommitMsg
+}
+
+// afgReceivedPrevoteTM holds `afg.received_prevote` telemetry message which is
+// supposed to be sent when grandpa client receives a prevote.
+type afgReceivedPrevoteTM afgReceivedTM
+
+// NewAfgReceivedPrevoteTM gets a new afgReceivedPrevoteTM struct.
+func NewAfgReceivedPrevoteTM(targetHash common.Hash, targetNumber, voter string) Message {
+	return &afgReceivedPrevoteTM{
+		TargetHash:   targetHash,
+		TargetNumber: targetNumber,
+		Voter:        voter,
+	}
+}
+
+func (t afgReceivedPrevoteTM) messageType() string {
+	return afgReceivedPrevoteMsg
+}
+
+// afgReceivedCommitTM holds `afg.received_commit` telemetry message which is
+// supposed to be sent when grandpa client receives a commit.
+type afgReceivedCommitTM struct {
+	TargetHash                 common.Hash `json:"target_hash"`
+	TargetNumber               string      `json:"target_number"`
+	ContainsPrecommitsSignedBy []string    `json:"contains_precommits_signed_by"`
+}
+
+// NewAfgReceivedCommitTM gets a new afgReceivedCommitTM struct.
+func NewAfgReceivedCommitTM(targetHash common.Hash, targetNumber string, containsPrecommitsSignedBy []string) Message {
+	return &afgReceivedCommitTM{
+		TargetHash:                 targetHash,
+		TargetNumber:               targetNumber,
+		ContainsPrecommitsSignedBy: containsPrecommitsSignedBy,
+	}
+}
+
+func (afgReceivedCommitTM) messageType() string {
+	return afgReceivedCommitMsg
+}

dot/telemetry/telemetry.go

@@ -29,11 +29,14 @@ import (
 
 // telemetry message types
 const (
-	notifyFinalizedMsg    = "notify.finalized"
-	blockImportMsg        = "block.import"
-	systemNetworkStateMsg = "system.network_state"
-	systemConnectedMsg    = "system.connected"
-	systemIntervalMsg     = "system.interval"
+	notifyFinalizedMsg      = "notify.finalized"
+	blockImportMsg          = "block.import"
+	systemNetworkStateMsg   = "system.network_state"
+	systemConnectedMsg      = "system.connected"
+	systemIntervalMsg       = "system.interval"
+	afgReceivedPrecommitMsg = "afg.received_precommit"
+	afgReceivedPrevoteMsg   = "afg.received_prevote"
+	afgReceivedCommitMsg    = "afg.received_commit"
 )
 
 type telemetryConnection struct {

dot/telemetry/telemetry_test.go

@@ -43,7 +43,7 @@ func TestMain(m *testing.M) {
 
 func TestHandler_SendMulti(t *testing.T) {
 	var wg sync.WaitGroup
-	wg.Add(6)
+	wg.Add(9)
 
 	resultCh = make(chan []byte)
 
@@ -92,6 +92,21 @@ func TestHandler_SendMulti(t *testing.T) {
 		wg.Done()
 	}()
 
+	go func() {
+		GetInstance().SendMessage(NewAfgReceivedCommitTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", []string{}))
+		wg.Done()
+	}()
+
+	go func() {
+		GetInstance().SendMessage(NewAfgReceivedPrecommitTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", ""))
+		wg.Done()
+	}()
+
+	go func() {
+		GetInstance().SendMessage(NewAfgReceivedPrevoteTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", ""))
+		wg.Done()
+	}()
+
 	wg.Wait()
 
 	expected1 := []byte(`{"authority":false,"chain":"chain","genesis_hash":"0x91b171bb158e2d3848fa23a9f1c25182fb8e20313b2c1eb49219da7a70ce90c3","implementation":"systemName","msg":"system.connected","name":"nodeName","network_id":"netID","startup_time":"startTime","ts":`)
@@ -100,13 +115,15 @@ func TestHandler_SendMulti(t *testing.T) {
 	expected4 := []byte(`{"best":"0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6","finalized_hash":"0x687197c11b4cf95374159843e7f46fbcd63558db981aaef01a8bac2a44a1d6b2","finalized_height":32256,"height":32375,"msg":"system.interval","ts":`) // nolint
 	expected5 := []byte(`{"best":"0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6","height":"32375","msg":"notify.finalized","ts":`)
 	expected6 := []byte(`{"hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","msg":"prepared_block_for_proposing","number":"1","ts":`)
-
-	expected := [][]byte{expected1, expected3, expected4, expected5, expected2, expected6}
+	expected7 := []byte(`{"contains_precommits_signed_by":[],"msg":"afg.received_commit","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`)
+	expected8 := []byte(`{"msg":"afg.received_precommit","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`)
+	expected9 := []byte(`{"msg":"afg.received_prevote","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`)
+	expected := [][]byte{expected1, expected3, expected4, expected5, expected2, expected7, expected6, expected8, expected9}
 
 	var actual [][]byte
 	for data := range resultCh {
 		actual = append(actual, data)
-		if len(actual) == 6 {
+		if len(actual) == 9 {
 			break
 		}
 	}

lib/grandpa/grandpa.go

@@ -487,6 +487,7 @@ func (s *Service) playGrandpaRound() error {
 	go s.sendVoteMessage(prevote, vm, roundComplete)
 
 	logger.Debug("receiving pre-commit messages...")
+	// through goroutine s.receiveMessages(ctx)
 	time.Sleep(s.interval)
 
 	if s.paused.Load().(bool) {
@@ -530,9 +531,10 @@ func (s *Service) sendVoteMessage(stage Subround, msg *VoteMessage, roundComplet
 
 		if err := s.sendMessage(msg); err != nil {
 			logger.Warn("could not send message", "stage", stage, "error", err)
+		} else {
+			logger.Trace("sent vote message", "stage", stage, "vote", msg)
 		}
 
-		logger.Trace("sent vote message", "stage", stage, "vote", msg)
 		select {
 		case <-roundComplete:
 			return

lib/grandpa/message_handler.go

@@ -24,6 +24,7 @@ import (
 	"reflect"
 
 	"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"
@@ -105,6 +106,23 @@ func (h *MessageHandler) handleNeighbourMessage(msg *NeighbourMessage) error {
 
 func (h *MessageHandler) handleCommitMessage(msg *CommitMessage) error {
 	logger.Debug("received commit message", "msg", msg)
+
+	contains_precommits_signed_by := make([]string, len(msg.AuthData))
+	for i, authData := range msg.AuthData {
+		contains_precommits_signed_by[i] = authData.AuthorityID.String()
+	}
+
+	err := telemetry.GetInstance().SendMessage(
+		telemetry.NewAfgReceivedCommitTM(
+			msg.Vote.Hash,
+			fmt.Sprintf("%d", msg.Vote.Number),
+			contains_precommits_signed_by,
+		),
+	)
+	if err != nil {
+		logger.Debug("problem sending afg.received_commit telemetry message", "err", err)
+	}
+
 	if has, _ := h.blockState.HasFinalisedBlock(msg.Round, h.grandpa.state.setID); has {
 		return nil
 	}

lib/grandpa/vote_message.go

@@ -20,7 +20,9 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 
+	"github.com/ChainSafe/gossamer/dot/telemetry"
 	"github.com/ChainSafe/gossamer/lib/blocktree"
 	"github.com/ChainSafe/gossamer/lib/crypto/ed25519"
 	"github.com/ChainSafe/gossamer/pkg/scale"
@@ -49,6 +51,31 @@ func (s *Service) receiveMessages(ctx context.Context) {
 			logger.Trace("received vote message", "msg", msg.msg, "from", msg.from)
 			vm := msg.msg
 
+			switch vm.Message.Stage {
+			case prevote:
+				err := telemetry.GetInstance().SendMessage(
+					telemetry.NewAfgReceivedPrevoteTM(
+						vm.Message.Hash,
+						fmt.Sprintf("%d", vm.Message.Number),
+						vm.Message.AuthorityID.String(),
+					),
+				)
+				if err != nil {
+					logger.Debug("problem sending afg.received_prevote telemetry message", "err", err)
+				}
+			case precommit:
+				err := telemetry.GetInstance().SendMessage(
+					telemetry.NewAfgReceivedPrecommitTM(
+						vm.Message.Hash,
+						fmt.Sprintf("%d", vm.Message.Number),
+						vm.Message.AuthorityID.String(),
+					),
+				)
+				if err != nil {
+					logger.Debug("problem sending afg.received_precommit telemetry message", "err", err)
+				}
+			}
+
 			v, err := s.validateMessage(msg.from, vm)
 			if err != nil {
 				logger.Debug("failed to validate vote message", "message", vm, "error", err)