[receiver/awsfirehosereceiver] Add support to ingest logs from services that send directly to firehose

.chloggen/add_firehose_logs.yaml

@@ -0,0 +1,25 @@
+# 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 logs sent from services directly to Firehose
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [36184]
+
+# (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]

receiver/awsfirehosereceiver/README.md

@@ -86,3 +86,7 @@ For example:
 ### otlp_v1
 The OTLP v1 format 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.
+
+### firehoselogs
+The record type for logs sent directly from AWS services to Firehose. 
+For example, VPC flow logs can be configured to send to Firehose directly. Please see [documentation](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs-firehose.html) for more details.

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", "firehoselogs", "invalid",
 	} {
 		t.Run(configType, func(t *testing.T) {
 			fileName := configType + "_config.yaml"

receiver/awsfirehosereceiver/factory.go

@@ -18,6 +18,7 @@ import (
 	"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/firehoselog"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/otlpmetricstream"
 )
 
@@ -32,6 +33,7 @@ var (
 		cwmetricstream.TypeStr:   true,
 		cwlog.TypeStr:            true,
 		otlpmetricstream.TypeStr: true,
+		firehoselog.TypeStr:      true,
 	}
 )
 
@@ -67,9 +69,11 @@ 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)
+	cwlogu := cwlog.NewUnmarshaler(logger)
+	firehoselogu := firehoselog.NewUnmarshaler(logger)
 	return map[string]unmarshaler.LogsUnmarshaler{
-		u.Type(): u,
+		cwlogu.Type():       cwlogu,
+		firehoselogu.Type(): firehoselogu,
 	}
 }
 

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

