Implement streaming in cloudtrail format of awslogsencoding extension

.chloggen/feat_streaming-for-cloudtrail.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: Add streaming support for CloudTrail logs.
+
+# 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,16 @@ 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
+
+CloudTrail logs support both streaming & non-streaming unmarshaling. The table below summarizes streaming support details for CloudTrail:
+
+| Log Type   | Sub Log Type/Source            | Offset Tracking                   |
+|------------|--------------------------------|-----------------------------------|
+| CloudTrail | S3 records                     | Number of records processed       |
+| CloudTrail | CloudWatch subscription filter | Always 0 (full payload processed) |
+| CloudTrail | Digest record                  | Always 0 (full payload 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{
@@ -250,3 +253,10 @@ func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
 
 	return logs, nil
 }
+
+func (e *encodingExtension) NewLogsDecoder(reader io.Reader, options ...encoding.DecoderOption) (encoding.LogsDecoder, error) {
+	if u, ok := e.unmarshaler.(awsunmarshaler.StreamingLogsUnmarshaler); ok {
+		return u.NewLogsDecoder(reader, options...)
+	}
+	return nil, fmt.Errorf("streaming not supported for format %q", e.format)
+}

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.146.0
 	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_cw_expected.yaml

@@ -85,15 +85,6 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
-                      - key: CipherSuite
-                        value:
-                          stringValue: TLS_AES_128_GCM_SHA256
-                      - key: bytesTransferredIn
-                        value:
-                          doubleValue: 576
-                      - key: SSEApplied
-                        value:
-                          stringValue: Default_SSE_S3
                       - key: AuthenticationMethod
                         value:
                           stringValue: AuthHeader
@@ -106,6 +97,15 @@ resourceLogs:
                       - key: SignatureVersion
                         value:
                           stringValue: SigV4
+                      - key: CipherSuite
+                        value:
+                          stringValue: TLS_AES_128_GCM_SHA256
+                      - key: bytesTransferredIn
+                        value:
+                          doubleValue: 576
+                      - key: SSEApplied
+                        value:
+                          stringValue: Default_SSE_S3
               - key: aws.resources
                 value:
                   arrayValue:

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

@@ -65,15 +65,15 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
-                      - key: key
-                        value:
-                          stringValue: AWSLogs/627286350134/s3/2026-01-28-09-26-59-AB82D7306CDE5EF8
                       - key: bucketName
                         value:
                           stringValue: elastic-sar-bucket-fe44ff52c91b2efa4bccb8e07c85a202
                       - key: Host
                         value:
                           stringValue: s3.eu-west-1.amazonaws.com
+                      - key: key
+                        value:
+                          stringValue: AWSLogs/627286350134/s3/2026-01-28-09-26-59-AB82D7306CDE5EF8
               - key: aws.response.elements
                 value:
                   kvlistValue:
@@ -85,6 +85,9 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
+                      - key: bytesTransferredIn
+                        value:
+                          doubleValue: 576
                       - key: SSEApplied
                         value:
                           stringValue: Default_SSE_S3
@@ -103,9 +106,6 @@ resourceLogs:
                       - key: CipherSuite
                         value:
                           stringValue: TLS_AES_128_GCM_SHA256
-                      - key: bytesTransferredIn
-                        value:
-                          doubleValue: 576
               - key: aws.resources
                 value:
                   arrayValue:

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:
@@ -111,9 +111,6 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
-                      - key: requestId
-                        value:
-                          stringValue: e4336db0-149f-4a6b-844d-EXAMPLEb9d16
                       - key: instancesSet
                         value:
                           kvlistValue:
@@ -172,6 +169,9 @@ resourceLogs:
                                                     - key: name
                                                       value:
                                                         stringValue: stopped
+                      - key: requestId
+                        value:
+                          stringValue: e4336db0-149f-4a6b-844d-EXAMPLEb9d16
             body: {}
             timeUnixNano: "1689801448000000000"
           - attributes:
@@ -359,6 +359,12 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
+                      - key: messageStructure
+                        value:
+                          stringValue: json
+                      - key: messageAttributes
+                        value:
+                          stringValue: HIDDEN_DUE_TO_SECURITY_REASONS
                       - key: topicArn
                         value:
                           stringValue: arn:aws:sns:us-east-1:123456789012:ExampleSNSTopic
@@ -368,12 +374,6 @@ resourceLogs:
                       - key: subject
                         value:
                           stringValue: HIDDEN_DUE_TO_SECURITY_REASONS
-                      - key: messageStructure
-                        value:
-                          stringValue: json
-                      - key: messageAttributes
-                        value:
-                          stringValue: HIDDEN_DUE_TO_SECURITY_REASONS
               - key: aws.response.elements
                 value:
                   kvlistValue:
@@ -418,6 +418,9 @@ resourceLogs:
                 value:
                   kvlistValue:
                     values:
+                      - key: state
+                        value:
+                          stringValue: Start
                       - key: eventSource
                         value:
                           stringValue: ssm.amazonaws.com
@@ -449,9 +452,6 @@ resourceLogs:
                                               - key: average
                                                 value:
                                                   doubleValue: 669
-                      - key: state
-                        value:
-                          stringValue: Start
             body: {}
             timeUnixNano: "1672627860000000000"
           - attributes:
@@ -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: