Add support for stream decoding

[axw]

Component(s)

extension/encoding

Is your feature request related to a problem? Please describe.

The encoding extension currently provides interfaces for extensions that embed the pdata unmarshaler interfaces, which look like as follows for logs (the same sort of interface exists for traces, metrics, profiles):

// Unmarshaler unmarshalls bytes into pdata.Logs.
type Unmarshaler interface {
	// UnmarshalLogs the given bytes into pdata.Logs.
	// If the error is not nil, the returned pdata.Logs cannot be used.
	UnmarshalLogs(buf []byte) (Logs, error)
}

This interface presents two challenges:

• We cannot stream into the unmarshaler.
• We cannot stream the results of the unmarshaling.

Together this means means we must have both the full unmarshaled input AND the full set of unmarshaled logs in memory before we can proceed to process any data.

Describe the solution you'd like

• I would like to be able to pass in an io.Reader so the unmarshaler to avoid reading the full input into memory.
• I would like an optional interface for unmarshalers to incrementally decode logs.

Concretely, I'm thinking that we should introduce interfaces like follows, similar to the encoding/json.Decoder API:

// LogsDecoderExtension is an extension that decodes logs from a stream.
type StreamingLogsUnmarshalerExtension interface {
	extension.Extension
	NewLogsDecoder(io.Reader) (LogsDecoder, error)
}

type LogsDecoder interface {
	// DecodeLogs decodes logs and appends them to the given plog.Logs.
	//
	// DecodeLogs returns io.EOF if there are no more logs to be decoded.
	DecodeLogs(to plog.Logs) error
}

We could also provide types to adapt the Unmarshaler interfaces to Decoder interfaces, which would look something like:

type LogsDecoderAdapter struct {
	plog.Unmarshaler

	mu sync.Mutex
	r io.Reader
}

func (a *LogsDecoderAdapter) DecodeLogs(to plog.Logs) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.r == nil {
		return io.EOF
	}
	data, err := io.ReadAll(a.r)
	if err != nil {
		return err
	}
	a.r = nil
	logs, err := a.UnmarshalLogs(data)
	if err != nil {
		return err
	}
	// TODO append logs to 'to'
	return nil
}

Describe alternatives you've considered

Initially I thought of using an iter.Seq coroutine approach, but that would make it awkward to deal with errors mid stream.

Additional context

No response

[github-actions]

Pinging code owners:

• extension/encoding: @atoulme @dao-jun @dmitryax @MovieStoreGuy @VihasMakwana

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

[github-actions]

Pinging code owners for extension/encoding
extension/encoding/avrologencodingextension
extension/encoding/awscloudwatchmetricstreamsencodingextension
extension/encoding/googlecloudlogentryencodingextension
extension/encoding/jaegerencodingextension
extension/encoding/jsonlogencodingextension
extension/encoding/otlpencodingextension
extension/encoding/skywalkingencodingextension
extension/encoding/textencodingextension
extension/encoding/zipkinencodingextension: @atoulme @dao-jun @dmitryax @MovieStoreGuy @VihasMakwana. See Adding Labels via Comments if you do not have permissions to add labels yourself. For example, comment '/label priority:p2 -needs-triaged' to set the priority and remove the needs-triaged label.

[MovieStoreGuy]

I am for the first approach since it can hide a lot of the boring details, I worry the adapter in practice might actual cause a lot of contention if the mutex is kept there.