[chore] remove converter type from stanza

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()