Add connectors to otelcol config (#6964)

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
 	}

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

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{