plugin/decision: encoder return event same size as limit immediately (#7928)

In the encoder, if the event size equals the limit it was added to the buffer.
Instead return it to avoid unnecessary downsize step that could lead to an infinite loop.

Signed-off-by: Sebastian Spaink <[email protected]>

Author: sspaink

v1/plugins/logs/encoder.go

@@ -135,7 +135,7 @@ func (enc *chunkEncoder) Encode(event EventV1, eventBytes []byte) ([][]byte, err
 		}
 	}
 
-	if int64(len(eventBytes)+enc.bytesWritten+1) <= enc.uncompressedLimit {
+	if int64(len(eventBytes)+enc.bytesWritten+1) < enc.uncompressedLimit {
 		return nil, enc.appendEvent(eventBytes)
 	}
 
@@ -167,16 +167,12 @@ func (enc *chunkEncoder) Encode(event EventV1, eventBytes []byte) ([][]byte, err
 		}
 
 		currentSize := len(result)
-		if currentSize < int(enc.limit) {
+		if currentSize <= int(enc.limit) {
 			// success! the incoming chunk doesn't have to lose the ND cache and can go into a chunk by itself
 			// scale up the uncompressed limit using the uncompressed event size as a base
-			err = enc.appendEvent(eventBytes)
-			if err != nil {
-				return nil, err
-			}
 			enc.uncompressedLimit = int64(len(eventBytes))
 			enc.scaleUp()
-			return nil, nil
+			return [][]byte{result}, nil
 		}
 
 		// The ND cache has to be dropped, record this size as a known maximum event size
@@ -185,7 +181,6 @@ func (enc *chunkEncoder) Encode(event EventV1, eventBytes []byte) ([][]byte, err
 		}
 
 		// 2. Drop the ND cache and see if the incoming event can fit within the current chunk without the cache (so we can maximize chunk size)
-		enc.initialize()
 		enc.incrMetric(encLogExUploadSizeLimitCounterName)
 		// If there's no ND builtins cache in the event, then we don't need to retry encoding anything.
 		if event.NDBuiltinCache == nil {

v1/plugins/logs/encoder_test.go

@@ -281,7 +281,7 @@ func TestChunkEncoderSizeLimit(t *testing.T) {
 		t.Fatal(err)
 	}
 	if len(chunks) != 1 {
-		t.Errorf("Unexpected result: %v", result)
+		t.Errorf("Unexpected chunk: %v", chunks)
 	}
 	// the incoming event doesn't fit, but the previous small event size is between 90-100% capacity so chunk is returned
 	// the incoming event is too large and will be dropped
@@ -324,6 +324,31 @@ func TestChunkEncoderSizeLimit(t *testing.T) {
 	if enc.metrics.Counter(logEncodingFailureCounterName).Value().(uint64) != 1 {
 		t.Errorf("Expected one encoding failure but got: %v", enc.metrics.Counter(logEncodingFailureCounterName).Value().(uint64))
 	}
+
+	// 179 is the size of the event compressed into a chunk by itself
+	// the eventBytes size is 197 uncompressed, bigger than the limit
+	// this tests that a chunk should be returned containing this single event
+	// because after compression it adheres to the limit
+	enc = newChunkEncoder(179).WithMetrics(metrics.New())
+	chunks, err = enc.Encode(event, eventBytes)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(chunks) != 1 {
+		t.Errorf("Unexpected chunk: %v", chunks)
+	}
+
+	// 198 is the size of eventBytes + 1, +1 accounting for the closing bracket
+	// this tests that a chunk should be returned containing this single event
+	// because after compression it adheres to the limit
+	enc = newChunkEncoder(198).WithMetrics(metrics.New())
+	chunks, err = enc.Encode(event, eventBytes)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(chunks) != 1 {
+		t.Errorf("Unexpected chunk: %v", chunks)
+	}
 }
 
 func TestChunkEncoderAdaptive(t *testing.T) {
@@ -348,7 +373,7 @@ func TestChunkEncoderAdaptive(t *testing.T) {
 			expectedMaxEventsInChunk:  1,
 			expectedScaleUpEvents:     1,
 			expectedScaleDownEvents:   0,
-			expectedEquiEvents:        999,
+			expectedEquiEvents:        998,
 		},
 		{
 			// 61 events can fit, but takes some guessing before it gets to the uncompressed limit 7200

v1/plugins/logs/eventBuffer.go

@@ -127,21 +127,22 @@ func (b *eventBuffer) Upload(ctx context.Context, client rest.Client, uploadPath
 	}
 
 	for range eventLen {
-		event := b.readEvent()
-		if event == nil {
+		bufItem := b.readBufItem()
+		if bufItem == nil {
 			break
 		}
 
 		var result [][]byte
-		if event.chunk != nil {
-			result = [][]byte{event.chunk}
+		if bufItem.chunk != nil {
+			result = [][]byte{bufItem.chunk}
 		} else {
+			event := bufItem.EventV1
 			eventBytes, err := json.Marshal(&event)
 			if err != nil {
 				return err
 			}
 
-			result, err = b.enc.Encode(*event.EventV1, eventBytes)
+			result, err = b.enc.Encode(*event, eventBytes)
 			if err != nil {
 				b.incrMetric(logEncodingFailureCounterName)
 				if b.logger != nil {
@@ -189,7 +190,7 @@ func (b *eventBuffer) uploadChunks(ctx context.Context, result [][]byte, client
 }
 
 // readEvent does a nonblocking read from the event buffer
-func (b *eventBuffer) readEvent() *bufferItem {
+func (b *eventBuffer) readBufItem() *bufferItem {
 	select {
 	case event := <-b.buffer:
 		return event

v1/plugins/logs/eventBuffer_test.go

@@ -143,8 +143,8 @@ func TestEventBuffer_Upload(t *testing.T) {
 			uploadSizeLimitBytes: 196, // Each test event is 195 bytes
 			handleFunc: func(w http.ResponseWriter, r *http.Request) {
 				events := decodeLogEvent(t, r.Body)
-				if len(events) != 2 {
-					t.Errorf("expected 2 events, got %d", len(events))
+				if len(events) != 1 {
+					t.Errorf("expected 1 events, got %d", len(events))
 				}
 				w.WriteHeader(http.StatusOK)
 			},