otel-collector/processor/o11ylogspipelineprocessor/processor.go

// A sync implementation of stanza based otel collector processor.Logs
// logstransform processor in opentelemetry-collector-contrib is async
package o11ylogspipelineprocessor

import (
	"context"
	"errors"
	"math"
	"runtime"
	"slices"
	"sync"

	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/adapter"
	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/pipeline"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/extension/xextension/storage"
	"go.opentelemetry.io/collector/pdata/plog"
	"go.opentelemetry.io/collector/processor"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/metric"
	"go.uber.org/zap"

	_ "github.com/hanzoai/otel-collector/pkg/parser/grok" // ensure grok parser gets registered.
	"github.com/hanzoai/otel-collector/processor/o11ylogspipelineprocessor/internal/metadata"
)

type logsPipelineProcessor struct {
	telemetrySettings component.TelemetrySettings
	processorConfig   *Config

	consumer consumer.Logs

	telemetryBuilder *metadata.TelemetryBuilder
	otelAttrs        metric.MeasurementOption

	pipe          *pipeline.DirectedPipeline
	firstOperator operator.Operator
	emitter       *helper.BatchingLogEmitter
	fromConverter *adapter.FromPdataConverter
	shutdownFns   []component.ShutdownFunc
}

func newLogsPipelineProcessor(
	processorConfig *Config,
	set processor.Settings,
	nextConsumer consumer.Logs,
) (*logsPipelineProcessor, error) {
	tb, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
	if err != nil {
		return nil, err
	}
	p := &logsPipelineProcessor{
		telemetrySettings: set.TelemetrySettings,
		processorConfig:   processorConfig,
		consumer:          nextConsumer,
		telemetryBuilder:  tb,
		otelAttrs: metric.WithAttributeSet(attribute.NewSet(
			attribute.String("processor", set.ID.String()),
			attribute.String("otel.signal", "logs"),
		)),
	}

	p.emitter = helper.NewBatchingLogEmitter(set.TelemetrySettings, p.consumeStanzaLogEntries)

	pipe, err := pipeline.Config{
		Operators:     processorConfig.OperatorConfigs(),
		DefaultOutput: p.emitter,
	}.Build(set.TelemetrySettings)
	if err != nil {
		return nil, err
	}
	p.pipe = pipe

	return p, nil
}

func (*logsPipelineProcessor) Capabilities() consumer.Capabilities {
	return consumer.Capabilities{MutatesData: true}
}

func (p *logsPipelineProcessor) Start(ctx context.Context, _ component.Host) error {
	p.telemetrySettings.Logger.Info("Starting logsPipelineProcessor")

	wkrCount := int(math.Max(1, float64(runtime.NumCPU())))
	p.fromConverter = adapter.NewFromPdataConverter(p.telemetrySettings, wkrCount)

	// Start in reverse order of the data flow (pipeline → converterLoop → fromConverter)
	// so that the pipeline is ready before entries start arriving.
	if err := p.startPipeline(); err != nil {
		return err
	}
	p.startConverterLoop(ctx)
	p.startFromConverter()

	return nil
}

func (p *logsPipelineProcessor) Shutdown(ctx context.Context) error {
	p.telemetrySettings.Logger.Info("Stopping logsPipelineProcessor")

	// Call shutdown functions in reverse order of registration: fromConverter
	// stops first (no new entries), then converterLoop drains, then the
	// pipeline stops.
	for _, fn := range slices.Backward(p.shutdownFns) {

		if err := fn(ctx); err != nil {
			return err
		}
	}

	return nil
}

func (p *logsPipelineProcessor) startPipeline() error {
	// This processor does not use stanza's persistent storage.
	if err := p.pipe.Start(storage.NewNopClient()); err != nil {
		return err
	}
	p.shutdownFns = append(p.shutdownFns, func(_ context.Context) error {
		return p.pipe.Stop()
	})

	ops := p.pipe.Operators()
	if len(ops) == 0 {
		return errors.New("processor requires at least one operator to be configured")
	}
	p.firstOperator = ops[0]

	return nil
}

func (p *logsPipelineProcessor) startConverterLoop(ctx context.Context) {
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go p.converterLoop(ctx, wg)
	p.shutdownFns = append(p.shutdownFns, func(_ context.Context) error {
		wg.Wait()
		return nil
	})
}

func (p *logsPipelineProcessor) startFromConverter() {
	p.fromConverter.Start()
	p.shutdownFns = append(p.shutdownFns, func(_ context.Context) error {
		p.fromConverter.Stop()
		return nil
	})
}

func (p *logsPipelineProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
	p.telemetryBuilder.ProcessorIncomingItems.Add(ctx, int64(ld.LogRecordCount()), p.otelAttrs)
	return p.fromConverter.Batch(ld)
}

// converterLoop reads stanza entries produced by fromConverter and feeds them
// serially into the first operator of the stanza pipeline.
func (p *logsPipelineProcessor) converterLoop(ctx context.Context, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		select {
		case <-ctx.Done():
			p.telemetrySettings.Logger.Debug("converter loop stopped")
			return

		case entries, ok := <-p.fromConverter.OutChannel():
			if !ok {
				p.telemetrySettings.Logger.Debug("fromConverter channel got closed")
				return
			}

			for _, e := range entries {
				if err := p.firstOperator.Process(ctx, e); err != nil {
					p.telemetrySettings.Logger.Error("processor encountered an issue with the pipeline", zap.Error(err))
					break
				}
			}
		}
	}
}

// consumeStanzaLogEntries is the BatchingLogEmitter callback: a batch of
// processed entries arrives here, gets converted back to plog, and is
// forwarded to the next consumer. Errors are logged but not propagated.
//
// We deliberately use the local convertEntriesToPlogs (utils.go) instead of
// adapter.ConvertEntries because the local version filters out attributes
// prefixed with o11ystanzaentry.InternalTempAttributePrefix, which the
// O11y operator suite uses for scratch state. Those keys must not be
// emitted to the next consumer.
func (p *logsPipelineProcessor) consumeStanzaLogEntries(ctx context.Context, entries []*entry.Entry) {
	pLogs := convertEntriesToPlogs(entries)
	p.telemetryBuilder.ProcessorOutgoingItems.Add(ctx, int64(pLogs.LogRecordCount()), p.otelAttrs)
	if err := p.consumer.ConsumeLogs(ctx, pLogs); err != nil {
		p.telemetrySettings.Logger.Error("processor encountered an issue with next consumer", zap.Error(err))
	}
}