Skip to content

Track pubsub topic validator receives - F-ssv-1044 #2843

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
julienharbulot wants to merge 4 commits into
base: stage
Choose a base branch
from
Open

Track pubsub topic validator receives - F-ssv-1044 #2843

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
57d7213
Track pubsub topic validator receives
julienharbulot May 6, 2026
716103e
Namespace pubsub topic metric attribute
julienharbulot May 6, 2026
b88c35e
Merge branch 'stage' into F-ssv-1044
julienharbulot May 13, 2026
a925ed5
network/topics: clarify pubsub-received metric semantics, rename shad…
iurii-ssv Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion network/topics/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -308,7 +308,13 @@ func (ctrl *topicsCtrl) setupTopicValidator(name string) error {

opts := []pubsub.ValidatorOpt{pubsub.WithValidatorTimeout(topicValidatorTimeout)}

err = ctrl.ps.RegisterTopicValidator(name, ctrl.msgValidator.ValidatorForTopic(name), opts...)
validator := ctrl.msgValidator.ValidatorForTopic(name)
wrappedValidator := func(ctx context.Context, p peer.ID, pmsg *pubsub.Message) pubsub.ValidationResult {

@momosh-ssv momosh-ssv Jun 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might be worth a test that goes through the wrapper itself — the new test calls recordPubsubMessageReceived directly, so the wiring here stays unverified.

Registering the validator and pushing one message through it, then asserting the counter, would close that gap.

recordPubsubMessageReceived(ctx, name)
return validator(ctx, p, pmsg)
}

err = ctrl.ps.RegisterTopicValidator(name, wrappedValidator, opts...)
if err != nil {
return fmt.Errorf("could not register topic validator: %w", err)
}
Expand Down
18 changes: 18 additions & 0 deletions network/topics/observability.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package topics

import (
"context"

"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/metric"
Expand All @@ -12,6 +14,8 @@ import (
const (
observabilityName = "github.com/ssvlabs/ssv/network/topics"
observabilityNamespace = "ssv.p2p.messages"

pubsubObservabilityNamespace = "ssv.p2p.pubsub.messages"
)

var (
Expand All @@ -29,13 +33,23 @@ var (
metric.WithUnit("{message}"),
metric.WithDescription("total number of outbound(broadcasted) messages")))

pubsubMessagesReceivedCounter = metrics.New(
meter.Int64Counter(
observability.InstrumentName(pubsubObservabilityNamespace, "received"),
metric.WithUnit("{message}"),
metric.WithDescription("total number of messages received by the pubsub topic validator")))

@iurii-ssv iurii-ssv May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Description is ambiguous next to inboundMessageCounter

The description here ("total number of messages received by the pubsub topic validator") and inboundMessageCounter's description ("total number of inbound messages", line 29) sound nearly identical to anyone reading the metric registry — yet the entire point of this PR is that the two counters measure different things (pre- vs. post-validation). Tightening the wording so the distinction is visible at a glance:

Suggested change
metric.WithDescription("total number of messages received by the pubsub topic validator")))
metric.WithDescription("total number of messages delivered to the pubsub topic validator, before SSV validation runs (compare with ssv_p2p_messages_in_total for the post-validation rate)")))


msgIDHandlerBufferFallbackCounter = metrics.New(
meter.Int64Counter(
observability.InstrumentName(observabilityNamespace, "msg_id_buffer_fallback"),
metric.WithUnit("{event}"),
metric.WithDescription("total number of msg_id add operations processed synchronously because the async buffer was full")))
)

func pubsubTopicAttribute(value string) attribute.KeyValue {
return attribute.String("topic", value)
}
Comment thread
julienharbulot marked this conversation as resolved.

func messageTopicAttribute(value string) attribute.KeyValue {
return attribute.String("ssv.p2p.message.topic", value)
}
Expand All @@ -46,3 +60,7 @@ func messageTypeAttribute(value uint64) attribute.KeyValue {
Value: observability.Uint64AttributeValue(value),
}
}

func recordPubsubMessageReceived(ctx context.Context, topic string) {

@iurii-ssv iurii-ssv May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: worth a 1-line comment to flag pre-validation timing

A future reader touching the wrapper in controller.go might assume this counts only successful deliveries. Calling out the timing explicitly here means they don't have to chase the call site to find out:

Suggested change
func recordPubsubMessageReceived(ctx context.Context, topic string) {
// recordPubsubMessageReceived is called from the topic validator wrapper before the inner SSV
// validator runs, so the counter increments for every message libp2p hands to the validator
// regardless of validation outcome (accept/ignore/reject/timeout).
func recordPubsubMessageReceived(ctx context.Context, topic string) {

pubsubMessagesReceivedCounter.Add(ctx, 1, metric.WithAttributes(pubsubTopicAttribute(topic)))
}
48 changes: 48 additions & 0 deletions network/topics/observability_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
package topics

import (
"testing"

"github.com/stretchr/testify/require"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/sdk/metric"
"go.opentelemetry.io/otel/sdk/metric/metricdata"
)

func TestRecordPubsubMessageReceived(t *testing.T) {
reader := metric.NewManualReader()
provider := metric.NewMeterProvider(metric.WithReader(reader))
previousProvider := otel.GetMeterProvider()
otel.SetMeterProvider(provider)

@momosh-ssv momosh-ssv Jun 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we consider moving the provider swap into a TestMain, like protocol/v2/ssv/queue/main_test.go does?

Seems that setting the global provider mid-test leaves the package-level counters re-pointed for whatever runs after, and restoring the previous (delegating) provider in cleanup may not actually rebind them.

Setting a ManualReader-backed provider once before m.Run() would match the existing pattern and avoid the order dependence.

t.Cleanup(func() {
otel.SetMeterProvider(previousProvider)
require.NoError(t, provider.Shutdown(t.Context()))
})

const topic = "ssv.v2.42"
recordPubsubMessageReceived(t.Context(), topic)
recordPubsubMessageReceived(t.Context(), topic)

var rm metricdata.ResourceMetrics
require.NoError(t, reader.Collect(t.Context(), &rm))

for _, scopeMetrics := range rm.ScopeMetrics {
for _, metric := range scopeMetrics.Metrics {
if metric.Name != "ssv.p2p.pubsub.messages.received" {
continue
}

sum, ok := metric.Data.(metricdata.Sum[int64])

@iurii-ssv iurii-ssv May 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: loop variable metric shadows the imported metric package

From go.opentelemetry.io/otel/sdk/metric (line 8). Compiles only because the package isn't referenced inside the loop body, but it's an avoidable foot-gun if anyone later edits this block. Renaming to m:

Suggested change
for _, metric := range scopeMetrics.Metrics {
if metric.Name != "ssv.p2p.pubsub.messages.received" {
continue
}
sum, ok := metric.Data.(metricdata.Sum[int64])
for _, m := range scopeMetrics.Metrics {
if m.Name != "ssv.p2p.pubsub.messages.received" {
continue
}
sum, ok := m.Data.(metricdata.Sum[int64])

require.True(t, ok)
require.Len(t, sum.DataPoints, 1)
require.EqualValues(t, 2, sum.DataPoints[0].Value)

topicAttr, ok := sum.DataPoints[0].Attributes.Value("topic")
require.True(t, ok)
require.Equal(t, topic, topicAttr.AsString())
Comment thread
julienharbulot marked this conversation as resolved.
Outdated
return
}
}

t.Fatal("pubsub received metric was not collected")
}