Initial interface and implementation of prometheus metrics handler

Author: bilaljaved

Gopkg.lock

@@ -20,7 +20,8 @@ type (
 		queueProcessors int
 		queueBuffer     int
 
-		logger goengine.Logger
+		logger  goengine.Logger
+		metrics goengine.Metrics
 	}
 
 	// ProcessHandler is a func used to trigger a notification but with the addition of providing a Trigger func so
@@ -29,7 +30,7 @@ type (
 )
 
 // newBackgroundProcessor create a new projectionNotificationProcessor
-func newBackgroundProcessor(queueProcessors, queueBuffer int, logger goengine.Logger) (*projectionNotificationProcessor, error) {
+func newBackgroundProcessor(queueProcessors, queueBuffer int, logger goengine.Logger, metrics goengine.Metrics) (*projectionNotificationProcessor, error) {
 	if queueProcessors <= 0 {
 		return nil, errors.New("queueProcessors must be greater then zero")
 	}
@@ -39,11 +40,15 @@ func newBackgroundProcessor(queueProcessors, queueBuffer int, logger goengine.Lo
 	if logger == nil {
 		logger = goengine.NopLogger
 	}
+	if metrics == nil {
+		metrics = goengine.NopMetrics
+	}
 
 	return &projectionNotificationProcessor{
 		queueProcessors: queueProcessors,
 		queueBuffer:     queueBuffer,
 		logger:          logger,
+		metrics:         metrics,
 	}, nil
 }
 
@@ -104,6 +109,8 @@ func (b *projectionNotificationProcessor) Queue(ctx context.Context, notificatio
 		return errors.New("goengine: unable to queue notification because the processor was stopped")
 	}
 
+	b.metrics.QueueNotification(notification)
+
 	b.queue <- notification
 	return nil
 }
@@ -117,12 +124,14 @@ func (b *projectionNotificationProcessor) startProcessor(ctx context.Context, ha
 			return
 		case notification := <-b.queue:
 			// Execute the notification
+			b.metrics.StartNotificationProcessing(notification)
 			if err := handler(ctx, notification, b.Queue); err != nil {
 				b.logger.Error("the ProcessHandler produced an error", func(e goengine.LoggerEntry) {
 					e.Error(err)
 					e.Any("notification", notification)
 				})
 			}
+			b.metrics.FinishNotificationProcessing(notification)
 		}
 	}
 }

driver/sql/projection_notification_processor.go

@@ -35,6 +35,7 @@ func NewAggregateProjector(
 	projectorStorage AggregateProjectorStorage,
 	projectionErrorHandler ProjectionErrorCallback,
 	logger goengine.Logger,
+	metrics goengine.Metrics,
 ) (*AggregateProjector, error) {
 	switch {
 	case db == nil:
@@ -58,7 +59,7 @@ func NewAggregateProjector(
 		e.String("projection", projection.Name())
 	})
 
-	processor, err := newBackgroundProcessor(10, 32, logger)
+	processor, err := newBackgroundProcessor(10, 32, logger, metrics)
 	if err != nil {
 		return nil, err
 	}

driver/sql/projector_aggregate.go

@@ -0,0 +1,84 @@
+package prometheus
+
+import (
+	"github.com/hellofresh/goengine/driver/sql"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const namespace = "goengine"
+
+type Metrics struct {
+	notificationCounter        *prometheus.CounterVec
+	queueDuration              *prometheus.HistogramVec
+	notificationHandleDuration *prometheus.HistogramVec
+}
+
+func NewMetrics() *Metrics {
+	return &Metrics{
+		// notificationCounter is used to expose 'notification_count' metric
+		notificationCounter: prometheus.NewCounterVec(
+			prometheus.CounterOpts{
+				Namespace: namespace,
+				Name:      "notification_count",
+				Help:      "counter for number of notifications received",
+			},
+			[]string{"is_event"},
+		),
+		// queueDuration is used to expose 'queue_duration_seconds' metrics
+		queueDuration: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: namespace,
+				Name:      "queue_duration_seconds",
+				Help:      "histogram of queue latencies",
+				Buckets:   []float64{0.1, 0.5, 0.9, 0.99}, //buckets for histogram
+			},
+			[]string{"retry", "success"},
+		),
+
+		// notificationHandleDuration is used to expose 'notification_handle_duration_seconds' metrics
+		notificationHandleDuration: prometheus.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Namespace: namespace,
+				Name:      "notification_handle_duration_seconds",
+				Help:      "histogram of event handled latencies",
+				Buckets:   []float64{0.1, 0.5, 0.9, 0.99}, //buckets for histogram
+			},
+			[]string{"retry", "success"},
+		),
+	}
+}
+
+// RegisterMetrics returns http handler for prometheus
+func (m *Metrics) RegisterMetrics(registry *prometheus.Registry) error {
+	err := registry.Register(m.notificationCounter)
+	if err != nil {
+		return err
+	}
+
+	err = registry.Register(m.queueDuration)
+	if err != nil {
+		return err
+	}
+
+	return registry.Register(m.notificationHandleDuration)
+}
+
+// ReceivedNotification
+func (m *Metrics) ReceivedNotification(isNotification bool) {
+
+}
+
+// QueueNotification returns http handler for prometheus
+func (m *Metrics) QueueNotification(notification *sql.ProjectionNotification) {
+
+}
+
+// StartNotificationProcessing is used to record start time of notification processing
+func (m *Metrics) StartNotificationProcessing(notification *sql.ProjectionNotification) {
+
+}
+
+// FinishNotificationProcessing is used to observe end time of notification queue and processing time
+func (m *Metrics) FinishNotificationProcessing(notification *sql.ProjectionNotification) {
+
+}

