[processor/routing] Expand error handling on failure to build exporte…

…rs (open-telemetry#8125)

* expand error handling on failure to build exporters

* update changelog

* add testcase scenario

* run gofmt and add shutdown

* fix imports

CHANGELOG.md

@@ -15,6 +15,7 @@
 - `internal/stanza`: Add support for arbitrary attribute types (#8081)
 - `resourcedetectionprocessor`: Add confighttp.HTTPClientSettings To Resource Detection Config Fixes (#7397)
 - `honeycombexporter`: Add validation for `sending_queue` setting (#8113)
+- `routingprocessor`: Expand error handling on failure to build exporters (#8125)
 
 ### 🛑 Breaking changes 🛑
 

processor/routingprocessor/factory_test.go

@@ -16,15 +16,21 @@ package routingprocessor
 
 import (
 	"context"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/configgrpc"
 	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/exporter/otlpexporter"
 	"go.opentelemetry.io/collector/model/pdata"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+	"go.opentelemetry.io/collector/service/servicetest"
+	"go.uber.org/zap"
 )
 
 func TestProcessorGetsCreatedWithValidConfiguration(t *testing.T) {
@@ -136,6 +142,56 @@ func TestShouldNotFailWhenNextIsProcessor(t *testing.T) {
 	assert.NotNil(t, exp)
 }
 
+func TestProcessorDoesNotFailToBuildExportersWithMultiplePipelines(t *testing.T) {
+	// prepare
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	processorFactory := NewFactory()
+	factories.Processors[typeStr] = processorFactory
+
+	otlpExporterFactory := otlpexporter.NewFactory()
+	factories.Exporters["otlp"] = otlpExporterFactory
+
+	otlpConfig := &otlpexporter.Config{
+		ExporterSettings: config.NewExporterSettings(config.NewComponentID("otlp")),
+		GRPCClientSettings: configgrpc.GRPCClientSettings{
+			Endpoint: "example.com:1234",
+		},
+	}
+
+	otlpTracesExporter, err := otlpExporterFactory.CreateTracesExporter(context.Background(), componenttest.NewNopExporterCreateSettings(), otlpConfig)
+	require.NoError(t, err)
+
+	otlpMetricsExporter, err := otlpExporterFactory.CreateMetricsExporter(context.Background(), componenttest.NewNopExporterCreateSettings(), otlpConfig)
+	require.NoError(t, err)
+
+	host := &mockHost{
+		Host: componenttest.NewNopHost(),
+		GetExportersFunc: func() map[config.DataType]map[config.ComponentID]component.Exporter {
+			return map[config.DataType]map[config.ComponentID]component.Exporter{
+				config.TracesDataType: {
+					config.NewComponentID("otlp/traces"): otlpTracesExporter,
+				},
+				config.MetricsDataType: {
+					config.NewComponentID("otlp/metrics"): otlpMetricsExporter,
+				},
+			}
+		},
+	}
+
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "config_multipipelines.yaml"), factories)
+	assert.NoError(t, err)
+
+	for _, cfg := range cfg.Processors {
+		exp := newProcessor(zap.NewNop(), cfg)
+		err = exp.Start(context.Background(), host)
+		// assert that no error is thrown due to multiple pipelines and exporters not using the routing processor
+		assert.NoError(t, err)
+		assert.NoError(t, exp.Shutdown(context.Background()))
+	}
+}
+
 func TestShutdown(t *testing.T) {
 	// prepare
 	factory := NewFactory()

processor/routingprocessor/router.go

@@ -334,8 +334,8 @@ func (r *router) registerExporters(hostExporters map[config.DataType]map[config.
 		},
 	} {
 		if err := reg.registerFunc(hostExporters[reg.typ]); err != nil {
-			if errors.Is(err, errDefaultExporterNotFound) {
-				r.logger.Warn("can't find the default exporter for the routing processor for this pipeline type. This is OK if you did not specify this processor for that pipeline type",
+			if errors.Is(err, errDefaultExporterNotFound) || errors.Is(err, errExporterNotFound) {
+				r.logger.Warn("can't find the exporter for the routing processor for this pipeline type. This is OK if you did not specify this processor for that pipeline type",
 					zap.Any("pipeline_type", reg.typ),
 					zap.Error(err),
 				)

processor/routingprocessor/testdata/config_multipipelines.yaml

@@ -0,0 +1,31 @@
+receivers:
+  nop:
+
+processors:
+  routing:
+    attribute_source: resource
+    from_attribute: X-Tenant
+    table:
+      - value: acme
+        exporters:
+          - otlp/traces
+
+exporters:
+  otlp/metrics:
+  otlp/traces:
+
+service:
+  pipelines:
+    traces:
+      receivers:
+        - nop
+      processors:
+        - routing
+      exporters:
+        - otlp/traces
+    metrics:
+      receivers:
+        - nop
+      processors: []
+      exporters:
+        - otlp/metrics