[chore] remove converter type from stanza (open-telemetry#36288)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

This PR removes the `Converter` type that was previously used mainly by
the stanza receiver adapter (see
open-telemetry#35669 (comment)
for more details). Two other receivers were still using the converter to
generate test data within the unit tests, so those have been adapted as
well with this PR

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Follow up to
open-telemetry#35453

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Adapted unit tests that were still using the converter

---------

Signed-off-by: Florian Bacher <[email protected]>

Author: bacherfl

pkg/stanza/adapter/converter.go

@@ -4,162 +4,19 @@
 package adapter // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
 
 import (
-	"context"
 	"encoding/binary"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"math"
-	"runtime"
 	"sort"
 	"sync"
 
 	"github.com/cespare/xxhash/v2"
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
-	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
 )
 
-// Converter converts a batch of entry.Entry into plog.Logs aggregating translated
-// entries into logs coming from the same Resource.
-//
-// The diagram below illustrates the internal communication inside the Converter:
-//
-//	          ┌─────────────────────────────────┐
-//	          │ Batch()                         │
-//	┌─────────┤  Ingests batches of log entries │
-//	│         │  and sends them onto workerChan │
-//	│         └─────────────────────────────────┘
-//	│
-//	│ ┌───────────────────────────────────────────────────┐
-//	├─► workerLoop()                                      │
-//	│ │ ┌─────────────────────────────────────────────────┴─┐
-//	├─┼─► workerLoop()                                      │
-//	│ │ │ ┌─────────────────────────────────────────────────┴─┐
-//	└─┼─┼─► workerLoop()                                      │
-//	  └─┤ │   consumes sent log entries from workerChan,      │
-//	    │ │   translates received entries to plog.LogRecords, │
-//	    └─┤   and sends them on flushChan                     │
-//	      └─────────────────────────┬─────────────────────────┘
-//	                                │
-//	                                ▼
-//	    ┌─────────────────────────────────────────────────────┐
-//	    │ flushLoop()                                         │
-//	    │   receives log records from flushChan and sends     │
-//	    │   them onto pLogsChan which is consumed by          │
-//	    │   downstream consumers via OutChannel()             │
-//	    └─────────────────────────────────────────────────────┘
-type Converter struct {
-	set component.TelemetrySettings
-
-	// pLogsChan is a channel on which aggregated logs will be sent to.
-	pLogsChan chan plog.Logs
-
-	stopOnce sync.Once
-
-	// converterChan is an internal communication channel signaling stop was called
-	// prevents sending to closed channels
-	converterChan chan struct{}
-
-	// workerChan is an internal communication channel that gets the log
-	// entries from Batch() calls and it receives the data in workerLoop().
-	workerChan chan []*entry.Entry
-	// workerCount configures the amount of workers started.
-	workerCount int
-
-	// flushChan is an internal channel used for transporting batched plog.Logs.
-	flushChan chan plog.Logs
-
-	// wg is a WaitGroup that makes sure that we wait for spun up goroutines exit
-	// when Stop() is called.
-	wg sync.WaitGroup
-
-	// flushWg is a WaitGroup that makes sure that we wait for flush loop to exit
-	// when Stop() is called.
-	flushWg sync.WaitGroup
-}
-
-type converterOption interface {
-	apply(*Converter)
-}
-
-func withWorkerCount(workerCount int) converterOption {
-	return workerCountOption{workerCount}
-}
-
-type workerCountOption struct {
-	workerCount int
-}
-
-func (o workerCountOption) apply(c *Converter) {
-	c.workerCount = o.workerCount
-}
-
-func NewConverter(set component.TelemetrySettings, opts ...converterOption) *Converter {
-	set.Logger = set.Logger.With(zap.String("component", "converter"))
-	c := &Converter{
-		set:           set,
-		workerChan:    make(chan []*entry.Entry),
-		workerCount:   int(math.Max(1, float64(runtime.NumCPU()/4))),
-		pLogsChan:     make(chan plog.Logs),
-		converterChan: make(chan struct{}),
-		flushChan:     make(chan plog.Logs),
-	}
-	for _, opt := range opts {
-		opt.apply(c)
-	}
-	return c
-}
-
-func (c *Converter) Start() {
-	c.set.Logger.Debug("Starting log converter", zap.Int("worker_count", c.workerCount))
-
-	c.wg.Add(c.workerCount)
-	for i := 0; i < c.workerCount; i++ {
-		go c.workerLoop()
-	}
-
-	c.flushWg.Add(1)
-	go c.flushLoop()
-}
-
-func (c *Converter) Stop() {
-	c.stopOnce.Do(func() {
-		close(c.converterChan)
-
-		// close workerChan and wait for entries to be processed
-		close(c.workerChan)
-		c.wg.Wait()
-
-		// close flushChan and wait for flush loop to finish
-		close(c.flushChan)
-		c.flushWg.Wait()
-
-		// close pLogsChan so callers can stop processing
-		close(c.pLogsChan)
-	})
-}
-
-// OutChannel returns the channel on which converted entries will be sent to.
-func (c *Converter) OutChannel() <-chan plog.Logs {
-	return c.pLogsChan
-}
-
-// workerLoop is responsible for obtaining log entries from Batch() calls,
-// converting them to plog.LogRecords batched by Resource, and sending them
-// on flushChan.
-func (c *Converter) workerLoop() {
-	defer c.wg.Done()
-
-	for entries := range c.workerChan {
-		// Send plogs directly to flushChan
-		c.flushChan <- ConvertEntries(entries)
-	}
-}
-
 func ConvertEntries(entries []*entry.Entry) plog.Logs {
 	resourceHashToIdx := make(map[uint64]int)
 	scopeIdxByResource := make(map[uint64]map[string]int)
@@ -197,47 +54,6 @@ func ConvertEntries(entries []*entry.Entry) plog.Logs {
 	return pLogs
 }
 
-func (c *Converter) flushLoop() {
-	defer c.flushWg.Done()
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	for pLogs := range c.flushChan {
-		if err := c.flush(ctx, pLogs); err != nil {
-			c.set.Logger.Debug("Problem sending log entries",
-				zap.Error(err),
-			)
-		}
-	}
-}
-
-// flush flushes provided plog.Logs entries onto a channel.
-func (c *Converter) flush(ctx context.Context, pLogs plog.Logs) error {
-	doneChan := ctx.Done()
-
-	select {
-	case <-doneChan:
-		return fmt.Errorf("flushing log entries interrupted, err: %w", ctx.Err())
-
-	case c.pLogsChan <- pLogs:
-	}
-
-	return nil
-}
-
-// Batch takes in an entry.Entry and sends it to an available worker for processing.
-func (c *Converter) Batch(e []*entry.Entry) error {
-	// in case Stop was called do not process batch
-	select {
-	case <-c.converterChan:
-		return errors.New("logs converter has been stopped")
-	default:
-	}
-
-	c.workerChan <- e
-	return nil
-}
-
 // convert converts one entry.Entry into plog.LogRecord allocating it.
 func convert(ent *entry.Entry) plog.LogRecord {
 	dest := plog.NewLogRecord()