[extension/awslogs_encoding] Add support for cloudwatch logs coming from subscription filter (open-telemetry#38821)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Add support for [cloudwatch logs coming from subscription
filter](https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html).

<!-- Issue number (e.g. open-telemetry#1234) or full URL to issue, if applicable. -->
#### Link to tracking issue

open-telemetry#38820

<!--Describe what testing was performed and which tests were added.-->
#### Testing

There are unit tests for new code.

<!--Describe the documentation added.-->
#### Documentation

Comments and unit tests should be enough.

<!--Please delete paragraphs that you did not use before submitting.-->

Author: constanca-m

.chloggen/aws-subscription-unmarshaler.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. filelogreceiver)
+component: awslogsencodingextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for cloudwatch logs coming from subscription filters.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38820]
+
+# (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: [user]

extension/encoding/awslogsencodingextension/cloudwatchlogssubscriptionfilter.go

@@ -12,6 +12,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding"
+	subscriptionfilter "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/subscription-filter"
 )
 
 var _ encoding.LogsUnmarshalerExtension = (*encodingExtension)(nil)
@@ -21,12 +22,12 @@ type encodingExtension struct {
 	format      string
 }
 
-func newExtension(cfg *Config, _ extension.Settings) (*encodingExtension, error) {
+func newExtension(cfg *Config, settings extension.Settings) (*encodingExtension, error) {
 	switch cfg.Format {
 	case formatCloudWatchLogsSubscriptionFilter:
 		return &encodingExtension{
-			unmarshaler: cloudWatchLogsSubscriptionFilterUnmarshaler{},
-			format:      cfg.Format,
+			unmarshaler: subscriptionfilter.NewSubscriptionFilterUnmarshaler(settings.BuildInfo),
+			format:      formatCloudWatchLogsSubscriptionFilter,
 		}, nil
 	default:
 		// Format will have been validated by Config.Validate,
@@ -44,10 +45,10 @@ func (*encodingExtension) Shutdown(_ context.Context) error {
 	return nil
 }
 
-func (e *encodingExtension) UnmarshalLogs(record []byte) (plog.Logs, error) {
-	logs, err := e.unmarshaler.UnmarshalLogs(record)
+func (e *encodingExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
+	logs, err := e.unmarshaler.UnmarshalLogs(buf)
 	if err != nil {
-		return plog.Logs{}, fmt.Errorf("failed to unmarshal logs as '%s' format: %w", e.format, err)
+		return plog.Logs{}, fmt.Errorf("failed to unmarshal logs as %q format: %w", e.format, err)
 	}
 	return logs, nil
 }

extension/encoding/awslogsencodingextension/extension.go

@@ -15,6 +15,9 @@ func TestNew_CloudWatchLogsSubscriptionFilter(t *testing.T) {
 	e, err := newExtension(&Config{Format: formatCloudWatchLogsSubscriptionFilter}, extensiontest.NewNopSettings(extensiontest.NopType))
 	require.NoError(t, err)
 	require.NotNil(t, e)
+
+	_, err = e.UnmarshalLogs([]byte("invalid"))
+	require.ErrorContains(t, err, `failed to unmarshal logs as "cloudwatch_logs_subscription_filter" format`)
 }
 
 func TestNew_Unimplemented(t *testing.T) {

extension/encoding/awslogsencodingextension/extension_test.go

@@ -3,7 +3,12 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/extension/encod
 go 1.23.0
 
 require (
+	github.com/aws/aws-lambda-go v1.47.0
+	github.com/goccy/go-json v0.10.5
+	github.com/klauspost/compress v1.18.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding v0.122.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.122.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.122.0
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v1.28.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/component/componenttest v0.122.2-0.20250319144947-41a9ea7f7402
@@ -12,10 +17,12 @@ require (
 	go.opentelemetry.io/collector/extension v1.28.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/extension/extensiontest v0.122.2-0.20250319144947-41a9ea7f7402
 	go.opentelemetry.io/collector/pdata v1.28.2-0.20250319144947-41a9ea7f7402
+	go.opentelemetry.io/collector/semconv v0.122.2-0.20250319144947-41a9ea7f7402
 	go.uber.org/goleak v1.3.0
 )
 
 require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
@@ -31,6 +38,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.122.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.28.2-0.20250319144947-41a9ea7f7402 // indirect
@@ -52,3 +60,9 @@ require (
 )
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding => ../
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../../../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

extension/encoding/awslogsencodingextension/go.mod

@@ -0,0 +1,130 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package subscriptionfilter // import "github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/unmarshaler/subscription-filter"
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/aws/aws-lambda-go/events"
+	gojson "github.com/goccy/go-json"
+	"github.com/klauspost/compress/gzip"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	conventions "go.opentelemetry.io/collector/semconv/v1.27.0"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/awslogsencodingextension/internal/metadata"
+)
+
+var (
+	errEmptyOwner     = errors.New("cloudwatch log with message type 'DATA_MESSAGE' has empty owner field")
+	errEmptyLogGroup  = errors.New("cloudwatch log with message type 'DATA_MESSAGE' has empty log group field")
+	errEmptyLogStream = errors.New("cloudwatch log with message type 'DATA_MESSAGE' has empty log stream field")
+)
+
+func validateLog(log events.CloudwatchLogsData) error {
+	switch log.MessageType {
+	case "DATA_MESSAGE":
+		if log.Owner == "" {
+			return errEmptyOwner
+		}
+		if log.LogGroup == "" {
+			return errEmptyLogGroup
+		}
+		if log.LogStream == "" {
+			return errEmptyLogStream
+		}
+	case "CONTROL_MESSAGE":
+	default:
+		return fmt.Errorf("cloudwatch log has invalid message type %q", log.MessageType)
+	}
+	return nil
+}
+
+type subscriptionFilterUnmarshaler struct {
+	buildInfo component.BuildInfo
+
+	// Pool the gzip readers, which are expensive to create.
+	gzipPool sync.Pool
+}
+
+func NewSubscriptionFilterUnmarshaler(buildInfo component.BuildInfo) plog.Unmarshaler {
+	return &subscriptionFilterUnmarshaler{
+		buildInfo: buildInfo,
+		gzipPool:  sync.Pool{},
+	}
+}
+
+var _ plog.Unmarshaler = (*subscriptionFilterUnmarshaler)(nil)
+
+// UnmarshalLogs deserializes the given record as CloudWatch Logs events
+// into a plog.Logs, grouping logs by owner (account ID), log group, and
+// log stream. Logs are assumed to be gzip-compressed as specified at
+// https://docs.aws.amazon.com/firehose/latest/dev/writing-with-cloudwatch-logs.html.
+func (f *subscriptionFilterUnmarshaler) UnmarshalLogs(compressedRecord []byte) (plog.Logs, error) {
+	var errDecompress error
+	gzipReader, ok := f.gzipPool.Get().(*gzip.Reader)
+	if !ok {
+		gzipReader, errDecompress = gzip.NewReader(bytes.NewReader(compressedRecord))
+	} else {
+		errDecompress = gzipReader.Reset(bytes.NewReader(compressedRecord))
+	}
+	if errDecompress != nil {
+		if gzipReader != nil {
+			f.gzipPool.Put(gzipReader)
+		}
+		return plog.Logs{}, fmt.Errorf("failed to decompress record: %w", errDecompress)
+	}
+	defer func() {
+		_ = gzipReader.Close()
+		f.gzipPool.Put(gzipReader)
+	}()
+
+	var cwLog events.CloudwatchLogsData
+	decoder := gojson.NewDecoder(gzipReader)
+	if err := decoder.Decode(&cwLog); err != nil {
+		return plog.Logs{}, fmt.Errorf("failed to decode decompressed record: %w", err)
+	}
+
+	if cwLog.MessageType == "CONTROL_MESSAGE" {
+		return plog.Logs{}, nil
+	}
+
+	if err := validateLog(cwLog); err != nil {
+		return plog.Logs{}, fmt.Errorf("invalid cloudwatch log: %w", err)
+	}
+
+	return f.createLogs(cwLog), nil
+}
+
+// createLogs create plog.Logs from the cloudwatchLog
+func (f *subscriptionFilterUnmarshaler) createLogs(
+	cwLog events.CloudwatchLogsData,
+) plog.Logs {
+	logs := plog.NewLogs()
+	rl := logs.ResourceLogs().AppendEmpty()
+	resourceAttrs := rl.Resource().Attributes()
+	resourceAttrs.PutStr(conventions.AttributeCloudProvider, conventions.AttributeCloudProviderAWS)
+	resourceAttrs.PutStr(conventions.AttributeCloudAccountID, cwLog.Owner)
+	resourceAttrs.PutEmptySlice(conventions.AttributeAWSLogGroupNames).AppendEmpty().SetStr(cwLog.LogGroup)
+	resourceAttrs.PutEmptySlice(conventions.AttributeAWSLogStreamNames).AppendEmpty().SetStr(cwLog.LogStream)
+	resourceAttrs.PutStr(conventions.AttributeAWSLogGroupNames, cwLog.LogGroup)
+	resourceAttrs.PutStr(conventions.AttributeAWSLogStreamNames, cwLog.LogStream)
+
+	sl := rl.ScopeLogs().AppendEmpty()
+	sl.Scope().SetName(metadata.ScopeName)
+	sl.Scope().SetVersion(f.buildInfo.Version)
+	for _, event := range cwLog.LogEvents {
+		logRecord := sl.LogRecords().AppendEmpty()
+		// pcommon.Timestamp is a time specified as UNIX Epoch time in nanoseconds
+		// but timestamp in cloudwatch logs are in milliseconds.
+		logRecord.SetTimestamp(pcommon.Timestamp(event.Timestamp * int64(time.Millisecond)))
+		logRecord.Body().SetStr(event.Message)
+	}
+	return logs
+}