HYPERFLEET-469: Add Prometheus metrics with hyperfleet_broker_ prefix

Add MetricsRecorder that instruments publisher and subscriber with
Prometheus counters and histograms following the HyperFleet Metrics
Standard:

- hyperfleet_broker_messages_consumed_total
- hyperfleet_broker_messages_published_total
- hyperfleet_broker_errors_total (conversion, handler, publish)
- hyperfleet_broker_message_duration_seconds

MetricsRecorder is a required parameter in NewPublisher and
NewSubscriber constructors. Passing nil returns an error.

Author: rafabene

README.md

@@ -14,6 +14,7 @@ The current implementation uses [Watermill](https://github.com/ThreeDotsLabs/wat
 - **Worker Pools**: Configurable parallel message processing for subscribers
 - **Subscription Management**: Flexible subscription IDs for load balancing (shared subscriptions) or fanout (separate subscriptions)
 - **Health Checks**: Built-in `Health()` method on `Publisher` for readiness probes (per [HyperFleet Health Endpoints standard](https://github.com/openshift-hyperfleet/architecture/blob/main/hyperfleet/standards/health-endpoints.md))
+- **Prometheus Metrics**: Built-in instrumentation with `MetricsRecorder` (per [HyperFleet Metrics Standard](https://github.com/openshift-hyperfleet/architecture/blob/main/hyperfleet/standards/metrics.md))
 - **Simple API**: Clean, easy-to-use interface that hides Watermill complexity
 
 ## Installation
@@ -48,17 +49,20 @@ This interface matches the HyperFleet adapter logger interface, ensuring consist
 ### Usage Examples
 
 ```go
+// Create metrics recorder (required)
+metrics := broker.NewMetricsRecorder("my-component", "v1.0.0", prometheus.DefaultRegisterer)
+
 // Use the default logger
 appLogger := logger.NewTestLogger()
-publisher, err := broker.NewPublisher(appLogger)
+publisher, err := broker.NewPublisher(appLogger, metrics)
 
 // Use JSON format for the default logger
 appLogger := logger.NewTestLogger(logger.WithFormat(logger.FormatJSON))
-publisher, err := broker.NewPublisher(appLogger)
+publisher, err := broker.NewPublisher(appLogger, metrics)
 
 // Use your own logger implementation with config
 myLogger := createMyApplicationLogger()
-publisher, err := broker.NewPublisher(myLogger, config)
+publisher, err := broker.NewPublisher(myLogger, metrics, config)
 ```
 
 ## Quick Start
@@ -78,12 +82,14 @@ import (
     "github.com/cloudevents/sdk-go/v2/event"
     "github.com/openshift-hyperfleet/hyperfleet-broker/broker"
     "github.com/openshift-hyperfleet/hyperfleet-broker/pkg/logger"
+    "github.com/prometheus/client_golang/prometheus"
 )
 
 func main() {
-    // Create logger and publisher
+    // Create logger, metrics, and publisher
     appLogger := logger.NewTestLogger()
-    publisher, err := broker.NewPublisher(appLogger)
+    metrics := broker.NewMetricsRecorder("example-publisher", "v1.0.0", prometheus.DefaultRegisterer)
+    publisher, err := broker.NewPublisher(appLogger, metrics)
     if err != nil {
         log.Fatalf("Failed to create publisher: %v", err)
     }
@@ -148,15 +154,17 @@ import (
     "github.com/cloudevents/sdk-go/v2/event"
     "github.com/openshift-hyperfleet/hyperfleet-broker/broker"
     "github.com/openshift-hyperfleet/hyperfleet-broker/pkg/logger"
+    "github.com/prometheus/client_golang/prometheus"
 )
 
 func main() {
-    // Create logger and subscriber with subscription ID
+    // Create logger, metrics, and subscriber with subscription ID
     // Subscribers with the same subscription ID share messages (load balancing)
     // Subscribers with different IDs receive all messages (fanout)
     appLogger := logger.NewTestLogger()
+    metrics := broker.NewMetricsRecorder("example-subscriber", "v1.0.0", prometheus.DefaultRegisterer)
     subscriptionID := "shared-subscription"
-    subscriber, err := broker.NewSubscriber(appLogger, subscriptionID)
+    subscriber, err := broker.NewSubscriber(appLogger, subscriptionID, metrics)
     if err != nil {
         log.Fatalf("Failed to create subscriber: %v", err)
     }
@@ -437,18 +445,55 @@ You can also provide configuration programmatically using a map:
 
 ```go
 appLogger := logger.NewTestLogger()
+metrics := broker.NewMetricsRecorder("my-component", "v1.0.0", prometheus.DefaultRegisterer)
 configMap := map[string]string{
     "broker.type": "rabbitmq",
     "broker.rabbitmq.url": "amqp://user:pass@localhost:5672/",
     "subscriber.parallelism": "5",
 }
 
-publisher, err := broker.NewPublisher(appLogger, configMap)
-subscriber, err := broker.NewSubscriber(appLogger, "my-subscription", configMap)
+publisher, err := broker.NewPublisher(appLogger, metrics, configMap)
+subscriber, err := broker.NewSubscriber(appLogger, "my-subscription", metrics, configMap)
 ```
 
 </details>
 
+## Prometheus Metrics
+
+The library provides Prometheus metrics via `MetricsRecorder`. All metrics follow the [HyperFleet Metrics Standard](https://github.com/openshift-hyperfleet/architecture/blob/main/hyperfleet/standards/metrics.md) and use the `hyperfleet_broker_` prefix.
+
+A `MetricsRecorder` is required when creating publishers and subscribers:
+
+```go
+metrics := broker.NewMetricsRecorder("my-component", "v1.0.0", prometheus.DefaultRegisterer)
+
+publisher, err := broker.NewPublisher(appLogger, metrics)
+subscriber, err := broker.NewSubscriber(appLogger, "my-subscription", metrics)
+```
+
+### Exported Metrics
+
+| Metric | Type | Labels | Description |
+|--------|------|--------|-------------|
+| `hyperfleet_broker_messages_consumed_total` | Counter | `topic`, `component`, `version` | Total messages consumed from the broker |
+| `hyperfleet_broker_messages_published_total` | Counter | `topic`, `component`, `version` | Total messages published to the broker |
+| `hyperfleet_broker_errors_total` | Counter | `topic`, `error_type`, `component`, `version` | Total message processing errors |
+| `hyperfleet_broker_message_duration_seconds` | Histogram | `topic`, `component`, `version` | Duration of message handler execution |
+
+### Error Types
+
+The `error_type` label on `hyperfleet_broker_errors_total` can have the following values:
+
+| Value | Description |
+|-------|-------------|
+| `conversion` | Failed to convert between CloudEvent and Watermill message format |
+| `handler` | The user-provided handler returned an error |
+| `publish` | The underlying broker rejected the publish operation |
+
+### Duration Buckets
+
+The `hyperfleet_broker_message_duration_seconds` histogram uses the following buckets (in seconds): `0.1, 0.5, 1, 2, 5, 10, 30, 60, 120`.
+
 ## Main Architectural Decisions
 
 ### 1. Watermill Abstraction

broker/broker.go

@@ -21,14 +21,17 @@ const (
 	DefaultShutdownTimeout       = 30 * time.Second
 )
 
-// NewPublisher creates a new publisher with a required logger and optional configuration.
+// NewPublisher creates a new publisher with the provided logger, metrics recorder, and optional configuration.
 // Usage:
-//   - NewPublisher(logger) - uses provided logger and loads config from file
-//   - NewPublisher(logger, configMap) - uses provided logger with config map
-func NewPublisher(log logger.Logger, configMap ...map[string]string) (Publisher, error) {
+//   - NewPublisher(logger, metrics) - uses provided logger and loads config from file
+//   - NewPublisher(logger, metrics, configMap) - uses provided logger with config map
+func NewPublisher(log logger.Logger, metrics *MetricsRecorder, configMap ...map[string]string) (Publisher, error) {
 	if log == nil {
 		return nil, fmt.Errorf("logger is required")
 	}
+	if metrics == nil {
+		return nil, fmt.Errorf("metrics is required")
+	}
 
 	var cfg *config
 	var err error
@@ -96,20 +99,24 @@ func NewPublisher(log logger.Logger, configMap ...map[string]string) (Publisher,
 		logger:       log,
 		healthCheck:  hc,
 		healthCloser: healthCloser,
+		metrics:      metrics,
 	}, nil
 }
 
-// NewSubscriber creates a new subscriber with a required logger and optional configuration.
+// NewSubscriber creates a new subscriber with the provided logger, subscription ID, metrics recorder, and optional configuration.
 // Usage:
-//   - NewSubscriber(logger, "id") - uses provided logger and loads config from file
-//   - NewSubscriber(logger, "id", configMap) - uses provided logger with config map
-func NewSubscriber(log logger.Logger, subscriptionID string, configMap ...map[string]string) (Subscriber, error) {
+//   - NewSubscriber(logger, "id", metrics) - uses provided logger and loads config from file
+//   - NewSubscriber(logger, "id", metrics, configMap) - uses provided logger with config map
+func NewSubscriber(log logger.Logger, subscriptionID string, metrics *MetricsRecorder, configMap ...map[string]string) (Subscriber, error) {
 	if subscriptionID == "" {
 		return nil, fmt.Errorf("subscriptionID is required")
 	}
 	if log == nil {
 		return nil, fmt.Errorf("logger is required")
 	}
+	if metrics == nil {
+		return nil, fmt.Errorf("metrics is required")
+	}
 
 	var cfg *config
 	var err error
@@ -165,6 +172,7 @@ func NewSubscriber(log logger.Logger, subscriptionID string, configMap ...map[st
 		subscriptionID: subscriptionID,
 		logger:         log,
 		errorChan:      make(chan *SubscriberError, ErrorChannelBufferSize),
+		metrics:        metrics,
 	}, nil
 }
 

broker/error_test.go

@@ -4,9 +4,15 @@ import (
 	"testing"
 
 	"github.com/openshift-hyperfleet/hyperfleet-broker/pkg/logger"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 )
 
+func newErrorTestMetrics(t *testing.T) *MetricsRecorder {
+	t.Helper()
+	return NewMetricsRecorder("test", "v0.0.0", prometheus.NewRegistry())
+}
+
 func TestNewPublisherErrorHandling(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -65,10 +71,11 @@ func TestNewPublisherErrorHandling(t *testing.T) {
 			var err error
 
 			mockLogger := logger.NewMockLogger()
+			metrics := newErrorTestMetrics(t)
 			if tt.configMap == nil {
-				pub, err = NewPublisher(mockLogger)
+				pub, err = NewPublisher(mockLogger, metrics)
 			} else {
-				pub, err = NewPublisher(mockLogger, tt.configMap)
+				pub, err = NewPublisher(mockLogger, metrics, tt.configMap)
 			}
 
 			if tt.expectError {
@@ -154,10 +161,11 @@ func TestNewSubscriberErrorHandling(t *testing.T) {
 			var err error
 
 			mockLogger := logger.NewMockLogger()
+			metrics := newErrorTestMetrics(t)
 			if tt.configMap == nil {
-				sub, err = NewSubscriber(mockLogger, tt.subscriptionID)
+				sub, err = NewSubscriber(mockLogger, tt.subscriptionID, metrics)
 			} else {
-				sub, err = NewSubscriber(mockLogger, tt.subscriptionID, tt.configMap)
+				sub, err = NewSubscriber(mockLogger, tt.subscriptionID, metrics, tt.configMap)
 			}
 
 			if tt.expectError {
@@ -192,7 +200,8 @@ func TestPublisherPublishErrorHandling(t *testing.T) {
 	}
 
 	mockLogger := logger.NewMockLogger()
-	_, err := NewPublisher(mockLogger, configMap)
+	metrics := newErrorTestMetrics(t)
+	_, err := NewPublisher(mockLogger, metrics, configMap)
 	assert.Error(t, err)
 }
 

broker/metrics.go

@@ -0,0 +1,86 @@
+package broker
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// MetricsRecorder holds Prometheus metrics for the broker library.
+type MetricsRecorder struct {
+	component string
+	version   string
+
+	messagesConsumed  *prometheus.CounterVec
+	messagesPublished *prometheus.CounterVec
+	errors            *prometheus.CounterVec
+	messageDuration   *prometheus.HistogramVec
+}
+
+// NewMetricsRecorder creates a new MetricsRecorder and registers all metrics
+// with the provided prometheus.Registerer. If registerer is nil, the default
+// prometheus.DefaultRegisterer is used.
+//
+// The component and version parameters are used as constant label values
+// on all metrics, per the HyperFleet Metrics Standard.
+func NewMetricsRecorder(component, version string, registerer prometheus.Registerer) *MetricsRecorder {
+	if registerer == nil {
+		registerer = prometheus.DefaultRegisterer
+	}
+
+	m := &MetricsRecorder{
+		component: component,
+		version:   version,
+
+		messagesConsumed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "hyperfleet_broker_messages_consumed_total",
+			Help: "Total number of messages consumed from the broker.",
+		}, []string{"topic", "component", "version"}),
+
+		messagesPublished: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "hyperfleet_broker_messages_published_total",
+			Help: "Total number of messages published to the broker.",
+		}, []string{"topic", "component", "version"}),
+
+		errors: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "hyperfleet_broker_errors_total",
+			Help: "Total number of message processing errors.",
+		}, []string{"topic", "error_type", "component", "version"}),
+
+		messageDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name: "hyperfleet_broker_message_duration_seconds",
+			Help: "Duration of message processing in seconds.",
+			// Event processing buckets per HyperFleet Metrics Standard
+			Buckets: []float64{0.1, 0.5, 1, 2, 5, 10, 30, 60, 120},
+		}, []string{"topic", "component", "version"}),
+	}
+
+	registerer.MustRegister(
+		m.messagesConsumed,
+		m.messagesPublished,
+		m.errors,
+		m.messageDuration,
+	)
+
+	return m
+}
+
+// RecordConsumed increments the messages consumed counter for the given topic.
+func (m *MetricsRecorder) RecordConsumed(topic string) {
+	m.messagesConsumed.WithLabelValues(topic, m.component, m.version).Inc()
+}
+
+// RecordPublished increments the messages published counter for the given topic.
+func (m *MetricsRecorder) RecordPublished(topic string) {
+	m.messagesPublished.WithLabelValues(topic, m.component, m.version).Inc()
+}
+
+// RecordError increments the errors counter for the given topic and error type.
+func (m *MetricsRecorder) RecordError(topic, errorType string) {
+	m.errors.WithLabelValues(topic, errorType, m.component, m.version).Inc()
+}
+
+// RecordDuration observes the message processing duration for the given topic.
+func (m *MetricsRecorder) RecordDuration(topic string, duration time.Duration) {
+	m.messageDuration.WithLabelValues(topic, m.component, m.version).Observe(duration.Seconds())
+}