oops

Signed-off-by: sspaink <[email protected]>

Author: sspaink

Diff for: v1/plugins/logs/eventBuffer.go

@@ -7,7 +7,6 @@ package logs
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"sync"
 
 	"github.com/open-policy-agent/opa/v1/logging"
@@ -44,6 +43,7 @@ func newEventBuffer(bufferSizeLimitEvents int64, client rest.Client, uploadPath
 
 func (b *eventBuffer) WithMetrics(m metrics.Metrics) *eventBuffer {
 	b.metrics = m
+	b.encoder.metrics = m
 	return b
 }
 
@@ -158,18 +158,22 @@ func (b *eventBuffer) readEvent() *bufferItem {
 	}
 }
 
-// processEvent serializes the event and determines if the ND cache needs to be dropped
-func (b *eventBuffer) processEvent(event *bufferItem) ([][]byte, error) {
+// Handle serializes the event and determines if the ND cache needs to be dropped
+func (b *eventBuffer) Handle(event *bufferItem) [][]byte {
+	if event == nil {
+		return nil
+	}
+
 	if event.chunk != nil { // this is a chunk that has failed to upload and is being retried
-		return [][]byte{event.chunk}, nil
+		return [][]byte{event.chunk}
 	}
 
 	serialized, err := json.Marshal(event.EventV1)
 
 	// The non-deterministic cache (NDBuiltinCache) could cause issues, if it is too big or can't be encoded try to drop it.
 	if err != nil || int64(len(serialized)) >= b.uploadSizeLimitBytes {
 		if event.NDBuiltinCache == nil {
-			return nil, fmt.Errorf("upload event size (%d) exceeds upload_size_limit_bytes (%d), dropping event with decision ID: %v",
+			b.logError("upload event size (%d) exceeds upload_size_limit_bytes (%d), dropping event with decision ID: %v",
 				int64(len(serialized)), b.uploadSizeLimitBytes, event.DecisionID)
 		}
 
@@ -179,10 +183,10 @@ func (b *eventBuffer) processEvent(event *bufferItem) ([][]byte, error) {
 		serialized, err = json.Marshal(event.EventV1)
 		if err != nil {
 			b.incrMetric(logEncodingFailureCounterName)
-			return nil, fmt.Errorf("encoding failure: %v, dropping event with decision ID: %v", err, event.DecisionID)
+			b.logError("encoding failure: %v, dropping event with decision ID: %v", err, event.DecisionID)
 		}
 		if int64(len(serialized)) > b.uploadSizeLimitBytes {
-			return nil, fmt.Errorf("upload event size (%d) exceeds upload_size_limit_bytes (%d), dropping event with decision ID: %v",
+			b.logError("upload event size (%d) exceeds upload_size_limit_bytes (%d), dropping event with decision ID: %v",
 				int64(len(serialized)), b.uploadSizeLimitBytes, event.DecisionID)
 		}
 
@@ -194,28 +198,17 @@ func (b *eventBuffer) processEvent(event *bufferItem) ([][]byte, error) {
 	if err != nil {
 		b.incrMetric(logEncodingFailureCounterName)
 		b.logError("encoding failure: %v, dropping event with decision ID: %v", err, event.DecisionID)
-		return nil, err
+		return nil
 	}
 
-	return result, nil
+	return result
 }
 
 // Input returns the buffer channel to be read from
 func (b *eventBuffer) Input() <-chan *bufferItem {
 	return b.buffer
 }
 
-// Handle takes an event and encodes it as a chunk
-func (b *eventBuffer) Handle(event *bufferItem) [][]byte {
-	result, err := b.processEvent(event)
-	if err != nil {
-		b.logError("%v", err)
-		return nil
-	}
-
-	return result
-}
-
 // Flush empties the encoder and returns whatever is ready
 func (b *eventBuffer) Flush() [][]byte {
 	result, err := b.encoder.Flush()

Diff for: v1/plugins/logs/plugin.go

@@ -984,11 +984,11 @@ func (p *Plugin) oneShot(ctx context.Context) error {
 		return &bufferEmpty{}
 	}
 
-	var finalErr error
 	for bs := oldBuffer.Pop(); bs != nil; bs = oldBuffer.Pop() {
-		err = uploadChunk(ctx, p.manager.Client(p.config.Service), *p.config.Resource, bs)
+		if err == nil {
+			err = uploadChunk(ctx, p.manager.Client(p.config.Service), *p.config.Resource, bs)
+		}
 		if err != nil {
-			finalErr = err
 			if p.limiter != nil {
 				events, decErr := newChunkDecoder(bs).decode()
 				if decErr != nil {
@@ -1007,7 +1007,7 @@ func (p *Plugin) oneShot(ctx context.Context) error {
 		}
 	}
 
-	return finalErr
+	return err
 }
 
 func (p *Plugin) reconfigure(ctx context.Context, config interface{}) {
@@ -1250,6 +1250,7 @@ func (p *Plugin) dropEvent(ctx context.Context, txn storage.Transaction, input a
 }
 
 func uploadChunk(ctx context.Context, client rest.Client, uploadPath string, data []byte) error {
+
 	resp, err := client.
 		WithHeader("Content-Type", "application/json").
 		WithHeader("Content-Encoding", "gzip").