[connector,exporter,processor,receiver] Error out on mismatched type

.chloggen/mx-psi_type-mismatch.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: processor, connector, exporter, receiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Explicitly error out at component creation time if there is a type mismatch.
+
+# One or more tracking issues or pull requests related to the change
+issues: [12305]
+
+# (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:
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

connector/connector.go

@@ -5,6 +5,7 @@
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/connector/internal"
@@ -305,62 +306,107 @@
 	if f.createTracesToTracesFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalTraces, pipeline.SignalTraces)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createTracesToTracesFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateTracesToMetrics(ctx context.Context, set Settings, cfg component.Config, next consumer.Metrics) (Traces, error) {
 	if f.createTracesToMetricsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalTraces, pipeline.SignalMetrics)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createTracesToMetricsFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateTracesToLogs(ctx context.Context, set Settings, cfg component.Config, next consumer.Logs) (Traces, error) {
 	if f.createTracesToLogsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalTraces, pipeline.SignalLogs)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createTracesToLogsFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateMetricsToTraces(ctx context.Context, set Settings, cfg component.Config, next consumer.Traces) (Metrics, error) {
 	if f.createMetricsToTracesFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalMetrics, pipeline.SignalTraces)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createMetricsToTracesFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateMetricsToMetrics(ctx context.Context, set Settings, cfg component.Config, next consumer.Metrics) (Metrics, error) {
 	if f.createMetricsToMetricsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalMetrics, pipeline.SignalMetrics)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createMetricsToMetricsFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateMetricsToLogs(ctx context.Context, set Settings, cfg component.Config, next consumer.Logs) (Metrics, error) {
 	if f.createMetricsToLogsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalMetrics, pipeline.SignalLogs)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createMetricsToLogsFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateLogsToTraces(ctx context.Context, set Settings, cfg component.Config, next consumer.Traces) (Logs, error) {
 	if f.createLogsToTracesFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalLogs, pipeline.SignalTraces)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createLogsToTracesFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateLogsToMetrics(ctx context.Context, set Settings, cfg component.Config, next consumer.Metrics) (Logs, error) {
 	if f.createLogsToMetricsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalLogs, pipeline.SignalMetrics)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createLogsToMetricsFunc(ctx, set, cfg, next)
 }
 
 func (f *factory) CreateLogsToLogs(ctx context.Context, set Settings, cfg component.Config, next consumer.Logs) (Logs, error) {
 	if f.createLogsToLogsFunc == nil {
 		return nil, internal.ErrDataTypes(set.ID, pipeline.SignalLogs, pipeline.SignalLogs)
 	}
+
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createLogsToLogsFunc(ctx, set, cfg, next)
 }
 

connector/connector_test.go

