DSET-4449: Do not count dropped buffer as processed (#52)

* DSET-4449: Do not count dropped buffer as processed

* Add more tests to verify statistics

* Fix the test

* Fix failing test in CI/CD

* Do not drop event

* Improve comment explaining the fix

* Update comments

* Increase version to 0.13.0

Author: martin-majlis-s1

RELEASE_NOTES.md

@@ -1,5 +1,10 @@
 # Release Notes
 
+## 0.13.0
+
+* fix: do not drop event when buffer is being published
+* fix: do not double count dropped buffers
+
 ## 0.12.0
 
 * fix: make client shutdown timeout configurable.

pkg/client/add_events.go

@@ -170,8 +170,12 @@ func (client *DataSetClient) listenAndSendBundlesForKey(key string, ch chan inte
 					buf = getBuffer(key)
 				} else {
 					client.Logger.Error("Cannot add bundle", zap.Error(err))
-					// TODO: what to do? For now, lets skip it
-					continue
+					// TODO: what to do?
+					// this error happens when the buffer is already publishing
+					// we can solve it by waiting little bit
+					// since we are also returning TooMuch we will get new buffer
+					// few lines below
+					time.Sleep(10 * time.Millisecond)
 				}
 			}
 

pkg/client/add_events_long_running_test.go

