Add cloudwatch encoding

extension/encoding/cloudwatchencoding/Makefile

@@ -0,0 +1 @@
+include ../../../Makefile.Common

extension/encoding/cloudwatchencoding/config.go

@@ -0,0 +1,31 @@
+package cloudwatchencoding
+
+import (
+	"fmt"
+	"go.opentelemetry.io/collector/component"
+)
+
+type contentEncoding string
+
+const (
+	NoEncoding  = ""
+	GZipEncoded = "gzip"
+)
+
+type Config struct {
+	Encoding contentEncoding `mapstructure:"content_encoding"`
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+func (c *Config) Validate() error {
+	switch c.Encoding {
+	case NoEncoding:
+	case GZipEncoded:
+	default:
+		return fmt.Errorf("unknown content encoding %s", c.Encoding)
+	}
+	return nil
+}

extension/encoding/cloudwatchencoding/config_test.go

@@ -0,0 +1,54 @@
+package cloudwatchencoding
+
+import (
+	"errors"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/cloudwatchencodingextension/internal/metadata"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"path/filepath"
+	"testing"
+)
+
+func TestLoadConfig(t *testing.T) {
+	t.Parallel()
+
+	defaultCfg := createDefaultConfig().(*Config)
+
+	tests := []struct {
+		id          component.ID
+		expected    component.Config
+		expectedErr error
+	}{
+		{
+			id:       component.NewID(metadata.Type),
+			expected: defaultCfg,
+		},
+		{
+			id:          component.NewIDWithName(metadata.Type, "invalid_encoding"),
+			expectedErr: errors.New("unknown content encoding invalid"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.id.String(), func(t *testing.T) {
+			cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+			require.NoError(t, err)
+
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig()
+			sub, err := cm.Sub(tt.id.String())
+			require.NoError(t, err)
+			require.NoError(t, sub.Unmarshal(cfg))
+
+			if tt.expectedErr != nil {
+				require.Equal(t, tt.expectedErr, component.ValidateConfig(cfg))
+				return
+			}
+
+			require.NoError(t, component.ValidateConfig(cfg))
+			require.Equal(t, tt.expected, cfg)
+		})
+	}
+
+}

extension/encoding/cloudwatchencoding/doc.go

@@ -0,0 +1,2 @@
+//go:generate mdatagen metadata.yaml
+package cloudwatchencoding

extension/encoding/cloudwatchencoding/extension.go

@@ -0,0 +1,79 @@
+package cloudwatchencoding
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"fmt"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/cloudwatch"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/extension"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
+	"io"
+)
+
+var (
+	_ encoding.LogsUnmarshalerExtension    = (*cloudwatchExtension)(nil)
+	_ encoding.MetricsUnmarshalerExtension = (*cloudwatchExtension)(nil)
+)
+
+type cloudwatchExtension struct {
+	config *Config
+	logger *zap.Logger
+}
+
+func createExtension(_ context.Context, settings extension.Settings, config component.Config) (extension.Extension, error) {
+	return &cloudwatchExtension{
+		config: config.(*Config),
+		logger: settings.Logger,
+	}, nil
+}
+
+func (c *cloudwatchExtension) Start(_ context.Context, _ component.Host) error {
+	return nil
+}
+
+func (c *cloudwatchExtension) Shutdown(_ context.Context) error {
+	return nil
+}
+
+func decompress(buf []byte, encoding contentEncoding) ([]byte, error) {
+	switch encoding {
+	case NoEncoding:
+		return buf, nil
+	case GZipEncoded:
+		reader, err := gzip.NewReader(bytes.NewReader(buf))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create gzip reader: %w", err)
+		}
+		defer reader.Close()
+
+		decompressed, err := io.ReadAll(reader)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read from gzip reader: %w", err)
+		}
+		return decompressed, nil
+	default:
+		// not possible, prevented by config.Validate
+		return nil, nil
+	}
+}
+
+func (c *cloudwatchExtension) UnmarshalLogs(buf []byte) (plog.Logs, error) {
+	data, err := decompress(buf, c.config.Encoding)
+	if err != nil {
+		return plog.Logs{}, err
+	}
+	return cloudwatch.UnmarshalLogs(data)
+}
+
+func (c *cloudwatchExtension) UnmarshalMetrics(buf []byte) (pmetric.Metrics, error) {
+	data, err := decompress(buf, c.config.Encoding)
+	if err != nil {
+		return pmetric.Metrics{}, err
+	}
+	return cloudwatch.UnmarshalMetrics(data)
+}

