Logging in case of failures

Author: bilaljaved

driver/sql/metrics.go

@@ -4,8 +4,8 @@ type (
 	// Metrics a structured metrics interface
 	Metrics interface {
 		ReceivedNotification(isNotification bool)
-		QueueNotification(notification *ProjectionNotification) bool
-		StartNotificationProcessing(notification *ProjectionNotification) bool
-		FinishNotificationProcessing(notification *ProjectionNotification, success bool) bool
+		QueueNotification(notification *ProjectionNotification)
+		StartNotificationProcessing(notification *ProjectionNotification)
+		FinishNotificationProcessing(notification *ProjectionNotification, success bool)
 	}
 )

driver/sql/metrics_nop.go

@@ -6,12 +6,10 @@ var NopMetrics Metrics = &nopMetrics{}
 type nopMetrics struct{}
 
 func (nm *nopMetrics) ReceivedNotification(isNotification bool) {}
-func (nm *nopMetrics) QueueNotification(notification *ProjectionNotification) bool {
-	return true
+func (nm *nopMetrics) QueueNotification(notification *ProjectionNotification) {
 }
-func (nm *nopMetrics) StartNotificationProcessing(notification *ProjectionNotification) bool {
-	return true
+func (nm *nopMetrics) StartNotificationProcessing(notification *ProjectionNotification) {
+
 }
-func (nm *nopMetrics) FinishNotificationProcessing(notification *ProjectionNotification, success bool) bool {
-	return true
+func (nm *nopMetrics) FinishNotificationProcessing(notification *ProjectionNotification, success bool) {
 }

extension/prometheus/prometheus.go

@@ -2,6 +2,7 @@ package prometheus
 
 import (
 	"fmt"
+	"github.com/hellofresh/goengine"
 	"strconv"
 	"sync"
 	"time"
@@ -22,6 +23,7 @@ type Metrics struct {
 	notificationQueueDuration      *prometheus.HistogramVec
 	notificationProcessingDuration *prometheus.HistogramVec
 	notificationStartTimes         sync.Map
+	logger                         goengine.Logger
 }
 
 // NewMetrics instantiate and return an object of Metrics
@@ -60,6 +62,10 @@ func NewMetrics() *Metrics {
 	}
 }
 
+func (m *Metrics) SetLogger(logger goengine.Logger) {
+	m.logger = logger
+}
+
 // RegisterMetrics returns http handler for prometheus
 func (m *Metrics) RegisterMetrics(registry *prometheus.Registry) error {
 	err := registry.Register(m.notificationCounter)
@@ -82,38 +88,53 @@ func (m *Metrics) ReceivedNotification(isNotification bool) {
 }
 
 // QueueNotification returns http handler for prometheus
-func (m *Metrics) QueueNotification(notification *sql.ProjectionNotification) bool {
-	return m.storeStartTime(notificationQueueKeyPrefix, notification)
+func (m *Metrics) QueueNotification(notification *sql.ProjectionNotification) {
+	if !m.storeStartTime(notificationQueueKeyPrefix, notification) {
+		m.logger.Warn("notification already queued", func(e goengine.LoggerEntry) {
+			e.Any("notification", notification)
+		})
+	}
+
 }
 
 // StartNotificationProcessing is used to record start time of notification processing
-func (m *Metrics) StartNotificationProcessing(notification *sql.ProjectionNotification) bool {
-	return m.storeStartTime(notificationProcessingKeyPrefix, notification)
+func (m *Metrics) StartNotificationProcessing(notification *sql.ProjectionNotification) {
+	if !m.storeStartTime(notificationProcessingKeyPrefix, notification) {
+		m.logger.Warn("notification processing already started", func(e goengine.LoggerEntry) {
+			e.Any("notification", notification)
+		})
+	}
 }
 
 // FinishNotificationProcessing is used to observe end time of notification queue and processing time
-func (m *Metrics) FinishNotificationProcessing(notification *sql.ProjectionNotification, success bool) bool {
+func (m *Metrics) FinishNotificationProcessing(notification *sql.ProjectionNotification, success bool) {
 	memAddress := fmt.Sprintf("%p", notification)
 	labels := prometheus.Labels{"success": strconv.FormatBool(success)}
 
-	queueStartTime, queueOk := m.notificationStartTimes.Load(notificationQueueKeyPrefix + memAddress)
+	if queueStartTime, ok := m.notificationStartTimes.Load(notificationQueueKeyPrefix + memAddress); ok {
+		m.notificationQueueDuration.With(labels).Observe(time.Since(queueStartTime.(time.Time)).Seconds())
+
+	} else {
 
-	processingStartTime, processingOk := m.notificationStartTimes.Load(notificationProcessingKeyPrefix + memAddress)
+		m.logger.Warn("notification queue start time not found", func(e goengine.LoggerEntry) {
+			e.Any("notification", notification)
+		})
+	}
 
-	if processingOk && queueOk {
+	if processingStartTime, ok := m.notificationStartTimes.Load(notificationProcessingKeyPrefix + memAddress); ok {
 		m.notificationProcessingDuration.With(labels).Observe(time.Since(processingStartTime.(time.Time)).Seconds())
-		m.notificationQueueDuration.With(labels).Observe(time.Since(queueStartTime.(time.Time)).Seconds())
-		return true
+	} else {
+		m.logger.Warn("notification processing start time not found", func(e goengine.LoggerEntry) {
+			e.Any("notification", notification)
+		})
 	}
 
-	return false
 }
 
 // storeStartTime stores the start time against each notification only if it's not already existent
 func (m *Metrics) storeStartTime(prefix string, notification *sql.ProjectionNotification) bool {
 	key := prefix + fmt.Sprintf("%p", notification)
 
 	_, alreadyExists := m.notificationStartTimes.LoadOrStore(key, time.Now())
-
 	return !alreadyExists
 }