Refactor unmarshalers to fit into encoding framework

The internal unmarshalers now implement plog.Unmarshaler
and pmetric.Unmarshaler. This will enable extracting them
later as encoding extensions; for now they remain embedded
within the receiver.

As a result of the interface change, the unmarshalers now
unmarshal a single record at a time, which means we cannot
merge resources/metrics as we go, but only after each record.

This also fixes a bug in the cwmetrics unmarshaller where
the unit of a metric was not considered part of its identity,
and so two metrics that differed only by unit would be merged.

Author: axw

.chloggen/firehose-encoding-extension.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: awsfirehosereceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for encoding extensions
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37113]
+
+# (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: Adds `encoding` config setting, and deprecates the `record_type` setting.
+
+# 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]

receiver/awsfirehosereceiver/README.md

@@ -45,25 +45,28 @@ See [documentation](https://github.com/open-telemetry/opentelemetry-collector/bl
 
 A `cert_file` and `key_file` are required.
 
-### record_type:
-The type of record being received from the delivery stream. Each unmarshaler handles a specific type, so the field allows the receiver to use the correct one.
+### encoding:
+
+The ID of an [encoding extension](https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/extension/encoding) for decoding logs or metrics.
+This configuration also supports the built-in encodings listed in the [Encodings](#encodings) section.
+If no encoding is specified, then the receiver will default to a signal-specific encoding: `cwmetrics` for metrics, and `cwlogs` for logs.
 
-default: `cwmetrics`
+### record_type:
 
-See the [Record Types](#record-types) section for all available options.
+Deprecated, use `encoding` instead. `record_type` will be removed in a future release; it is an alias for `encoding`.
 
 ### access_key (Optional):
 The access key to be checked on each request received. This can be set when creating or updating the delivery stream.
 See [documentation](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-http) for details.
 
-## Record Types
+## Encodings
 
 ### cwmetrics
-The record type for the CloudWatch metric stream. Expects the format for the records to be JSON.
+The encoding for the CloudWatch metric stream. Expects the format for the records to be JSON.
 See [documentation](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Metric-Streams.html) for details.
 
 ### cwlogs
-The record type for the CloudWatch log stream. Expects the format for the records to be JSON.
+The encoding for the CloudWatch log stream. Expects the format for the records to be JSON.
 For example:
 
 ```json
@@ -84,5 +87,5 @@ For example:
 ```
 
 ### otlp_v1
-The OTLP v1 format as produced by CloudWatch metric streams.
+The OTLP v1 encoding as produced by CloudWatch metric streams.
 See [documentation](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-metric-streams-formats-opentelemetry-100.html) for details.

receiver/awsfirehosereceiver/config.go

@@ -8,15 +8,22 @@ import (
 
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configopaque"
+	"go.uber.org/zap"
 )
 
 type Config struct {
 	// ServerConfig is used to set up the Firehose delivery
 	// endpoint. The Firehose delivery stream expects an HTTPS
 	// endpoint, so TLSSettings must be used to enable that.
 	confighttp.ServerConfig `mapstructure:",squash"`
-	// RecordType is the key used to determine which unmarshaler to use
-	// when receiving the requests.
+	// Encoding identifies the encoding of records received from
+	// Firehose. Defaults to telemetry-specific encodings: "cwlog"
+	// for logs, and "cwmetrics" for metrics.
+	Encoding string `mapstructure:"encoding"`
+	// RecordType is an alias for Encoding for backwards compatibility.
+	// It is an error to specify both encoding and record_type.
+	//
+	// Deprecated: use Encoding instead.
 	RecordType string `mapstructure:"record_type"`
 	// AccessKey is checked against the one received with each request.
 	// This can be set when creating or updating the Firehose delivery
@@ -30,10 +37,14 @@ func (c *Config) Validate() error {
 	if c.Endpoint == "" {
 		return errors.New("must specify endpoint")
 	}
-	// If a record type is specified, it must be valid.
-	// An empty string is acceptable, however, because it will use a telemetry-type-specific default.
-	if c.RecordType != "" {
-		return validateRecordType(c.RecordType)
+	if c.RecordType != "" && c.Encoding != "" {
+		return errors.New("record_type must not be set when encoding is set")
 	}
 	return nil
 }
+
+func handleDeprecatedConfig(cfg *Config, logger *zap.Logger) {
+	if cfg.RecordType != "" {
+		logger.Warn("record_type is deprecated, and will be removed in a future version. Use encoding instead.")
+	}
+}

receiver/awsfirehosereceiver/config_test.go

@@ -19,7 +19,7 @@ import (
 
 func TestLoadConfig(t *testing.T) {
 	for _, configType := range []string{
-		"cwmetrics", "cwlogs", "otlp_v1", "invalid",
+		"cwmetrics", "cwlogs", "otlp_v1",
 	} {
 		t.Run(configType, func(t *testing.T) {
 			fileName := configType + "_config.yaml"
@@ -34,24 +34,36 @@ func TestLoadConfig(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			err = component.ValidateConfig(cfg)
-			if configType == "invalid" {
-				assert.Error(t, err)
-			} else {
-				assert.NoError(t, err)
-				require.Equal(t, &Config{
-					RecordType: configType,
-					AccessKey:  "some_access_key",
-					ServerConfig: confighttp.ServerConfig{
-						Endpoint: "0.0.0.0:4433",
-						TLSSetting: &configtls.ServerConfig{
-							Config: configtls.Config{
-								CertFile: "server.crt",
-								KeyFile:  "server.key",
-							},
+			assert.NoError(t, err)
+			require.Equal(t, &Config{
+				RecordType: configType,
+				AccessKey:  "some_access_key",
+				ServerConfig: confighttp.ServerConfig{
+					Endpoint: "0.0.0.0:4433",
+					TLSSetting: &configtls.ServerConfig{
+						Config: configtls.Config{
+							CertFile: "server.crt",
+							KeyFile:  "server.key",
 						},
 					},
-				}, cfg)
-			}
+				},
+			}, cfg)
 		})
 	}
 }
+
+func TestLoadConfigInvalid(t *testing.T) {
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "invalid_config.yaml"))
+	require.NoError(t, err)
+
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+
+	sub, err := cm.Sub(component.NewIDWithName(metadata.Type, "").String())
+	require.NoError(t, err)
+	require.NoError(t, sub.Unmarshal(cfg))
+
+	err = component.ValidateConfig(cfg)
+	require.Error(t, err)
+	assert.EqualError(t, err, "record_type must not be set when encoding is set")
+}

receiver/awsfirehosereceiver/factory.go

@@ -5,34 +5,19 @@ package awsfirehosereceiver // import "github.com/open-telemetry/opentelemetry-c
 
 import (
 	"context"
-	"errors"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/receiver"
-	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwmetricstream"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/otlpmetricstream"
 )
 
 const (
 	defaultEndpoint = "localhost:4433"
 )
 
-var (
-	errUnrecognizedRecordType = errors.New("unrecognized record type")
-	availableRecordTypes      = map[string]bool{
-		cwmetricstream.TypeStr:   true,
-		cwlog.TypeStr:            true,
-		otlpmetricstream.TypeStr: true,
-	}
-)
-
 // NewFactory creates a receiver factory for awsfirehose. Currently, only
 // available in metrics pipelines.
 func NewFactory() receiver.Factory {
@@ -43,34 +28,6 @@ func NewFactory() receiver.Factory {
 		receiver.WithLogs(createLogsReceiver, metadata.LogsStability))
 }
 
-// validateRecordType checks the available record types for the
-// passed in one and returns an error if not found.
-func validateRecordType(recordType string) error {
-	if _, ok := availableRecordTypes[recordType]; !ok {
-		return errUnrecognizedRecordType
-	}
-	return nil
-}
-
-// defaultMetricsUnmarshalers creates a map of the available metrics
-// unmarshalers.
-func defaultMetricsUnmarshalers(logger *zap.Logger) map[string]unmarshaler.MetricsUnmarshaler {
-	cwmsu := cwmetricstream.NewUnmarshaler(logger)
-	otlpv1msu := otlpmetricstream.NewUnmarshaler(logger)
-	return map[string]unmarshaler.MetricsUnmarshaler{
-		cwmsu.Type():     cwmsu,
-		otlpv1msu.Type(): otlpv1msu,
-	}
-}
-
-// defaultLogsUnmarshalers creates a map of the available logs unmarshalers.
-func defaultLogsUnmarshalers(logger *zap.Logger) map[string]unmarshaler.LogsUnmarshaler {
-	u := cwlog.NewUnmarshaler(logger)
-	return map[string]unmarshaler.LogsUnmarshaler{
-		u.Type(): u,
-	}
-}
-
 // createDefaultConfig creates a default config with the endpoint set
 // to port 8443 and the record type set to the CloudWatch metric stream.
 func createDefaultConfig() component.Config {
@@ -88,7 +45,9 @@ func createMetricsReceiver(
 	cfg component.Config,
 	nextConsumer consumer.Metrics,
 ) (receiver.Metrics, error) {
-	return newMetricsReceiver(cfg.(*Config), set, defaultMetricsUnmarshalers(set.Logger), nextConsumer)
+	c := cfg.(*Config)
+	handleDeprecatedConfig(c, set.Logger)
+	return newMetricsReceiver(c, set, nextConsumer)
 }
 
 // createMetricsReceiver implements the CreateMetricsReceiver function type.
@@ -98,5 +57,7 @@ func createLogsReceiver(
 	cfg component.Config,
 	nextConsumer consumer.Logs,
 ) (receiver.Logs, error) {
-	return newLogsReceiver(cfg.(*Config), set, defaultLogsUnmarshalers(set.Logger), nextConsumer)
+	c := cfg.(*Config)
+	handleDeprecatedConfig(c, set.Logger)
+	return newLogsReceiver(c, set, nextConsumer)
 }

receiver/awsfirehosereceiver/factory_test.go

@@ -11,8 +11,6 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/receiver/receivertest"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/otlpmetricstream"
 )
 
 func TestValidConfig(t *testing.T) {
@@ -41,10 +39,3 @@ func TestCreateLogsReceiver(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, r)
 }
-
-func TestValidateRecordType(t *testing.T) {
-	require.NoError(t, validateRecordType(defaultMetricsRecordType))
-	require.NoError(t, validateRecordType(defaultLogsRecordType))
-	require.NoError(t, validateRecordType(otlpmetricstream.TypeStr))
-	require.Error(t, validateRecordType("nop"))
-}

receiver/awsfirehosereceiver/go.mod

@@ -4,6 +4,8 @@ go 1.22.0
 
 require (
 	github.com/gogo/protobuf v1.3.2
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics v0.117.1-0.20250114172347-71aae791d7f8
+	github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil v0.117.1-0.20250114172347-71aae791d7f8
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v0.117.1-0.20250114172347-71aae791d7f8
 	go.opentelemetry.io/collector/component/componentstatus v0.117.1-0.20250114172347-71aae791d7f8
@@ -24,6 +26,7 @@ require (
 )
 
 require (
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fsnotify/fsnotify v1.8.0 // indirect
@@ -41,6 +44,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.117.1-0.20250114172347-71aae791d7f8 // indirect
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/rs/cors v1.11.1 // indirect
@@ -75,3 +79,13 @@ retract (
 	v0.76.1
 	v0.65.0
 )
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/pdatautil => ../../internal/pdatautil
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil => ../../pkg/pdatautil
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/internal/exp/metrics => ../../internal/exp/metrics
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest => ../../pkg/pdatatest
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden => ../../pkg/golden