[chore] Do regular flush with combined batches

pkg/stanza/operator/transformer/recombine/recombine.go

@@ -178,7 +178,7 @@ func (r *Transformer) flushLoop() {
 				if timeSinceFirstEntry < r.forceFlushTimeout {
 					continue
 				}
-				if err := r.flushSource(source, true); err != nil {
+				if err := r.flushSource(source, true, nil); err != nil {
 					r.Errorf("there was error flushing combined logs %s", err)
 				}
 			}
@@ -198,7 +198,10 @@ func (r *Transformer) Stop() error {
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	r.flushUncombined(ctx)
+	err := r.flushUncombined(ctx)
+	if err != nil {
+		return err
+	}
 
 	close(r.chClose)
 
@@ -241,7 +244,7 @@ func (r *Transformer) Process(ctx context.Context, e *entry.Entry) error {
 	// This is the first entry in the next batch
 	case matches && r.matchIndicatesFirst():
 		// Flush the existing batch
-		err := r.flushSource(s, true)
+		err := r.flushSource(s, true, nil)
 		if err != nil {
 			return err
 		}
@@ -255,7 +258,7 @@ func (r *Transformer) Process(ctx context.Context, e *entry.Entry) error {
 	// When matching on first entry, never batch partial first. Just emit immediately
 	case !matches && r.matchIndicatesFirst() && r.batchMap[s] == nil:
 		r.addToBatch(ctx, e, s)
-		return r.flushSource(s, true)
+		return r.flushSource(s, true, nil)
 	}
 
 	// This is neither the first entry of a new log,
@@ -305,7 +308,7 @@ func (r *Transformer) addToBatch(_ context.Context, e *entry.Entry, source strin
 	batch.recombined.WriteString(s)
 
 	if (r.maxLogSize > 0 && int64(batch.recombined.Len()) > r.maxLogSize) || len(batch.entries) >= r.maxBatchSize {
-		if err := r.flushSource(source, false); err != nil {
+		if err := r.flushSource(source, false, nil); err != nil {
 			r.Errorf("there was error flushing combined logs %s", err)
 		}
 	}
@@ -315,19 +318,20 @@ func (r *Transformer) addToBatch(_ context.Context, e *entry.Entry, source strin
 // flushUncombined flushes all the logs in the batch individually to the
 // next output in the pipeline. This is only used when there is an error
 // or at shutdown to avoid dropping the logs.
-func (r *Transformer) flushUncombined(ctx context.Context) {
+func (r *Transformer) flushUncombined(ctx context.Context) error {
 	for source := range r.batchMap {
-		for _, entry := range r.batchMap[source].entries {
-			r.Write(ctx, entry)
+		err := r.flushSource(source, true, ctx)
+		if err != nil {
+			return err
 		}
-		r.removeBatch(source)
 	}
 	r.ticker.Reset(r.forceFlushTimeout)
+	return nil
 }
 
 // flushSource combines the entries currently in the batch into a single entry,
 // then forwards them to the next operator in the pipeline
-func (r *Transformer) flushSource(source string, deleteSource bool) error {
+func (r *Transformer) flushSource(source string, deleteSource bool, ctx context.Context) error {
 	batch := r.batchMap[source]
 	// Skip flushing a combined log if the batch is empty
 	if batch == nil {
@@ -355,7 +359,7 @@ func (r *Transformer) flushSource(source string, deleteSource bool) error {
 		return err
 	}
 
-	r.Write(context.Background(), base)
+	r.Write(ctx, base)
 	if deleteSource {
 		r.removeBatch(source)
 	} else {

pkg/stanza/operator/transformer/recombine/recombine_test.go

@@ -735,7 +735,7 @@ func TestSourceBatchDelete(t *testing.T) {
 	require.NoError(t, recombine.Process(ctx, start))
 	require.NoError(t, recombine.Process(ctx, next))
 	require.Equal(t, 1, len(recombine.batchMap))
-	require.NoError(t, recombine.flushSource("file1", true))
+	require.NoError(t, recombine.flushSource("file1", true, nil))
 	require.Equal(t, 0, len(recombine.batchMap))
 	require.NoError(t, recombine.Stop())
 }