diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index e2bb47466c..886b18249e 100644 --- a/pkg/metrics/metrics.go +++ b/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] } diff --git a/pkg/node/node.go b/pkg/node/node.go index c9c946a51b..561407090a 100644 --- a/pkg/node/node.go +++ b/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)) } } }