feat: Adopt encoding extension streaming support for AWS Logs encoding extension

.chloggen/feat_streaming-for-aws-logs.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/awslogs_encoding
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adopt encoder streaming support for AWS Logs Encoding Extension
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [45567]
+
+# (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: []

extension/encoding/awslogsencodingextension/README.md

@@ -171,6 +171,23 @@ otelcol --config=config.yaml --feature-gates --feature-gates=<FEATURE_GATE_ID>
 | `userIdentity.arn`              | `aws.principal.arn`            | `aws.user_identity.principal.arn`             |
 | `userIdentity.type`             | `aws.principal.type`           | `aws.user_identity.principal.type`            |
 
+## Streaming Support
+
+All sub formats support both streaming & non-streaming unmarshaling.
+The table below summarizes streaming support details for each log type, along with the offset tracking mechanism,
+
+| Log Type            | Sub Log Type/Source            | Offset Tracking                   |
+|---------------------|--------------------------------|-----------------------------------|
+| CloudTrail          | Generic records                | Number of records processed       |
+| CloudTrail          | Digest record                  | Always 0 (full payload processed) |
+| ELB Access Logs     | ALB/NLB/CLB                    | Bytes processed                   |
+| Network Firewall    | Alert/Flow/TLS                 | Bytes processed                   |
+| S3 Access Logs      | -                              | Bytes processed                   |
+| Subscription filter | -                              | Always 0 (full payload processed) |
+| VPC Flow Logs       | S3 plain text                  | Bytes processed                   |
+| VPC Flow Logs       | CloudWatch subscription filter | Always 0 (full payload processed) |
+| WAF Logs            | -                              | Bytes processed                   |
+
 ## Produced Records per Format
 
 ### VPC flow log record fields

extension/encoding/awslogsencodingextension/extension.go

@@ -55,7 +55,10 @@ func init() {
 		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/45459"))
 }
 
-var _ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
+var (
+	_ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
+	_ encoding.LogsDecoderExtension     = (*encodingExtension)(nil)
+)
 
 type encodingExtension struct {
 	cfg *Config
@@ -70,8 +73,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatCloudWatchLogsSubscriptionFilter, constants.FormatCloudWatchLogsSubscriptionFilterV1:
 		if cfg.Format == constants.FormatCloudWatchLogsSubscriptionFilterV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatCloudWatchLogsSubscriptionFilterV1)),
-				zap.String("new_format", string(constants.FormatCloudWatchLogsSubscriptionFilter)),
+				zap.String("old_format", constants.FormatCloudWatchLogsSubscriptionFilterV1),
+				zap.String("new_format", constants.FormatCloudWatchLogsSubscriptionFilter),
 			)
 		}
 		return &encodingExtension{
@@ -81,8 +84,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatVPCFlowLog, constants.FormatVPCFlowLogV1:
 		if cfg.Format == constants.FormatVPCFlowLogV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatVPCFlowLogV1)),
-				zap.String("new_format", string(constants.FormatVPCFlowLog)),
+				zap.String("old_format", constants.FormatVPCFlowLogV1),
+				zap.String("new_format", constants.FormatVPCFlowLog),
 			)
 		}
 
@@ -100,8 +103,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatS3AccessLog, constants.FormatS3AccessLogV1:
 		if cfg.Format == constants.FormatS3AccessLogV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatS3AccessLogV1)),
-				zap.String("new_format", string(constants.FormatS3AccessLog)),
+				zap.String("old_format", constants.FormatS3AccessLogV1),
+				zap.String("new_format", constants.FormatS3AccessLog),
 			)
 		}
 		return &encodingExtension{
@@ -111,8 +114,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatWAFLog, constants.FormatWAFLogV1:
 		if cfg.Format == constants.FormatWAFLogV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatWAFLogV1)),
-				zap.String("new_format", string(constants.FormatWAFLog)),
+				zap.String("old_format", constants.FormatWAFLogV1),
+				zap.String("new_format", constants.FormatWAFLog),
 			)
 		}
 		return &encodingExtension{
@@ -122,8 +125,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatCloudTrailLog, constants.FormatCloudTrailLogV1:
 		if cfg.Format == constants.FormatCloudTrailLogV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatCloudTrailLogV1)),
