[receiver/awsfirehosereceiver] Add support for CloudWatch logs (open-telemetry#35077)

**Description:** Add support for CloudWatch logs in JSON format. 

The receiver can now be configured to receiver either `cwmetrics` or
`cwlogs` record type for CloudWatch metrics logs, respectively. The
former is only valid for metrics receivers and the latter for logs
receivers.

Some changes were necessary to the default record type handling to
conform to the contracts required by the generated tests and ensure the
default config returned by the factory was usable. Previously the
default record type was `cwmetrics` but this is not sensible for logs
receivers.

To accommodate this, the default record type is now
telemetry-type-specific. That is, for metrics it is `cwmetrics` and for
logs it is `cwlogs`. Each receiver makes this determination on its own.
Therefore an empty string is now the default returned by the factory and
an empty string is also acceptable in the users config. It is no longer
required to specify a record type, although it is still supported, so
this is not a breaking change.

**Link to tracking Issue:** n/a

**Testing:** Unit tests covering new code as well as manual smoke
testing against AWS Firehose and CloudWatch themselves.

**Documentation:** Component README update.

Author: gavincabbage

.chloggen/awsfirehosereceiver-cwlogs.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 CloudWatch logs
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35077]
+
+# (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: []

receiver/awsfirehosereceiver/README.md

@@ -3,7 +3,7 @@
 <!-- status autogenerated section -->
 | Status        |           |
 | ------------- |-----------|
-| Stability     | [alpha]: metrics   |
+| Stability     | [alpha]: metrics, logs   |
 | Distributions | [contrib] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fawsfirehose%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fawsfirehose) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fawsfirehose%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fawsfirehose) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@Aneurysm9](https://www.github.com/Aneurysm9) |
@@ -62,3 +62,24 @@ See [documentation](https://docs.aws.amazon.com/firehose/latest/dev/create-desti
 The record type 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.
+For example:
+
+```json
+{
+  "messageType": "DATA_MESSAGE",
+  "owner": "111122223333",
+  "logGroup": "my-log-group",
+  "logStream": "my-log-stream",
+  "subscriptionFilters": ["my-subscription-filter"],
+  "logEvents": [
+    {
+      "id": "123",
+      "timestamp": 1725544035523,
+      "message": "My log message."
+    }
+  ]
+}
+```
+

receiver/awsfirehosereceiver/config.go

@@ -30,8 +30,10 @@ func (c *Config) Validate() error {
 	if c.Endpoint == "" {
 		return errors.New("must specify endpoint")
 	}
-	if c.RecordType == "" {
-		return errors.New("must specify record type")
+	// 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)
 	}
-	return validateRecordType(c.RecordType)
+	return nil
 }

receiver/awsfirehosereceiver/config_test.go

@@ -4,6 +4,7 @@
 package awsfirehosereceiver
 
 import (
+	"fmt"
 	"path/filepath"
 	"testing"
 
@@ -18,29 +19,40 @@ import (
 )
 
 func TestLoadConfig(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(component.NewIDWithName(metadata.Type, "").String())
-	require.NoError(t, err)
-	require.NoError(t, sub.Unmarshal(cfg))
-
-	assert.NoError(t, component.ValidateConfig(cfg))
-
-	require.Equal(t, &Config{
-		RecordType: "cwmetrics",
-		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)
+	for _, configType := range []string{
+		"cwmetrics", "cwlogs", "invalid",
+	} {
+		t.Run(configType, func(t *testing.T) {
+			fileName := fmt.Sprintf("%s_config.yaml", configType)
+			cm, err := confmaptest.LoadConf(filepath.Join("testdata", fileName))
+			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)
+			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",
+							},
+						},
+					},
+				}, cfg)
+			}
+		})
+	}
 }

receiver/awsfirehosereceiver/factory.go

@@ -16,19 +16,20 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
 	"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"
 )
 
 const (
-	defaultRecordType = cwmetricstream.TypeStr
-	defaultEndpoint   = "0.0.0.0:4433"
-	defaultPort       = 4433
+	defaultEndpoint = "0.0.0.0:4433"
+	defaultPort     = 4433
 )
 
 var (
 	errUnrecognizedRecordType = errors.New("unrecognized record type")
 	availableRecordTypes      = map[string]bool{
 		cwmetricstream.TypeStr: true,
+		cwlog.TypeStr:          true,
 	}
 )
 
