From e12df8f0a17462c310f8260cb370211e76ad507d Mon Sep 17 00:00:00 2001
From: Dan Jaglowski <jaglows3@gmail.com>
Date: Wed, 18 Jan 2023 09:32:43 -0500
Subject: [PATCH] Add connectors to otelcol config

---
 otelcol/config.go         | 22 ++++++++++++++++++++++
 otelcol/configprovider.go |  1 +
 otelcol/unmarshaler.go    |  3 +--
 3 files changed, 24 insertions(+), 2 deletions(-)

diff --git a/otelcol/config.go b/otelcol/config.go
index ed196e2381a..1d78ee5c23f 100644
--- a/otelcol/config.go
+++ b/otelcol/config.go
@@ -19,6 +19,7 @@ import (
 	"fmt"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/featuregate"
 	"go.opentelemetry.io/collector/service"
 )
 
@@ -27,6 +28,20 @@ var (
 	errMissingReceivers = errors.New("no receiver configuration specified in config")
 )
 
+const (
+	connectorsFeatureGateID = "otelcol.enableConnectors"
+	connectorsFeatureStage  = featuregate.StageAlpha
+)
+
+func init() {
+	featuregate.GlobalRegistry().MustRegisterID(
+		connectorsFeatureGateID,
+		connectorsFeatureStage,
+		featuregate.WithRegisterDescription("Enables 'connectors', a new type of component for transmitting signals between pipelines."),
+		featuregate.WithRegisterReferenceURL("https://github.com/open-telemetry/opentelemetry-collector/issues/2336"),
+	)
+}
+
 // Config defines the configuration for the various elements of collector or agent.
 type Config struct {
 	// Receivers is a map of ComponentID to Receivers.
@@ -38,6 +53,9 @@ type Config struct {
 	// Processors is a map of ComponentID to Processors.
 	Processors map[component.ID]component.Config
 
+	// Connectors is a map of ComponentID to connectors.
+	Connectors map[component.ID]component.Config
+
 	// Extensions is a map of ComponentID to extensions.
 	Extensions map[component.ID]component.Config
 
@@ -90,6 +108,10 @@ func (cfg *Config) Validate() error {
 		}
 	}
 
+	if len(cfg.Connectors) != 0 && !featuregate.GlobalRegistry().IsEnabled(connectorsFeatureGateID) {
+		return fmt.Errorf("connectors require feature gate: %s", connectorsFeatureGateID)
+	}
+
 	if err := cfg.Service.Validate(); err != nil {
 		return err
 	}
diff --git a/otelcol/configprovider.go b/otelcol/configprovider.go
index 411b11d6444..16a10013d18 100644
--- a/otelcol/configprovider.go
+++ b/otelcol/configprovider.go
@@ -102,6 +102,7 @@ func (cm *configProvider) Get(ctx context.Context, factories Factories) (*Config
 		Receivers:  cfg.Receivers.Configs(),
 		Processors: cfg.Processors.Configs(),
 		Exporters:  cfg.Exporters.Configs(),
+		Connectors: cfg.Connectors.Configs(),
 		Extensions: cfg.Extensions.Configs(),
 		Service:    cfg.Service,
 	}, nil
diff --git a/otelcol/unmarshaler.go b/otelcol/unmarshaler.go
index ec0071a6d9f..04256de9bc4 100644
--- a/otelcol/unmarshaler.go
+++ b/otelcol/unmarshaler.go
@@ -17,7 +17,6 @@ package otelcol // import "go.opentelemetry.io/collector/otelcol"
 import (
 	"go.uber.org/zap/zapcore"
 
-	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/confmap"
 	"go.opentelemetry.io/collector/connector"
@@ -47,7 +46,7 @@ func unmarshal(v *confmap.Conf, factories Factories) (*configSettings, error) {
 		Receivers:  configunmarshaler.NewConfigs(factories.Receivers),
 		Processors: configunmarshaler.NewConfigs(factories.Processors),
 		Exporters:  configunmarshaler.NewConfigs(factories.Exporters),
-		Connectors: configunmarshaler.NewConfigs(map[component.Type]connector.Factory{}),
+		Connectors: configunmarshaler.NewConfigs(factories.Connectors),
 		Extensions: configunmarshaler.NewConfigs(factories.Extensions),
 		// TODO: Add a component.ServiceFactory to allow this to be defined by the Service.
 		Service: service.Config{