extension/encoding/cloudwatchencoding/extension_test.go

@@ -0,0 +1,18 @@
+package cloudwatchencoding
+
+import (
+	"context"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"testing"
+)
+
+func TestExtension_Start_Shutdown(t *testing.T) {
+	extension := &cloudwatchExtension{}
+
+	err := extension.Start(context.Background(), componenttest.NewNopHost())
+	require.NoError(t, err)
+
+	err = extension.Shutdown(context.Background())
+	require.NoError(t, err)
+}

extension/encoding/cloudwatchencoding/factory.go

@@ -0,0 +1,15 @@
+package cloudwatchencoding
+
+import (
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/cloudwatchencodingextension/internal/metadata"
+	"go.opentelemetry.io/collector/extension"
+)
+
+func NewFactory() extension.Factory {
+	return extension.NewFactory(
+		metadata.Type,
+		createDefaultConfig,
+		createExtension,
+		metadata.ExtensionStability,
+	)
+}

extension/encoding/cloudwatchencoding/factory_test.go

@@ -0,0 +1,15 @@
+package cloudwatchencoding
+
+import (
+	"context"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/extension/extensiontest"
+	"testing"
+)
+
+func TestCreateExtension(t *testing.T) {
+	cfg := createDefaultConfig().(*Config)
+	ext, err := createExtension(context.Background(), extensiontest.NewNopSettings(), cfg)
+	require.NoError(t, err)
+	require.NotNil(t, ext)
+}

extension/encoding/cloudwatchencoding/go.mod

@@ -0,0 +1,60 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding/cloudwatchencodingextension
+
+go 1.23
+
+require (
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/encoding v0.117.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil v0.117.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/cloudwatch v0.0.0-00010101000000-000000000000
+	github.com/stretchr/testify v1.10.0
+	github.com/tidwall/gjson v1.18.0
+	go.opentelemetry.io/collector/component v0.117.0
+	go.opentelemetry.io/collector/component/componenttest v0.117.0
+	go.opentelemetry.io/collector/confmap v1.23.0
+	go.opentelemetry.io/collector/extension v0.117.0
+	go.opentelemetry.io/collector/extension/extensiontest v0.117.0
+	go.opentelemetry.io/collector/pdata v1.23.0
+	go.opentelemetry.io/collector/semconv v0.117.0
+	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.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
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/knadh/koanf/maps v0.1.1 // indirect
+	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.2 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	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/internal/exp/metrics v0.117.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.117.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/tidwall/match v1.1.1 // indirect
+	github.com/tidwall/pretty v1.2.0 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.117.0 // indirect
+	go.opentelemetry.io/collector/pdata/pprofile v0.117.0 // indirect
+	go.opentelemetry.io/otel v1.32.0 // indirect
+	go.opentelemetry.io/otel/metric v1.32.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.32.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.32.0 // indirect
+	go.opentelemetry.io/otel/trace v1.32.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20241104194629-dd2ea8efbc28 // indirect
+	google.golang.org/grpc v1.69.2 // indirect
+	google.golang.org/protobuf v1.36.2 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/cloudwatch => ../../../pkg/translator/cloudwatch

extension/encoding/cloudwatchencoding/metadata.yaml

@@ -0,0 +1,9 @@
+type: cloudwatch_encoding
+
+status:
+  class: extension
+  stability:
+    alpha: [extension]
+  distributions: []
+  codeowners:
+    active: []

extension/encoding/cloudwatchencoding/testdata/config.yaml

@@ -0,0 +1,4 @@
+cloudwatch_encoding:
+
+cloudwatch_encoding/invalid_encoding:
+  content_encoding: invalid

pkg/translator/cloudwatch/Makefile

@@ -0,0 +1 @@
+include ../../../Makefile.Common