@@ -19,7 +19,7 @@ import (
 
 var (
 	testType = component.MustNewType("test")
-	testID   = component.MustNewIDWithName("type", "name")
+	testID   = component.MustNewIDWithName("test", "name")
 )
 
 func TestNewFactoryNoOptions(t *testing.T) {
@@ -147,33 +147,33 @@ func TestNewFactoryWithAllTypes(t *testing.T) {
 	assert.EqualValues(t, &defaultCfg, factory.CreateDefaultConfig())
 
 	assert.Equal(t, component.StabilityLevelAlpha, factory.TracesToTracesStability())
-	_, err := factory.CreateTracesToTraces(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err := factory.CreateTracesToTraces(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelDevelopment, factory.TracesToMetricsStability())
-	_, err = factory.CreateTracesToMetrics(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateTracesToMetrics(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelAlpha, factory.TracesToLogsStability())
-	_, err = factory.CreateTracesToLogs(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateTracesToLogs(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 
 	assert.Equal(t, component.StabilityLevelBeta, factory.MetricsToTracesStability())
-	_, err = factory.CreateMetricsToTraces(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateMetricsToTraces(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelBeta, factory.MetricsToMetricsStability())
-	_, err = factory.CreateMetricsToMetrics(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateMetricsToMetrics(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelStable, factory.MetricsToLogsStability())
-	_, err = factory.CreateMetricsToLogs(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateMetricsToLogs(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 
 	assert.Equal(t, component.StabilityLevelDeprecated, factory.LogsToTracesStability())
-	_, err = factory.CreateLogsToTraces(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateLogsToTraces(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelUnmaintained, factory.LogsToMetricsStability())
-	_, err = factory.CreateLogsToMetrics(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateLogsToMetrics(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	require.NoError(t, err)
 	assert.Equal(t, component.StabilityLevelUnmaintained, factory.LogsToLogsStability())
-	_, err = factory.CreateLogsToLogs(context.Background(), Settings{}, &defaultCfg, consumertest.NewNop())
+	_, err = factory.CreateLogsToLogs(context.Background(), Settings{ID: testID}, &defaultCfg, consumertest.NewNop())
 	assert.NoError(t, err)
 }
 

exporter/exporter.go

@@ -5,6 +5,7 @@
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
@@ -132,6 +133,10 @@
 		return nil, pipeline.ErrSignalNotSupported
 	}
 
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createTracesFunc(ctx, set, cfg)
 }
 
@@ -140,6 +145,10 @@
 		return nil, pipeline.ErrSignalNotSupported
 	}
 
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createMetricsFunc(ctx, set, cfg)
 }
 
@@ -148,6 +157,10 @@
 		return nil, pipeline.ErrSignalNotSupported
 	}
 
+	if set.ID.Type() != f.Type() {
+		return nil, fmt.Errorf("component type mismatch: component ID %q does not have type %q", set.ID, f.Type())
+	}
+
 	return f.createLogsFunc(ctx, set, cfg)
 }
 

exporter/exporter_test.go

@@ -15,24 +15,27 @@ import (
 	"go.opentelemetry.io/collector/pipeline"
 )
 
+var (
+	testType = component.MustNewType("test")
+	testID   = component.NewID(testType)
+)
+
 func TestNewFactory(t *testing.T) {
-	testType := component.MustNewType("test")
 	defaultCfg := struct{}{}
 	f := NewFactory(
 		testType,
 		func() component.Config { return &defaultCfg })
 	assert.EqualValues(t, testType, f.Type())
 	assert.EqualValues(t, &defaultCfg, f.CreateDefaultConfig())
-	_, err := f.CreateTraces(context.Background(), Settings{}, &defaultCfg)
+	_, err := f.CreateTraces(context.Background(), Settings{ID: testID}, &defaultCfg)
 	require.ErrorIs(t, err, pipeline.ErrSignalNotSupported)
-	_, err = f.CreateMetrics(context.Background(), Settings{}, &defaultCfg)
+	_, err = f.CreateMetrics(context.Background(), Settings{ID: testID}, &defaultCfg)
 	require.ErrorIs(t, err, pipeline.ErrSignalNotSupported)
-	_, err = f.CreateLogs(context.Background(), Settings{}, &defaultCfg)
+	_, err = f.CreateLogs(context.Background(), Settings{ID: testID}, &defaultCfg)
 	require.ErrorIs(t, err, pipeline.ErrSignalNotSupported)
 }
 
 func TestNewFactoryWithOptions(t *testing.T) {
-	testType := component.MustNewType("test")
 	defaultCfg := struct{}{}
 	f := NewFactory(
 		testType,
@@ -44,15 +47,15 @@ func TestNewFactoryWithOptions(t *testing.T) {
 	assert.EqualValues(t, &defaultCfg, f.CreateDefaultConfig())
 
 	assert.Equal(t, component.StabilityLevelDevelopment, f.TracesStability())
-	_, err := f.CreateTraces(context.Background(), Settings{}, &defaultCfg)
+	_, err := f.CreateTraces(context.Background(), Settings{ID: testID}, &defaultCfg)
 	require.NoError(t, err)
 
 	assert.Equal(t, component.StabilityLevelAlpha, f.MetricsStability())
-	_, err = f.CreateMetrics(context.Background(), Settings{}, &defaultCfg)
+	_, err = f.CreateMetrics(context.Background(), Settings{ID: testID}, &defaultCfg)
 	require.NoError(t, err)
 
 	assert.Equal(t, component.StabilityLevelDeprecated, f.LogsStability())
-	_, err = f.CreateLogs(context.Background(), Settings{}, &defaultCfg)
+	_, err = f.CreateLogs(context.Background(), Settings{ID: testID}, &defaultCfg)
 	assert.NoError(t, err)
 }
 

exporter/exportertest/contract_checker.go

@@ -151,7 +151,7 @@ func checkTraces(t *testing.T, params CheckConsumeContractParams, mockReceiver c
 	ctx := context.Background()
 	var exp exporter.Traces
 	var err error
-	exp, err = params.ExporterFactory.CreateTraces(ctx, NewNopSettings(NopType), params.ExporterConfig)
+	exp, err = params.ExporterFactory.CreateTraces(ctx, NewNopSettings(params.ExporterFactory.Type()), params.ExporterConfig)
 	require.NoError(t, err)
 	require.NotNil(t, exp)
 
@@ -191,7 +191,7 @@ func checkLogs(t *testing.T, params CheckConsumeContractParams, mockReceiver com
 	ctx := context.Background()
 	var exp exporter.Logs
 	var err error
-	exp, err = params.ExporterFactory.CreateLogs(ctx, NewNopSettings(NopType), params.ExporterConfig)
+	exp, err = params.ExporterFactory.CreateLogs(ctx, NewNopSettings(params.ExporterFactory.Type()), params.ExporterConfig)
 	require.NoError(t, err)
 	require.NotNil(t, exp)