[receiver/awsfirehose] Add support for encoding extensions (#37262)

#### Description

Add support for using encoding extensions for unmarshalling records
transmitted via Amazon Data Firehose.
    
The "record_type" config is now deprecated and has been replaced by
"encoding". This new config setting supports all of the existing
encodings (cwlogs, cwmetrics otlp_v1) as well as support for loading
additional encodings via extensions.

#### Link to tracking issue

Fixes
#37113

#### Testing

Should be a non-functional change, so mostly relying on existing unit
tests to catch issues. Tests have been added for new extension
functionality. Manually tested creating a Firehose delivery stream and
using the `text_encoding` extension:

1. Ran collector with following config:

```yaml
receivers:
  awsfirehose:
    endpoint: localhost:1234
    encoding:  text_encoding

exporters:
  debug:
    verbosity: detailed

extensions:
  text_encoding:

service:
  extensions: [text_encoding]
  pipelines:
    logs:
      receivers: [awsfirehose]
      processors: []
      exporters: [debug]
```

2. Exposed to the internet with ngrok
3. Created a Firehose delivery stream pointed at ngrok HTTPS endpoint
4. Used AWS CLI to send a record: `aws firehose put-record
--delivery-stream-name=axwtest --record Data=$(echo -n abc | base64)`
5. Observed log record being exported by the debug exporter:

```
2025-02-11T14:09:17.090+0800    info    Logs    {"kind": "exporter", "data_type": "logs", "name": "debug", "resource logs": 1, "log records": 1}
2025-02-11T14:09:17.090+0800    info    ResourceLog #0
Resource SchemaURL: 
ScopeLogs #0
ScopeLogs SchemaURL: 
InstrumentationScope  
LogRecord #0
ObservedTimestamp: 2025-02-11 06:09:17.090506322 +0000 UTC
Timestamp: 1970-01-01 00:00:00 +0000 UTC
SeverityText: 
SeverityNumber: Unspecified(0)
Body: Str(abc)
Trace ID: 
Span ID: 
Flags: 0
```

#### Documentation

Updated README.

---------

Co-authored-by: Anthony Mirabella <[email protected]>
Co-authored-by: Pablo Baeyens <[email protected]>

.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,24 @@ import (
 
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configopaque"
+	"go.uber.org/zap"
 )
 
+var errRecordTypeEncodingSet = errors.New("record_type must not be set when encoding is set")
+
 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: [v0.121.0] 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 +39,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 errRecordTypeEncodingSet
 	}
 	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

@@ -20,7 +20,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"
@@ -35,24 +35,35 @@ func TestLoadConfig(t *testing.T) {
 			require.NoError(t, sub.Unmarshal(cfg))
 
 			err = xconfmap.Validate(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 = xconfmap.Validate(cfg)
+	assert.ErrorIs(t, err, errRecordTypeEncodingSet)
+}

receiver/awsfirehosereceiver/factory.go

@@ -5,35 +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/pdata/plog"
-	"go.opentelemetry.io/collector/pdata/pmetric"
 	"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/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 {
@@ -44,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]pmetric.Unmarshaler {
-	cwmsu := cwmetricstream.NewUnmarshaler(logger)
-	otlpv1msu := otlpmetricstream.NewUnmarshaler(logger)
-	return map[string]pmetric.Unmarshaler{
-		cwmsu.Type():     cwmsu,
-		otlpv1msu.Type(): otlpv1msu,
-	}
-}
-
-// defaultLogsUnmarshalers creates a map of the available logs unmarshalers.
-func defaultLogsUnmarshalers(logger *zap.Logger) map[string]plog.Unmarshaler {
-	u := cwlog.NewUnmarshaler(logger)
-	return map[string]plog.Unmarshaler{
-		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 {
@@ -89,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.
@@ -99,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

@@ -13,7 +13,6 @@ import (
 	"go.opentelemetry.io/collector/receiver/receivertest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/metadata"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/otlpmetricstream"
 )
 
 func TestValidConfig(t *testing.T) {
@@ -42,10 +41,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/logs_receiver.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"go.opentelemetry.io/collector/pdata/plog"
@@ -18,11 +19,14 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog"
 )
 
-const defaultLogsRecordType = cwlog.TypeStr
+const defaultLogsEncoding = cwlog.TypeStr
 
 // logsConsumer implements the firehoseConsumer
 // to use a logs consumer and unmarshaler.
 type logsConsumer struct {
+	config   *Config
+	settings receiver.Settings
+
 	// consumer passes the translated logs on to the
 	// next consumer.
 	consumer consumer.Logs
@@ -38,21 +42,12 @@ var _ firehoseConsumer = (*logsConsumer)(nil)
 func newLogsReceiver(
 	config *Config,
 	set receiver.Settings,
-	unmarshalers map[string]plog.Unmarshaler,
 	nextConsumer consumer.Logs,
 ) (receiver.Logs, error) {
-	recordType := config.RecordType
-	if recordType == "" {
-		recordType = defaultLogsRecordType
-	}
-	configuredUnmarshaler := unmarshalers[recordType]
-	if configuredUnmarshaler == nil {
-		return nil, fmt.Errorf("%w: recordType = %s", errUnrecognizedRecordType, recordType)
-	}
-
 	c := &logsConsumer{
-		consumer:    nextConsumer,
-		unmarshaler: configuredUnmarshaler,
+		config:   config,
+		settings: set,
+		consumer: nextConsumer,
 	}
 	return &firehoseReceiver{
 		settings: set,
@@ -61,8 +56,32 @@ func newLogsReceiver(
 	}, nil
 }
 
-// Consume uses the configured unmarshaler to unmarshal each record
-// into a plog.Logs and pass it to the next consumer, one record at a time.
+// Start sets the consumer's log unmarshaler to either a built-in
+// unmarshaler or one loaded from an encoding extension.
+func (c *logsConsumer) Start(_ context.Context, host component.Host) error {
+	encoding := c.config.Encoding
+	if encoding == "" {
+		encoding = c.config.RecordType
+		if encoding == "" {
+			encoding = defaultLogsEncoding
+		}
+	}
+	if encoding == cwlog.TypeStr {
+		// TODO: make cwlogs an encoding extension
+		c.unmarshaler = cwlog.NewUnmarshaler(c.settings.Logger)
+	} else {
+		unmarshaler, err := loadEncodingExtension[plog.Unmarshaler](host, encoding, "logs")
+		if err != nil {
+			return fmt.Errorf("failed to load encoding extension: %w", err)
+		}
+		c.unmarshaler = unmarshaler
+	}
+	return nil
+}
+
+// Consume uses the configured unmarshaler to deserialize each record,
+// with each resulting plog.Logs being sent to the next consumer as
+// they are unmarshalled.
 func (c *logsConsumer) Consume(ctx context.Context, nextRecord nextRecordFunc, commonAttributes map[string]string) (int, error) {
 	for {
 		record, err := nextRecord()