Track pubsub topic validator receives - F-ssv-1044

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)
 		}

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,13 +34,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 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(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)))
+}

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")
+}