Return status of functions to add unit tests

Author: bilaljaved

driver/sql/metrics.go

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

driver/sql/metrics_nop.go

@@ -5,8 +5,13 @@ 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) {
+func (nm *nopMetrics) ReceivedNotification(isNotification bool) {}
+func (nm *nopMetrics) QueueNotification(notification *ProjectionNotification) bool {
+	return true
+}
+func (nm *nopMetrics) StartNotificationProcessing(notification *ProjectionNotification) bool {
+	return true
+}
+func (nm *nopMetrics) FinishNotificationProcessing(notification *ProjectionNotification, success bool) bool {
+	return true
 }

extension/prometheus/prometheus.go

@@ -10,7 +10,11 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-const namespace = "goengine"
+const (
+	namespace                       = "goengine"
+	notificationQueueKeyPrefix      = "q"
+	notificationProcessingKeyPrefix = "p"
+)
 
 // Metrics is an object for exposing prometheus metrics
 type Metrics struct {
@@ -78,24 +82,38 @@ func (m *Metrics) ReceivedNotification(isNotification bool) {
 }
 
 // QueueNotification returns http handler for prometheus
-func (m *Metrics) QueueNotification(notification *sql.ProjectionNotification) {
-	key := "q" + fmt.Sprintf("%p", notification)
-	m.notificationStartTimes.Store(key, time.Now())
+func (m *Metrics) QueueNotification(notification *sql.ProjectionNotification) bool {
+	return m.storeStartTime(notificationQueueKeyPrefix, notification)
 }
 
 // StartNotificationProcessing is used to record start time of notification processing
-func (m *Metrics) StartNotificationProcessing(notification *sql.ProjectionNotification) {
-	key := "p" + fmt.Sprintf("%p", notification)
-	m.notificationStartTimes.Store(key, time.Now())
+func (m *Metrics) StartNotificationProcessing(notification *sql.ProjectionNotification) bool {
+	return m.storeStartTime(notificationProcessingKeyPrefix, notification)
 }
 
 // FinishNotificationProcessing is used to observe end time of notification queue and processing time
-func (m *Metrics) FinishNotificationProcessing(notification *sql.ProjectionNotification, success bool) {
+func (m *Metrics) FinishNotificationProcessing(notification *sql.ProjectionNotification, success bool) bool {
 	memAddress := fmt.Sprintf("%p", notification)
-	queueStartTime, _ := m.notificationStartTimes.Load("q" + memAddress)
-	processingStartTime, _ := m.notificationStartTimes.Load("p" + memAddress)
 	labels := prometheus.Labels{"success": strconv.FormatBool(success)}
 
-	m.notificationQueueDuration.With(labels).Observe(time.Since(queueStartTime.(time.Time)).Seconds())
-	m.notificationProcessingDuration.With(labels).Observe(time.Since(processingStartTime.(time.Time)).Seconds())
+	queueStartTime, queueOk := m.notificationStartTimes.Load(notificationQueueKeyPrefix + memAddress)
+
+	processingStartTime, processingOk := m.notificationStartTimes.Load(notificationProcessingKeyPrefix + memAddress)
+
+	if processingOk && queueOk {
+		m.notificationProcessingDuration.With(labels).Observe(time.Since(processingStartTime.(time.Time)).Seconds())
+		m.notificationQueueDuration.With(labels).Observe(time.Since(queueStartTime.(time.Time)).Seconds())
+		return true
+	}
+
+	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
 }

extension/prometheus/prometheus_test.go

@@ -1,34 +1,43 @@
 // +build unit
 
-package prometheus
+package prometheus_test
 
 import (
-	"fmt"
 	"testing"
-	"time"
 
+	"github.com/hellofresh/goengine/extension/prometheus"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/hellofresh/goengine/driver/sql"
 )
 
 func TestPrometheusMetrics(t *testing.T) {
-	metrics := NewMetrics()
+	metrics := prometheus.NewMetrics()
 
 	testSQLProjection := sql.ProjectionNotification{
 		No:          1,
 		AggregateID: "C56A4180-65AA-42EC-A945-5FD21DEC0538",
 	}
 
-	metrics.QueueNotification(&testSQLProjection)
-	metrics.StartNotificationProcessing(&testSQLProjection)
+	ok := metrics.FinishNotificationProcessing(&testSQLProjection, true)
+	assert.False(t, ok)
 
-	memAddress := fmt.Sprintf("%p", &testSQLProjection)
-	queueStartTime, ok := metrics.notificationStartTimes.Load("q" + memAddress)
+	ok = metrics.QueueNotification(&testSQLProjection)
 	assert.True(t, ok)
-	assert.IsType(t, time.Time{}, queueStartTime)
 
-	processingStartTime, _ := metrics.notificationStartTimes.Load("p" + memAddress)
+	ok = metrics.QueueNotification(&testSQLProjection)
+	assert.False(t, ok)
+
+	ok = metrics.FinishNotificationProcessing(&testSQLProjection, true)
+	assert.False(t, ok)
+
+	ok = metrics.StartNotificationProcessing(&testSQLProjection)
+	assert.True(t, ok)
+
+	ok = metrics.StartNotificationProcessing(&testSQLProjection)
+	assert.False(t, ok)
+
+	ok = metrics.FinishNotificationProcessing(&testSQLProjection, true)
 	assert.True(t, ok)
-	assert.IsType(t, time.Time{}, processingStartTime)
+
 }