feat(telemetry): send telemetry messages when GRANDPA receieves commit or vote messages

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 (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

@@ -16,11 +16,14 @@ import (
 
 // telemetry message types
 const (
+	notifyFinalizedMsg           = "notify.finalized"
+	blockImportMsg               = "block.import"
+	systemNetworkStateMsg        = "system.network_state"
 	systemConnectedMsg           = "system.connected"
 	systemIntervalMsg            = "system.interval"
-	systemNetworkStateMsg        = "system.network_state"
-	blockImportMsg               = "block.import"
-	notifyFinalizedMsg           = "notify.finalized"
+	afgReceivedPrecommitMsg      = "afg.received_precommit"
+	afgReceivedPrevoteMsg        = "afg.received_prevote"
+	afgReceivedCommitMsg         = "afg.received_commit"
 	txPoolImportMsg              = "txpool.import"
 	preparedBlockForProposingMsg = "prepared_block_for_proposing"
 )

dot/telemetry/telemetry_test.go

@@ -53,6 +53,9 @@ func TestHandler_SendMulti(t *testing.T) {
 		[]byte(`{"best":"0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6","height":"32375","msg":"notify.finalized","ts":`),
 		[]byte(`{"hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","msg":"prepared_block_for_proposing","number":"1","ts":`),
 		[]byte(`{"future":2,"msg":"txpool.import","ready":1,"ts":`),
+		[]byte(`{"contains_precommits_signed_by":[],"msg":"afg.received_commit","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`),
+		[]byte(`{"msg":"afg.received_precommit","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`),
+		[]byte(`{"msg":"afg.received_prevote","target_hash":"0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c","target_number":"1","ts":`),
 	}
 
 	messages := []Message{
@@ -76,6 +79,10 @@ func TestHandler_SendMulti(t *testing.T) {
 			common.MustHexToHash("0x687197c11b4cf95374159843e7f46fbcd63558db981aaef01a8bac2a44a1d6b2"),
 		),
 
+		NewAfgReceivedCommitTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", []string{}),
+		NewAfgReceivedPrecommitTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", ""),
+		NewAfgReceivedPrevoteTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1", ""),
+
 		NewNotifyFinalizedTM(common.MustHexToHash("0x07b749b6e20fd5f1159153a2e790235018621dd06072a62bcd25e8576f6ff5e6"), "32375"),
 		NewPreparedBlockForProposingTM(common.MustHexToHash("0x5814aec3e28527f81f65841e034872f3a30337cf6c33b2d258bba6071e37e27c"), "1"),
 	}

lib/grandpa/grandpa.go

@@ -475,6 +475,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) {
@@ -518,9 +519,10 @@ func (s *Service) sendVoteMessage(stage Subround, msg *VoteMessage, roundComplet
 
 		if err := s.sendMessage(msg); err != nil {
 			logger.Warnf("could not send message for stage %s: %s", stage, err)
+		} else {
+			logger.Tracef("sent vote message for stage %s: %s", stage, msg.Message)
 		}
 
-		logger.Tracef("sent vote message for stage %s: %s", stage, msg.Message)
 		select {
 		case <-roundComplete:
 			return

lib/grandpa/message_handler.go

@@ -11,6 +11,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"
@@ -91,7 +92,24 @@ func (h *MessageHandler) handleNeighbourMessage(msg *NeighbourMessage) error {
 }
 
 func (h *MessageHandler) handleCommitMessage(msg *CommitMessage) error {
-	logger.Debugf("received commit message %v", msg)
+	logger.Debugf("received commit message, msg: %+v", 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.Debugf("problem sending afg.received_commit telemetry message, err: %s", err)
+	}
+
 	if has, _ := h.blockState.HasFinalisedBlock(msg.Round, h.grandpa.state.setID); has {
 		return nil
 	}

lib/grandpa/vote_message.go

@@ -7,7 +7,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"
@@ -36,6 +38,31 @@ func (s *Service) receiveMessages(ctx context.Context) {
 			logger.Tracef("received vote message %v from %s", msg.msg, 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.Debugf("problem sending afg.received_prevote telemetry message, err: %s", 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.Debugf("problem sending afg.received_precommit telemetry message err: %s", err)
+				}
+			}
+
 			v, err := s.validateMessage(msg.from, vm)
 			if err != nil {
 				logger.Debugf("failed to validate vote message %v: %s", vm, err)