Fix unbounded memory allocation. #275

mikecdavis · 2020-07-01T01:28:27Z

Summary

Add bounded capacity to InMemoryQueue directly
Fix potential concurrency issues in InMemoryQueue when checking queue size
Fix QueueEventDispatcher size gauge metric
Fix leaking goroutines in QueueEventDispatcher
Run go mod tidy

Issues

OASIS-6735
OASIS-6738

aliabbasrizvi

Looks good.

aliabbasrizvi · 2020-07-01T05:48:02Z

pkg/event/dispatcher.go

@@ -136,7 +136,6 @@ func (ed *QueueEventDispatcher) flushEvents() {

 		if err == nil {
 			if success {
-				ed.logger.Debug(fmt.Sprintf("Dispatched log event %+v", event))


Was the message too chatty/not useful?

This line is wasting resources. The event payload (which is large) is getting parsed into this string representation regardless of the log level. Our logging interface should really have a Debugf(...), etc method for these types of formatted messages.

Might want to keep this line, but don't parse the whole event. It can be useful.

aliabbasrizvi · 2020-07-01T06:09:22Z

pkg/event/queue.go

 )

 // Queue represents a queue
 type Queue interface {
-	Add(item interface{})
+	Add(item interface{}) // Should return a bool


Yeah probably a TODO item, but unfortunately it would break the interface otherwise I would have done it here. Likely we'd have to define another interface and add backwards compatibility but wrapping any old implementations into the new one.

aliabbasrizvi · 2020-07-01T06:12:13Z

pkg/event/queue.go

+
+	if len(q.Queue) >= q.MaxSize {
+		q.logger.Warning("MaxQueueSize has been met. Discarding event")
+		return


Shouldn't this be returning error or something? Or is that planned in a follow up?

We should yes, and this is what prompted me to add that comment on the struct to return a boolean. So for now planned follow-up.

aliabbasrizvi · 2020-07-01T06:18:15Z

pkg/event/queue_test.go

@@ -22,14 +22,25 @@ import (
 	"testing"
 )

+func TestInMemoryQueue_Add(t *testing.T) {


nit. This could probably be more descriptive i.e. TestInMemoryQueue_Add_Queue_Size_Bounded_By_MaxSize?

pawels-optimizely · 2020-07-01T13:07:05Z

pkg/event/dispatcher.go

@@ -94,24 +95,23 @@ type QueueEventDispatcher struct {
 // DispatchEvent queues event with callback and calls flush in a go routine.
 func (ed *QueueEventDispatcher) DispatchEvent(event LogEvent) (bool, error) {
 	ed.eventQueue.Add(event)
-	go func() {
-		ed.flushEvents()


pawels-optimizely · 2020-07-01T13:16:35Z

pkg/event/dispatcher.go

-	ed.queueSize.Set(float64(ed.eventQueue.Size()))
-	for ed.eventQueue.Size() > 0 {
+	queueSize := ed.eventQueue.Size()
+	for ; queueSize > 0; queueSize = ed.eventQueue.Size() {


I think I understand, but can you provide a quick comment explaining why this has to be in the loop?

Since the Size() was only evaluated at the beginning and the end of the loop. We failed to capture any queue size that may have evolved while the loop was being processed. Under heavy load, the queue can grow and this loop will continue to iterate, but the size would not be captured. I discovered this while investigating runaway memory allocation. Our metrics were reporting a constant size in the single digits, but in reality it was growing into the hundreds of thousands.

pawels-optimizely · 2020-07-01T13:25:25Z

pkg/event/dispatcher.go

@@ -159,7 +158,7 @@ func (ed *QueueEventDispatcher) flushEvents() {
 			ed.retryFlushCounter.Add(1)
 		}
 	}
-	ed.queueSize.Set(float64(ed.eventQueue.Size()))
+	ed.queueSize.Set(float64(queueSize))


shouldn't it get the right size all the time when we use ed.eventQueue.Size() ? (just asking)

eventQueue.Size() is a blocking call and I didn't feel it necessary to call it again here since it was just called from the for loop and we presumably exited with a size of zero. If the queue had changed since then it will be captured with the next call to flushEvents().

msohailhussain · 2020-07-02T20:50:02Z

pkg/event/dispatcher.go

 		Dispatcher:        NewHTTPEventDispatcher(sdkKey, nil, nil),
 		queueSize:         dispatcherMetricsRegistry.GetGauge(metrics.DispatcherQueueSize),
 		retryFlushCounter: dispatcherMetricsRegistry.GetCounter(metrics.DispatcherRetryFlush),
 		failFlushCounter:  dispatcherMetricsRegistry.GetCounter(metrics.DispatcherFailedFlush),
 		sucessFlush:       dispatcherMetricsRegistry.GetCounter(metrics.DispatcherSuccessFlush),
-		logger:            logging.GetLogger(sdkKey, "QueueEventDispatcher"),
+		logger:            logger,
+		processing:        semaphore.NewWeighted(int64(1)),


can we make it const

msohailhussain · 2020-07-02T20:50:37Z

pkg/event/dispatcher.go

-		ed.eventFlushLock.Unlock()
-	}()
+	// Limit flushing to a single worker
+	if !ed.processing.TryAcquire(1) {


to use that const value here.

mjc1283 · 2020-07-02T22:15:43Z

pkg/event/dispatcher.go

@@ -136,7 +136,6 @@ func (ed *QueueEventDispatcher) flushEvents() {

 		if err == nil {
 			if success {
-				ed.logger.Debug(fmt.Sprintf("Dispatched log event %+v", event))


Might want to keep this line, but don't parse the whole event. It can be useful.

mikecdavis requested review from mjc1283, thomaszurkan-optimizely and pawels-optimizely July 1, 2020 01:28

mikecdavis requested a review from a team as a code owner July 1, 2020 01:28

optimizely-codereview-bot assigned mikecdavis Jul 1, 2020

aliabbasrizvi approved these changes Jul 1, 2020

View reviewed changes

aliabbasrizvi reviewed Jul 1, 2020

View reviewed changes

pawels-optimizely reviewed Jul 1, 2020

View reviewed changes

Mike Davis added 4 commits July 2, 2020 08:44

Implement bounded queue.

e4e9c13

Limit the number of concurrent goroutines.

f2eb2aa

Fix linter.

c399e6c

Bump the bench queue sizes.

e580b28

mikecdavis force-pushed the mikecdavis/OASIS-6738-cap-dispatcher-queue branch from 5ef675f to e580b28 Compare July 2, 2020 15:45

msohailhussain reviewed Jul 2, 2020

View reviewed changes

mjc1283 approved these changes Jul 2, 2020

View reviewed changes

Mike Davis added 2 commits July 2, 2020 15:33

Make PR updates.

0b67e48

Add logging back.

8bf7d68

mikecdavis merged commit 2e7c275 into master Jul 2, 2020

mikecdavis deleted the mikecdavis/OASIS-6738-cap-dispatcher-queue branch July 2, 2020 23:28

Fix unbounded memory allocation. #275

Fix unbounded memory allocation. #275

Uh oh!

Conversation

mikecdavis commented Jul 1, 2020

Summary

Issues

Uh oh!

aliabbasrizvi left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aliabbasrizvi Jul 1, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

aliabbasrizvi Jul 1, 2020 •

edited

Loading