Add otel.receiver.aws_firehose component

CHANGELOG.md

@@ -27,6 +27,9 @@ Main (unreleased)
 - 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

@@ -86,6 +86,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,88 @@
+// 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",
+	},
+}
+
+// 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,86 @@
+package awsfirehose_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"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)
+}

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

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

docs/sources/flow/reference/components/otelcol.receiver.aws_firehose.md

@@ -0,0 +1,165 @@
+---
+aliases:
+- /docs/grafana-cloud/agent/flow/reference/components/otelcol.receiver.aws_firehose/
+- /docs/grafana-cloud/monitor-infrastructure/agent/flow/reference/components/otelcol.receiver.aws_firehose/
+- /docs/grafana-cloud/monitor-infrastructure/integrations/agent/flow/reference/components/otelcol.receiver.aws_firehose/
+- /docs/grafana-cloud/send-data/agent/flow/reference/components/otelcol.receiver.aws_firehose/
+canonical: https://grafana.com/docs/agent/latest/flow/reference/components/otelcol.receiver.aws_firehose/
+description: Learn about otelcol.receiver.aws_firehose
+label:
+  stage: experimental
+title: otelcol.receiver.aws_firehose
+---
+
+# otelcol.receiver.aws_firehose
+
+`otelcol.receiver.aws_firehose` receives metrics from [Cloudwatch Metrics Streams](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Metric-Streams.html) using AWS Kinesis Data Firehose and forwards it to other `otelcol.*` components.
+
+Set the output format of the Metrics Stream to JSON. Make sure the receiver is accessible by AWS on port 443. You can set the output format with a load balancer.
+
+{{% admonition type="note" %}}
+`otelcol.receiver.aws_firehose` is a wrapper over the upstream OpenTelemetry Collector `awsfirehose` receiver. If necessary, bug reports or feature requests are redirected to the upstream repository.
+{{% /admonition %}}
+
+You can specify multiple `otelcol.receiver.aws_firehose` components by giving them different labels.
+
+## Usage
+
+```river
+otelcol.receiver.aws_firehose "LABEL" {
+  endpoint = "HOST:PORT"
+  output {
+    metrics = [...]
+  }
+}
+```
+
+## Arguments
+
+`otelcol.receiver.aws_firehose` supports the following arguments:
+
+Name | Type | Description | Default | Required
+---- | ---- | ----------- | ------- | --------
+`record_type` | `string` | The type of record received from the delivery stream. | `cwmetrics` | no
+`access_key` | `secret` | The access key to be checked on each request received. | | no
+`endpoint` | `string` | `host:port` to listen for traffic on. | `"0.0.0.0:4433"` | no
+`max_request_body_size` | `string` | Maximum request body size the HTTP server will allow. No limit when unset. | | no
+`include_metadata` | `boolean` | Propagate incoming connection metadata to downstream consumers. | | no
+
+`access_key` can be set when creating or updating the delivery stream. See the [AWS Firehose documentation](https://docs.aws.amazon.com/firehose/latest/dev/create-destination.html#create-destination-http) for more details.
+
+The supported values for `record_type`  are:
+* `cwmetrics`: The record type for the CloudWatch metric stream. Expects the format for the records to be JSON. See the [CloudWatch documentation][cloudwatch-metric-streams] for details.
+[cloudwatch-metric-streams]: https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Metric-Streams.html
+
+## Blocks
+
+The following blocks are supported inside the definition of
+`otelcol.receiver.aws_firehose`:
+
+Hierarchy | Block | Description | Required
+--------- | ----- | ----------- | --------
+tls | [tls][] | Configures TLS for the HTTP server. | no
+cors | [cors][] | Configures CORS for the HTTP server. | no
+debug_metrics | [debug_metrics][] | Configures the metrics that this component generates to monitor its state. | no
+output | [output][] | Configures where to send received traces. | yes
+
+The `>` symbol indicates deeper levels of nesting. For example, `grpc > tls`
+refers to a `tls` block defined inside a `grpc` block.
+
+[tls]: #tls-block
+[cors]: #cors-block
+[debug_metrics]: #debug_metrics-block
+[output]: #output-block
+
+### tls block
+
+The `tls` block configures TLS settings used for a server. If the `tls` block
+isn't provided, TLS won't be used for connections to the server.
+
+{{< docs/shared lookup="flow/reference/components/otelcol-tls-config-block.md" source="agent" version="<AGENT_VERSION>" >}}
+
+### cors block
+
+The `cors` block configures CORS settings for an HTTP server.
+
+The following arguments are supported:
+
+Name | Type | Description | Default | Required
+---- | ---- | ----------- | ------- | --------
+`allowed_origins` | `list(string)` | Allowed values for the `Origin` header. | | no
+`allowed_headers` | `list(string)` | Accepted headers from CORS requests. | `["X-Requested-With"]` | no
+`max_age` | `number` | Configures the `Access-Control-Max-Age` response header. | | no
+
+The `allowed_headers` argument specifies which headers are acceptable from a
+CORS request. The following headers are always implicitly allowed:
+
+* `Accept`
+* `Accept-Language`
+* `Content-Type`
+* `Content-Language`
+
+If `allowed_headers` includes `"*"`, all headers are permitted.
+
+### debug_metrics block
+
+{{< docs/shared lookup="flow/reference/components/otelcol-debug-metrics-block.md" source="agent" version="<AGENT_VERSION>" >}}
+
+### output block
+
+{{< docs/shared lookup="flow/reference/components/output-block.md" source="agent" version="<AGENT_VERSION>" >}}
+
+## Exported fields
+
+`otelcol.receiver.aws_firehose` does not export any fields.
+
+## Component health
+
+`otelcol.receiver.aws_firehose` is only reported as unhealthy if given an invalid
+configuration.
+
+## Debug information
+
+`otelcol.receiver.aws_firehose` does not expose any component-specific debug
+information.
+
+## Example
+
+This example forwards received metrics through a batch processor before finally
+sending it to an OTLP-capable endpoint:
+
+```river
+otelcol.receiver.aws_firehose "default" {
+  endpoint = "0.0.0.0:4433"
+  output {
+    metrics = [otelcol.processor.batch.default.input]
+  }
+}
+
+otelcol.processor.batch "default" {
+  output {
+    metrics = [otelcol.exporter.otlp.default.input]
+  }
+}
+
+otelcol.exporter.otlp "default" {
+  client {
+    endpoint = env("OTLP_ENDPOINT")
+  }
+}
+```
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`otelcol.receiver.aws_firehose` can accept arguments from the following components:
+
+- Components that export [OpenTelemetry `otelcol.Consumer`]({{< relref "../compatibility/#opentelemetry-otelcolconsumer-exporters" >}})
+
+
+{{< admonition type="note" >}}
+Connecting some components may not be sensible or components may require further configuration to make the connection work correctly.
+Refer to the linked documentation for more details.
+{{< /admonition >}}
+
+<!-- END GENERATED COMPATIBLE COMPONENTS -->

go.mod

@@ -603,7 +603,7 @@ require (
 	sigs.k8s.io/structured-merge-diff/v4 v4.3.0 // indirect
 )
 
-require github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab
+require github.com/ianlancetaylor/demangle v0.0.0-20230524184225-eabc099b10ab // indirect
 
 require (
 	connectrpc.com/connect v1.14.0
@@ -660,6 +660,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka v0.87.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/metadataproviders v0.87.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheusremotewrite v0.87.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver v0.87.0
 	github.com/openshift/api v3.9.0+incompatible // indirect
 	github.com/openshift/client-go v0.0.0-20210521082421-73d9475a9142 // indirect
 	github.com/pjbgf/sha1cd v0.3.0 // indirect

go.sum

@@ -1795,6 +1795,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsampling
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.87.0/go.mod h1:iUrecf5kSV8pGF7OaM/brFOJs4OMEhogclBncGT5QtI=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.87.0 h1:1TK0+QULqgwwvE8JJxpQlugRdUw6knt0vYMkI65Jac0=
 github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor v0.87.0/go.mod h1:UYmAgkCKmtMyt40ffRherZJcU3zeCJjq4nZ7hVxApT4=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver v0.87.0 h1:U4vc1LAm8G5cncjL2snVjqE0ryYE7ZkzoXTS2hHEkwQ=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awsfirehosereceiver v0.87.0/go.mod h1:Wn6JfZybv9J46+9ZHwZkayDpR2vJQvKzlM+bMXxgGII=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.87.0 h1:8LN1Ky+Q6L6dmzm3k7Bec4fmlYs1OuJ7vaMuVnVIBLo=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/jaegerreceiver v0.87.0/go.mod h1:xb97OESLQFviQ0ikbUmneISHuRG91Uf+97EymDW4yus=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/kafkareceiver v0.87.0 h1:+rsOWSP9SDxbnwmRmQcsdZZJJeHvuKjPFN10jQXgsQI=