Add the fluentforward receiver

CHANGELOG.md

@@ -15,6 +15,7 @@ Main (unreleased)
 - (_Experimental_) Add an `array.group_by` stdlib function to group items in an array by a key. (@wildum)
 - Add the `otelcol.exporter.faro` exporter to export traces and logs to Faro endpoint. (@mar4uk)
 - Add the `otelcol.receiver.faro` receiver to receive traces and logs from the Grafana Faro Web SDK. (@mar4uk)
+- Add the `otelcol.receiver.fluentforward` receiver to receive logs via Fluent Forward Protocol. (@rucciva)
 
 - Add entropy support for `loki.secretfilter` (@romain-gaillard)
 

docs/sources/reference/compatibility/_index.md

@@ -375,6 +375,7 @@ The following components, grouped by namespace, _consume_ OpenTelemetry `otelcol
 - [otelcol.receiver.faro](../components/otelcol/otelcol.receiver.faro)
 - [otelcol.receiver.file_stats](../components/otelcol/otelcol.receiver.file_stats)
 - [otelcol.receiver.filelog](../components/otelcol/otelcol.receiver.filelog)
+- [otelcol.receiver.fluentforward](../components/otelcol/otelcol.receiver.fluentforward)
 - [otelcol.receiver.influxdb](../components/otelcol/otelcol.receiver.influxdb)
 - [otelcol.receiver.jaeger](../components/otelcol/otelcol.receiver.jaeger)
 - [otelcol.receiver.kafka](../components/otelcol/otelcol.receiver.kafka)

docs/sources/reference/components/otelcol/otelcol.receiver.fluentforward.md

@@ -0,0 +1,113 @@
+---
+canonical: https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.receiver.fluentforward/
+description: Learn about otelcol.receiver.fluentforward
+labels:
+  stage: experimental
+  products:
+    - oss
+title: otelcol.receiver.fluentforward
+---
+
+# `otelcol.receiver.fluentforward`
+
+{{< docs/shared lookup="stability/experimental.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+`otelcol.receiver.fluentforward` accepts log messages over a TCP connection via the [Fluent Forward Protocol](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1) and forwards them as logs to other `otelcol.*` components.
+
+{{< admonition type="note" >}}
+`otelcol.receiver.fluentforward` is a wrapper over the upstream OpenTelemetry Collector [`fluentforward`][] receiver.
+Bug reports or feature requests will be redirected to the upstream repository, if necessary.
+
+[`fluentforward`]: <https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/{{>< param "OTEL_VERSION" >}}/receiver/fluentforwardreceiver
+{{< /admonition >}}
+
+You can specify multiple `otelcol.receiver.fluentforward` components by giving them different labels.
+
+## Usage
+
+```alloy
+otelcol.receiver.fluentforward "<LABEL>" {
+  endpoint = "<IP_ADDRESS:PORT>"
+
+  output {
+    logs    = [...]
+  }
+}
+```
+
+## Arguments
+
+You can use the following arguments with `otelcol.receiver.fluentforward`:
+
+| Name                 | Type     | Description                                                                                                 | Default   | Required |
+| -------------------- | -------- | ----------------------------------------------------------------------------------------------------------- | --------- | -------- |
+| `endpoint`     | `string` | The `<HOST:PORT>` or `unix://<path to socket>` address to listen to for logs messages.                                                   |           | yes      |
+
+## Blocks
+
+You can use the following blocks with `otelcol.receiver.fluentforward`:
+
+| Block                                  | Description                                                                                     | Required |
+| -------------------------------------- | ----------------------------------------------------------------------------------------------- | -------- |
+| [`output`][output]                     | Configures where to send received telemetry data.                                               | yes      |
+| [`debug_metrics`][debug_metrics]       | Configures the metrics that this component generates to monitor its state.                      | no       |
+
+[debug_metrics]: #debug_metrics
+[output]: #output
+
+### `output`
+
+<span class="badge docs-labels__stage docs-labels__item">Required</span>
+
+{{< docs/shared lookup="reference/components/output-block-logs.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+### `debug_metrics`
+
+{{< docs/shared lookup="reference/components/otelcol-debug-metrics-block.md" source="alloy" version="<ALLOY_VERSION>" >}}
+
+## Exported fields
+
+`otelcol.receiver.fluentforward` doesn't export any fields.
+
+## Component health
+
+`otelcol.receiver.fluentforward` is only reported as unhealthy if given an invalid configuration.
+
+## Debug information
+
+`otelcol.receiver.fluentforward` doesn't expose any component-specific debug information.
+
+## Debug metrics
+
+`otelcol.receiver.fluentforward` doesn't expose any component-specific debug metrics.
+
+## Example
+
+This example receives log messages using Fluent Forward Protocol on TCP port 8006 and logs them.
+
+```alloy
+otelcol.receiver.fluentforward "default" {
+    endpoint = "localhost:8006"
+    output {
+        logs = [otelcol.exporter.debug.default.input]
+    }
+}
+
+otelcol.exporter.debug "default" {}
+```
+
+<!-- START GENERATED COMPATIBLE COMPONENTS -->
+
+## Compatible components
+
+`otelcol.receiver.fluentforward` can accept arguments from the following components:
+
+- Components that export [OpenTelemetry `otelcol.Consumer`](../../../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

@@ -1010,6 +1010,7 @@ require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/extension/ackextension v0.128.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk v0.128.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/faro v0.128.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver v0.128.0 // indirect
 	github.com/tonistiigi/dchapes-mode v0.0.0-20250318174251-73d941a28323 // indirect
 	github.com/tonistiigi/go-csvvalue v0.0.0-20240710180619-ddb21b71c0b4 // indirect
 	github.com/twmb/franz-go v1.18.1 // indirect

go.sum

@@ -2330,6 +2330,8 @@ github.com/open-telemetry/opentelemetry-collector-contrib/receiver/filelogreceiv
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/filelogreceiver v0.128.0/go.mod h1:cVKK4o+7i0w1yD36ojYC+KwSj4WFBHG0TazYQu5lC/A=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/filestatsreceiver v0.128.0 h1:cHUha259kwbefs3KebyKHVedGqMP75wwPBPNUwFOtKY=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/filestatsreceiver v0.128.0/go.mod h1:i3jLyma0OruFlVTJIlCh5QT291qZW5ZyZgcUEMrm0Fc=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver v0.128.0 h1:EkSbn2vBOD5u389jTiQpssrdSUT7rQdltzYBYFBGR0g=
+github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver v0.128.0/go.mod h1:SwARiAxarYR+aElfGriPQ5WqHf5YfIkGiE0KVJonoSo=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver v0.128.0 h1:LUrfwhcqP0k/lwtT8nxKX7L11RxU7xvw/NwXN/r91/A=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/hostmetricsreceiver v0.128.0/go.mod h1:1jWL8z4+7JhoamkAzfSAzVLp+joDOgcf0HuAkMbK0AE=
 github.com/open-telemetry/opentelemetry-collector-contrib/receiver/influxdbreceiver v0.128.0 h1:robqWLRsdalVuZhjW6oBqiHn0am5VRqQPJ5FfPZaHUk=

internal/component/all/all.go

@@ -105,6 +105,7 @@ import (
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/faro"                    // Import otelcol.receiver.faro
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/file_stats"              // Import otelcol.receiver.file_stats
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/filelog"                 // Import otelcol.receiver.filelog
+	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/fluentforward"           // Import otelcol.receiver.fluentforward
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/influxdb"                // Import otelcol.receiver.influxdb
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/jaeger"                  // Import otelcol.receiver.jaeger
 	_ "github.com/grafana/alloy/internal/component/otelcol/receiver/kafka"                   // Import otelcol.receiver.kafka

internal/component/otelcol/receiver/fluentforward/fluentforward.go

@@ -0,0 +1,86 @@
+package fluentforward
+
+import (
+	"fmt"
+
+	"github.com/grafana/alloy/internal/component"
+	"github.com/grafana/alloy/internal/component/otelcol"
+	otelcolConfig "github.com/grafana/alloy/internal/component/otelcol/config"
+	"github.com/grafana/alloy/internal/component/otelcol/receiver"
+	"github.com/grafana/alloy/internal/featuregate"
+	"github.com/grafana/alloy/syntax"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver"
+	collectorComponent "go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pipeline"
+)
+
+func init() {
+	component.Register(component.Registration{
+		Name:      "otelcol.receiver.fluentforward",
+		Stability: featuregate.StabilityExperimental,
+		Args:      Arguments{},
+		Build: func(opts component.Options, args component.Arguments) (component.Component, error) {
+			factory := fluentforwardreceiver.NewFactory()
+			return receiver.New(opts, factory, args.(Arguments))
+		},
+	})
+}
+
+var (
+	_ receiver.Arguments = (*Arguments)(nil)
+	_ syntax.Defaulter   = (*Arguments)(nil)
+	_ syntax.Validator   = (*Arguments)(nil)
+)
+
+type Arguments struct {
+	// The address to listen on for incoming Fluent Forward events.  Should be
+	// of the form `<ip addr>:<port>` (TCP) or `unix://<socket_path>` (Unix
+	// domain socket).
+	Endpoint string `alloy:"endpoint,attr"`
+
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcolConfig.DebugMetricsArguments `alloy:"debug_metrics,block,optional"`
+
+	// Output configures where to send received data. Required.
+	Output *ConsumerArguments `alloy:"output,block"`
+}
+
+func (a *Arguments) Validate() error {
+	if a.Endpoint == "" {
+		return fmt.Errorf("endpoint must not be empty")
+	}
+	return nil
+}
+
+func (a *Arguments) SetToDefault() {
+	a.DebugMetrics.SetToDefault()
+}
+
+type ConsumerArguments struct {
+	Logs []otelcol.Consumer `alloy:"logs,attr,optional"`
+}
+
+func (a Arguments) Convert() (collectorComponent.Config, error) {
+	cfg := &fluentforwardreceiver.Config{
+		ListenAddress: a.Endpoint,
+	}
+	return cfg, nil
+}
+
+func (a Arguments) DebugMetricsConfig() otelcolConfig.DebugMetricsArguments {
+	return a.DebugMetrics
+}
+
+func (a Arguments) Exporters() map[pipeline.Signal]map[collectorComponent.ID]collectorComponent.Component {
+	return nil
+}
+
+func (a Arguments) Extensions() map[collectorComponent.ID]collectorComponent.Component {
+	return nil
+}
+
+func (a Arguments) NextConsumers() *otelcol.ConsumerArguments {
+	return &otelcol.ConsumerArguments{
+		Logs: a.Output.Logs,
+	}
+}

internal/component/otelcol/receiver/fluentforward/fluentforward_test.go

@@ -0,0 +1,43 @@
+package fluentforward_test
+
+import (
+	"testing"
+
+	"github.com/grafana/alloy/internal/component/otelcol/receiver/fluentforward"
+	"github.com/grafana/alloy/syntax"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfig(t *testing.T) {
+	alloyCfg := `
+		endpoint = "localhost:1514"
+		output {
+		}
+	`
+	var args fluentforward.Arguments
+	err := syntax.Unmarshal([]byte(alloyCfg), &args)
+	require.NoError(t, err)
+	require.NoError(t, args.Validate())
+
+	comCfg, err := args.Convert()
+	require.NoError(t, err)
+
+	fluentComCfg, ok := comCfg.(*fluentforwardreceiver.Config)
+	require.True(t, ok)
+
+	assert.Equal(t, "localhost:1514", fluentComCfg.ListenAddress)
+}
+
+func TestConfigDefault(t *testing.T) {
+	args := fluentforward.Arguments{}
+	args.SetToDefault()
+
+	fCfg, err := args.Convert()
+	require.NoError(t, err)
+	cfg := fluentforwardreceiver.NewFactory().CreateDefaultConfig()
+	assert.Equal(t, cfg, fCfg)
+
+	assert.Error(t, args.Validate())
+}

internal/converter/internal/otelcolconvert/converter_fluentforwardreceiver.go

@@ -0,0 +1,52 @@
+package otelcolconvert
+
+import (
+	"fmt"
+
+	"github.com/grafana/alloy/internal/component/otelcol/receiver/fluentforward"
+	"github.com/grafana/alloy/internal/converter/diag"
+	"github.com/grafana/alloy/internal/converter/internal/common"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/fluentforwardreceiver"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componentstatus"
+)
+
+func init() {
+	converters = append(converters, fluentforwardReceiverConverter{})
+}
+
+type fluentforwardReceiverConverter struct{}
+
+func (fluentforwardReceiverConverter) Factory() component.Factory {
+	return fluentforwardreceiver.NewFactory()
+}
+
+func (fluentforwardReceiverConverter) InputComponentName() string {
+	return "otelcol.receiver.fluentforward"
+}
+
+func (fluentforwardReceiverConverter) ConvertAndAppend(state *State, id componentstatus.InstanceID, cfg component.Config) diag.Diagnostics {
+	var diags diag.Diagnostics
+
+	label := state.AlloyComponentLabel()
+
+	args := toOtelcolReceiverfluentforward(cfg.(*fluentforwardreceiver.Config))
+	block := common.NewBlockWithOverride([]string{"otelcol", "receiver", "fluentforward"}, label, args)
+
+	diags.Add(
+		diag.SeverityLevelInfo,
+		fmt.Sprintf("Converted %s into %s", StringifyInstanceID(id), StringifyBlock(block)),
+	)
+
+	state.Body().AppendBlock(block)
+	return diags
+}
+
+func toOtelcolReceiverfluentforward(cfg *fluentforwardreceiver.Config) *fluentforward.Arguments {
+	args := &fluentforward.Arguments{
+		Endpoint:     cfg.ListenAddress,
+		DebugMetrics: common.DefaultValue[fluentforward.Arguments]().DebugMetrics,
+	}
+
+	return args
+}

internal/converter/internal/otelcolconvert/testdata/fluentforwarder.alloy

@@ -0,0 +1,9 @@
+otelcol.receiver.fluentforward "default" {
+	endpoint = "localhost:1514"
+}
+
+otelcol.exporter.otlp "default" {
+	client {
+		endpoint = "database:4317"
+	}
+}

internal/converter/internal/otelcolconvert/testdata/fluentforwarder.yaml

@@ -0,0 +1,14 @@
+receivers:
+  fluentforward:
+    endpoint: "localhost:1514"
+
+exporters:
+  otlp:
+    endpoint: database:4317
+
+service:
+  pipelines:
+    logs:
+      receivers: [fluentforward]
+      processors: []
+      exporters: [otlp]