Add otel.receiver.aws_firehose component

CHANGELOG.md

@@ -44,6 +44,9 @@ v0.40.0 (2024-02-27)
 - A new `otelcol.processor.resourcedetection` component which inserts resource attributes
   to OTLP telemetry based on the host on which Grafana Agent is running. (@ptodev)
 
+- A new `otelcol.receiver.aws_firehose` component which receives log entries
+  from AWS Firehose. (@Obito1903)
+
 - Expose track_timestamps_staleness on Prometheus scraping, to fix the issue where container metrics live for 5 minutes after the container disappears. (@ptodev)
 
 - Introduce the `remotecfg` service that enables loading configuration from a

component/all/all.go

@@ -88,6 +88,7 @@ import (
 	_ "github.com/grafana/agent/component/otelcol/processor/span"                   // Import otelcol.processor.span
 	_ "github.com/grafana/agent/component/otelcol/processor/tail_sampling"          // Import otelcol.processor.tail_sampling
 	_ "github.com/grafana/agent/component/otelcol/processor/transform"              // Import otelcol.processor.transform
+	_ "github.com/grafana/agent/component/otelcol/receiver/awsfirehose"             // Import otelcol.receiver.awsfirehose
 	_ "github.com/grafana/agent/component/otelcol/receiver/jaeger"                  // Import otelcol.receiver.jaeger
 	_ "github.com/grafana/agent/component/otelcol/receiver/kafka"                   // Import otelcol.receiver.kafka
 	_ "github.com/grafana/agent/component/otelcol/receiver/loki"                    // Import otelcol.receiver.loki

component/otelcol/receiver/awsfirehose/awsfirehose.go

@@ -0,0 +1,90 @@
+// Package awsfirehose provides an otelcol.receiver.aws_firehose component.
+package awsfirehose
+
+import (
+	"github.com/grafana/agent/component"
+	"github.com/grafana/agent/component/otelcol"
+	"github.com/grafana/agent/component/otelcol/receiver"
+	"github.com/grafana/river/rivertypes"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver"
+	otelcomponent "go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configopaque"
+	otelextension "go.opentelemetry.io/collector/extension"
+)
+
+func init() {
+	component.Register(component.Registration{
+		Name: "otelcol.receiver.aws_firehose",
+		Args: Arguments{},
+
+		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
+			fact := awsfirehosereceiver.NewFactory()
+			return receiver.New(opts, fact, args.(Arguments))
+		},
+	})
+}
+
+// Arguments configures the otelcol.receiver.awsfirehose component.
+type Arguments struct {
+	// 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.
+	RecordType string `river:"record_type,attr,optional"`
+
+	// The access key to be checked on each request received.
+	AccessKey rivertypes.Secret `river:"access_key,attr,optional"`
+
+	HTTPServer otelcol.HTTPServerArguments `river:",squash"`
+
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
+	// Output configures where to send received data. Required.
+	Output *otelcol.ConsumerArguments `river:"output,block"`
+}
+
+var _ receiver.Arguments = Arguments{}
+
+// DefaultArguments holds default settings for otelcol.receiver.awsfirehose.
+var DefaultArguments = Arguments{
+	RecordType: "cwmetrics",
+	HTTPServer: otelcol.HTTPServerArguments{
+		Endpoint: "0.0.0.0:4433",
+	},
+
+	DebugMetrics: otelcol.DefaultDebugMetricsArguments,
+}
+
+// SetToDefault implements river.Defaulter.
+func (args *Arguments) SetToDefault() {
+	*args = DefaultArguments
+}
+
+// Convert implements receiver.Arguments.
+func (args Arguments) Convert() (otelcomponent.Config, error) {
+	return &awsfirehosereceiver.Config{
+		RecordType:         args.RecordType,
+		AccessKey:          configopaque.String(args.AccessKey),
+		HTTPServerSettings: *args.HTTPServer.Convert(),
+	}, nil
+}
+
+// Extensions implements receiver.Arguments.
+func (args Arguments) Extensions() map[otelcomponent.ID]otelextension.Extension {
+	return nil
+}
+
+// Exporters implements receiver.Arguments.
+func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.ID]otelcomponent.Component {
+	return nil
+}
+
+// NextConsumers implements receiver.Arguments.
+func (args Arguments) NextConsumers() *otelcol.ConsumerArguments {
+	return args.Output
+}
+
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}

component/otelcol/receiver/awsfirehose/awsfirehose_test.go

