Introduce dynamic routing connector

connector/dynamicroutingconnector/Makefile

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

connector/dynamicroutingconnector/config.go

@@ -0,0 +1,54 @@
+// 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 (
+	"errors"
+	"sort"
+	"time"
+
+	"go.opentelemetry.io/collector/pipeline"
+)
+
+type Config struct {
+	DefaultPipelines   []pipeline.ID   `mapstructure:"default_pipelines"`
+	EvaluationInterval time.Duration   `mapstructure:"evalaution_interval"`
+	Pipelines          [][]pipeline.ID `mapstructure:"pipelines"`
+	Thresholds         []int           `mapstructure:"thresholds"`
+	PrimaryMetadataKey string          `mapstructure:"primary_metadata_key"`
+	MetadataKeys       []string        `mapstructure:"metadata_keys"`
+}
+
+func (c *Config) Validate() error {
+	if len(c.Pipelines) == 0 {
+		return errors.New("atleast one pipeline needs to be defined")
+	}
+	if len(c.Pipelines)+1 != len(c.Thresholds) {
+		return errors.New("pipelines need to be defined for each threshold bucket, including +inf")
+	}
+	if !sort.IntsAreSorted(c.Thresholds) {
+		return errors.New("thresolds is expected to be in increasing order")
+	}
+
+	for i := 1; i < len(c.Thresholds); i++ {
+		if c.Thresholds[i] == c.Thresholds[i-1] {
+			return errors.New("thresholds are expected to be unique")
+		}
+	}
+	return nil
+}

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{}
+}

connector/dynamicroutingconnector/go.mod

@@ -0,0 +1,60 @@
+module github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector
+
+go 1.24.0
+
+require (
+	github.com/axiomhq/hyperloglog v0.2.5
+	github.com/cespare/xxhash/v2 v2.3.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.141.0
+	github.com/stretchr/testify v1.11.1
+	go.opentelemetry.io/collector/client v1.47.0
+	go.opentelemetry.io/collector/component v1.47.0
+	go.opentelemetry.io/collector/component/componenttest v0.141.0
+	go.opentelemetry.io/collector/confmap v1.47.0
+	go.opentelemetry.io/collector/connector v0.141.0
+	go.opentelemetry.io/collector/connector/connectortest v0.141.0
+	go.opentelemetry.io/collector/consumer v1.47.0
+	go.opentelemetry.io/collector/consumer/consumertest v0.141.0
+	go.opentelemetry.io/collector/pdata v1.47.0
+	go.opentelemetry.io/collector/pipeline v1.47.0
+	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
+	github.com/gobwas/glob v0.2.3 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/hashicorp/go-version v1.7.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/kamstrup/intmap v0.5.1 // indirect
+	github.com/knadh/koanf/maps v0.1.2 // indirect
+	github.com/knadh/koanf/providers/confmap v1.0.0 // indirect
+	github.com/knadh/koanf/v2 v2.3.0 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.141.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/connector/xconnector v0.141.0 // indirect
+	go.opentelemetry.io/collector/consumer/xconsumer v0.141.0 // indirect
+	go.opentelemetry.io/collector/featuregate v1.47.0 // indirect
+	go.opentelemetry.io/collector/internal/fanoutconsumer v0.141.0 // indirect
+	go.opentelemetry.io/collector/pdata/pprofile v0.141.0 // indirect
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.141.0 // indirect
+	go.opentelemetry.io/otel v1.38.0 // indirect
+	go.opentelemetry.io/otel/metric v1.38.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
+	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
+	go.opentelemetry.io/otel/trace v1.38.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
+	golang.org/x/sys v0.37.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)