Log Records passed to log processor OnEmit is non-mutable

[MrAlias]

Blocked by:

• Logs: Allow the registered processor to be an independent pipeline opentelemetry-specification#4010

Description

The specification requires that log processors be provided a LogRecordProcessor:

OnEmit

[...]

Parameters:

• logRecord - a ReadWriteLogRecord for the emitted LogRecord.

Our current implementation Defines the Processor interface as such:

type Processor interface {
	// OnEmit is called when a Record is emitted.
	//
	// OnEmit will be called independent of Enabled. Implementations need to
	// validate the arguments themselves before processing.
	//
	// Implementation should not interrupt the record processing
	// if the context is canceled.
	//
	// All retry logic must be contained in this function. The SDK does not
	// implement any retry logic. All errors returned by this function are
	// considered unrecoverable and will be reported to a configured error
	// Handler.
	//
	// Before modifying a Record, the implementation must use Record.Clone
	// to create a copy that shares no state with the original.
	OnEmit(ctx context.Context, record Record) error
/* ... */
}

Any modification made to the record passed to a processor's OnEmit method will not be applied to the subsequent processor. The record is not mutable, it does not comply with the specification requirement.

[MrAlias]

One solution is to have OnEmit accept a reference:

OnEmit(ctx context.Context, record *Record) error

This will mean that every record will need to be allocated to the heap.

[MrAlias]

Another solution is to have OnEmit return a Record

OnEmit(ctx context.Context, record Record) (Record, error)

The returned Record can be the same record passed, or a mutated clone of the Record. That returned value will be passed to the next processor in the SDK's processing chain.

[pellared]

I could work on this but it would be nice to have #5093 merged first to reduce the amount of conflicts.

[MrAlias]

I could work on this but it would be nice to have #5093 merged first to reduce the amount of conflicts.

Merged

[pellared]

I remember what was my idea.

It was to use Processor as decorators in order to make modifications. A processor can modify the received record before passing to the next processor:

type mutatingProcessor struct {
  wrapped log.Processor
}

func (mutatingProcessor p) OnEmit(ctx context.Context, record Record) error {
  r.SetBody(log.StringValue("REDACTED"))
  return p.wrapped.OnEmit(ctx, r.Clone())
}

Thanks to it you can have even two "processor" pipelines which do not interfere with each other. So you can have one processor pipeline which e.g. has a processor redacts secrets and wraps a simple processor with stdout exporter and then a second pipeline which uses a batch processor (no redacting) with OTLP exporter:

log.NewLoggerProvider(
  log.WithProcessor(mutatingProcessor{ log.NewSimpleProcessor(stdoutExp) }), // stdout exporter with simple processor, the log records' body are redacted
  log.WithProcessor(log.NewBatchProcessor(otlpExp)), // otlp exporter with a batch processor
)

@MrAlias, can we close this issue? I do not see how this implementation would break the specification and I think it is a safer and more usable design.

[MrAlias]

No, I don't think we can close the issue. We are not compatible with the specification. The specification states the log records are mutable and that processors are executed in order. The processors after a mutation should receive the mutated record.

[pellared]

The processors after a mutation should receive the mutated record.

Where is it defined?

What do you think about updating the specification?

[MrAlias]

Where is it defined?

In the definition of a read/write record.

[pellared]

I think it is "in spirit" of the specification. I would rather ask the Specification SIG if it is an acceptable design and clarify the specification.

[pellared]

I created open-telemetry/opentelemetry-specification#4010