Make DataType a separate type within Component

.chloggen/datatype-newid.yaml

@@ -0,0 +1,32 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: breaking
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: component
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Changes DataType to be an enum of Otel signals.  Creates a new module pipeline to centralize types related to pipelines.  Creates a new struct pipeline.ID - made up of a DataType and an optional string name.  This pipeline.ID replaces the previous use of component.ID to identify Pipelines.  Moves InstanceID to pipeline to break import cycles."
+
+# One or more tracking issues or pull requests related to the change
+issues: [9429]
+
+# (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: |
+  Related API changes:
+  connector.LogsRouterAndConsumer, connector.MetricsRouterAndConsumer, and connector.TracesRouterAndConsumer `Consumer` methods now accepts pipeline.ID instead of component.ID.  Their PipelineIDs function return pipeline.ID instead of component.ID.
+  NewStatusWatcherExtensionFactory now accepts a pipeline.InstanceID instead of component.InstanceID.
+  The generated component tests now use PipelineIDs for testing connectors.
+  extension.ComponentStatusChanged now takes in a *pipeline.InstanceID instead of a *component.InstanceID.
+  extensiontest.NewStatusWatcherExtensionFactory now takes in a *pipeline.InstanceID instead of a *component.InstanceID.
+  extensions.NotifyComponentStatusChange now takes in a *pipeline.InstanceID instead of a *component.InstanceID.
+
+# 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']

cmd/mdatagen/go.mod

@@ -49,7 +49,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.27.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.25.0 // indirect
-	golang.org/x/sys v0.20.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240520151616-dc85e6b867a5 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect
@@ -85,3 +85,5 @@ retract (
 	v0.76.1
 	v0.65.0
 )
+
+replace go.opentelemetry.io/collector/pipeline => ../../pipeline

cmd/mdatagen/templates/component_test.go.tmpl

@@ -44,6 +44,8 @@ import (
 	"go.opentelemetry.io/collector/connector/connectortest"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/pipeline"
+
 	{{- end }}
 	{{- if or isExporter isProcessor }}
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -372,7 +374,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "logs_to_logs",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewLogsRouter(map[component.ID]consumer.Logs{component.NewID(component.DataTypeLogs): consumertest.NewNop()})
+				router := connector.NewLogsRouter(map[pipeline.ID]consumer.Logs{pipeline.ID(component.DataTypeLogs): consumertest.NewNop()})
 				return factory.CreateLogsToLogs(ctx, set, cfg, router)
 			},
 		},
@@ -381,7 +383,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "logs_to_metrics",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewMetricsRouter(map[component.ID]consumer.Metrics{component.NewID(component.DataTypeMetrics): consumertest.NewNop()})
+				router := connector.NewMetricsRouter(map[pipeline.ID]consumer.Metrics{pipeline.NewID(component.DataTypeMetrics): consumertest.NewNop()})
 				return factory.CreateLogsToMetrics(ctx, set, cfg, router)
 			},
 		},
@@ -390,7 +392,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "logs_to_traces",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewTracesRouter(map[component.ID]consumer.Traces{component.NewID(component.DataTypeTraces): consumertest.NewNop()})
+				router := connector.NewTracesRouter(map[pipeline.ID]consumer.Traces{pipeline.NewID(component.DataTypeTraces): consumertest.NewNop()})
 				return factory.CreateLogsToTraces(ctx, set, cfg, router)
 			},
 		},
@@ -399,7 +401,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "metrics_to_logs",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewLogsRouter(map[component.ID]consumer.Logs{component.NewID(component.DataTypeLogs): consumertest.NewNop()})
+				router := connector.NewLogsRouter(map[pipeline.ID]consumer.Logs{pipeline.NewID(component.DataTypeLogs): consumertest.NewNop()})
 				return factory.CreateMetricsToLogs(ctx, set, cfg, router)
 			},
 		},
@@ -408,7 +410,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "metrics_to_metrics",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewMetricsRouter(map[component.ID]consumer.Metrics{component.NewID(component.DataTypeMetrics): consumertest.NewNop()})
+				router := connector.NewMetricsRouter(map[pipeline.ID]consumer.Metrics{pipeline.NewID(component.DataTypeMetrics): consumertest.NewNop()})
 				return factory.CreateMetricsToMetrics(ctx, set, cfg, router)
 			},
 		},
@@ -417,7 +419,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "metrics_to_traces",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewTracesRouter(map[component.ID]consumer.Traces{component.NewID(component.DataTypeTraces): consumertest.NewNop()})
+				router := connector.NewTracesRouter(map[pipeline.ID]consumer.Traces{pipeline.NewID(component.DataTypeTraces): consumertest.NewNop()})
 				return factory.CreateMetricsToTraces(ctx, set, cfg, router)
 			},
 		},
