Introduce end to end streaming support for encoding extensions (#45567)

### Description

This PR introduce interfaces for end-to-end streaming support for
encoding extensions. This PR contains,

- Introduce streaming contracts for encoding extensions
- Abstracted experimental helpers for streaming with a dedicated module
`pkg/xstreamencoding`
- To prove concept, adopting streaming support for
`textencodingextension`

The streaming interface contract is as below,

```go
// LogsDecoder unmarshals logs from a stream, returning one batch per DecodeLogs call.
type LogsDecoder interface {
	// DecodeLogs is expected to be called iteratively to read all derived plog.Logs batches from the stream.
	// The last batch of logs should be returned with a nil error. io.EOF error should follow on the subsequent call.
	DecodeLogs() (plog.Logs, error)
	// OffSet returns the current offset read from the stream.
	// The exact meaning of the offset may vary by decoder (e.g. bytes, lines, records).
	// You may use this value with WithOffset option to resume reading from the same offset when retrying after a failure.
	Offset() int64
}

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

// MetricsDecoder unmarshals metrics from a stream, returning one batch per DecodeMetrics call.
type MetricsDecoder interface {
	// DecodeMetrics is expected to be called iteratively to read all derived pmetric.Metrics batches from the stream.
	// The last batch of metrics should be returned with a nil error. io.EOF error should follow on the subsequent call.
	DecodeMetrics() (pmetric.Metrics, error)
	// OffSet returns the current offset read from the stream.
	// The exact meaning of the offset may vary by decoder (e.g. bytes, lines, records).
	// You may use this value with WithOffset option to resume reading from the same offset when retrying after a failure.
	Offset() int64
}

// MetricsDecoderExtension is an extension that unmarshals metrics from a stream.
type MetricsDecoderExtension interface {
	extension.Extension
	NewMetricsDecoder(reader io.Reader, options ...DecoderOption) (MetricsDecoder, error)
}
```

### Why Streaming  ?

The need of streaming arises when dealing with blobs. For example,
consider S3 objects as a signal source. These blobs can be large. If
there is no end to end streaming, then OTel collector sourcing &
decoding signals from blobs requires high memory allocation.

With end-to-end streaming, the need for high memory goes away. And it
allows OTel collector to process signals in batches and emit them to
downstream consumers.

```mermaid
flowchart TB
    subgraph WITHOUT_STREAMING["Without Streaming "]
        S3A["S3 Blob"]
        OTelA["OTel Collector -  Load entire blob into memory"]
        ConsumerA["Downstream Consumer"]

        S3A --> OTelA --> ConsumerA
    end

    subgraph WITH_STREAMING["With Streaming"]
        S3B["S3 Blob"]
        OTelB["OTel Collector - Stream & process batches"]
        Batch1["Batch 1"]
        Batch2["Batch 2"]
        Batch3["Batch 3"]
        ConsumerB["Downstream Consumer"]

        S3B --> OTelB
        OTelB --> Batch1 --> ConsumerB
        OTelB --> Batch2 --> ConsumerB
        OTelB --> Batch3 --> ConsumerB
    end
```

### Proof of concept: OTel collector as a Lambda

Setup using combined changes of this PR, AWS logs adoption -
#45804
& AWS metrics adoptions -
#45805

- **Setup** : A collector deployment in AWS Lambda with [AWS Lambda
Receiver](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/receiver/awslambdareceiver)
- **Challenge** : AWS CloudTrail logs are stored as a JSON. Internally
there's a record array that fits streaming solutionProcess. Further, AWS
Lambda has defined memory limits. In the POC this is set to 512MB
- **Test input**: Predictable S3 blobs with CloudTrail logs from
[Data-gen](https://github.com/Kavindu-Dodan/data-gen)


I ran the test round with CloudTrail logs of 30MB each file. This
translates to around ~30K trails.

#### Without streaming

With no end-to-end streaming, Lambda's max memory consumption is at
**360MB/512MB**.

<details>
<summary> Non streaming memory consumption</summary>
<img width="401" height="547" alt="image"
src="https://github.com/user-attachments/assets/9d647057-928f-4be2-b7b1-1d8bbd164541"
/>
</details>

#### With streaming - using components from this PR

With end to end streaming, memory consumption max was around
**189MB/512MB** This leaves space for spikes which can go upto 50MB per
known CloudTrail S3 file limits.

<details>
<summary> Streaming memory consumption</summary>
<img width="401" height="547" alt="image"
src="https://github.com/user-attachments/assets/a6f3103b-0375-40c4-85bc-dbdf2f32ba19"
/>
</details>

#### POC conclusions

- Streaming consumed 48% less memory
- This is compromising 14% of CPU runtime (this is the tradeoff)

Above shows why streaming is important and can be a good tradeoff to
save costs.

### Link to tracking issue

Fixes
#38780

#### Testing

Updated unit tests

### Documentation

Code level documentation is added with example usage pattern for
`textencodingextension`


___

If reviewing, recommends checking out and see details (`gh pr checkout
45567`)

---------

Signed-off-by: Kavindu Dodanduwa <kavindu.dodanduwa@elastic.co>

Author: Kavindu-Dodan

.chloggen/config.yaml

@@ -179,6 +179,7 @@ components:
     - pkg/translator/zipkin
     - pkg/winperfcounters
     - pkg/xk8stest
+    - pkg/xstreamencoding
     - processor/akamaidetector
     - processor/alibabaecsdetector
     - processor/attributes

.chloggen/feat_encoding-extension-streaming.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: extension/encoding
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Introduce streaming support for encoding extensions
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38780]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

.github/CODEOWNERS

@@ -192,6 +192,7 @@ pkg/translator/splunk/                                           @open-telemetry
 pkg/translator/zipkin/                                           @open-telemetry/collector-contrib-approvers @MovieStoreGuy @andrzej-stencel @crobert-1
 pkg/winperfcounters/                                             @open-telemetry/collector-contrib-approvers @dashpole @Mrod1598 @alxbl @pjanotti
 pkg/xk8stest/                                                    @open-telemetry/collector-contrib-approvers @crobert-1
+pkg/xstreamencoding/                                             @open-telemetry/collector-contrib-approvers @Kavindu-Dodan @axw
 processor/attributesprocessor/                                   @open-telemetry/collector-contrib-approvers @boostchicken
 processor/coralogixprocessor/                                    @open-telemetry/collector-contrib-approvers @crobert-1 @povilasv @iblancasa
 processor/cumulativetodeltaprocessor/                            @open-telemetry/collector-contrib-approvers @TylerHelmuth

.github/ISSUE_TEMPLATE/beta_stability.yaml

@@ -190,6 +190,7 @@ body:
       - pkg/translator/zipkin
       - pkg/winperfcounters
       - pkg/xk8stest
+      - pkg/xstreamencoding
       - processor/attributes
       - processor/coralogix
       - processor/cumulativetodelta

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -193,6 +193,7 @@ body:
       - pkg/translator/zipkin
       - pkg/winperfcounters
       - pkg/xk8stest
+      - pkg/xstreamencoding
       - processor/attributes
       - processor/coralogix
       - processor/cumulativetodelta

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -187,6 +187,7 @@ body:
       - pkg/translator/zipkin
       - pkg/winperfcounters
       - pkg/xk8stest
+      - pkg/xstreamencoding
       - processor/attributes
       - processor/coralogix
       - processor/cumulativetodelta

.github/ISSUE_TEMPLATE/other.yaml

@@ -187,6 +187,7 @@ body:
       - pkg/translator/zipkin
       - pkg/winperfcounters
       - pkg/xk8stest
+      - pkg/xstreamencoding
       - processor/attributes
       - processor/coralogix
       - processor/cumulativetodelta

.github/ISSUE_TEMPLATE/unmaintained.yaml

@@ -192,6 +192,7 @@ body:
       - pkg/translator/zipkin
       - pkg/winperfcounters
       - pkg/xk8stest
+      - pkg/xstreamencoding
       - processor/attributes
       - processor/coralogix
       - processor/cumulativetodelta

.github/component_labels.txt

@@ -173,6 +173,7 @@ pkg/translator/splunk pkg/translator/splunk
 pkg/translator/zipkin pkg/translator/zipkin
 pkg/winperfcounters pkg/winperfcounters
 pkg/xk8stest pkg/xk8stest
+pkg/xstreamencoding pkg/xstreamencoding
 processor/attributesprocessor processor/attributes
 processor/coralogixprocessor processor/coralogix
 processor/cumulativetodeltaprocessor processor/cumulativetodelta

extension/encoding/encoding.go

@@ -4,13 +4,20 @@
 package encoding // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding"
 
 import (
+	"io"
+
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 )
 
+const (
+	defaultFlushBytes = 1024 * 1024 // 1MB
+	defaultFlushItems = 1000        // 1000 items
+)
+
 // LogsMarshalerExtension is an extension that marshals logs.
 type LogsMarshalerExtension interface {
 	extension.Extension
@@ -23,6 +30,23 @@ type LogsUnmarshalerExtension interface {
 	plog.Unmarshaler
 }
 
+// LogsDecoder unmarshals logs from a stream, returning one batch per DecodeLogs call.
+type LogsDecoder interface {
+	// DecodeLogs is expected to be called iteratively to read all derived plog.Logs batches from the stream.
+	// The last batch of logs should be returned with a nil error. io.EOF error should follow on the subsequent call.
+	DecodeLogs() (plog.Logs, error)
+	// Offset returns the offset after the most recent batch read from the stream, or the initial offset.
+	// The exact meaning of the offset may vary by decoder (e.g. bytes, lines, records).
+	// You may use this value with WithOffset option to resume reading from the same offset when retrying after a failure.
+	Offset() int64
+}
+
+// LogsDecoderExtension is an extension that unmarshals logs from a stream.
+type LogsDecoderExtension interface {
+	extension.Extension
+	NewLogsDecoder(reader io.Reader, options ...DecoderOption) (LogsDecoder, error)
+}
+
 // MetricsMarshalerExtension is an extension that marshals metrics.
 type MetricsMarshalerExtension interface {
 	extension.Extension
@@ -35,6 +59,23 @@ type MetricsUnmarshalerExtension interface {
 	pmetric.Unmarshaler
 }
 
+// MetricsDecoder unmarshals metrics from a stream, returning one batch per DecodeMetrics call.
+type MetricsDecoder interface {
+	// DecodeMetrics is expected to be called iteratively to read all derived pmetric.Metrics batches from the stream.
+	// The last batch of metrics should be returned with a nil error. io.EOF error should follow on the subsequent call.
+	DecodeMetrics() (pmetric.Metrics, error)
+	// Offset returns the offset after the most recent batch read from the stream, or the initial offset.
+	// The exact meaning of the offset may vary by decoder (e.g. bytes, lines, records).
+	// You may use this value with WithOffset option to resume reading from the same offset when retrying after a failure.
+	Offset() int64
+}
+
+// MetricsDecoderExtension is an extension that unmarshals metrics from a stream.
+type MetricsDecoderExtension interface {
+	extension.Extension
+	NewMetricsDecoder(reader io.Reader, options ...DecoderOption) (MetricsDecoder, error)
+}
+
 // TracesMarshalerExtension is an extension that marshals traces.
 type TracesMarshalerExtension interface {
 	extension.Extension
@@ -58,3 +99,53 @@ type ProfilesUnmarshalerExtension interface {
 	extension.Extension
 	pprofile.Unmarshaler
 }
+
+// DecoderOptions configures the behavior of stream decoding.
+// FlushBytes and FlushItems control how often the decoder should flush decoded data from the stream.
+// Offset defines the initial stream offset for the stream.
+// Use NewDecoderOptions to construct with default options.
+type DecoderOptions struct {
+	FlushBytes int64
+	FlushItems int64
+	Offset     int64
+}
+
+func NewDecoderOptions(opts ...DecoderOption) DecoderOptions {
+	options := DecoderOptions{
+		FlushBytes: defaultFlushBytes,
+		FlushItems: defaultFlushItems,
+		Offset:     0,
+	}
+
+	for _, o := range opts {
+		o(&options)
+	}
+	return options
+}
+
+// DecoderOption defines the functional option for DecoderOptions.
+type DecoderOption func(*DecoderOptions)
+
+// WithFlushBytes sets the number of bytes after stream decoder should flush.
+// Use WithFlushBytes(0) to disable flushing by byte count.
+func WithFlushBytes(b int64) DecoderOption {
+	return func(o *DecoderOptions) {
+		o.FlushBytes = b
+	}
+}
+
+// WithFlushItems sets the number of items after stream decoder should flush.
+// Use WithFlushItems(0) to disable flushing by item count.
+func WithFlushItems(i int64) DecoderOption {
+	return func(o *DecoderOptions) {
+		o.FlushItems = i
+	}
+}
+
+// WithOffset defines the initial stream offset for the stream.
+// The exact meaning of the offset may vary by decoder (e.g. bytes, lines, records).
+func WithOffset(offset int64) DecoderOption {
+	return func(o *DecoderOptions) {
+		o.Offset = offset
+	}
+}