feat: implement streaming contract for CloudTrail logs

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

.chloggen/feat_aws-logs-extension-adopt-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/awslogs_encoding
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adopt encoding extension streaming contract for AWS Logs Extension encoding
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [46214]
+
+# (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
+
+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                   | Note                                                                                                        |
+|------------|---------------------|-----------------------------------|-------------------------------------------------------------------------------------------------------------|
+| CloudTrail | Generic records     | Number of records processed       | Number of records are used as CloudTrail log arrives JSON and streaming is done on internal `Records` array |
+| CloudTrail | Digest record       | Always 0 (full payload processed) |                                                                                                             |
+
 ## Produced Records per Format
 
 ### VPC flow log record fields

extension/encoding/awslogsencodingextension/extension.go

@@ -35,6 +35,11 @@ const (
 	parquetEncoding = "parquet"
 )
 
+var (
+	_ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
+	_ encoding.LogsDecoderExtension     = (*encodingExtension)(nil)
+)
+
 var (
 	vpcFlowStartISO8601FormatFeatureGate    *featuregate.Gate
 	cloudTrailUserIdentityPrefixFeatureGate *featuregate.Gate
@@ -55,8 +60,6 @@ func init() {
 		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/45459"))
 }
 
-var _ encoding.LogsUnmarshalerExtension = (*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,6 +170,36 @@ func (*encodingExtension) Shutdown(_ context.Context) error {
 	return nil
 }
 
+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)
+	}
+
+	defer func() {
+		if encodingReader == gzipEncoding {
+			r := reader.(*gzip.Reader)
+			_ = r.Close()
+			e.gzipPool.Put(r)
+		}
+	}()
+
+	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)
+	}
+
+	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)
+}
+
 func (e *encodingExtension) getGzipReader(buf []byte) (io.Reader, error) {
 	var err error
 	gzipReader, ok := e.gzipPool.Get().(*gzip.Reader)
@@ -228,25 +261,3 @@ func (e *encodingExtension) getReaderFromFormat(buf []byte) (string, io.Reader,
 		return "", nil, fmt.Errorf("unimplemented: format %q has no reader", e.format)
 	}
 }
-
-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)
-	}
-
-	defer func() {
-		if encodingReader == gzipEncoding {
-			r := reader.(*gzip.Reader)
-			_ = r.Close()
-			e.gzipPool.Put(r)
-		}
-	}()
-
-	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)
-	}
-
-	return logs, nil
-}

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:
@@ -679,4 +679,4 @@ resourceLogs:
               value:
                 stringValue: aws.cloudtrail
           name: github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension
-          version: test-version
+          version: test-version