feat: Add observability metrics for event queue (#223)

* feat: Add observability metrics for event queue

Author: pawels-optimizely

pkg/client/factory.go

@@ -25,6 +25,7 @@ import (
 	"github.com/optimizely/go-sdk/pkg/config"
 	"github.com/optimizely/go-sdk/pkg/decision"
 	"github.com/optimizely/go-sdk/pkg/event"
+	"github.com/optimizely/go-sdk/pkg/metrics"
 	"github.com/optimizely/go-sdk/pkg/registry"
 	"github.com/optimizely/go-sdk/pkg/utils"
 )
@@ -41,6 +42,7 @@ type OptimizelyFactory struct {
 	eventProcessor     event.Processor
 	userProfileService decision.UserProfileService
 	overrideStore      decision.ExperimentOverrideStore
+	metricsRegistry    metrics.Registry
 }
 
 // OptionFunc is used to provide custom client configuration to the OptimizelyFactory.
@@ -57,6 +59,13 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 		return nil, errors.New("unable to instantiate client: no project config manager, SDK key, or a Datafile provided")
 	}
 
+	var metricsRegistry metrics.Registry
+	if f.metricsRegistry != nil {
+		metricsRegistry = f.metricsRegistry
+	} else {
+		metricsRegistry = metrics.NewNoopRegistry()
+	}
+
 	var ctx context.Context
 	if f.ctx != nil {
 		ctx = f.ctx
@@ -85,6 +94,7 @@ func (f OptimizelyFactory) Client(clientOptions ...OptionFunc) (*OptimizelyClien
 		if f.eventDispatcher != nil {
 			eventProcessorOptions = append(eventProcessorOptions, event.WithEventDispatcher(f.eventDispatcher))
 		}
+		eventProcessorOptions = append(eventProcessorOptions, event.WithEventDispatcherMetrics(metricsRegistry))
 		appClient.EventProcessor = event.NewBatchEventProcessor(eventProcessorOptions...)
 	}
 
@@ -180,6 +190,13 @@ func WithContext(ctx context.Context) OptionFunc {
 	}
 }
 
+// WithMetricsRegistry allows user to pass in their own implementation of a metrics collector
+func WithMetricsRegistry(metricsRegistry metrics.Registry) OptionFunc {
+	return func(f *OptimizelyFactory) {
+		f.metricsRegistry = metricsRegistry
+	}
+}
+
 // StaticClient returns a client initialized with a static project config.
 func (f OptimizelyFactory) StaticClient() (*OptimizelyClient, error) {
 	var configManager config.ProjectConfigManager

pkg/client/factory_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/optimizely/go-sdk/pkg/config/datafileprojectconfig"
 	"github.com/optimizely/go-sdk/pkg/decision"
 	"github.com/optimizely/go-sdk/pkg/event"
+	"github.com/optimizely/go-sdk/pkg/metrics"
 	"github.com/optimizely/go-sdk/pkg/utils"
 
 	"github.com/stretchr/testify/assert"
@@ -53,10 +54,6 @@ func (f *MockDispatcher) DispatchEvent(event event.LogEvent) (bool, error) {
 	return true, nil
 }
 
-func (f *MockDispatcher) GetMetrics() event.Metrics {
-	return nil
-}
-
 func TestFactoryClientReturnsDefaultClient(t *testing.T) {
 	factory := OptimizelyFactory{}
 
@@ -175,3 +172,16 @@ func TestClientWithEventDispatcher(t *testing.T) {
 	dispatcher := optimizelyClient.EventProcessor.(*event.BatchEventProcessor).EventDispatcher
 	assert.Equal(t, dispatcher, mockEventDispatcher)
 }
+
+func TestClientMetrics(t *testing.T) {
+	factory := OptimizelyFactory{SDKKey: "1212"}
+
+	metricsRegistry := metrics.NewNoopRegistry()
+
+	mockEventDispatcher := new(MockDispatcher)
+	optimizelyClient, err := factory.Client(WithEventDispatcher(mockEventDispatcher), WithMetricsRegistry(metricsRegistry))
+	assert.NoError(t, err)
+
+	eventProcessor := optimizelyClient.EventProcessor.(*event.BatchEventProcessor)
+	assert.NotNil(t, eventProcessor)
+}

pkg/event/dispatcher.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/optimizely/go-sdk/pkg/logging"
+	"github.com/optimizely/go-sdk/pkg/metrics"
 	"github.com/optimizely/go-sdk/pkg/utils"
 )
 
@@ -36,19 +37,13 @@ var dispatcherLogger = logging.GetLogger("EventDispatcher")
 // Dispatcher dispatches events
 type Dispatcher interface {
 	DispatchEvent(event LogEvent) (bool, error)
-	GetMetrics() Metrics
 }
 
 // HTTPEventDispatcher is the HTTP implementation of the Dispatcher interface
 type HTTPEventDispatcher struct {
 	requester *utils.HTTPRequester
 }
 
-// GetMetrics is the metric accessor
-func (ed *HTTPEventDispatcher) GetMetrics() Metrics {
-	return nil
-}
-
 // DispatchEvent dispatches event with callback
 func (ed *HTTPEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 
@@ -77,7 +72,11 @@ type QueueEventDispatcher struct {
 	eventFlushLock sync.Mutex
 	Dispatcher     Dispatcher
 
-	metrics Metrics
+	// metrics
+	queueSize         metrics.Gauge
+	sucessFlush       metrics.Counter
+	failFlushCounter  metrics.Counter
+	retryFlushCounter metrics.Counter
 }
 
 // DispatchEvent queues event with callback and calls flush in a go routine.
@@ -89,17 +88,6 @@ func (ed *QueueEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 	return true, nil
 }
 
-// GetMetrics is the metric accessor
-func (ed *QueueEventDispatcher) GetMetrics() Metrics {
-	ed.eventFlushLock.Lock()
-
-	defer func() {
-		ed.eventFlushLock.Unlock()
-	}()
-	return ed.metrics
-
-}
-
 // flush the events
 func (ed *QueueEventDispatcher) flushEvents() {
 
@@ -110,12 +98,11 @@ func (ed *QueueEventDispatcher) flushEvents() {
 	}()
 
 	retryCount := 0
-
-	ed.metrics.SetQueueSize(ed.eventQueue.Size())
+	ed.queueSize.Set(float64(ed.eventQueue.Size()))
 	for ed.eventQueue.Size() > 0 {
 		if retryCount > maxRetries {
 			dispatcherLogger.Error(fmt.Sprintf("event failed to send %d times. It will retry on next event sent", maxRetries), nil)
-			ed.metrics.IncrFailFlushCount()
+			ed.failFlushCounter.Add(1)
 			break
 		}
 
@@ -129,7 +116,7 @@ func (ed *QueueEventDispatcher) flushEvents() {
 			// remove it
 			dispatcherLogger.Error("invalid type passed to event Dispatcher", nil)
 			ed.eventQueue.Remove(1)
-			ed.metrics.IncrFailFlushCount()
+			ed.failFlushCounter.Add(1)
 			continue
 		}
 
@@ -140,16 +127,15 @@ func (ed *QueueEventDispatcher) flushEvents() {
 				dispatcherLogger.Debug(fmt.Sprintf("Dispatched log event %+v", event))
 				ed.eventQueue.Remove(1)
 				retryCount = 0
-				ed.metrics.IncrSuccessFlushCount()
+				ed.sucessFlush.Add(1)
 			} else {
 				dispatcherLogger.Warning("dispatch event failed")
 				// we failed.  Sleep some seconds and try again.
 				time.Sleep(sleepTime)
 				// increase retryCount.  We exit if we have retried x times.
 				// we will retry again next event that is added.
 				retryCount++
-				ed.metrics.IncrRetryFlushCount()
-
+				ed.retryFlushCounter.Add(1)
 			}
 		} else {
 			dispatcherLogger.Error("Error dispatching ", err)
@@ -158,17 +144,29 @@ func (ed *QueueEventDispatcher) flushEvents() {
 			// increase retryCount.  We exit if we have retried x times.
 			// we will retry again next event that is added.
 			retryCount++
-			ed.metrics.IncrRetryFlushCount()
+			ed.retryFlushCounter.Add(1)
 		}
 	}
-	ed.metrics.SetQueueSize(ed.eventQueue.Size())
+	ed.queueSize.Set(float64(ed.eventQueue.Size()))
 }
 
 // NewQueueEventDispatcher creates a Dispatcher that queues in memory and then sends via go routine.
-func NewQueueEventDispatcher() *QueueEventDispatcher {
+func NewQueueEventDispatcher(metricsRegistry metrics.Registry) *QueueEventDispatcher {
+
+	var dispatcherMetricsRegistry metrics.Registry
+	if metricsRegistry != nil {
+		dispatcherMetricsRegistry = metricsRegistry
+	} else {
+		dispatcherMetricsRegistry = metrics.NewNoopRegistry() // protective code to set
+	}
+
 	return &QueueEventDispatcher{
 		eventQueue: NewInMemoryQueue(defaultQueueSize),
 		Dispatcher: &HTTPEventDispatcher{requester: utils.NewHTTPRequester()},
-		metrics:    &DefaultMetrics{},
+
+		queueSize:         dispatcherMetricsRegistry.GetGauge(metrics.DispatcherQueueSize),
+		retryFlushCounter: dispatcherMetricsRegistry.GetCounter(metrics.DispatcherRetryFlush),
+		failFlushCounter:  dispatcherMetricsRegistry.GetCounter(metrics.DispatcherFailedFlush),
+		sucessFlush:       dispatcherMetricsRegistry.GetCounter(metrics.DispatcherSuccessFlush),
 	}
 }