change LogEmitter to process entries in batches

Instead of adding each entry in a batch separately to the LogEmitter's buffer, add the whole batch in one operation.

Author: andrzej-stencel

pkg/stanza/operator/helper/emitter.go

@@ -96,8 +96,24 @@ func (e *LogEmitter) Stop() error {
 
 // ProcessBatch emits the entries to the consumerFunc
 func (e *LogEmitter) ProcessBatch(ctx context.Context, entries []*entry.Entry) error {
-	for i := range entries {
-		_ = e.Process(ctx, entries[i])
+	if oldBatch := e.appendEntries(entries); len(oldBatch) > 0 {
+		e.consumerFunc(ctx, oldBatch)
+	}
+
+	return nil
+}
+
+// appendEntry appends the entry to the current batch. If maxBatchSize is reached, a new batch will be made, and the old batch
+// (which should be flushed) will be returned
+func (e *LogEmitter) appendEntries(entries []*entry.Entry) []*entry.Entry {
+	e.batchMux.Lock()
+	defer e.batchMux.Unlock()
+
+	e.batch = append(e.batch, entries...)
+	if uint(len(e.batch)) >= e.maxBatchSize {
+		var oldBatch []*entry.Entry
+		oldBatch, e.batch = e.batch, make([]*entry.Entry, 0, e.maxBatchSize)
+		return oldBatch
 	}
 
 	return nil