@@ -37,7 +37,7 @@ func NewUnmarshaler(logger *zap.Logger) *Unmarshaler {
 // Unmarshal deserializes the records into cWLogs and uses the
 // resourceLogsBuilder to group them into a single plog.Logs.
 // Skips invalid cWLogs received in the record and
-func (u Unmarshaler) Unmarshal(records [][]byte) (plog.Logs, error) {
+func (u Unmarshaler) Unmarshal(records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (plog.Logs, error) {
 	md := plog.NewLogs()
 	builders := make(map[resourceAttributes]*resourceLogsBuilder)
 	for recordIndex, compressedRecord := range records {

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

@@ -61,7 +61,7 @@ func TestUnmarshal(t *testing.T) {
 			require.NoError(t, err)
 			records := [][]byte{compressedRecord}
 
-			got, err := unmarshaler.Unmarshal(records)
+			got, err := unmarshaler.Unmarshal(records, nil, "", 0)
 			if testCase.wantErr != nil {
 				require.Error(t, err)
 				require.Equal(t, testCase.wantErr, err)

receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog/firehoselog.go

@@ -0,0 +1,10 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package firehoselog // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog"
+
+type firehoseLog struct {
+	Timestamp   int64  `json:"timestamp"`
+	FirehoseARN string `json:"firehoseARN"`
+	Message     string `json:"message"`
+}

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

@@ -0,0 +1,47 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package firehoselog // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog"
+
+import (
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"time"
+)
+
+const (
+	attributeAWSFirehoseARN = "aws.firehose.arn"
+)
+
+// resourceAttributes are the firehose log attributes that define a unique resource.
+type resourceAttributes struct {
+	firehoseARN string
+}
+
+// resourceLogsBuilder provides convenient access to the 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(attributeAWSFirehoseARN, ra.firehoseARN)
+}
+
+// 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 firehoseLog) {
+	logLine := rlb.rls.AppendEmpty()
+	// pcommon.Timestamp is a time specified as UNIX Epoch time in nanoseconds
+	// but timestamp in cloudwatch logs are in milliseconds.
+	logLine.SetTimestamp(pcommon.Timestamp(log.Timestamp * int64(time.Millisecond)))
+	logLine.Body().SetStr(log.Message)
+}

receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog/testdata/firehose_log.json

@@ -0,0 +1 @@
+{"requestId":"test-request-id-firehoselogs","timestamp":1646096348000,"records":[{"data":"eyJ2ZXJzaW9uIjoiMS4xMDAwMDAiLCJhY2NvdW50X2lkIjoiNDI4MTUyNTAyNDY3IiwicmVnaW9uIjoidXMtZWFzdC0xIiwidnBjX2lkIjoidnBjLWU4NzIyZDkyIiwicXVlcnlfdGltZXN0YW1wIjoiMjAyMi0wNC0xNVQxNTo1NTozMFoiLCJxdWVyeV9uYW1lIjoiMTcyLjE0NC41Mi4xMTYuaW4tYWRkci5hcnBhLiIsInF1ZXJ5X3R5cGUiOiJQVFIiLCJxdWVyeV9jbGFzcyI6IklOIiwicmNvZGUiOiIiLCJhbnN3ZXJzIjpbXSwic3JjYWRkciI6IjE3Mi4zMS44NC40MyIsInNyY3BvcnQiOiI0Mzc4NiIsInRyYW5zcG9ydCI6IlVEUCIsInNyY2lkcyI6eyJpbnN0YW5jZSI6ImktMDY0NmMwNDM1NTU0Y2M1ZWQifX0="}]}

receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog/unmarshaler.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package firehoselog // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog"
+
+import (
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver/internal/unmarshaler"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.uber.org/zap"
+)
+
+const (
+	TypeStr = "firehoselogs"
+)
+
+// Unmarshaler for the Firehose log record format.
+type Unmarshaler struct {
+	logger *zap.Logger
+}
+
+var _ unmarshaler.LogsUnmarshaler = (*Unmarshaler)(nil)
+
+// NewUnmarshaler creates a new instance of the Unmarshaler.
+func NewUnmarshaler(logger *zap.Logger) *Unmarshaler {
+	return &Unmarshaler{logger}
+}
+
+// Unmarshal deserializes the records into firehoselogs and uses the
+// resourceLogsBuilder to group them into a single plog.Logs.
+func (u Unmarshaler) Unmarshal(records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (plog.Logs, error) {
+	md := plog.NewLogs()
+	builders := make(map[resourceAttributes]*resourceLogsBuilder)
+	for _, record := range records {
+		var log firehoseLog
+		log.Message = string(record)
+		log.Timestamp = timestamp
+		attrs := resourceAttributes{
+			firehoseARN: firehoseARN,
+		}
+
+		lb, ok := builders[attrs]
+		if !ok {
+			lb = newResourceLogsBuilder(md, attrs)
+			builders[attrs] = lb
+		}
+		lb.AddLog(log)
+	}
+	return md, nil
+}
+
+// Type of the serialized messages.
+func (u Unmarshaler) Type() string {
+	return TypeStr
+}

receiver/awsfirehosereceiver/internal/unmarshaler/firehoselog/unmarshaler_test.go

@@ -0,0 +1,26 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package firehoselog
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func TestUnmarshal(t *testing.T) {
+	body, err := os.ReadFile("testdata/firehose_log.json")
+	require.NoError(t, err)
+
+	unmarshaler := NewUnmarshaler(zap.NewNop())
+	var records [][]byte
+	records = append(records, body)
+	got, err := unmarshaler.Unmarshal(records, nil, "arn:aws:firehose:us-east-1:123456789:deliverystream/testStream", 1646096348000)
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.Equal(t, 1, got.ResourceLogs().Len())
+	require.Equal(t, 1, got.ResourceLogs().At(0).ScopeLogs().Len())
+}

receiver/awsfirehosereceiver/internal/unmarshaler/unmarshaler.go

@@ -20,7 +20,7 @@ type MetricsUnmarshaler interface {
 // LogsUnmarshaler deserializes the message body
 type LogsUnmarshaler interface {
 	// Unmarshal deserializes the records into logs.
-	Unmarshal(records [][]byte) (plog.Logs, error)
+	Unmarshal(records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (plog.Logs, error)
 
 	// Type of the serialized messages.
 	Type() string

receiver/awsfirehosereceiver/internal/unmarshaler/unmarshalertest/nop_logs_unmarshaler.go

@@ -37,7 +37,7 @@ func NewErrLogs(err error) *NopLogsUnmarshaler {
 }
 
 // Unmarshal deserializes the records into logs.
-func (u *NopLogsUnmarshaler) Unmarshal([][]byte) (plog.Logs, error) {
+func (u *NopLogsUnmarshaler) Unmarshal([][]byte, map[string]string, string, int64) (plog.Logs, error) {
 	return u.logs, u.err
 }
 

receiver/awsfirehosereceiver/internal/unmarshaler/unmarshalertest/nop_logs_unmarshaler_test.go

@@ -13,7 +13,7 @@ import (
 
 func TestNewNopLogs(t *testing.T) {
 	unmarshaler := NewNopLogs()
-	got, err := unmarshaler.Unmarshal(nil)
+	got, err := unmarshaler.Unmarshal(nil, nil, "", 0)
 	require.NoError(t, err)
 	require.NotNil(t, got)
 	require.Equal(t, typeStr, unmarshaler.Type())
@@ -23,7 +23,7 @@ func TestNewWithLogs(t *testing.T) {
 	logs := plog.NewLogs()
 	logs.ResourceLogs().AppendEmpty()
 	unmarshaler := NewWithLogs(logs)
-	got, err := unmarshaler.Unmarshal(nil)
+	got, err := unmarshaler.Unmarshal(nil, nil, "", 0)
 	require.NoError(t, err)
 	require.NotNil(t, got)
 	require.Equal(t, logs, got)
@@ -33,7 +33,7 @@ func TestNewWithLogs(t *testing.T) {
 func TestNewErrLogs(t *testing.T) {
 	wantErr := errors.New("test error")
 	unmarshaler := NewErrLogs(wantErr)
-	got, err := unmarshaler.Unmarshal(nil)
+	got, err := unmarshaler.Unmarshal(nil, nil, "", 0)
 	require.Error(t, err)
 	require.Equal(t, wantErr, err)
 	require.NotNil(t, got)

receiver/awsfirehosereceiver/logs_receiver.go

@@ -61,8 +61,8 @@ func newLogsReceiver(
 // Consume uses the configured unmarshaler to deserialize the records into a
 // single plog.Logs. It will send the final result
 // to the next consumer.
-func (mc *logsConsumer) Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string) (int, error) {
-	md, err := mc.unmarshaler.Unmarshal(records)
+func (mc *logsConsumer) Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (int, error) {
+	md, err := mc.unmarshaler.Unmarshal(records, commonAttributes, firehoseARN, timestamp)
 	if err != nil {
 		return http.StatusBadRequest, err
 	}

receiver/awsfirehosereceiver/logs_receiver_test.go

@@ -94,7 +94,7 @@ func TestLogsConsumer(t *testing.T) {
 				unmarshaler: unmarshalertest.NewErrLogs(testCase.unmarshalerErr),
 				consumer:    consumertest.NewErr(testCase.consumerErr),
 			}
-			gotStatus, gotErr := mc.Consume(context.TODO(), nil, nil)
+			gotStatus, gotErr := mc.Consume(context.TODO(), nil, nil, "", 0)
 			require.Equal(t, testCase.wantStatus, gotStatus)
 			require.Equal(t, testCase.wantErr, gotErr)
 		})
@@ -110,7 +110,7 @@ func TestLogsConsumer(t *testing.T) {
 		}
 		gotStatus, gotErr := mc.Consume(context.TODO(), nil, map[string]string{
 			"CommonAttributes": "Test",
-		})
+		}, "", 0)
 		require.Equal(t, http.StatusOK, gotStatus)
 		require.NoError(t, gotErr)
 		gotRms := rc.result.ResourceLogs()

receiver/awsfirehosereceiver/metrics_receiver.go

@@ -63,7 +63,7 @@ func newMetricsReceiver(
 // single pmetric.Metrics. If there are common attributes available, then it will
 // attach those to each of the pcommon.Resources. It will send the final result
 // to the next consumer.
-func (mc *metricsConsumer) Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string) (int, error) {
+func (mc *metricsConsumer) Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (int, error) {
 	md, err := mc.unmarshaler.Unmarshal(records)
 	if err != nil {
 		return http.StatusBadRequest, err

receiver/awsfirehosereceiver/metrics_receiver_test.go

@@ -95,7 +95,7 @@ func TestMetricsConsumer(t *testing.T) {
 				unmarshaler: unmarshalertest.NewErrMetrics(testCase.unmarshalerErr),
 				consumer:    consumertest.NewErr(testCase.consumerErr),
 			}
-			gotStatus, gotErr := mc.Consume(context.TODO(), nil, nil)
+			gotStatus, gotErr := mc.Consume(context.TODO(), nil, nil, "", 0)
 			require.Equal(t, testCase.wantStatus, gotStatus)
 			require.Equal(t, testCase.wantErr, gotErr)
 		})
@@ -111,7 +111,7 @@ func TestMetricsConsumer(t *testing.T) {
 		}
 		gotStatus, gotErr := mc.Consume(context.TODO(), nil, map[string]string{
 			"CommonAttributes": "Test",
-		})
+		}, "", 0)
 		require.Equal(t, http.StatusOK, gotStatus)
 		require.NoError(t, gotErr)
 		gotRms := rc.result.ResourceMetrics()

receiver/awsfirehosereceiver/receiver.go

@@ -26,6 +26,7 @@ const (
 	headerFirehoseRequestID        = "X-Amz-Firehose-Request-Id"
 	headerFirehoseAccessKey        = "X-Amz-Firehose-Access-Key"
 	headerFirehoseCommonAttributes = "X-Amz-Firehose-Common-Attributes"
+	headerFirehoseSourceArn        = "X-Amz-Firehose-Source-Arn"
 	headerContentType              = "Content-Type"
 	headerContentLength            = "Content-Length"
 )
@@ -41,7 +42,7 @@ var (
 // The firehoseConsumer is responsible for using the unmarshaler and the consumer.
 type firehoseConsumer interface {
 	// Consume unmarshalls and consumes the records.
-	Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string) (int, error)
+	Consume(ctx context.Context, records [][]byte, commonAttributes map[string]string, firehoseARN string, timestamp int64) (int, error)
 }
 
 // firehoseReceiver
@@ -220,7 +221,15 @@ func (fmr *firehoseReceiver) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		)
 	}
 
-	statusCode, err := fmr.consumer.Consume(ctx, records, commonAttributes)
+	firehoseARN, err := fmr.getFirehoseArn(r)
+	if err != nil {
+		fmr.settings.Logger.Error(
+			"Unable to get firehose source ARN from request header. Will not attach attributes.",
+			zap.Error(err),
+		)
+	}
+
+	statusCode, err := fmr.consumer.Consume(ctx, records, commonAttributes, firehoseARN, fr.Timestamp)
 	if err != nil {
 		fmr.settings.Logger.Error(
 			"Unable to consume records",
@@ -272,6 +281,11 @@ func (fmr *firehoseReceiver) getCommonAttributes(r *http.Request) (map[string]st
 	return attributes, nil
 }
 
+// getFirehoseArn returns firehose ARN from the request header
+func (fmr *firehoseReceiver) getFirehoseArn(r *http.Request) (string, error) {
+	return r.Header.Get(headerFirehoseSourceArn), nil
+}
+
 // sendResponse writes a response to Firehose in the expected format.
 func (fmr *firehoseReceiver) sendResponse(w http.ResponseWriter, requestID string, statusCode int, err error) {
 	var errorMessage string

receiver/awsfirehosereceiver/receiver_test.go

@@ -40,7 +40,7 @@ func newNopFirehoseConsumer(statusCode int, err error) *nopFirehoseConsumer {
 	return &nopFirehoseConsumer{statusCode, err}
 }
 
-func (nfc *nopFirehoseConsumer) Consume(context.Context, [][]byte, map[string]string) (int, error) {
+func (nfc *nopFirehoseConsumer) Consume(context.Context, [][]byte, map[string]string, string, int64) (int, error) {
 	return nfc.statusCode, nfc.err
 }
 

receiver/awsfirehosereceiver/testdata/firehoselogs_config.yaml

@@ -0,0 +1,7 @@
+awsfirehose:
+  endpoint: 0.0.0.0:4433
+  record_type: firehoselogs
+  access_key: "some_access_key"
+  tls:
+    cert_file: server.crt
+    key_file: server.key