telemetry: Count and report the number of duplicate proposals and MsgDigestSkipTag messages received (#4605)

Author: cce

agreement/proposalStore.go

@@ -18,8 +18,16 @@ package agreement
 
 import (
 	"fmt"
+
+	"github.com/algorand/go-algorand/util/metrics"
 )
 
+var proposalAlreadyFilledCounter = metrics.MakeCounter(
+	metrics.MetricName{Name: "algod_agreement_proposal_already_filled", Description: "Number of times a duplicate proposal payload was received before validation"})
+
+var proposalAlreadyAssembledCounter = metrics.MakeCounter(
+	metrics.MetricName{Name: "algod_agreement_proposal_already_assembled", Description: "Number of times a duplicate proposal payload was received after validation"})
+
 // An blockAssembler contains the proposal data associated with some
 // proposal-value.
 //
@@ -52,10 +60,12 @@ type blockAssembler struct {
 // an error if the pipelining operation is redundant.
 func (a blockAssembler) pipeline(p unauthenticatedProposal) (blockAssembler, error) {
 	if a.Assembled {
+		proposalAlreadyAssembledCounter.Inc(nil)
 		return a, fmt.Errorf("blockAssembler.pipeline: already assembled")
 	}
 
 	if a.Filled {
+		proposalAlreadyFilledCounter.Inc(nil)
 		return a, fmt.Errorf("blockAssembler.pipeline: already filled")
 	}
 

logging/telemetryspec/event.go

@@ -300,6 +300,8 @@ type PeerConnectionDetails struct {
 	Endpoint string `json:",omitempty"`
 	// MessageDelay is the avarage relative message delay. Not being used for incoming connection.
 	MessageDelay int64 `json:",omitempty"`
+	// DuplicateFilterCount is the number of times this peer has sent us a message hash to filter that it had already sent before.
+	DuplicateFilterCount int64
 }
 
 // CatchpointGenerationEvent event

network/wsNetwork.go

@@ -1749,9 +1749,10 @@ func (wn *WebsocketNetwork) sendPeerConnectionsTelemetryStatus() {
 	var connectionDetails telemetryspec.PeersConnectionDetails
 	for _, peer := range peers {
 		connDetail := telemetryspec.PeerConnectionDetails{
-			ConnectionDuration: uint(now.Sub(peer.createTime).Seconds()),
-			TelemetryGUID:      peer.TelemetryGUID,
-			InstanceName:       peer.InstanceName,
+			ConnectionDuration:   uint(now.Sub(peer.createTime).Seconds()),
+			TelemetryGUID:        peer.TelemetryGUID,
+			InstanceName:         peer.InstanceName,
+			DuplicateFilterCount: peer.duplicateFilterCount,
 		}
 		if peer.outgoing {
 			connDetail.Address = justHost(peer.conn.RemoteAddr().String())

network/wsPeer.go

@@ -75,6 +75,7 @@ var networkMessageQueueMicrosTotal = metrics.MakeCounter(metrics.MetricName{Name
 
 var duplicateNetworkMessageReceivedTotal = metrics.MakeCounter(metrics.DuplicateNetworkMessageReceivedTotal)
 var duplicateNetworkMessageReceivedBytesTotal = metrics.MakeCounter(metrics.DuplicateNetworkMessageReceivedBytesTotal)
+var duplicateNetworkFilterReceivedTotal = metrics.MakeCounter(metrics.DuplicateNetworkFilterReceivedTotal)
 var outgoingNetworkMessageFilteredOutTotal = metrics.MakeCounter(metrics.OutgoingNetworkMessageFilteredOutTotal)
 var outgoingNetworkMessageFilteredOutBytesTotal = metrics.MakeCounter(metrics.OutgoingNetworkMessageFilteredOutBytesTotal)
 
@@ -184,6 +185,9 @@ type wsPeer struct {
 
 	incomingMsgFilter *messageFilter
 	outgoingMsgFilter *messageFilter
+	// duplicateFilterCount counts how many times the remote peer has sent us a message hash
+	// to filter that it had already sent before.
+	duplicateFilterCount int64
 
 	processed chan struct{}
 
@@ -576,7 +580,14 @@ func (wp *wsPeer) handleFilterMessage(msg IncomingMessage) {
 	var digest crypto.Digest
 	copy(digest[:], msg.Data)
 	//wp.net.log.Debugf("add filter %v", digest)
-	wp.outgoingMsgFilter.CheckDigest(digest, true, true)
+	has := wp.outgoingMsgFilter.CheckDigest(digest, true, true)
+	if has {
+		// Count that this peer has sent us duplicate filter messages: this means it received the same
+		// large message concurrently from several peers, and then sent the filter message to us after
+		// each large message finished transferring.
+		duplicateNetworkFilterReceivedTotal.Inc(nil)
+		atomic.AddInt64(&wp.duplicateFilterCount, 1)
+	}
 }
 
 func (wp *wsPeer) writeLoopSend(msgs sendMessages) disconnectReason {

util/metrics/metrics.go

@@ -49,6 +49,8 @@ var (
 	DuplicateNetworkMessageReceivedTotal = MetricName{Name: "algod_network_duplicate_message_received_total", Description: "Total number of duplicate messages that were received from the network"}
 	// DuplicateNetworkMessageReceivedBytesTotal The total number ,in bytes, of the duplicate messages that were received from the network
 	DuplicateNetworkMessageReceivedBytesTotal = MetricName{Name: "algod_network_duplicate_message_received_bytes_total", Description: "The total number ,in bytes, of the duplicate messages that were received from the network"}
+	// DuplicateNetworkFilterReceivedTotal Total number of duplicate filter messages (tag MsgDigestSkipTag) that were received from the network
+	DuplicateNetworkFilterReceivedTotal = MetricName{Name: "algod_network_duplicate_filter_received_total", Description: "Total number of duplicate filter messages that were received from the network"}
 	// OutgoingNetworkMessageFilteredOutTotal Total number of messages that were not sent per peer request
 	OutgoingNetworkMessageFilteredOutTotal = MetricName{Name: "algod_outgoing_network_message_filtered_out_total", Description: "Total number of messages that were not sent per peer request"}
 	// OutgoingNetworkMessageFilteredOutBytesTotal Total number of bytes saved by not sending messages that were asked not to be sent by peer