extension/prometheus/prometheus.go

@@ -0,0 +1,15 @@
+package goengine
+
+import (
+	"github.com/hellofresh/goengine/driver/sql"
+)
+
+type (
+	// Metrics a structured metrics interface
+	Metrics interface {
+		ReceivedNotification(isNotification bool)
+		QueueNotification(notification *sql.ProjectionNotification)
+		StartNotificationProcessing(notification *sql.ProjectionNotification)
+		FinishNotificationProcessing(notification *sql.ProjectionNotification)
+	}
+)

metrics.go

@@ -0,0 +1,12 @@
+package goengine
+
+import "github.com/hellofresh/goengine/driver/sql"
+
+var NopMetrics Metrics = &nopMetrics{}
+
+type nopMetrics struct{}
+
+func (nm *nopMetrics) ReceivedNotification(isNotification bool)                              {}
+func (nm *nopMetrics) QueueNotification(notification *sql.ProjectionNotification)            {}
+func (nm *nopMetrics) StartNotificationProcessing(notification *sql.ProjectionNotification)  {}
+func (nm *nopMetrics) FinishNotificationProcessing(notification *sql.ProjectionNotification) {}

metrics_nop.go

@@ -17,17 +17,21 @@ type SingleStreamManager struct {
 	persistenceStrategy driverSQL.PersistenceStrategy
 	messageFactory      driverSQL.MessageFactory
 
-	logger goengine.Logger
+	logger  goengine.Logger
+	metrics goengine.Metrics
 }
 
 // NewSingleStreamManager return a new instance of the SingleStreamManager
-func NewSingleStreamManager(db *sql.DB, logger goengine.Logger) (*SingleStreamManager, error) {
+func NewSingleStreamManager(db *sql.DB, logger goengine.Logger, metrics goengine.Metrics) (*SingleStreamManager, error) {
 	if db == nil {
 		return nil, goengine.InvalidArgumentError("db")
 	}
 	if logger == nil {
 		logger = goengine.NopLogger
 	}
+	if metrics == nil {
+		metrics = goengine.NopMetrics
+	}
 
 	payloadTransformer := json.NewPayloadTransformer()
 
@@ -49,6 +53,7 @@ func NewSingleStreamManager(db *sql.DB, logger goengine.Logger) (*SingleStreamMa
 		persistenceStrategy: persistenceStrategy,
 		messageFactory:      messageFactory,
 		logger:              logger,
+		metrics:             metrics,
 	}, nil
 }
 
@@ -145,5 +150,6 @@ func (m *SingleStreamManager) NewAggregateProjector(
 		projectorStorage,
 		projectionErrorHandler,
 		m.logger,
+		m.metrics,
 	)
 }