@@ -426,7 +428,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "traces_to_logs",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewLogsRouter(map[component.ID]consumer.Logs{component.NewID(component.DataTypeLogs): consumertest.NewNop()})
+				router := connector.NewLogsRouter(map[pipeline.ID]consumer.Logs{pipeline.NewID(component.DataTypeLogs): consumertest.NewNop()})
 				return factory.CreateTracesToLogs(ctx, set, cfg, router)
 			},
 		},
@@ -435,7 +437,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "traces_to_metrics",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewMetricsRouter(map[component.ID]consumer.Metrics{component.NewID(component.DataTypeMetrics): consumertest.NewNop()})
+				router := connector.NewMetricsRouter(map[pipeline.ID]consumer.Metrics{pipeline.NewID(component.DataTypeMetrics): consumertest.NewNop()})
 				return factory.CreateTracesToMetrics(ctx, set, cfg, router)
 			},
 		},
@@ -444,7 +446,7 @@ func TestComponentLifecycle(t *testing.T) {
 		{
 			name: "traces_to_traces",
 			createFn: func(ctx context.Context, set connector.Settings, cfg component.Config) (component.Component, error) {
-				router := connector.NewTracesRouter(map[component.ID]consumer.Traces{component.NewID(component.DataTypeTraces): consumertest.NewNop()})
+				router := connector.NewTracesRouter(map[pipeline.ID]consumer.Traces{pipeline.NewID(component.DataTypeTraces): consumertest.NewNop()})
 				return factory.CreateTracesToTraces(ctx, set, cfg, router)
 			},
 		},

cmd/otelcorecol/go.mod

@@ -94,6 +94,7 @@ require (
 	go.opentelemetry.io/collector/extension/auth v0.102.1 // indirect
 	go.opentelemetry.io/collector/featuregate v1.9.0 // indirect
 	go.opentelemetry.io/collector/pdata v1.9.0 // indirect
+	go.opentelemetry.io/collector/pipeline v0.102.1 // indirect
 	go.opentelemetry.io/collector/semconv v0.102.1 // indirect
 	go.opentelemetry.io/collector/service v0.102.1 // indirect
 	go.opentelemetry.io/contrib/config v0.7.0 // indirect
@@ -218,3 +219,5 @@ replace go.opentelemetry.io/collector/processor/memorylimiterprocessor => ../../
 replace go.opentelemetry.io/collector/semconv => ../../semconv
 
 replace go.opentelemetry.io/collector/service => ../../service
+
+replace go.opentelemetry.io/collector/pipeline => ../../pipeline

component/component.go

@@ -189,10 +189,3 @@ type CreateDefaultConfigFunc func() Config
 func (f CreateDefaultConfigFunc) CreateDefaultConfig() Config {
 	return f()
 }
-
-// InstanceID uniquely identifies a component instance
-type InstanceID struct {
-	ID          ID
-	Kind        Kind
-	PipelineIDs map[ID]struct{}
-}

component/config.go

@@ -4,6 +4,7 @@
 package component // import "go.opentelemetry.io/collector/component"
 
 import (
+	"errors"
 	"fmt"
 	"reflect"
 	"regexp"
@@ -97,6 +98,8 @@ func callValidateIfPossible(v reflect.Value) error {
 }
 
 // Type is the component type as it is used in the config.
+// Examples of a Type include the names of receivers (otlp, filelog, etc),
+// processors (batch, memory_limit, etc), or exporters (debug, rabbitmq)
 type Type struct {
 	name string
 }
@@ -147,20 +150,44 @@ func MustNewType(strType string) Type {
 
 // DataType is a special Type that represents the data types supported by the collector. We currently support
 // collecting metrics, traces and logs, this can expand in the future.
-type DataType = Type
-
-func mustNewDataType(strType string) DataType {
-	return MustNewType(strType)
-}
+type DataType string
 
 // Currently supported data types. Add new data types here when new types are supported in the future.
-var (
+const (
 	// DataTypeTraces is the data type tag for traces.
-	DataTypeTraces = mustNewDataType("traces")
+	DataTypeTraces DataType = "traces"
 
 	// DataTypeMetrics is the data type tag for metrics.
-	DataTypeMetrics = mustNewDataType("metrics")
+	DataTypeMetrics DataType = "metrics"
 
 	// DataTypeLogs is the data type tag for logs.
-	DataTypeLogs = mustNewDataType("logs")
+	DataTypeLogs DataType = "logs"
 )
+
+func (dt DataType) String() string {
+	return string(dt)
+}
+
+func (dt DataType) MarshalText() (text []byte, err error) {
+	return []byte(dt), nil
+}
+
+func (dt *DataType) UnmarshalText(text []byte) error {
+	if len(text) == 0 {
+		return errors.New("id must not be empty")
+	}
+	strText := string(text)
+	switch strText {
+	case "metrics":
+		*dt = DataTypeMetrics
+		return nil
+	case "logs":
+		*dt = DataTypeLogs
+		return nil
+	case "traces":
+		*dt = DataTypeTraces
+		return nil
+	default:
+		return fmt.Errorf("invalid data type %q", strText)
+	}
+}

component/config_test.go

@@ -420,3 +420,41 @@ func TestNewType(t *testing.T) {
 		})
 	}
 }
+
+func TestNewDataType(t *testing.T) {
+	var dt DataType
+	err := dt.UnmarshalText([]byte("logs"))
+	require.NoError(t, err)
+	assert.Equal(t, DataTypeLogs, dt)
+
+	var dt2 DataType
+	// hopefully this reminds us to update DataType when profiles get included
+	err = dt2.UnmarshalText([]byte("profiles"))
+	assert.Equal(t, DataType(""), dt2)
+	assert.Equal(t, errors.New("invalid data type \"profiles\""), err)
+}
+
+func TestDataTypeStringMarshal(t *testing.T) {
+	assert.Equal(t, "metrics", DataTypeMetrics.String())
+	text, err := DataTypeMetrics.MarshalText()
+	assert.Equal(t, []byte("metrics"), text)
+	assert.Nil(t, err)
+}
+
+func TestDataTypeFromSignal(t *testing.T) {
+	var dt DataType
+	err := dt.UnmarshalText([]byte("metrics"))
+	assert.Nil(t, err)
+	assert.Equal(t, DataTypeMetrics, dt)
+
+	var dt2 DataType
+	err = dt2.UnmarshalText([]byte("traces"))
+	assert.Equal(t, DataTypeTraces, dt2)
+	assert.Nil(t, err)
+
+	var dt3 DataType
+	err = dt3.UnmarshalText([]byte(""))
+	assert.Equal(t, errors.New("id must not be empty"), err)
+	assert.Equal(t, DataType(""), dt3)
+
+}

component/go.mod

@@ -29,10 +29,10 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.0 // indirect
-	golang.org/x/net v0.24.0 // indirect
+	golang.org/x/net v0.25.0 // indirect
 	golang.org/x/sys v0.20.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
+	golang.org/x/text v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240520151616-dc85e6b867a5 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

component/identifiable.go

@@ -9,12 +9,13 @@ import (
 	"strings"
 )
 
-// typeAndNameSeparator is the separator that is used between type and name in type/name composite keys.
-const typeAndNameSeparator = "/"
+// TypeAndNameSeparator is the separator that is used between type and name in type/name composite keys.
+const TypeAndNameSeparator = "/"
 
 // ID represents the identity for a component. It combines two values:
 // * type - the Type of the component.
-// * name - the name of that component.
+// * name - the name of that component.  This can be an empty string.
+// Components are defined in configuration by type[/name] - for examples [traces/1] or [oltp/blah]
 // The component ID (combination type + name) is unique for a given component.Kind.
 type ID struct {
 	typeVal Type   `mapstructure:"-"`
@@ -62,7 +63,7 @@ func (id ID) MarshalText() (text []byte, err error) {
 // UnmarshalText implements the encoding.TextUnmarshaler interface.
 func (id *ID) UnmarshalText(text []byte) error {
 	idStr := string(text)
-	items := strings.SplitN(idStr, typeAndNameSeparator, 2)
+	items := strings.SplitN(idStr, TypeAndNameSeparator, 2)
 	var typeStr, nameStr string
 	if len(items) >= 1 {
 		typeStr = strings.TrimSpace(items[0])
@@ -73,14 +74,14 @@ func (id *ID) UnmarshalText(text []byte) error {
 	}
 
 	if typeStr == "" {
-		return fmt.Errorf("in %q id: the part before %s should not be empty", idStr, typeAndNameSeparator)
+		return fmt.Errorf("in %q id: the part before %s should not be empty", idStr, TypeAndNameSeparator)
 	}
 
 	if len(items) > 1 {
 		// "name" part is present.
 		nameStr = strings.TrimSpace(items[1])
 		if nameStr == "" {
-			return fmt.Errorf("in %q id: the part after %s should not be empty", idStr, typeAndNameSeparator)
+			return fmt.Errorf("in %q id: the part after %s should not be empty", idStr, TypeAndNameSeparator)
 		}
 	}
 
@@ -99,5 +100,5 @@ func (id ID) String() string {
 		return id.typeVal.String()
 	}
 
-	return id.typeVal.String() + typeAndNameSeparator + id.nameVal
+	return id.typeVal.String() + TypeAndNameSeparator + id.nameVal
 }