Add a roundrobin connector, that can help single thread components to…

… scale

.chloggen/roundrobinconnector.yaml

@@ -0,0 +1,22 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: roundrobinconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add a roundrobin connector, that can help single thread components to scale
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32853]
+
+# 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, api]

cmd/otelcontribcol/builder-config.yaml

@@ -214,6 +214,7 @@ connectors:
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/exceptionsconnector v0.99.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/failoverconnector v0.99.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector v0.99.0
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector v0.99.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector v0.99.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/servicegraphconnector v0.99.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector v0.99.0
@@ -434,6 +435,7 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/exceptionsconnector => ../../connector/exceptionsconnector
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/failoverconnector => ../../connector/failoverconnector
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector => ../../connector/grafanacloudconnector
+  - github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector => ../../connector/roundrobinconnector
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector => ../../connector/routingconnector
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/servicegraphconnector => ../../connector/servicegraphconnector
   - github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector => ../../connector/spanmetricsconnector

connector/roundrobinconnector/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

connector/roundrobinconnector/README.md

@@ -0,0 +1,71 @@
+# Round-Robin Connector
+<!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Distributions | [contrib] |
+| Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aconnector%2Froundrobin%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aconnector%2Froundrobin) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aconnector%2Froundrobin%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aconnector%2Froundrobin) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@bogdandrutu](https://www.github.com/bogdandrutu) |
+
+[beta]: https://github.com/open-telemetry/opentelemetry-collector#beta
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+
+## Supported Pipeline Types
+
+| [Exporter Pipeline Type] | [Receiver Pipeline Type] | [Stability Level] |
+| ------------------------ | ------------------------ | ----------------- |
+| traces | traces | [beta] |
+| metrics | metrics | [beta] |
+| logs | logs | [beta] |
+
+[Exporter Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#exporter-pipeline-type
+[Receiver Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#receiver-pipeline-type
+[Stability Level]: https://github.com/open-telemetry/opentelemetry-collector#stability-levels
+<!-- end autogenerated section -->
+
+The `roundrobin` connector can fork pipelines of the same type and equally split the load between them.
+
+## Configuration
+
+If you are not already familiar with connectors, you may find it helpful to first visit the [Connectors README].
+
+The `roundrobin` connector does not have any configuration settings.
+
+```yaml
+receivers:
+  otlp:
+exporters:
+  prometheusremotewrite/1:
+  prometheusremotewrite/2:
+connectors:
+  roundrobin:
+```
+
+Preprocess data, then export using multiple exporter instances to scale the throughput if the exporter 
+does not support scale well (e.g. prometheusremotewrite).
+
+```yaml
+receivers:
+  otlp:
+processors:
+  resourcedetection:
+  batch:
+exporters:
+  prometheusremotewrite/1:
+  prometheusremotewrite/2:
+connectors:
+  roundrobin:
+service:
+  pipelines:
+    metrics:
+      receivers: [otlp]
+      processors: [resourcedetection, batch]
+      exporters: [roundrobin]
+    metrics/1:
+      receivers: [roundrobin]
+      exporters: [prometheusremotewrite/1]
+    metrics/2:
+      receivers: [roundrobin]
+      exporters: [prometheusremotewrite/2]
+```
+
+[Connectors README]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md

connector/roundrobinconnector/config.go

@@ -0,0 +1,7 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package roundrobinconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector"
+
+// Config for the connector
+type Config struct{}

connector/roundrobinconnector/connector.go

@@ -0,0 +1,86 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package roundrobinconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector"
+
+import (
+	"context"
+	"sync/atomic"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func allConsumers[T any](r router[T]) ([]T, error) {
+	pipeIDs := r.PipelineIDs()
+	consumers := make([]T, len(pipeIDs))
+	for i, pipeID := range pipeIDs {
+		cons, err := r.Consumer(pipeID)
+		if err != nil {
+			return nil, err
+		}
+		consumers[i] = cons
+	}
+	return consumers, nil
+}
+
+type router[T any] interface {
+	PipelineIDs() []component.ID
+	Consumer(pipelineIDs ...component.ID) (T, error)
+}
+
+func newLogs(nextConsumer consumer.Logs) (connector.Logs, error) {
+	nextConsumers, err := allConsumers[consumer.Logs](nextConsumer.(connector.LogsRouterAndConsumer))
+	if err != nil {
+		return nil, err
+	}
+	return &roundRobin{nextLogs: nextConsumers}, nil
+}
+
+func newMetrics(nextConsumer consumer.Metrics) (connector.Metrics, error) {
+	nextConsumers, err := allConsumers[consumer.Metrics](nextConsumer.(connector.MetricsRouterAndConsumer))
+	if err != nil {
+		return nil, err
+	}
+	return &roundRobin{nextMetrics: nextConsumers}, nil
+}
+
+func newTraces(nextConsumer consumer.Traces) (connector.Traces, error) {
+	nextConsumers, err := allConsumers[consumer.Traces](nextConsumer.(connector.TracesRouterAndConsumer))
+	if err != nil {
+		return nil, err
+	}
+	return &roundRobin{nextTraces: nextConsumers}, nil
+}
+
+// roundRobin is used to pass signals directly from one pipeline to one of the configured once in a round-robin mode.
+// This is useful when there is a need to scale (shard) data processing and downstream components do not
+// handle concurrent requests very well.
+type roundRobin struct {
+	component.StartFunc
+	component.ShutdownFunc
+	nextConsumer atomic.Uint64
+	nextMetrics  []consumer.Metrics
+	nextLogs     []consumer.Logs
+	nextTraces   []consumer.Traces
+}
+
+func (rr *roundRobin) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (rr *roundRobin) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	return rr.nextLogs[rr.nextConsumer.Add(1)%uint64(len(rr.nextLogs))].ConsumeLogs(ctx, ld)
+}
+
+func (rr *roundRobin) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
+	return rr.nextMetrics[rr.nextConsumer.Add(1)%uint64(len(rr.nextMetrics))].ConsumeMetrics(ctx, md)
+}
+
+func (rr *roundRobin) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
+	return rr.nextTraces[rr.nextConsumer.Add(1)%uint64(len(rr.nextTraces))].ConsumeTraces(ctx, td)
+}

connector/roundrobinconnector/connector_test.go

@@ -0,0 +1,140 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package roundrobinconnector
+
+import (
+	"context"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/connector/connectortest"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+func newPipelineMap[T any](tp component.Type, consumers ...T) map[component.ID]T {
+	ret := make(map[component.ID]T, len(consumers))
+	for i, cons := range consumers {
+		ret[component.NewIDWithName(tp, strconv.Itoa(i))] = cons
+	}
+	return ret
+}
+
+func TestLogsRoundRobin(t *testing.T) {
+	f := NewFactory()
+	cfg := f.CreateDefaultConfig()
+	assert.Equal(t, &Config{}, cfg)
+
+	ctx := context.Background()
+	set := connectortest.NewNopCreateSettings()
+	host := componenttest.NewNopHost()
+
+	sink1 := new(consumertest.LogsSink)
+	sink2 := new(consumertest.LogsSink)
+	sink3 := new(consumertest.LogsSink)
+	logs, err := f.CreateLogsToLogs(ctx, set, cfg, connector.NewLogsRouter(newPipelineMap[consumer.Logs](component.DataTypeLogs, sink1, sink2, sink3)))
+	assert.NoError(t, err)
+	assert.NotNil(t, logs)
+
+	assert.NoError(t, logs.Start(ctx, host))
+
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+
+	assert.Equal(t, 1, len(sink1.AllLogs()))
+	assert.Equal(t, 1, len(sink2.AllLogs()))
+	assert.Equal(t, 1, len(sink3.AllLogs()))
+
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+	assert.NoError(t, logs.ConsumeLogs(ctx, plog.NewLogs()))
+
+	assert.Equal(t, 2, len(sink1.AllLogs()))
+	assert.Equal(t, 2, len(sink2.AllLogs()))
+	assert.Equal(t, 2, len(sink3.AllLogs()))
+
+	assert.NoError(t, logs.Shutdown(ctx))
+}
+
+func TestMetricsRoundRobin(t *testing.T) {
+	f := NewFactory()
+	cfg := f.CreateDefaultConfig()
+	assert.Equal(t, &Config{}, cfg)
+
+	ctx := context.Background()
+	set := connectortest.NewNopCreateSettings()
+	host := componenttest.NewNopHost()
+
+	sink1 := new(consumertest.MetricsSink)
+	sink2 := new(consumertest.MetricsSink)
+	sink3 := new(consumertest.MetricsSink)
+	metrics, err := f.CreateMetricsToMetrics(ctx, set, cfg, connector.NewMetricsRouter(newPipelineMap[consumer.Metrics](component.DataTypeMetrics, sink1, sink2, sink3)))
+	assert.NoError(t, err)
+	assert.NotNil(t, metrics)
+
+	assert.NoError(t, metrics.Start(ctx, host))
+
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+
+	assert.Equal(t, 1, len(sink1.AllMetrics()))
+	assert.Equal(t, 1, len(sink2.AllMetrics()))
+	assert.Equal(t, 1, len(sink3.AllMetrics()))
+
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+	assert.NoError(t, metrics.ConsumeMetrics(ctx, pmetric.NewMetrics()))
+
+	assert.Equal(t, 2, len(sink1.AllMetrics()))
+	assert.Equal(t, 2, len(sink2.AllMetrics()))
+	assert.Equal(t, 2, len(sink3.AllMetrics()))
+
+	assert.NoError(t, metrics.Shutdown(ctx))
+}
+
+func TestTracesRoundRobin(t *testing.T) {
+	f := NewFactory()
+	cfg := f.CreateDefaultConfig()
+	assert.Equal(t, &Config{}, cfg)
+
+	ctx := context.Background()
+	set := connectortest.NewNopCreateSettings()
+	host := componenttest.NewNopHost()
+
+	sink1 := new(consumertest.TracesSink)
+	sink2 := new(consumertest.TracesSink)
+	sink3 := new(consumertest.TracesSink)
+	traces, err := f.CreateTracesToTraces(ctx, set, cfg, connector.NewTracesRouter(newPipelineMap[consumer.Traces](component.DataTypeTraces, sink1, sink2, sink3)))
+	assert.NoError(t, err)
+	assert.NotNil(t, traces)
+
+	assert.NoError(t, traces.Start(ctx, host))
+
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+
+	assert.Equal(t, 1, len(sink1.AllTraces()))
+	assert.Equal(t, 1, len(sink2.AllTraces()))
+	assert.Equal(t, 1, len(sink3.AllTraces()))
+
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+	assert.NoError(t, traces.ConsumeTraces(ctx, ptrace.NewTraces()))
+
+	assert.Equal(t, 2, len(sink1.AllTraces()))
+	assert.Equal(t, 2, len(sink2.AllTraces()))
+	assert.Equal(t, 2, len(sink3.AllTraces()))
+
+	assert.NoError(t, traces.Shutdown(ctx))
+}

connector/roundrobinconnector/doc.go

@@ -0,0 +1,4 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package roundrobinconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/roundrobinconnector"