-				zap.String("new_format", string(constants.FormatCloudTrailLog)),
+				zap.String("old_format", constants.FormatCloudTrailLogV1),
+				zap.String("new_format", constants.FormatCloudTrailLog),
 			)
 		}
 		return &encodingExtension{
@@ -135,8 +138,8 @@ func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension,
 	case constants.FormatELBAccessLog, constants.FormatELBAccessLogV1:
 		if cfg.Format == constants.FormatELBAccessLogV1 {
 			settings.Logger.Warn("using old format value. This format will be removed in version 0.138.0.",
-				zap.String("old_format", string(constants.FormatELBAccessLogV1)),
-				zap.String("new_format", string(constants.FormatELBAccessLog)),
+				zap.String("old_format", constants.FormatELBAccessLogV1),
+				zap.String("new_format", constants.FormatELBAccessLog),
 			)
 		}
 		return &encodingExtension{
@@ -167,37 +170,30 @@ func (*encodingExtension) Shutdown(_ context.Context) error {
 	return nil
 }
 
-func (e *encodingExtension) getGzipReader(buf []byte) (io.Reader, error) {
-	var err error
-	gzipReader, ok := e.gzipPool.Get().(*gzip.Reader)
-	if !ok {
-		gzipReader, err = gzip.NewReader(bytes.NewReader(buf))
-	} else {
-		err = gzipReader.Reset(bytes.NewBuffer(buf))
+func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
+	encodingReader, reader, err := e.getReaderFromFormat(buf)
+	if err != nil {
+		return plog.Logs{}, fmt.Errorf("failed to get reader for %q logs: %w", e.format, err)
 	}
 
-	if err != nil {
-		if gzipReader != nil {
-			e.gzipPool.Put(gzipReader)
+	defer func() {
+		if encodingReader == gzipEncoding {
+			r := reader.(*gzip.Reader)
+			_ = r.Close()
+			e.gzipPool.Put(r)
 		}
-		return nil, fmt.Errorf("failed to decompress content: %w", err)
-	}
+	}()
 
-	return gzipReader, nil
-}
+	logs, err := e.unmarshaler.UnmarshalAWSLogs(reader)
+	if err != nil {
+		return plog.Logs{}, fmt.Errorf("failed to unmarshal logs as %q format: %w", e.format, err)
+	}
 
-// isGzipData checks if the buffer contains gzip-compressed data by examining magic bytes
-func isGzipData(buf []byte) bool {
-	return len(buf) > 2 && buf[0] == 0x1f && buf[1] == 0x8b
+	return logs, nil
 }
 
-// getReaderForData returns the appropriate reader and encoding type based on data format
-func (e *encodingExtension) getReaderForData(buf []byte) (string, io.Reader, error) {
-	if isGzipData(buf) {
-		reader, err := e.getGzipReader(buf)
-		return gzipEncoding, reader, err
-	}
-	return bytesEncoding, bytes.NewReader(buf), nil
+func (e *encodingExtension) NewLogsDecoder(reader io.Reader, options ...encoding.DecoderOption) (encoding.LogsDecoder, error) {
+	return e.unmarshaler.NewLogsDecoder(reader, options...)
 }
 
 func (e *encodingExtension) getReaderFromFormat(buf []byte) (string, io.Reader, error) {
@@ -229,24 +225,35 @@ func (e *encodingExtension) getReaderFromFormat(buf []byte) (string, io.Reader,
 	}
 }
 
-func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
-	encodingReader, reader, err := e.getReaderFromFormat(buf)
-	if err != nil {
-		return plog.Logs{}, fmt.Errorf("failed to get reader for %q logs: %w", e.format, err)
+// getReaderForData returns the appropriate reader and encoding type based on data format
+func (e *encodingExtension) getReaderForData(buf []byte) (string, io.Reader, error) {
+	if isGzipData(buf) {
+		reader, err := e.getGzipReader(buf)
+		return gzipEncoding, reader, err
 	}
+	return bytesEncoding, bytes.NewReader(buf), nil
+}
 
-	defer func() {
-		if encodingReader == gzipEncoding {
-			r := reader.(*gzip.Reader)
-			_ = r.Close()
-			e.gzipPool.Put(r)
-		}
-	}()
+func (e *encodingExtension) getGzipReader(buf []byte) (io.Reader, error) {
+	var err error
+	gzipReader, ok := e.gzipPool.Get().(*gzip.Reader)
+	if !ok {
+		gzipReader, err = gzip.NewReader(bytes.NewReader(buf))
+	} else {
+		err = gzipReader.Reset(bytes.NewBuffer(buf))
+	}
 
-	logs, err := e.unmarshaler.UnmarshalAWSLogs(reader)
 	if err != nil {
-		return plog.Logs{}, fmt.Errorf("failed to unmarshal logs as %q format: %w", e.format, err)
+		if gzipReader != nil {
+			e.gzipPool.Put(gzipReader)
+		}
+		return nil, fmt.Errorf("failed to decompress content: %w", err)
 	}
 
-	return logs, nil
+	return gzipReader, nil
+}
+
+// isGzipData checks if the buffer contains gzip-compressed data by examining magic bytes
+func isGzipData(buf []byte) bool {
+	return len(buf) > 2 && buf[0] == 0x1f && buf[1] == 0x8b
 }

extension/encoding/awslogsencodingextension/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding v0.146.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.146.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.146.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xstreamencoding v0.0.0-20260218150750-8f6cb4673d5f
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/collector/component v1.52.0
 	go.opentelemetry.io/collector/component/componenttest v0.146.1
@@ -62,3 +63,5 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../../pkg/pdatatest
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../../pkg/golden
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/xstreamencoding => ../../../pkg/xstreamencoding

extension/encoding/awslogsencodingextension/internal/unmarshaler/cloudtraillog/testdata/cloudtrail_log_expected.yaml

@@ -6,10 +6,10 @@ resourceLogs:
             stringValue: aws
         - key: cloud.region
           value:
-            stringValue: us-east-1
+            stringValue: us-west-2
         - key: cloud.account.id
           value:
-            stringValue: "123456789012"
+            stringValue: "111122223333"
     scopeLogs:
       - logRecords:
           - attributes:
@@ -124,29 +124,29 @@ resourceLogs:
                                     values:
                                       - kvlistValue:
                                           values:
-                                            - key: instanceId
-                                              value:
-                                                stringValue: i-EXAMPLEaff4840c22
-                                            - key: currentState
+                                            - key: previousState
                                               value:
                                                 kvlistValue:
                                                   values:
                                                     - key: code
                                                       value:
-                                                        doubleValue: 0
+                                                        doubleValue: 80
                                                     - key: name
                                                       value:
-                                                        stringValue: pending
-                                            - key: previousState
+                                                        stringValue: stopped
+                                            - key: instanceId
+                                              value:
+                                                stringValue: i-EXAMPLEaff4840c22
+                                            - key: currentState
                                               value:
                                                 kvlistValue:
                                                   values:
                                                     - key: code
                                                       value:
-                                                        doubleValue: 80
+                                                        doubleValue: 0
                                                     - key: name
                                                       value:
-                                                        stringValue: stopped
+                                                        stringValue: pending
                                       - kvlistValue:
                                           values:
                                             - key: instanceId
@@ -166,12 +166,12 @@ resourceLogs:
                                               value:
                                                 kvlistValue:
                                                   values:
-                                                    - key: code
-                                                      value:
-                                                        doubleValue: 80
                                                     - key: name
                                                       value:
                                                         stringValue: stopped
+                                                    - key: code
+                                                      value:
+                                                        doubleValue: 80
             body: {}
             timeUnixNano: "1689801448000000000"
           - attributes:
@@ -359,9 +359,6 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
-                      - key: topicArn
-                        value:
-                          stringValue: arn:aws:sns:us-east-1:123456789012:ExampleSNSTopic
                       - key: message
                         value:
                           stringValue: HIDDEN_DUE_TO_SECURITY_REASONS
@@ -374,6 +371,9 @@ resourceLogs:
                       - key: messageAttributes
                         value:
                           stringValue: HIDDEN_DUE_TO_SECURITY_REASONS
+                      - key: topicArn
+                        value:
+                          stringValue: arn:aws:sns:us-east-1:123456789012:ExampleSNSTopic
               - key: aws.response.elements
                 value:
                   kvlistValue:
@@ -611,15 +611,15 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
+                      - key: MFAUsed
+                        value:
+                          boolValue: true
                       - key: MobileVersion
                         value:
                           stringValue: "No"
                       - key: LoginTo
                         value:
                           stringValue: https://console.aws.amazon.com/console/home?region=us
-                      - key: MFAUsed
-                        value:
-                          boolValue: true
             body: {}
             timeUnixNano: "1749997800000000000"
           - attributes: