[processor/logstransformprocessor]: Race condition on shutdown

[BinaryFissionGames]

Component(s)

processor/logstransform

What happened?

Description

In the logstransform processor, it looks like the pipeline is shutdown, then the converter, then the fromConverter, which triggers the shutdown of the various goroutines here.

It's possible the pipeline and converter are quickly shutdown, while converterLoop is in the middle of processing entries:

opentelemetry-collector-contrib/processor/logstransformprocessor/processor.go

Lines 153 to 159 in ff1325c

	for _, e := range entries {
	// Add item to the first operator of the pipeline manually
	if err := ltp.firstOperator.Process(ctx, e); err != nil {
	ltp.logger.Error("processor encountered an issue with the pipeline", zap.Error(err))
	break
	}
	}

Since we don't guarantee that converter loop is stopped before shutting down the pipeline, it's possible for entries to be sent down the shut-down pipeline, where it can end up panicking the LogEmitter by sending on a closed channel (e.flush here sends on a channel that is closed when Close is called).

opentelemetry-collector-contrib/pkg/stanza/adapter/emitter.go

Lines 114 to 120 in ff1325c

	func (e *LogEmitter) Process(ctx context.Context, ent *entry.Entry) error {
	if oldBatch := e.appendEntry(ent); len(oldBatch) > 0 {
	e.flush(ctx, oldBatch)
	}
	return nil
	}

Steps to Reproduce

Shutting down the collector while a high-throughput pipeline is using the logstransform processor.

Expected Result

No panic

Actual Result

panic: send on closed channel

goroutine 101 [running]:
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter.(*LogEmitter).flush(...)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/adapter/emitter.go:159
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter.(*LogEmitter).Process(0xc0007f25a0, {0x87b4960, 0xc000a8ab40}, 0x0?)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/adapter/emitter.go:116 +0xb8
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper.(*WriterOperator).Write(0xc00030a700, {0x87b4960, 0xc000a8ab40}, 0xc00462c480?)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/helper/writer.go:53 +0xd6
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper.(*ParserOperator).ProcessWithCallback(0xc00030a700, {0x87b4960, 0xc000a8ab40}, 0x10?, 0x1?, 0x0)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/helper/parser.go:122 +0xa5
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper.(*ParserOperator).ProcessWith(...)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/helper/parser.go:98
github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/parser/csv.(*Parser).Process(0xc000a71f00, {0x87b4960, 0xc000a8ab40}, 0x0)
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/[email protected]/operator/parser/csv/csv.go:140 +0x176
github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor.(*logsTransformProcessor).converterLoop(0xc0008d9800, {0x87b4960, 0xc000a8ab40})
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/processor.go:155 +0x349
created by github.com/open-telemetry/opentelemetry-collector-contrib/processor/logstransformprocessor.(*logsTransformProcessor).Start
	/home/runner/go/pkg/mod/github.com/open-telemetry/opentelemetry-collector-contrib/processor/[email protected]/processor.go:115 +0x470

Collector version

v0.92.0

Environment information

No response

OpenTelemetry Collector configuration

No response

Log output

No response

Additional context

No response

[github-actions]

Pinging code owners:

• processor/logstransform: @djaglowski @dehaansa

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[BinaryFissionGames]

I'm working on a fix for this right now, seems like we just need to be more meticulous in making sure everything shuts down in the right order (from producer -> consumer)