diff --git a/network/topics/controller.go b/network/topics/controller.go index be69588a57..2181eddb45 100644 --- a/network/topics/controller.go +++ b/network/topics/controller.go @@ -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 { + 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) } diff --git a/network/topics/observability.go b/network/topics/observability.go index 08ac2a5808..2c05a9327a 100644 --- a/network/topics/observability.go +++ b/network/topics/observability.go @@ -1,6 +1,8 @@ package topics import ( + "context" + "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -12,6 +14,9 @@ import ( const ( observabilityName = "github.com/ssvlabs/ssv/network/topics" observabilityNamespace = "ssv.p2p.messages" + + pubsubObservabilityNamespace = "ssv.p2p.pubsub.messages" + pubsubTopicAttributeKey = "ssv.p2p.pubsub.topic" ) var ( @@ -29,6 +34,12 @@ 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 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"), @@ -36,6 +47,10 @@ var ( 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(pubsubTopicAttributeKey, value) +} + func messageTopicAttribute(value string) attribute.KeyValue { return attribute.String("ssv.p2p.message.topic", value) } @@ -46,3 +61,10 @@ func messageTypeAttribute(value uint64) attribute.KeyValue { Value: observability.Uint64AttributeValue(value), } } + +// 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))) +} diff --git a/network/topics/observability_test.go b/network/topics/observability_test.go new file mode 100644 index 0000000000..8e9b30ce60 --- /dev/null +++ b/network/topics/observability_test.go @@ -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) + 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 _, 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(pubsubTopicAttributeKey) + require.True(t, ok) + require.Equal(t, topic, topicAttr.AsString()) + return + } + } + + t.Fatal("pubsub received metric was not collected") +}