From 1c720e70aea797cef0b28f9f9155e8f7c92a3824 Mon Sep 17 00:00:00 2001 From: esuwu Date: Mon, 16 Jun 2025 16:11:41 +0300 Subject: [PATCH 1/3] Added a channel monitoring metric --- pkg/metrics/metrics.go | 38 ++++++++++++++++++++++++++++++++------ pkg/node/node.go | 10 ++++++++++ 2 files changed, 42 insertions(+), 6 deletions(-) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index e89fdf8666..a983b4be95 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -23,12 +23,15 @@ const ( reportInterval = time.Second bufferSize = 2000 - eventInv = "Inv" - eventReceived = "Received" - eventApplied = "Applied" - eventAppended = "Appended" - eventDeclined = "Declined" - eventMined = "Mined" + eventInv = "Inv" + eventReceived = "Received" + eventApplied = "Applied" + eventAppended = "Appended" + eventDeclined = "Declined" + eventMined = "Mined" + eventScore = "Score" + eventUtx = "Utx" + eventFSMChannelLen = "FSMChannelLength" ) /* @@ -252,6 +255,15 @@ func FSMScore(fsm string, score *proto.Score, source string) { reportFSM(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 { @@ -428,6 +440,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 @@ -553,6 +570,15 @@ func reportFSM(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..f5c0c2aaff 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" @@ -151,12 +152,21 @@ func (a *Node) logErrors(err error) { } } +func monitorChanLength(messageCh <-chan peer.ProtoMessage) { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + for range ticker.C { + metrics.FSMChannelLength(len(messageCh)) + } +} + func (a *Node) Run( ctx context.Context, p peer.Parent, internalMessageCh <-chan messages.InternalMessage, networkMsgCh <-chan network.InfoMessage, syncPeer *network.SyncPeer, ) { go a.runOutgoingConnections(ctx) go a.runInternalMetrics(ctx, p.MessageCh) + go monitorChanLength(p.MessageCh) go a.runIncomingConnections(ctx) tasksCh := make(chan tasks.AsyncTask, 10) From ed6a78cea08ed1112beb4399a846bad1ae2bfa58 Mon Sep 17 00:00:00 2001 From: Nikolay Eskov Date: Thu, 19 Jun 2025 02:54:58 +0300 Subject: [PATCH 2/3] Fix issue. --- pkg/node/node.go | 24 ++++++------------------ 1 file changed, 6 insertions(+), 18 deletions(-) diff --git a/pkg/node/node.go b/pkg/node/node.go index f5c0c2aaff..843e1b0afa 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -152,21 +152,12 @@ func (a *Node) logErrors(err error) { } } -func monitorChanLength(messageCh <-chan peer.ProtoMessage) { - ticker := time.NewTicker(time.Second) - defer ticker.Stop() - for range ticker.C { - metrics.FSMChannelLength(len(messageCh)) - } -} - func (a *Node) Run( ctx context.Context, p peer.Parent, internalMessageCh <-chan messages.InternalMessage, networkMsgCh <-chan network.InfoMessage, syncPeer *network.SyncPeer, ) { go a.runOutgoingConnections(ctx) go a.runInternalMetrics(ctx, p.MessageCh) - go monitorChanLength(p.MessageCh) go a.runIncomingConnections(ctx) tasksCh := make(chan tasks.AsyncTask, 10) @@ -239,19 +230,16 @@ func (a *Node) runIncomingConnections(ctx context.Context) { } func (a *Node) runInternalMetrics(ctx context.Context, ch chan peer.ProtoMessage) { + 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 := len(ch) + metrics.FSMChannelLength(l) + metricInternalChannelSize.Set(float64(l)) } } } From b6ae632361228c0094db7d36ba04b22c8f387b06 Mon Sep 17 00:00:00 2001 From: Nikolay Eskov Date: Thu, 19 Jun 2025 02:59:29 +0300 Subject: [PATCH 3/3] Add 'lenProvider' interface. --- pkg/node/node.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/pkg/node/node.go b/pkg/node/node.go index 843e1b0afa..561407090a 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -156,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) @@ -229,7 +231,15 @@ 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 { @@ -237,7 +247,7 @@ func (a *Node) runInternalMetrics(ctx context.Context, ch chan peer.ProtoMessage case <-ctx.Done(): return case <-ticker.C: - l := len(ch) + l := protoMessagesChan.Len() metrics.FSMChannelLength(l) metricInternalChannelSize.Set(float64(l)) }