Add connector.Builder

.chloggen/connector-builder.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: connector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add Builder
+
+# One or more tracking issues or pull requests related to the change
+issues: [6867]
+
+# (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:

connector/connector.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"fmt"
 
+	"go.uber.org/zap"
+
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 )
@@ -85,7 +87,10 @@ type Logs interface {
 
 // CreateSettings configures Connector creators.
 type CreateSettings struct {
-	TelemetrySettings component.TelemetrySettings
+	// ID returns the ID of the component that will be created.
+	ID component.ID
+
+	component.TelemetrySettings
 
 	// BuildInfo can be used by components for informational purposes
 	BuildInfo component.BuildInfo
@@ -466,3 +471,173 @@ func MakeFactoryMap(factories ...Factory) (map[component.Type]Factory, error) {
 	}
 	return fMap, nil
 }
+
+// Builder processor is a helper struct that given a set of Configs and Factories helps with creating processors.
+type Builder struct {
+	cfgs      map[component.ID]component.Config
+	factories map[component.Type]Factory
+}
+
+// NewBuilder creates a new connector.Builder to help with creating components form a set of configs and factories.
+func NewBuilder(cfgs map[component.ID]component.Config, factories map[component.Type]Factory) *Builder {
+	return &Builder{cfgs: cfgs, factories: factories}
+}
+
+// CreateTracesToTraces creates a Traces connector based on the settings and config.
+func (b *Builder) CreateTracesToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Traces, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.TracesToTracesStability())
+	return f.CreateTracesToTraces(ctx, set, cfg, next)
+}
+
+// CreateTracesToMetrics creates a Traces connector based on the settings and config.
+func (b *Builder) CreateTracesToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Traces, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.TracesToMetricsStability())
+	return f.CreateTracesToMetrics(ctx, set, cfg, next)
+}
+
+// CreateTracesToLogs creates a Traces connector based on the settings and config.
+func (b *Builder) CreateTracesToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Traces, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.TracesToLogsStability())
+	return f.CreateTracesToLogs(ctx, set, cfg, next)
+}
+
+// CreateMetricsToTraces creates a Metrics connector based on the settings and config.
+func (b *Builder) CreateMetricsToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Metrics, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.MetricsToTracesStability())
+	return f.CreateMetricsToTraces(ctx, set, cfg, next)
+}
+
+// CreateMetricsToMetrics creates a Metrics connector based on the settings and config.
+func (b *Builder) CreateMetricsToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Metrics, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.MetricsToMetricsStability())
+	return f.CreateMetricsToMetrics(ctx, set, cfg, next)
+}
+
+// CreateMetricsToLogs creates a Metrics connector based on the settings and config.
+func (b *Builder) CreateMetricsToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Metrics, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.MetricsToLogsStability())
+	return f.CreateMetricsToLogs(ctx, set, cfg, next)
+}
+
+// CreateLogsToTraces creates a Logs connector based on the settings and config.
+func (b *Builder) CreateLogsToTraces(ctx context.Context, set CreateSettings, next consumer.Traces) (Logs, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.LogsToTracesStability())
+	return f.CreateLogsToTraces(ctx, set, cfg, next)
+}
+
+// CreateLogsToMetrics creates a Logs connector based on the settings and config.
+func (b *Builder) CreateLogsToMetrics(ctx context.Context, set CreateSettings, next consumer.Metrics) (Logs, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.LogsToMetricsStability())
+	return f.CreateLogsToMetrics(ctx, set, cfg, next)
+}
+
+// CreateLogsToLogs creates a Logs connector based on the settings and config.
+func (b *Builder) CreateLogsToLogs(ctx context.Context, set CreateSettings, next consumer.Logs) (Logs, error) {
+	cfg, existsCfg := b.cfgs[set.ID]
+	if !existsCfg {
+		return nil, fmt.Errorf("connector %q is not configured", set.ID)
+	}
+
+	f, existsFactory := b.factories[set.ID.Type()]
+	if !existsFactory {
+		return nil, fmt.Errorf("connector factory not available for: %q", set.ID)
+	}
+
+	logStabilityLevel(set.Logger, f.LogsToLogsStability())
+	return f.CreateLogsToLogs(ctx, set, cfg, next)
+}
+
+func (b *Builder) Factory(componentType component.Type) component.Factory {
+	return b.factories[componentType]
+}
+
+// logStabilityLevel logs the stability level of a component. The log level is set to info for
+// undefined, unmaintained, deprecated and development. The log level is set to debug
+// for alpha, beta and stable.
+func logStabilityLevel(logger *zap.Logger, sl component.StabilityLevel) {
+	if sl >= component.StabilityLevelAlpha {
+		logger.Debug(sl.LogMessage())
+	} else {
+		logger.Info(sl.LogMessage())
+	}
+}