Skip to content

Commit

Permalink
[chore] remove converter type from stanza (#36288)
Browse files Browse the repository at this point in the history 
<!--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
#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. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Follow up to
#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]>
  • Loading branch information
@bacherfl
bacherfl authored Dec 5, 2024
1 parent ad7f371 commit 396c63d
Show file tree
Hide file tree
Showing 5 changed files with 46 additions and 440 deletions.
184 changes: 0 additions & 184 deletions pkg/stanza/adapter/converter.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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()
Expand Down
Loading

0 comments on commit 396c63d

Please sign in to comment.