Merge pull request #37 from hellofresh/patch/projector-metrics

OWMS-2318 projector metrics

Author: bilaljaved

Gopkg.lock

@@ -0,0 +1,18 @@
+package sql
+
+type (
+	// Metrics a structured metrics interface
+	Metrics interface {
+		// ReceivedNotification sends the metric to keep count of notifications received by goengine
+		ReceivedNotification(isNotification bool)
+		// QueueNotification is called when a notification is queued.
+		// It saves start time for an event on aggregate when it's queued
+		QueueNotification(notification *ProjectionNotification)
+		// StartNotificationProcessing is called when a notification processing is started
+		// It saves start time for an event on aggregate when it's picked to be processed by background processor
+		StartNotificationProcessing(notification *ProjectionNotification)
+		// FinishNotificationProcessing is called when a notification processing is finished
+		// It actually sends metrics calculating duration for which a notification spends in queue and then processed by background processor
+		FinishNotificationProcessing(notification *ProjectionNotification, success bool)
+	}
+)

driver/sql/metrics.go

@@ -0,0 +1,12 @@
+package sql
+
+// NopMetrics is default Metrics handler in case nil is passed
+var NopMetrics Metrics = &nopMetrics{}
+
+type nopMetrics struct{}
+
+func (nm *nopMetrics) ReceivedNotification(isNotification bool)                         {}
+func (nm *nopMetrics) QueueNotification(notification *ProjectionNotification)           {}
+func (nm *nopMetrics) StartNotificationProcessing(notification *ProjectionNotification) {}
+func (nm *nopMetrics) FinishNotificationProcessing(notification *ProjectionNotification, success bool) {
+}

driver/sql/metrics_nop.go

@@ -181,6 +181,7 @@ func setup(
 			return driverSQL.ProjectionFail
 		},
 		goengine.NopLogger,
+		driverSQL.NopMetrics,
 	)
 	require.NoError(b, err, "failed to create aggregate projector")
 

driver/sql/postgres/projector_aggregate_bench_test.go

@@ -20,7 +20,8 @@ type (
 		queueProcessors int
 		queueBuffer     int
 
-		logger goengine.Logger
+		logger  goengine.Logger
+		metrics 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 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 = 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,11 +124,17 @@ 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, false)
+
+			} else {
+				b.metrics.FinishNotificationProcessing(notification, true)
 			}
 		}
 	}

driver/sql/projection_notification_processor.go

@@ -35,6 +35,7 @@ func NewAggregateProjector(
 	projectorStorage AggregateProjectorStorage,
 	projectionErrorHandler ProjectionErrorCallback,
 	logger goengine.Logger,
+	metrics 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

@@ -22,7 +22,8 @@ type Listener struct {
 	minReconnectInterval time.Duration
 	maxReconnectInterval time.Duration
 
-	logger goengine.Logger
+	logger  goengine.Logger
+	metrics sql.Metrics
 }
 
 // NewListener returns a new notification listener
@@ -32,6 +33,7 @@ func NewListener(
 	minReconnectInterval time.Duration,
 	maxReconnectInterval time.Duration,
 	logger goengine.Logger,
+	metrics sql.Metrics,
 ) (*Listener, error) {
 	switch {
 	case strings.TrimSpace(dbDSN) == "":
@@ -48,12 +50,17 @@ func NewListener(
 		logger = goengine.NopLogger
 	}
 
+	if metrics == nil {
+		metrics = sql.NopMetrics
+	}
+
 	return &Listener{
 		dbDSN:                dbDSN,
 		dbChannel:            dbChannel,
 		minReconnectInterval: minReconnectInterval,
 		maxReconnectInterval: maxReconnectInterval,
 		logger:               logger,
+		metrics:              metrics,
 	}, nil
 }
 
@@ -83,13 +90,15 @@ func (s *Listener) Listen(ctx context.Context, exec sql.ProjectionTrigger) error
 
 	// Execute an initial run of the projection.
 	// This is done after db listen is started to avoid losing a set of messages while the Listener creates a db connection.
+	s.metrics.ReceivedNotification(false)
 	if err := exec(ctx, nil); err != nil {
 		return err
 	}
 
 	for {
 		select {
 		case n := <-listener.Notify:
+			s.metrics.ReceivedNotification(n != nil)
 			// Unmarshal the notification
 			notification := s.unmarshalNotification(n)