[PoC] Dynamic routing connector for routing based on estimated counts

Author: lahsivjar

connector/dynamicroutingconnector/Makefile

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

connector/dynamicroutingconnector/config.go

@@ -0,0 +1,56 @@
+// 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 {
+	// TODO(lahsivjar): Revisit the decision to route to default pipeline
+	// if NO metadata key results in empty str OR if primary key doesn't exist.
+	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/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/generated_component_test.go

@@ -0,0 +1,77 @@
+module github.com/elastic/opentelemetry-collector-components/connector/dynamicroutingconnector
+
+go 1.24.0
+
+toolchain go1.24.2
+
+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.135.0
+	github.com/stretchr/testify v1.11.1
+	go.opentelemetry.io/collector/client v1.41.0
+	go.opentelemetry.io/collector/component v1.41.0
+	go.opentelemetry.io/collector/component/componenttest v0.135.0
+	go.opentelemetry.io/collector/confmap v1.41.0
+	go.opentelemetry.io/collector/connector v0.135.0
+	go.opentelemetry.io/collector/connector/connectortest v0.135.0
+	go.opentelemetry.io/collector/consumer v1.41.0
+	go.opentelemetry.io/collector/consumer/consumertest v0.135.0
+	go.opentelemetry.io/collector/pdata v1.41.0
+	go.opentelemetry.io/collector/pipeline v1.41.0
+	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.0
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // 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/gogo/protobuf v1.3.2 // 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.2.2 // 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.135.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/connector/xconnector v0.135.0 // indirect
+	go.opentelemetry.io/collector/consumer/xconsumer v0.135.0 // indirect
+	go.opentelemetry.io/collector/featuregate v1.41.0 // indirect
+	go.opentelemetry.io/collector/internal/fanoutconsumer v0.135.0 // indirect
+	go.opentelemetry.io/collector/internal/telemetry v0.135.0 // indirect
+	go.opentelemetry.io/collector/pdata/pprofile v0.135.0 // indirect
+	go.opentelemetry.io/collector/pipeline/xpipeline v0.135.0 // indirect
+	go.opentelemetry.io/contrib/bridges/otelzap v0.12.0 // indirect
+	go.opentelemetry.io/otel v1.38.0 // indirect
+	go.opentelemetry.io/otel/log v0.14.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/net v0.43.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/text v0.28.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250707201910-8d1bb00bc6a7 // indirect
+	google.golang.org/grpc v1.75.0 // indirect
+	google.golang.org/protobuf v1.36.8 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
+replace github.com/elastic/opentelemetry-collector-components/internal/sharedcomponent => ../../internal/sharedcomponent
+
+replace github.com/elastic/opentelemetry-collector-components/processor/elastictraceprocessor => ../../processor/elastictraceprocessor
+
+replace github.com/elastic/opentelemetry-collector-components/processor/lsmintervalprocessor => ../../processor/lsmintervalprocessor