@@ -157,6 +157,21 @@ func TestAddEventsManyLogsShouldSucceed(t *testing.T) {
 	err = sc.Shutdown()
 	assert.Nil(t, err, err)
 
+	stats := sc.Statistics()
+	assert.Equal(t, uint64(ExpectedLogs), stats.Events.Enqueued())
+	assert.Equal(t, uint64(ExpectedLogs), stats.Events.Processed())
+	assert.Equal(t, uint64(0), stats.Events.Waiting())
+	assert.Equal(t, uint64(0), stats.Events.Dropped())
+	assert.Equal(t, uint64(0), stats.Events.Broken())
+	assert.Equal(t, 1.0, stats.Events.SuccessRate())
+
+	assert.Equal(t, uint64(0), stats.Buffers.Waiting())
+	assert.Equal(t, uint64(0), stats.Buffers.Dropped())
+	assert.Equal(t, uint64(0), stats.Buffers.Broken())
+	assert.Equal(t, 1.0, stats.Buffers.SuccessRate())
+
+	assert.Equal(t, 1.0, stats.Transfer.SuccessRate())
+
 	assert.Equal(t, seenKeys, expectedKeys)
 	assert.Equal(t, int(processedEvents.Load()), int(ExpectedLogs), "processed items")
 	assert.Equal(t, int(len(seenKeys)), int(ExpectedLogs), "unique items")

pkg/client/add_events_test.go

@@ -135,11 +135,29 @@ func TestAddEventsRetry(t *testing.T) {
 	assert.Nil(t, err)
 	err = sc.Shutdown()
 	assert.Nil(t, err)
-
 	assert.True(t, wasSuccessful.Load())
-	assert.Nil(t, err)
-
 	assert.Equal(t, attempt.Load(), succeedInAttempt)
+
+	stats := sc.Statistics()
+	assert.Equal(t, uint64(1), stats.Events.Enqueued())
+	assert.Equal(t, uint64(1), stats.Events.Processed())
+	assert.Equal(t, uint64(0), stats.Events.Waiting())
+	assert.Equal(t, uint64(0), stats.Events.Dropped())
+	assert.Equal(t, uint64(0), stats.Events.Broken())
+	assert.Equal(t, 1.0, stats.Events.SuccessRate())
+	assert.Equal(t, uint64(1), stats.Buffers.Enqueued())
+	assert.Equal(t, uint64(1), stats.Buffers.Processed())
+	assert.Equal(t, uint64(0), stats.Buffers.Waiting())
+	assert.Equal(t, uint64(0), stats.Buffers.Dropped())
+	assert.Equal(t, uint64(0), stats.Buffers.Broken())
+	assert.Equal(t, 1.0, stats.Buffers.SuccessRate())
+	assert.Equal(t, 1.0/float64(succeedInAttempt), stats.Transfer.SuccessRate())
+	assert.Equal(t, uint64(1), stats.Transfer.BuffersProcessed())
+	/* TODO: on my Mac it's 337 in GitHub action on ubuntu-latest it's 339
+	assert.Equal(t, uint64(0x3f3), stats.Transfer.BytesSent())
+	assert.Equal(t, uint64(0x151), stats.Transfer.BytesAccepted())
+	assert.Equal(t, 337.0, stats.Transfer.AvgBufferBytes())
+	*/
 }
 
 func TestAddEventsRetryAfterSec(t *testing.T) {
@@ -238,8 +256,11 @@ func TestAddEventsRetryAfterSec(t *testing.T) {
 	wasSuccessful.Store(false)
 	assert.Nil(t, err2)
 	assert.Nil(t, sc.LastError())
-	// info2 := httpmock.GetCallCountInfo()
-	// assert.CmpDeeply(info2, map[string]int{"POST https://example.com/api/addEvents": 3})
+
+	stats := sc.Statistics()
+	assert.Equal(t, uint64(2), stats.Buffers.Enqueued())
+	assert.Equal(t, uint64(0), stats.Buffers.Waiting())
+	assert.Equal(t, uint64(0), stats.Buffers.Dropped())
 }
 
 func TestAddEventsRetryAfterTime(t *testing.T) {
@@ -388,12 +409,27 @@ func TestAddEventsLargeEvent(t *testing.T) {
 	assert.Nil(t, err)
 	err = sc.Shutdown()
 	assert.Nil(t, err)
-
 	assert.True(t, wasSuccessful.Load())
-	assert.Nil(t, err)
 	assert.Nil(t, sc.LastError())
-	// info := httpmock.GetCallCountInfo()
-	// assert.CmpDeeply(info, map[string]int{"POST https://example.com/api/addEvents": 1})
+
+	stats := sc.Statistics()
+	assert.Equal(t, uint64(1), stats.Events.Enqueued())
+	assert.Equal(t, uint64(1), stats.Events.Processed())
+	assert.Equal(t, uint64(0), stats.Events.Waiting())
+	assert.Equal(t, uint64(0), stats.Events.Dropped())
+	assert.Equal(t, uint64(0), stats.Events.Broken())
+	assert.Equal(t, 1.0, stats.Events.SuccessRate())
+	assert.Equal(t, uint64(2), stats.Buffers.Enqueued())
+	assert.Equal(t, uint64(2), stats.Buffers.Processed())
+	assert.Equal(t, uint64(0), stats.Buffers.Waiting())
+	assert.Equal(t, uint64(0), stats.Buffers.Dropped())
+	assert.Equal(t, uint64(0), stats.Buffers.Broken())
+	assert.Equal(t, 1.0, stats.Buffers.SuccessRate())
+	assert.Equal(t, 1.0, stats.Transfer.SuccessRate())
+	assert.Equal(t, uint64(2), stats.Transfer.BuffersProcessed())
+	assert.Equal(t, uint64(0x5f006d), stats.Transfer.BytesSent())
+	assert.Equal(t, uint64(0x5f006d), stats.Transfer.BytesAccepted())
+	assert.Equal(t, 3113014.5, stats.Transfer.AvgBufferBytes())
 }
 
 func TestAddEventsLargeEventThatNeedEscaping(t *testing.T) {
@@ -579,6 +615,21 @@ func TestAddEventsDoNotRetryForever(t *testing.T) {
 	assert.Nil(t, err)
 	err = sc.Shutdown()
 
+	stats := sc.Statistics()
+	assert.Equal(t, uint64(1), stats.Events.Enqueued())
+	assert.Equal(t, uint64(1), stats.Events.Processed())
+	assert.Equal(t, uint64(0), stats.Events.Waiting())
+	assert.Equal(t, uint64(0), stats.Events.Dropped())
+	assert.Equal(t, uint64(0), stats.Events.Broken())
+	assert.Equal(t, 1.0, stats.Events.SuccessRate())
+	assert.Equal(t, uint64(1), stats.Buffers.Enqueued())
+	assert.Equal(t, uint64(0), stats.Buffers.Processed())
+	assert.Equal(t, uint64(0), stats.Buffers.Waiting())
+	assert.Equal(t, uint64(1), stats.Buffers.Dropped())
+	assert.Equal(t, uint64(0), stats.Buffers.Broken())
+	assert.Equal(t, 0.0, stats.Buffers.SuccessRate())
+	assert.Equal(t, 0.0, stats.Transfer.SuccessRate())
+
 	assert.NotNil(t, err)
 	assert.Errorf(t, err, "some buffers were dropped during finishing - 1")
 	assert.GreaterOrEqual(t, attempt.Load(), int32(2))

pkg/client/client.go

@@ -307,8 +307,9 @@ func (client *DataSetClient) listenAndSendBufferForSession(session string, ch ch
 				client.buffersProcessed.Add(1)
 				continue
 			}
-			client.sendBufferWithRetryPolicy(buf)
-			client.buffersProcessed.Add(1)
+			if client.sendBufferWithRetryPolicy(buf) {
+				client.buffersProcessed.Add(1)
+			}
 		} else {
 			client.Logger.Error(
 				"Cannot convert message to Buffer",
@@ -322,7 +323,7 @@ func (client *DataSetClient) listenAndSendBufferForSession(session string, ch ch
 }
 
 // Sends buffer to DataSet. If not succeeds and try is possible (it retryable), try retry until possible (timeout)
-func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) {
+func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) bool {
 	// Do not use NewExponentialBackOff since it calls Reset and the code here must
 	// call Reset after changing the InitialInterval (this saves an unnecessary call to Now).
 	expBackoff := backoff.ExponentialBackOff{
@@ -347,7 +348,7 @@ func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) {
 				lastHttpStatus = HttpErrorHasErrorMessage
 				client.LastHttpStatus.Store(lastHttpStatus)
 				client.onBufferDrop(buf, lastHttpStatus, err)
-				break // exit loop (failed to send buffer)
+				return false // exit loop (failed to send buffer)
 			}
 			lastHttpStatus = HttpErrorCannotConnect
 			client.LastHttpStatus.Store(lastHttpStatus)
@@ -375,7 +376,7 @@ func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) {
 		if isOkStatus(lastHttpStatus) {
 			// everything was fine, there is no need for retries
 			client.bytesAPIAccepted.Add(uint64(payloadLen))
-			break // exit loop (buffer sent)
+			return true // exit loop (buffer sent)
 		}
 
 		backoffDelay := expBackoff.NextBackOff()
@@ -384,7 +385,7 @@ func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) {
 			// throw away the batch
 			err = fmt.Errorf("max elapsed time expired %w", err)
 			client.onBufferDrop(buf, lastHttpStatus, err)
-			break // exit loop (failed to send buffer)
+			return false // exit loop (failed to send buffer)
 		}
 
 		if isRetryableStatus(lastHttpStatus) {
@@ -406,7 +407,7 @@ func (client *DataSetClient) sendBufferWithRetryPolicy(buf *buffer.Buffer) {
 		} else {
 			err = fmt.Errorf("non recoverable error %w", err)
 			client.onBufferDrop(buf, lastHttpStatus, err)
-			break // exit loop (failed to send buffer)
+			return false // exit loop (failed to send buffer)
 		}
 		retryNum++
 	}
@@ -420,9 +421,8 @@ func (client *DataSetClient) statisticsSweeper() {
 	}
 }
 
-func (client *DataSetClient) logStatistics() {
-	mb := float64(1024 * 1024)
-
+// Statistics returns statistics about events, buffers processing from the start time
+func (client *DataSetClient) Statistics() *Statistics {
 	// for how long are events being processed
 	firstAt := time.Unix(0, client.firstReceivedAt.Load())
 	lastAt := time.Unix(0, client.lastAcceptedAt.Load())
@@ -431,54 +431,100 @@ func (client *DataSetClient) logStatistics() {
 
 	// if nothing was processed, do not log statistics
 	if processingInSec <= 0 {
-		return
+		return nil
 	}
 
 	// log buffer stats
 	bProcessed := client.buffersProcessed.Load()
 	bEnqueued := client.buffersEnqueued.Load()
 	bDropped := client.buffersDropped.Load()
 	bBroken := client.buffersBroken.Load()
-	client.Logger.Info(
-		"Buffers' Queue Stats:",
-		zap.Uint64("processed", bProcessed),
-		zap.Uint64("enqueued", bEnqueued),
-		zap.Uint64("dropped", bDropped),
-		zap.Uint64("broken", bBroken),
-		zap.Uint64("waiting", bEnqueued-bProcessed-bDropped-bBroken),
-		zap.Float64("processingS", processingInSec),
-	)
+
+	buffersStats := QueueStats{
+		bEnqueued,
+		bProcessed,
+		bDropped,
+		bBroken,
+		processingDur,
+	}
 
 	// log events stats
 	eProcessed := client.eventsProcessed.Load()
 	eEnqueued := client.eventsEnqueued.Load()
 	eDropped := client.eventsDropped.Load()
 	eBroken := client.eventsBroken.Load()
+
+	eventsStats := QueueStats{
+		eEnqueued,
+		eProcessed,
+		eDropped,
+		eBroken,
+		processingDur,
+	}
+
+	// log transferred stats
+	bAPISent := client.bytesAPISent.Load()
+	bAPIAccepted := client.bytesAPIAccepted.Load()
+	transferStats := TransferStats{
+		bAPISent,
+		bAPIAccepted,
+		bProcessed,
+		processingDur,
+	}
+
+	return &Statistics{
+		Buffers:  buffersStats,
+		Events:   eventsStats,
+		Transfer: transferStats,
+	}
+}
+
+func (client *DataSetClient) logStatistics() {
+	stats := client.Statistics()
+	if stats == nil {
+		return
+	}
+
+	b := stats.Buffers
+	client.Logger.Info(
+		"Buffers' Queue Stats:",
+		zap.Uint64("processed", b.Processed()),
+		zap.Uint64("enqueued", b.Enqueued()),
+		zap.Uint64("dropped", b.Dropped()),
+		zap.Uint64("broken", b.Broken()),
+		zap.Uint64("waiting", b.Waiting()),
+		zap.Float64("successRate", b.SuccessRate()),
+		zap.Float64("processingS", b.ProcessingTime().Seconds()),
+		zap.Duration("processing", b.ProcessingTime()),
+	)
+
+	// log events stats
+	e := stats.Events
 	client.Logger.Info(
 		"Events' Queue Stats:",
-		zap.Uint64("processed", eProcessed),
-		zap.Uint64("enqueued", eEnqueued),
-		zap.Uint64("dropped", eDropped),
-		zap.Uint64("broken", eBroken),
-		zap.Uint64("waiting", eEnqueued-eProcessed-eDropped-eBroken),
-		zap.Float64("processingS", processingInSec),
+		zap.Uint64("processed", e.Processed()),
+		zap.Uint64("enqueued", e.Enqueued()),
+		zap.Uint64("dropped", e.Dropped()),
+		zap.Uint64("broken", e.Broken()),
+		zap.Uint64("waiting", e.Waiting()),
+		zap.Float64("successRate", e.SuccessRate()),
+		zap.Float64("processingS", e.ProcessingTime().Seconds()),
+		zap.Duration("processing", e.ProcessingTime()),
 	)
 
 	// log transferred stats
-	bAPISent := float64(client.bytesAPISent.Load())
-	bAPIAccepted := float64(client.bytesAPIAccepted.Load())
-	throughput := bAPIAccepted / mb / processingInSec
-	successRate := (bAPIAccepted + 1) / (bAPISent + 1)
-	perBuffer := (bAPIAccepted) / float64(bProcessed)
+	mb := float64(1024 * 1024)
+	t := stats.Transfer
 	client.Logger.Info(
 		"Transfer Stats:",
-		zap.Float64("bytesSentMB", bAPISent/mb),
-		zap.Float64("bytesAcceptedMB", bAPIAccepted/mb),
-		zap.Float64("throughputMBpS", throughput),
-		zap.Float64("perBufferMB", perBuffer/mb),
-		zap.Float64("successRate", successRate),
-		zap.Float64("processingS", processingInSec),
-		zap.Duration("processing", processingDur),
+		zap.Float64("bytesSentMB", float64(t.BytesSent())/mb),
+		zap.Float64("bytesAcceptedMB", float64(t.BytesAccepted())/mb),
+		zap.Float64("throughputMBpS", t.ThroughputBpS()/mb),
+		zap.Uint64("buffersProcessed", t.BuffersProcessed()),
+		zap.Float64("perBufferMB", t.AvgBufferBytes()/mb),
+		zap.Float64("successRate", t.SuccessRate()),
+		zap.Float64("processingS", t.ProcessingTime().Seconds()),
+		zap.Duration("processing", t.ProcessingTime()),
 	)
 }