@@ -0,0 +1,140 @@
+package awsfirehose_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/grafana/agent/component/otelcol"
+	"github.com/grafana/agent/component/otelcol/receiver/awsfirehose"
+	"github.com/grafana/agent/pkg/flow/componenttest"
+	"github.com/grafana/agent/pkg/util"
+	"github.com/grafana/river"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver"
+	"github.com/phayes/freeport"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRun(t *testing.T) {
+	httpAddr := getFreeAddr(t)
+
+	ctx := componenttest.TestContext(t)
+	l := util.TestLogger(t)
+
+	ctrl, err := componenttest.NewControllerFromID(l, "otelcol.receiver.aws_firehose")
+	require.NoError(t, err)
+
+	cfg := fmt.Sprintf(`
+		endpoint = "%s"
+
+		output { /* no-op */ }
+	`, httpAddr)
+
+	var args awsfirehose.Arguments
+	require.NoError(t, river.Unmarshal([]byte(cfg), &args))
+
+	go func() {
+		err := ctrl.Run(ctx, args)
+		require.NoError(t, err)
+	}()
+
+	require.NoError(t, ctrl.WaitRunning(time.Second))
+}
+
+func TestArguments_UnmarshalRiver(t *testing.T) {
+	t.Run("grpc", func(t *testing.T) {
+		httpAddr := getFreeAddr(t)
+		in := fmt.Sprintf(`
+		endpoint = "%s"
+		cors {
+			allowed_origins = ["https://*.test.com", "https://test.com"]
+		}
+
+		record_type = "cwmetrics"
+
+		debug_metrics {
+			disable_high_cardinality_metrics = true
+		}
+
+		output { /* no-op */ }
+		`, httpAddr)
+
+		var args awsfirehose.Arguments
+		require.NoError(t, river.Unmarshal([]byte(in), &args))
+		require.Equal(t, args.DebugMetricsConfig().DisableHighCardinalityMetrics, true)
+		ext, err := args.Convert()
+		require.NoError(t, err)
+		otelArgs, ok := (ext).(*awsfirehosereceiver.Config)
+
+		require.True(t, ok)
+
+		// Check the arguments
+		require.Equal(t, otelArgs.HTTPServerSettings.Endpoint, httpAddr)
+		require.Equal(t, len(otelArgs.HTTPServerSettings.CORS.AllowedOrigins), 2)
+		require.Equal(t, otelArgs.HTTPServerSettings.CORS.AllowedOrigins[0], "https://*.test.com")
+		require.Equal(t, otelArgs.HTTPServerSettings.CORS.AllowedOrigins[1], "https://test.com")
+		require.Equal(t, otelArgs.RecordType, "cwmetrics")
+	})
+}
+
+func getFreeAddr(t *testing.T) string {
+	t.Helper()
+
+	portNumber, err := freeport.GetFreePort()
+	require.NoError(t, err)
+
+	return fmt.Sprintf("localhost:%d", portNumber)
+}
+
+func TestDebugMetricsConfig(t *testing.T) {
+	tests := []struct {
+		testName string
+		agentCfg string
+		expected otelcol.DebugMetricsArguments
+	}{
+		{
+			testName: "default",
+			agentCfg: `
+			output {}
+			`,
+			expected: otelcol.DebugMetricsArguments{
+				DisableHighCardinalityMetrics: true,
+			},
+		},
+		{
+			testName: "explicit_false",
+			agentCfg: `
+			debug_metrics {
+				disable_high_cardinality_metrics = false
+			}
+			output {}
+			`,
+			expected: otelcol.DebugMetricsArguments{
+				DisableHighCardinalityMetrics: false,
+			},
+		},
+		{
+			testName: "explicit_true",
+			agentCfg: `
+			debug_metrics {
+				disable_high_cardinality_metrics = true
+			}
+			output {}
+			`,
+			expected: otelcol.DebugMetricsArguments{
+				DisableHighCardinalityMetrics: true,
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.testName, func(t *testing.T) {
+			var args awsfirehose.Arguments
+			require.NoError(t, river.Unmarshal([]byte(tc.agentCfg), &args))
+			_, err := args.Convert()
+			require.NoError(t, err)
+
+			require.Equal(t, tc.expected, args.DebugMetricsConfig())
+		})
+	}
+}

docs/sources/flow/reference/compatibility/_index.md

@@ -335,6 +335,7 @@ The following components, grouped by namespace, _consume_ OpenTelemetry `otelcol
 - [otelcol.processor.span](../components/otelcol.processor.span)
 - [otelcol.processor.tail_sampling](../components/otelcol.processor.tail_sampling)
 - [otelcol.processor.transform](../components/otelcol.processor.transform)
+- [otelcol.receiver.aws_firehose]({{< relref "../components/otelcol.receiver.aws_firehose.md" >}})
 - [otelcol.receiver.jaeger](../components/otelcol.receiver.jaeger)
 - [otelcol.receiver.kafka](../components/otelcol.receiver.kafka)
 - [otelcol.receiver.loki](../components/otelcol.receiver.loki)