[chore] Remove unsage of pipeline.MustNewID in status helper (#40283)

Part of
open-telemetry/opentelemetry-collector#12835
effort.

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

extension/healthcheckv2extension/extension_test.go

@@ -20,6 +20,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 	"go.opentelemetry.io/collector/extension/extensiontest"
+	"go.opentelemetry.io/collector/pipeline"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/common/testutil"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/status"
@@ -40,7 +41,7 @@ func TestComponentStatus(t *testing.T) {
 
 	require.NoError(t, ext.Start(context.Background(), componenttest.NewNopHost()))
 
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	// StatusStarting will be sent immediately.
 	for _, id := range traces.InstanceIDs() {

extension/healthcheckv2extension/go.mod

@@ -18,6 +18,7 @@ require (
 	go.opentelemetry.io/collector/extension v1.32.1-0.20250523042642-a867641d12bd
 	go.opentelemetry.io/collector/extension/extensioncapabilities v0.126.1-0.20250523042642-a867641d12bd
 	go.opentelemetry.io/collector/extension/extensiontest v0.126.1-0.20250523042642-a867641d12bd
+	go.opentelemetry.io/collector/pipeline v0.126.1-0.20250523042642-a867641d12bd
 	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
@@ -59,7 +60,6 @@ require (
 	go.opentelemetry.io/collector/featuregate v1.32.1-0.20250523042642-a867641d12bd // indirect
 	go.opentelemetry.io/collector/internal/telemetry v0.126.1-0.20250523042642-a867641d12bd // indirect
 	go.opentelemetry.io/collector/pdata v1.32.1-0.20250523042642-a867641d12bd // indirect
-	go.opentelemetry.io/collector/pipeline v0.126.1-0.20250523042642-a867641d12bd // indirect
 	go.opentelemetry.io/contrib/bridges/otelzap v0.10.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect

extension/healthcheckv2extension/internal/grpc/grpc_test.go

@@ -15,6 +15,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/config/confignet"
+	"go.opentelemetry.io/collector/pipeline"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
@@ -39,8 +40,8 @@ func TestCheck(t *testing.T) {
 		},
 	}
 	var server *Server
-	traces := testhelpers.NewPipelineMetadata("traces")
-	metrics := testhelpers.NewPipelineMetadata("metrics")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
+	metrics := testhelpers.NewPipelineMetadata(pipeline.SignalMetrics)
 
 	type teststep struct {
 		step           func()
@@ -754,8 +755,8 @@ func TestWatch(t *testing.T) {
 		},
 	}
 	var server *Server
-	traces := testhelpers.NewPipelineMetadata("traces")
-	metrics := testhelpers.NewPipelineMetadata("metrics")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
+	metrics := testhelpers.NewPipelineMetadata(pipeline.SignalMetrics)
 
 	// statusUnchanged is a sentinel value to signal that a step does not result
 	// in a status change. This is important, because checking for a status

extension/healthcheckv2extension/internal/http/server_test.go

@@ -21,6 +21,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"go.opentelemetry.io/collector/pipeline"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckv2extension/internal/common"
 	internalhelpers "github.com/open-telemetry/opentelemetry-collector-contrib/extension/healthcheckv2extension/internal/testhelpers"
@@ -154,15 +155,14 @@ type teststep struct {
 
 func TestStatus(t *testing.T) {
 	var server *Server
-	traces := testhelpers.NewPipelineMetadata("traces")
-	metrics := testhelpers.NewPipelineMetadata("metrics")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
+	metrics := testhelpers.NewPipelineMetadata(pipeline.SignalMetrics)
 
 	tests := []struct {
 		name                  string
 		config                *Config
 		legacyConfig          LegacyConfig
 		componentHealthConfig *common.ComponentHealthConfig
-		pipelines             map[string]*testhelpers.PipelineMetadata
 		teststeps             []teststep
 	}{
 		{
@@ -2582,7 +2582,6 @@ func TestStatus(t *testing.T) {
 					Path:    "/status",
 				},
 			},
-			pipelines: testhelpers.NewPipelines("traces"),
 			teststeps: []teststep{
 				{
 					step: func() {

extension/opampextension/go.mod

@@ -116,7 +116,7 @@ require (
 	github.com/tklauser/numcpus v0.8.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/collector/pdata v1.32.1-0.20250523042642-a867641d12bd
-	go.opentelemetry.io/collector/pipeline v0.126.1-0.20250523042642-a867641d12bd // indirect
+	go.opentelemetry.io/collector/pipeline v0.126.1-0.20250523042642-a867641d12bd
 	go.opentelemetry.io/otel v1.35.0
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.35.0 // indirect

extension/opampextension/opamp_agent_test.go

@@ -24,6 +24,7 @@ import (
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 	"go.opentelemetry.io/collector/extension"
 	"go.opentelemetry.io/collector/extension/extensiontest"
+	"go.opentelemetry.io/collector/pipeline"
 	"go.opentelemetry.io/collector/service"
 	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
 	"go.uber.org/zap"
@@ -545,7 +546,7 @@ func TestHealthReportingForwardComponentHealthToAggregator(t *testing.T) {
 
 	assert.NoError(t, o.Start(context.Background(), componenttest.NewNopHost()))
 
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	// StatusStarting will be sent immediately.
 	for _, id := range traces.InstanceIDs() {

pkg/status/aggregator_test.go

@@ -20,7 +20,7 @@ import (
 
 func TestAggregateStatus(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	t.Run("zero value", func(t *testing.T) {
 		st, ok := agg.AggregateStatus(status.ScopeAll, status.Concise)
@@ -69,7 +69,7 @@ func TestAggregateStatus(t *testing.T) {
 
 func TestAggregateStatusVerbose(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 	tracesKey := toPipelineKey(traces.PipelineID)
 
 	t.Run("zero value", func(t *testing.T) {
@@ -131,7 +131,7 @@ func TestAggregateStatusVerbose(t *testing.T) {
 
 func TestAggregateStatusPriorityRecoverable(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityRecoverable)
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	testhelpers.SeedAggregator(agg, traces.InstanceIDs(), componentstatus.StatusOK)
 
@@ -174,7 +174,7 @@ func TestAggregateStatusPriorityRecoverable(t *testing.T) {
 
 func TestPipelineAggregateStatus(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	t.Run("non existent pipeline", func(t *testing.T) {
 		st, ok := agg.AggregateStatus("doesnotexist", status.Concise)
@@ -210,7 +210,7 @@ func TestPipelineAggregateStatus(t *testing.T) {
 
 func TestPipelineAggregateStatusVerbose(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	t.Run("non existent pipeline", func(t *testing.T) {
 		st, ok := agg.AggregateStatus("doesnotexist", status.Verbose)
@@ -303,8 +303,8 @@ func TestStreaming(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
 	defer agg.Close()
 
-	traces := testhelpers.NewPipelineMetadata("traces")
-	metrics := testhelpers.NewPipelineMetadata("metrics")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
+	metrics := testhelpers.NewPipelineMetadata(pipeline.SignalMetrics)
 
 	traceEvents, traceUnsub := agg.Subscribe(status.Scope(traces.PipelineID.String()), status.Concise)
 	defer traceUnsub()
@@ -364,7 +364,7 @@ func TestStreamingVerbose(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
 	defer agg.Close()
 
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 	tracesKey := toPipelineKey(traces.PipelineID)
 
 	allEvents, unsub := agg.Subscribe(status.ScopeAll, status.Verbose)
@@ -424,7 +424,7 @@ func TestUnsubscribe(t *testing.T) {
 	agg := status.NewAggregator(status.PriorityPermanent)
 	defer agg.Close()
 
-	traces := testhelpers.NewPipelineMetadata("traces")
+	traces := testhelpers.NewPipelineMetadata(pipeline.SignalTraces)
 
 	traceEvents, traceUnsub := agg.Subscribe(status.Scope(traces.PipelineID.String()), status.Concise)
 	allEvents, allUnsub := agg.Subscribe(status.ScopeAll, status.Concise)

pkg/status/testhelpers/helpers.go

@@ -26,25 +26,16 @@ func (p *PipelineMetadata) InstanceIDs() []*componentstatus.InstanceID {
 }
 
 // NewPipelineMetadata returns a metadata for a hypothetical pipeline.
-func NewPipelineMetadata(typestr string) *PipelineMetadata {
-	pipelineID := pipeline.MustNewID(typestr)
+func NewPipelineMetadata(signal pipeline.Signal) *PipelineMetadata {
+	pipelineID := pipeline.NewID(signal)
 	return &PipelineMetadata{
 		PipelineID:  pipelineID,
-		ReceiverID:  componentstatus.NewInstanceID(component.NewIDWithName(component.MustNewType(typestr), "in"), component.KindReceiver).WithPipelines(pipelineID),
+		ReceiverID:  componentstatus.NewInstanceID(component.NewIDWithName(component.MustNewType(signal.String()), "in"), component.KindReceiver).WithPipelines(pipelineID),
 		ProcessorID: componentstatus.NewInstanceID(component.MustNewID("batch"), component.KindProcessor).WithPipelines(pipelineID),
-		ExporterID:  componentstatus.NewInstanceID(component.NewIDWithName(component.MustNewType(typestr), "out"), component.KindExporter).WithPipelines(pipelineID),
+		ExporterID:  componentstatus.NewInstanceID(component.NewIDWithName(component.MustNewType(signal.String()), "out"), component.KindExporter).WithPipelines(pipelineID),
 	}
 }
 
-// NewPipelines returns a map of hypothetical pipelines identified by their stringified typeVal.
-func NewPipelines(typestrs ...string) map[string]*PipelineMetadata {
-	result := make(map[string]*PipelineMetadata, len(typestrs))
-	for _, typestr := range typestrs {
-		result[typestr] = NewPipelineMetadata(typestr)
-	}
-	return result
-}
-
 // SeedAggregator records a status event for each instanceID.
 func SeedAggregator(
 	agg *status.Aggregator,