Metrics report fsm channel length (#1723)

* Added a channel monitoring metric

* Fix issue.

* Add 'lenProvider' interface.

---------

Co-authored-by: Nikolay Eskov <mr.eskov1@yandex.ru>

Author: esuwu

pkg/metrics/metrics.go

@@ -22,14 +22,15 @@ const (
 	reportInterval = time.Second
 	bufferSize     = 2000
 
-	eventInv      = "Inv"
-	eventReceived = "Received"
-	eventApplied  = "Applied"
-	eventAppended = "Appended"
-	eventDeclined = "Declined"
-	eventMined    = "Mined"
-	eventScore    = "Score"
-	eventUtx      = "Utx"
+	eventInv           = "Inv"
+	eventReceived      = "Received"
+	eventApplied       = "Applied"
+	eventAppended      = "Appended"
+	eventDeclined      = "Declined"
+	eventMined         = "Mined"
+	eventScore         = "Score"
+	eventUtx           = "Utx"
+	eventFSMChannelLen = "FSMChannelLength"
 )
 
 /*
@@ -220,6 +221,15 @@ func Utx(utxCount int) {
 	reportUtx(t, f)
 }
 
+func FSMChannelLength(length int) {
+	if rep == nil {
+		return
+	}
+	t := emptyTags().node().withEvent(eventFSMChannelLen)
+	f := emptyFields().withChannelLength(length)
+	reportFSMChannelLength(t, f)
+}
+
 type tags map[string]string
 
 func emptyTags() tags {
@@ -345,6 +355,11 @@ func (f fields) withHeight(height proto.Height) fields {
 	return f
 }
 
+func (f fields) withChannelLength(chLength int) fields {
+	f["channel-size"] = chLength
+	return f
+}
+
 type reporter struct {
 	c         influx.Client
 	id        int
@@ -470,6 +485,15 @@ func reportUtx(t tags, f fields) {
 	rep.in <- p
 }
 
+func reportFSMChannelLength(t tags, f fields) {
+	p, err := influx.NewPoint("fsm-channel", t, f, time.Now())
+	if err != nil {
+		zap.S().Warnf("Failed to create metrics point 'fsm-channel': %v", err)
+		return
+	}
+	rep.in <- p
+}
+
 func shortID(id proto.BlockID) string {
 	return id.String()[:6]
 }

pkg/node/node.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
+	"github.com/wavesplatform/gowaves/pkg/metrics"
 	"github.com/wavesplatform/gowaves/pkg/node/fsm"
 	"github.com/wavesplatform/gowaves/pkg/node/fsm/tasks"
 	"github.com/wavesplatform/gowaves/pkg/node/messages"
@@ -155,8 +156,10 @@ func (a *Node) Run(
 	ctx context.Context, p peer.Parent, internalMessageCh <-chan messages.InternalMessage,
 	networkMsgCh <-chan network.InfoMessage, syncPeer *network.SyncPeer,
 ) {
+	protoMessagesChan := chanLenProvider[peer.ProtoMessage](p.MessageCh)
+
 	go a.runOutgoingConnections(ctx)
-	go a.runInternalMetrics(ctx, p.MessageCh)
+	go a.runInternalMetrics(ctx, protoMessagesChan)
 	go a.runIncomingConnections(ctx)
 
 	tasksCh := make(chan tasks.AsyncTask, 10)
@@ -228,20 +231,25 @@ func (a *Node) runIncomingConnections(ctx context.Context) {
 	}
 }
 
-func (a *Node) runInternalMetrics(ctx context.Context, ch chan peer.ProtoMessage) {
+type lenProvider interface {
+	Len() int
+}
+
+type chanLenProvider[T any] <-chan T
+
+func (c chanLenProvider[T]) Len() int { return len(c) }
+
+func (a *Node) runInternalMetrics(ctx context.Context, protoMessagesChan lenProvider) {
+	ticker := time.NewTicker(metricInternalChannelSizeUpdateInterval)
+	defer ticker.Stop()
 	for {
-		timer := time.NewTimer(metricInternalChannelSizeUpdateInterval)
 		select {
 		case <-ctx.Done():
-			if !timer.Stop() {
-				select {
-				case <-timer.C:
-				default:
-				}
-			}
 			return
-		case <-timer.C:
-			metricInternalChannelSize.Set(float64(len(ch)))
+		case <-ticker.C:
+			l := protoMessagesChan.Len()
+			metrics.FSMChannelLength(l)
+			metricInternalChannelSize.Set(float64(l))
 		}
 	}
 }