@@ -38,7 +39,8 @@ func NewFactory() receiver.Factory {
 	return receiver.NewFactory(
 		metadata.Type,
 		createDefaultConfig,
-		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability))
+		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
+		receiver.WithLogs(createLogsReceiver, metadata.LogsStability))
 }
 
 // validateRecordType checks the available record types for the
@@ -59,11 +61,18 @@ func defaultMetricsUnmarshalers(logger *zap.Logger) map[string]unmarshaler.Metri
 	}
 }
 
+// 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 {
 	return &Config{
-		RecordType: defaultRecordType,
 		ServerConfig: confighttp.ServerConfig{
 			Endpoint: testutil.EndpointForPort(defaultPort),
 		},
@@ -79,3 +88,13 @@ func createMetricsReceiver(
 ) (receiver.Metrics, error) {
 	return newMetricsReceiver(cfg.(*Config), set, defaultMetricsUnmarshalers(set.Logger), nextConsumer)
 }
+
+// createMetricsReceiver implements the CreateMetricsReceiver function type.
+func createLogsReceiver(
+	_ context.Context,
+	set receiver.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (receiver.Logs, error) {
+	return newLogsReceiver(cfg.(*Config), set, defaultLogsUnmarshalers(set.Logger), nextConsumer)
+}

receiver/awsfirehosereceiver/factory_test.go

@@ -29,7 +29,19 @@ func TestCreateMetrics(t *testing.T) {
 	require.NotNil(t, r)
 }
 
+func TestCreateLogsReceiver(t *testing.T) {
+	r, err := createLogsReceiver(
+		context.Background(),
+		receivertest.NewNopSettings(),
+		createDefaultConfig(),
+		consumertest.NewNop(),
+	)
+	require.NoError(t, err)
+	require.NotNil(t, r)
+}
+
 func TestValidateRecordType(t *testing.T) {
-	require.NoError(t, validateRecordType(defaultRecordType))
+	require.NoError(t, validateRecordType(defaultMetricsRecordType))
+	require.NoError(t, validateRecordType(defaultLogsRecordType))
 	require.Error(t, validateRecordType("nop"))
 }

receiver/awsfirehosereceiver/generated_component_test.go

@@ -0,0 +1,48 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package compression // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/compression"
+
+import (
+	"bytes"
+	"compress/gzip"
+)
+
+// Zip returns a gzip-compressed representation of the input bytes.
+func Zip(data []byte) ([]byte, error) {
+	var b bytes.Buffer
+	w := gzip.NewWriter(&b)
+
+	_, err := w.Write(data)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = w.Flush(); err != nil {
+		return nil, err
+	}
+
+	if err = w.Close(); err != nil {
+		return nil, err
+	}
+
+	return b.Bytes(), nil
+}
+
+// Unzip expects gzip-compressed input bytes and returns their uncompressed form.
+func Unzip(data []byte) ([]byte, error) {
+	b := bytes.NewBuffer(data)
+
+	r, err := gzip.NewReader(b)
+	if err != nil {
+		return nil, err
+	}
+
+	var rv bytes.Buffer
+	_, err = rv.ReadFrom(r)
+	if err != nil {
+		return nil, err
+	}
+
+	return rv.Bytes(), nil
+}

receiver/awsfirehosereceiver/internal/metadata/generated_status.go

@@ -0,0 +1,17 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cwlog // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog"
+
+type cWLog struct {
+	MessageType         string   `json:"messageType"`
+	Owner               string   `json:"owner"`
+	LogGroup            string   `json:"logGroup"`
+	LogStream           string   `json:"logStream"`
+	SubscriptionFilters []string `json:"subscriptionFilters"`
+	LogEvents           []struct {
+		ID        string `json:"id"`
+		Timestamp int64  `json:"timestamp"`
+		Message   string `json:"message"`
+	} `json:"logEvents"`
+}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/compression/compression.go

@@ -0,0 +1,50 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package cwlog // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/cwlog"
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
+)
+
+const (
+	attributeAWSCloudWatchLogGroupName  = "aws.cloudwatch.log_group_name"
+	attributeAWSCloudWatchLogStreamName = "aws.cloudwatch.log_stream_name"
+)
+
+// resourceAttributes are the CloudWatch log attributes that define a unique resource.
+type resourceAttributes struct {
+	owner, logGroup, logStream string
+}
+
+// resourceLogsBuilder provides convenient access to the a Resource's LogRecordSlice.
+type resourceLogsBuilder struct {
+	rls plog.LogRecordSlice
+}
+
+// setAttributes applies the resourceAttributes to the provided Resource.
+func (ra *resourceAttributes) setAttributes(resource pcommon.Resource) {
+	attrs := resource.Attributes()
+	attrs.PutStr(conventions.AttributeCloudAccountID, ra.owner)
+	attrs.PutStr(attributeAWSCloudWatchLogGroupName, ra.logGroup)
+	attrs.PutStr(attributeAWSCloudWatchLogStreamName, ra.logStream)
+}
+
+// newResourceLogsBuilder to capture logs for the Resource defined by the provided attributes.
+func newResourceLogsBuilder(logs plog.Logs, attrs resourceAttributes) *resourceLogsBuilder {
+	rls := logs.ResourceLogs().AppendEmpty()
+	attrs.setAttributes(rls.Resource())
+	return &resourceLogsBuilder{rls.ScopeLogs().AppendEmpty().LogRecords()}
+}
+
+// AddLog events to the LogRecordSlice. Resource attributes are captured when creating
+// the resourceLogsBuilder, so we only need to consider the LogEvents themselves.
+func (rlb *resourceLogsBuilder) AddLog(log cWLog) {
+	for _, event := range log.LogEvents {
+		logLine := rlb.rls.AppendEmpty()
+		logLine.SetTimestamp(pcommon.Timestamp(event.Timestamp))
+		logLine.Body().SetStr(event.Message)
+	}
+}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/cwlog.go

@@ -0,0 +1,4 @@
+{"CHANGE":-0.09,"PRICE":4.96,"TICKER_SYMBOL":"KIN","SECTOR":"ENERGY"}
+{"CHANGE":-1.47,"PRICE":134.74,"TICKER_SYMBOL":"DFG","SECTOR":"TECHNOLOGY"}
+{"CHANGE":1.96,"PRICE":57.53,"TICKER_SYMBOL":"SAC","SECTOR":"ENERGY"}
+{"CHANGE":0.04,"PRICE":32.84,"TICKER_SYMBOL":"PJN","SECTOR":"RETAIL"}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/logsbuilder.go

@@ -0,0 +1,2 @@
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695232","timestamp":1725544035523,"message":"Hello world, here is our first log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695233","timestamp":1725554035523,"message":"Hello world, here is our second log message!"}]}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/testdata/invalid_records

@@ -0,0 +1,6 @@
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695232","timestamp":1725544035523,"message":"Hello world, here is our first log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695233","timestamp":1725554035523,"message":"Hello world, here is our second log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test2","logStream":"test1","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695234","timestamp":1725564035523,"message":"Hello world, here is our third log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test2","logStream":"test2","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695235","timestamp":1725574035523,"message":"Hello world, here is our fourth log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test2","logStream":"test1","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695236","timestamp":1725584035523,"message":"Hello world, here is our fifth log message!"}]}
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test2","logStream":"test2","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695237","timestamp":1725594035523,"message":"Hello world, here is our sixth log message!"}]}

receiver/awsfirehosereceiver/internal/unmarshaler/cwlog/testdata/multiple_records

@@ -0,0 +1 @@
+{"messageType":"DATA_MESSAGE","owner":"123","logGroup":"test","logStream":"test","subscriptionFilters":["test"],"logEvents":[{"id":"38480917865042697267627490045603633139480491071049695232","timestamp":1725544035523,"message":"Hello world, here is our first log message!"}]}