Introduce dynamic routing connector

connector/dynamicroutingconnector/Makefile

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

connector/dynamicroutingconnector/config.go

@@ -0,0 +1,62 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dynamicroutingconnector // import "github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector"
+
+import (
+	"cmp"
+	"errors"
+	"math"
+	"slices"
+	"time"
+
+	"go.opentelemetry.io/collector/pipeline"
+)
+
+type Config struct {
+	PrimaryMetadataKeys []string          `mapstructure:"primary_metadata_keys"`
+	DefaultPipelines    []pipeline.ID     `mapstructure:"default_pipelines"`
+	EvaluationInterval  time.Duration     `mapstructure:"evaluation_interval"`
+	DynamicPipelines    []DynamicPipeline `mapstructure:"dynamic_pipelines"`
+	MetadataKeys        []string          `mapstructure:"metadata_keys"`
+}
+
+type DynamicPipeline struct {
+	Pipelines []pipeline.ID `mapstructure:"pipelines"`
+	MaxCount  float64       `mapstructure:"max_count"`
+}
+
+func (c *Config) Validate() error {
+	if len(c.PrimaryMetadataKeys) == 0 {
+		return errors.New("atleast one primary_metadata_key must be defined")
+	}
+	if len(c.DefaultPipelines) == 0 {
+		return errors.New("default pipeline must be specified")
+	}
+	if len(c.DynamicPipelines) == 0 {
+		return errors.New("atleast one pipeline needs to be defined")
+	}
+	if !math.IsInf(c.DynamicPipelines[len(c.DynamicPipelines)-1].MaxCount, 1) {
+		return errors.New("last dynamic pipeline must have max count set to positive infinity (.inf)")
+	}
+	if !slices.IsSortedFunc(c.DynamicPipelines, func(a, b DynamicPipeline) int {
+		return cmp.Compare(a.MaxCount, b.MaxCount)
+	}) {
+		return errors.New("pipelines must be defined in ascending order of max_count")
+	}
+	return nil
+}

connector/dynamicroutingconnector/config_test.go

@@ -0,0 +1,109 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dynamicroutingconnector
+
+import (
+	"math"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector/internal/metadata"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"go.opentelemetry.io/collector/confmap/xconfmap"
+	"go.opentelemetry.io/collector/pipeline"
+)
+
+func TestConfig(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		expected component.Config
+		errMsg   string
+	}{
+		{
+			name:   "empty",
+			errMsg: "primary_metadata_key must be defined",
+		},
+		{
+			name:   "no-dynamic-pipelines",
+			errMsg: "atleast one pipeline needs to be defined",
+		},
+		{
+			name:   "invalid-dynamic-pipelines/no-inf",
+			errMsg: "last dynamic pipeline must have max count set to positive infinity",
+		},
+		{
+			name:   "invalid-dynamic-pipelines/out-of-order",
+			errMsg: "pipelines must be defined in ascending order of max_count",
+		},
+		{
+			name: "invalid-dynamic-pipelines/valid",
+			expected: &Config{
+				PrimaryMetadataKeys: []string{"x-tenant"},
+				DefaultPipelines: []pipeline.ID{
+					pipeline.NewIDWithName(pipeline.SignalLogs, "default"),
+				},
+				EvaluationInterval: time.Minute,
+				DynamicPipelines: []DynamicPipeline{
+					{
+						Pipelines: []pipeline.ID{
+							pipeline.NewIDWithName(pipeline.SignalLogs, "test1"),
+						},
+						MaxCount: 10,
+					},
+					{
+						Pipelines: []pipeline.ID{
+							pipeline.NewIDWithName(pipeline.SignalLogs, "test2"),
+						},
+						MaxCount: 100,
+					},
+					{
+						Pipelines: []pipeline.ID{
+							pipeline.NewIDWithName(pipeline.SignalLogs, "final"),
+						},
+						MaxCount: math.Inf(1),
+					},
+				},
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			configPath := filepath.Join("testdata", "configs", "config.yaml")
+			cm, err := confmaptest.LoadConf(configPath)
+			require.NoError(t, err)
+
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig()
+
+			sub, err := cm.Sub(metadata.Type.String() + "/" + tc.name)
+			require.NoError(t, err)
+			require.NoError(t, sub.Unmarshal(cfg))
+
+			err = xconfmap.Validate(cfg)
+			if tc.errMsg == "" {
+				assert.NoError(t, err)
+				assert.Equal(t, tc.expected, cfg)
+			} else {
+				assert.ErrorContains(t, err, tc.errMsg)
+			}
+		})
+	}
+}

connector/dynamicroutingconnector/doc.go

@@ -0,0 +1,21 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//go:generate mdatagen metadata.yaml
+
+// Package dynamicroutingconnector provides a connector for dynamically routing requests to different pipelines depending on the configuration.
+package dynamicroutingconnector // import "github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector"

connector/dynamicroutingconnector/factory.go

@@ -0,0 +1,70 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package dynamicroutingconnector // import "github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/consumer"
+
+	"github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector/internal/metadata"
+)
+
+// NewFactory returns a connector.Factory.
+func NewFactory() connector.Factory {
+	return connector.NewFactory(
+		metadata.Type,
+		createDefaultConfig,
+		connector.WithTracesToTraces(createTracesToTraces, metadata.TracesToTracesStability),
+		connector.WithLogsToLogs(createLogsToLogs, metadata.LogsToLogsStability),
+		connector.WithMetricsToMetrics(createMetricsToMetrics, metadata.MetricsToMetricsStability),
+	)
+}
+
+func createTracesToTraces(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	traces consumer.Traces,
+) (connector.Traces, error) {
+	return newTracesConnector(set, cfg, traces)
+}
+
+func createLogsToLogs(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	logs consumer.Logs,
+) (connector.Logs, error) {
+	return newLogsConnector(set, cfg, logs)
+}
+
+func createMetricsToMetrics(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	metrics consumer.Metrics,
+) (connector.Metrics, error) {
+	return newMetricsConnector(set, cfg, metrics)
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{}
+}