Add support for connector configurations (open-telemetry#6789)

* Add connectors to configuration of collector

* changelog

.chloggen/connector-config.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: cmd/builder
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for connector configurations
+
+# One or more tracking issues or pull requests related to the change
+issues: [6789]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/builder/internal/builder/config.go

@@ -42,6 +42,7 @@ type Config struct {
 	Extensions   []Module     `mapstructure:"extensions"`
 	Receivers    []Module     `mapstructure:"receivers"`
 	Processors   []Module     `mapstructure:"processors"`
+	Connectors   []Module     `mapstructure:"connectors"`
 	Replaces     []string     `mapstructure:"replaces"`
 	Excludes     []string     `mapstructure:"excludes"`
 }
@@ -91,7 +92,13 @@ func NewDefaultConfig() Config {
 
 // Validate checks whether the current configuration is valid
 func (c *Config) Validate() error {
-	return multierr.Combine(validateModules(c.Extensions), validateModules(c.Receivers), validateModules(c.Exporters), validateModules(c.Processors))
+	return multierr.Combine(
+		validateModules(c.Extensions),
+		validateModules(c.Receivers),
+		validateModules(c.Exporters),
+		validateModules(c.Processors),
+		validateModules(c.Connectors),
+	)
 }
 
 // SetGoPath sets go path
@@ -134,6 +141,11 @@ func (c *Config) ParseModules() error {
 		return err
 	}
 
+	c.Connectors, err = parseModules(c.Connectors)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 

cmd/builder/internal/builder/templates/components.go.tmpl

@@ -3,11 +3,15 @@
 package main
 
 import (
+	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/otelcol"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/receiver"
+	{{- range .Connectors}}
+	{{.Name}} "{{.Import}}"
+	{{- end}}
 	{{- range .Exporters}}
 	{{.Name}} "{{.Import}}"
 	{{- end}}
@@ -62,5 +66,14 @@ func components() (otelcol.Factories, error) {
 		return otelcol.Factories{}, err
 	}
 
+	factories.Connectors, err = connector.MakeFactoryMap(
+		{{- range .Connectors}}
+		{{.Name}}.NewFactory(),
+		{{- end}}
+	)
+	if err != nil {
+		return otelcol.Factories{}, err
+	}
+
 	return factories, nil
 }

cmd/builder/internal/builder/templates/components_test.go.tmpl

@@ -26,4 +26,7 @@ func TestValidateConfigs(t *testing.T) {
 	for _, factory := range factories.Extensions {
 		assert.NoError(t, componenttest.CheckConfigStruct(factory.CreateDefaultConfig()))
 	}
+	for _, factory := range factories.Connectors {
+		assert.NoError(t, componenttest.CheckConfigStruct(factory.CreateDefaultConfig()))
+	}
 }

cmd/builder/internal/builder/templates/go.mod.tmpl

@@ -5,6 +5,9 @@ module {{.Distribution.Module}}
 go 1.18
 
 require (
+	{{- range .Connectors}}
+	{{if .GoMod}}{{.GoMod}}{{end}}
+	{{- end}}
 	{{- range .Extensions}}
 	{{if .GoMod}}{{.GoMod}}{{end}}
 	{{- end}}
@@ -20,6 +23,9 @@ require (
 	go.opentelemetry.io/collector v{{.Distribution.OtelColVersion}}
 )
 
+{{- range .Connectors}}
+{{if ne .Path ""}}replace {{.GoMod}} => {{.Path}}{{end}}
+{{- end}}
 {{- range .Extensions}}
 {{if ne .Path ""}}replace {{.GoMod}} => {{.Path}}{{end}}
 {{- end}}

cmd/builder/internal/command.go

@@ -165,6 +165,7 @@ func applyCfgFromFile(flags *flag.FlagSet, cfgFromFile builder.Config) {
 	cfg.Extensions = cfgFromFile.Extensions
 	cfg.Receivers = cfgFromFile.Receivers
 	cfg.Processors = cfgFromFile.Processors
+	cfg.Connectors = cfgFromFile.Connectors
 	cfg.Replaces = cfgFromFile.Replaces
 	cfg.Excludes = cfgFromFile.Excludes
 

cmd/builder/internal/config/default.yaml

@@ -17,4 +17,5 @@ extensions:
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.69.1
   - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.69.1
-
+connectors:
+  - gomod: go.opentelemetry.io/collector/connector/forwardconnector v0.69.1

cmd/otelcorecol/builder-config.yaml

@@ -17,12 +17,15 @@ extensions:
 processors:
   - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.69.1
   - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.69.1
+connectors:
+  - gomod: go.opentelemetry.io/collector/connector/forwardconnector v0.69.1
 
 replaces:
   - go.opentelemetry.io/collector => ../../
   - go.opentelemetry.io/collector/component => ../../component
   - go.opentelemetry.io/collector/confmap => ../../confmap
   - go.opentelemetry.io/collector/consumer => ../../consumer
+  - go.opentelemetry.io/collector/connector/forwardconnector => ../../connector/forwardconnector
   - go.opentelemetry.io/collector/exporter/loggingexporter => ../../exporter/loggingexporter
   - go.opentelemetry.io/collector/exporter/otlpexporter => ../../exporter/otlpexporter
   - go.opentelemetry.io/collector/exporter/otlphttpexporter => ../../exporter/otlphttpexporter

cmd/otelcorecol/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/stretchr/testify v1.8.1
 	go.opentelemetry.io/collector v0.69.1
 	go.opentelemetry.io/collector/component v0.69.1
+	go.opentelemetry.io/collector/connector/forwardconnector v0.69.1
 	go.opentelemetry.io/collector/exporter/loggingexporter v0.69.1
 	go.opentelemetry.io/collector/exporter/otlpexporter v0.69.1
 	go.opentelemetry.io/collector/exporter/otlphttpexporter v0.69.1
@@ -98,6 +99,8 @@ replace go.opentelemetry.io/collector/confmap => ../../confmap
 
 replace go.opentelemetry.io/collector/consumer => ../../consumer
 
+replace go.opentelemetry.io/collector/connector/forwardconnector => ../../connector/forwardconnector
+
 replace go.opentelemetry.io/collector/exporter/loggingexporter => ../../exporter/loggingexporter
 
 replace go.opentelemetry.io/collector/exporter/otlpexporter => ../../exporter/otlpexporter

otelcol/factories.go

@@ -16,6 +16,7 @@ package otelcol // import "go.opentelemetry.io/collector/otelcol"
 
 import (
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/processor"
@@ -36,4 +37,7 @@ type Factories struct {
 
 	// Extensions maps extension type names in the config to the respective factory.
 	Extensions map[component.Type]extension.Factory
+
+	// Connectors maps connector type names in the config to the respective factory.
+	Connectors map[component.Type]connector.Factory
 }

otelcol/otelcoltest/nop_factories.go

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 package otelcoltest // import "go.opentelemetry.io/collector/otelcol/otelcoltest"
+
 import (
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/connector/connectortest"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/extension"
@@ -46,5 +49,9 @@ func NopFactories() (otelcol.Factories, error) {
 		return otelcol.Factories{}, err
 	}
 
+	if factories.Connectors, err = connector.MakeFactoryMap(connectortest.NewNopFactory()); err != nil {
+		return otelcol.Factories{}, err
+	}
+
 	return factories, err
 }

otelcol/otelcoltest/nop_factories_test.go

@@ -45,4 +45,9 @@ func TestNopFactories(t *testing.T) {
 	nopExtensionFactory, ok := nopFactories.Extensions["nop"]
 	require.True(t, ok)
 	require.Equal(t, component.Type("nop"), nopExtensionFactory.Type())
+
+	require.Equal(t, 1, len(nopFactories.Connectors))
+	nopConnectorFactory, ok := nopFactories.Connectors["nop"]
+	require.True(t, ok)
+	require.Equal(t, component.Type("nop"), nopConnectorFactory.Type())
 }

service/host.go

@@ -16,6 +16,7 @@ package service // import "go.opentelemetry.io/collector/service"
 
 import (
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/processor"
@@ -30,6 +31,7 @@ type serviceHost struct {
 	receivers         *receiver.Builder
 	processors        *processor.Builder
 	exporters         *exporter.Builder
+	connectors        *connector.Builder
 	extensions        *extension.Builder
 
 	buildInfo component.BuildInfo
@@ -53,6 +55,8 @@ func (host *serviceHost) GetFactory(kind component.Kind, componentType component
 		return host.processors.Factory(componentType)
 	case component.KindExporter:
 		return host.exporters.Factory(componentType)
+	case component.KindConnector:
+		return host.connectors.Factory(componentType)
 	case component.KindExtension:
 		return host.extensions.Factory(componentType)
 	}

service/service.go

@@ -25,6 +25,7 @@ import (
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configtelemetry"
+	"go.opentelemetry.io/collector/connector"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/featuregate"
@@ -49,6 +50,9 @@ type Settings struct {
 	// Exporters builder for exporters.
 	Exporters *exporter.Builder
 
+	// Connectors builder for connectors.
+	Connectors *connector.Builder
+
 	// Extensions builder for extensions.
 	Extensions *extension.Builder
 
@@ -82,6 +86,7 @@ func New(ctx context.Context, set Settings, cfg Config) (*Service, error) {
 			receivers:         set.Receivers,
 			processors:        set.Processors,
 			exporters:         set.Exporters,
+			connectors:        set.Connectors,
 			extensions:        set.Extensions,
 			buildInfo:         set.BuildInfo,
 			asyncErrorChannel: set.AsyncErrorChannel,

service/service_test.go

@@ -32,6 +32,7 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.opentelemetry.io/collector/config/configtelemetry"
+	"go.opentelemetry.io/collector/connector/connectortest"
 	"go.opentelemetry.io/collector/exporter/exportertest"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/extension/extensiontest"
@@ -156,6 +157,9 @@ func TestServiceGetFactory(t *testing.T) {
 	assert.Nil(t, srv.host.GetFactory(component.KindExporter, "wrongtype"))
 	assert.Equal(t, set.Exporters.Factory("nop"), srv.host.GetFactory(component.KindExporter, "nop"))
 
+	assert.Nil(t, srv.host.GetFactory(component.KindConnector, "wrongtype"))
+	assert.Equal(t, set.Connectors.Factory("nop"), srv.host.GetFactory(component.KindConnector, "nop"))
+
 	assert.Nil(t, srv.host.GetFactory(component.KindExtension, "wrongtype"))
 	assert.Equal(t, set.Extensions.Factory("nop"), srv.host.GetFactory(component.KindExtension, "nop"))
 
@@ -393,6 +397,7 @@ func newNopSettings() Settings {
 		Receivers:  receivertest.NewNopBuilder(),
 		Processors: processortest.NewNopBuilder(),
 		Exporters:  exportertest.NewNopBuilder(),
+		Connectors: connectortest.NewNopBuilder(),
 		Extensions: extensiontest.NewNopBuilder(),
 	}
 }