Add metric for message processing latency

Fix #1839

internal/events/eventer.go

@@ -35,6 +35,8 @@ import (
 	promgo "github.com/prometheus/client_golang/prometheus"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/metric"
 
 	"github.com/stacklok/minder/internal/config"
 )
@@ -48,6 +50,7 @@ const (
 
 	GoChannelDriver = "go-channel"
 	SQLDriver       = "sql"
+	BuiltKey        = "built_at"
 )
 
 const (
@@ -98,10 +101,16 @@ type Eventer struct {
 	// webhookSubscriber will subscribe to the webhook topic and handle incoming events
 	webhookSubscriber message.Subscriber
 	// TODO: We'll have a Final publisher that will publish to the final topic
+	metrics *messageMetrics
 
 	closer driverCloser
 }
 
+type messageMetrics struct {
+	// message processing time duration histogram
+	messageProcessingTimeHistogram metric.Int64Histogram
+}
+
 var _ Registrar = (*Eventer)(nil)
 var _ message.Publisher = (*Eventer)(nil)
 
@@ -149,6 +158,15 @@ func Setup(ctx context.Context, cfg *config.EventConfig) (*Eventer, error) {
 		return nil, fmt.Errorf("failed to decorate subscriber: %w", err)
 	}
 
+	meter := otel.Meter("eventer")
+	histogram, err := meter.Int64Histogram("messages.processing_time",
+		metric.WithDescription("Duration between a message being enqueued and dequeued for processing"),
+		metric.WithUnit("ms"),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create message processing histogram: %w", err)
+	}
+
 	return &Eventer{
 		router:            router,
 		webhookPublisher:  pubWithMetrics,
@@ -161,6 +179,9 @@ func Setup(ctx context.Context, cfg *config.EventConfig) (*Eventer, error) {
 			// driver close
 			cl()
 		},
+		metrics: &messageMetrics{
+			messageProcessingTimeHistogram: histogram,
+		},
 	}, nil
 }
 
@@ -261,6 +282,7 @@ func (e *Eventer) Publish(topic string, messages ...*message.Message) error {
 				"topic":        topic,
 				"handler":      details.Name(),
 			})
+			msg.Metadata.Set(BuiltKey, time.Now().Format(time.RFC3339))
 		}
 	}
 
@@ -280,6 +302,12 @@ func (e *Eventer) Register(
 		topic,
 		e.webhookSubscriber,
 		func(msg *message.Message) error {
+			if builtAt := msg.Metadata.Get(BuiltKey); builtAt != "" {
+				if parsedTime, err := time.Parse(time.RFC3339, builtAt); err == nil {
+					processingTime := time.Since(parsedTime)
+					e.metrics.messageProcessingTimeHistogram.Record(msg.Context(), processingTime.Milliseconds())
+				}
+			}
 			if err := handler(msg); err != nil {
 				e.router.Logger().Error("Found error handling message", err, watermill.LogFields{
 					"message_uuid": msg.UUID,

internal/events/eventer_test.go

@@ -71,15 +71,15 @@ func TestEventer(t *testing.T) {
 	}{
 		{
 			name:    "single topic",
-			publish: []eventPair{{"a", &message.Message{}}},
+			publish: []eventPair{{"a", &message.Message{Metadata: map[string]string{}}}},
 			want:    map[string][]message.Message{"a": {}},
 			consumers: []fakeConsumer{
 				{topics: []string{"a"}},
 			},
 		},
 		{
 			name:    "two subscribers",
-			publish: []eventPair{{"a", &message.Message{}}, {"b", &message.Message{}}, {"a", &message.Message{}}},
+			publish: []eventPair{{"a", &message.Message{Metadata: map[string]string{}}}, {"b", &message.Message{Metadata: map[string]string{}}}, {"a", &message.Message{Metadata: map[string]string{}}}},
 			want: map[string][]message.Message{
 				"a": {{}, {}},
 				"b": {{}},
@@ -91,7 +91,7 @@ func TestEventer(t *testing.T) {
 		},
 		{
 			name:    "two subscribers to topic",
-			publish: []eventPair{{"a", &message.Message{}}, {"b", &message.Message{}}},
+			publish: []eventPair{{"a", &message.Message{Metadata: map[string]string{}}}, {"b", &message.Message{Metadata: map[string]string{}}}},
 			want: map[string][]message.Message{
 				"a":     {{}},
 				"b":     {{}},