[rotuingprocessor] Instrument routing processor with non-routed signals counters.

Author: kovrus

.chloggen/routing-processor.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. filelogreceiver)
+component: processor/routingprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Instrument the routing processor with non-routed spans/metricpoints/logrecords counters (OTel SDK).
+
+# One or more tracking issues related to the change
+issues: [21476]
+
+# (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:

processor/routingprocessor/factory.go

@@ -30,6 +30,15 @@ const (
 	typeStr = "routing"
 	// The stability level of the processor.
 	stability = component.StabilityLevelBeta
+
+	scopeName = "github.com/open-telemetry/opentelemetry-collector-contrib/processor/routingprocessor"
+	nameSep   = "/"
+
+	processorKey             = "processor"
+	metricSep                = "_"
+	nonRoutedSpansKey        = "non_routed_spans"
+	nonRoutedMetricPointsKey = "non_routed_metric_points"
+	nonRoutedLogRecordsKey   = "non_routed_log_records"
 )
 
 // NewFactory creates a factory for the routing processor.
@@ -52,17 +61,17 @@ func createDefaultConfig() component.Config {
 
 func createTracesProcessor(_ context.Context, params processor.CreateSettings, cfg component.Config, nextConsumer consumer.Traces) (processor.Traces, error) {
 	warnIfNotLastInPipeline(nextConsumer, params.Logger)
-	return newTracesProcessor(params.TelemetrySettings, cfg), nil
+	return newTracesProcessor(params.TelemetrySettings, cfg)
 }
 
 func createMetricsProcessor(_ context.Context, params processor.CreateSettings, cfg component.Config, nextConsumer consumer.Metrics) (processor.Metrics, error) {
 	warnIfNotLastInPipeline(nextConsumer, params.Logger)
-	return newMetricProcessor(params.TelemetrySettings, cfg), nil
+	return newMetricProcessor(params.TelemetrySettings, cfg)
 }
 
 func createLogsProcessor(_ context.Context, params processor.CreateSettings, cfg component.Config, nextConsumer consumer.Logs) (processor.Logs, error) {
 	warnIfNotLastInPipeline(nextConsumer, params.Logger)
-	return newLogProcessor(params.TelemetrySettings, cfg), nil
+	return newLogProcessor(params.TelemetrySettings, cfg)
 }
 
 func warnIfNotLastInPipeline(nextConsumer interface{}, logger *zap.Logger) {

processor/routingprocessor/factory_test.go

@@ -31,9 +31,17 @@ import (
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/processor/processorhelper"
 	"go.opentelemetry.io/collector/processor/processortest"
+	"go.opentelemetry.io/otel/metric/noop"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 )
 
+var noopTelemetrySettings = component.TelemetrySettings{
+	TracerProvider: trace.NewNoopTracerProvider(),
+	MeterProvider:  noop.NewMeterProvider(),
+	Logger:         zap.NewNop(),
+}
+
 func TestProcessorGetsCreatedWithValidConfiguration(t *testing.T) {
 	// prepare
 	factory := NewFactory()
@@ -174,7 +182,9 @@ func TestProcessorDoesNotFailToBuildExportersWithMultiplePipelines(t *testing.T)
 			require.NoError(t, err)
 			require.NoError(t, component.UnmarshalConfig(sub, cfg))
 
-			exp := newMetricProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, cfg)
+			exp, err := newMetricProcessor(noopTelemetrySettings, cfg)
+			require.NoError(t, err)
+
 			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)

processor/routingprocessor/go.mod

@@ -12,6 +12,9 @@ require (
 	go.opentelemetry.io/collector/exporter v0.77.0
 	go.opentelemetry.io/collector/exporter/otlpexporter v0.77.0
 	go.opentelemetry.io/collector/pdata v1.0.0-rcv0011
+	go.opentelemetry.io/otel v1.15.1
+	go.opentelemetry.io/otel/metric v0.38.1
+	go.opentelemetry.io/otel/trace v1.15.1
 	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.24.0
 	google.golang.org/grpc v1.55.0
@@ -43,9 +46,6 @@ require (
 	go.opentelemetry.io/collector/featuregate v0.77.0 // indirect
 	go.opentelemetry.io/collector/receiver v0.77.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.41.1 // indirect
-	go.opentelemetry.io/otel v1.15.1 // indirect
-	go.opentelemetry.io/otel/metric v0.38.1 // indirect
-	go.opentelemetry.io/otel/trace v1.15.1 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/exp v0.0.0-20221205204356-47842c84f3db // indirect
 	golang.org/x/net v0.10.0 // indirect

processor/routingprocessor/logs.go

@@ -23,6 +23,8 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/processor"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
@@ -39,12 +41,26 @@ type logProcessor struct {
 
 	extractor extractor
 	router    router[exporter.Logs, ottllog.TransformContext]
+
+	nonRoutedLogRecordsCounter metric.Int64Counter
 }
 
-func newLogProcessor(settings component.TelemetrySettings, config component.Config) *logProcessor {
+func newLogProcessor(settings component.TelemetrySettings, config component.Config) (*logProcessor, error) {
 	cfg := rewriteRoutingEntriesToOTTL(config.(*Config))
 
-	logParser, _ := ottllog.NewParser(common.Functions[ottllog.TransformContext](), settings)
+	logParser, err := ottllog.NewParser(common.Functions[ottllog.TransformContext](), settings)
+	if err != nil {
+		return nil, err
+	}
+
+	meter := settings.MeterProvider.Meter(scopeName + nameSep + "logs")
+	nonRoutedLogRecordsCounter, err := meter.Int64Counter(
+		typeStr+metricSep+processorKey+metricSep+nonRoutedLogRecordsKey,
+		metric.WithDescription("Number of log records that were not routed to some or all exporters"),
+	)
+	if err != nil {
+		return nil, err
+	}
 
 	return &logProcessor{
 		logger: settings.Logger,
@@ -55,8 +71,9 @@ func newLogProcessor(settings component.TelemetrySettings, config component.Conf
 			settings,
 			logParser,
 		),
-		extractor: newExtractor(cfg.FromAttribute, settings.Logger),
-	}
+		extractor:                  newExtractor(cfg.FromAttribute, settings.Logger),
+		nonRoutedLogRecordsCounter: nonRoutedLogRecordsCounter,
+	}, nil
 }
 
 func (p *logProcessor) Start(_ context.Context, host component.Host) error {
@@ -111,6 +128,7 @@ func (p *logProcessor) route(ctx context.Context, l plog.Logs) error {
 					return err
 				}
 				p.group("", groups, p.router.defaultExporters, rlogs)
+				p.recordNonRoutedResourceLogs(ctx, key, rlogs)
 				continue
 			}
 			if !isMatch {
@@ -123,6 +141,7 @@ func (p *logProcessor) route(ctx context.Context, l plog.Logs) error {
 		if matchCount == 0 {
 			// no route conditions are matched, add resource logs to default exporters group
 			p.group("", groups, p.router.defaultExporters, rlogs)
+			p.recordNonRoutedResourceLogs(ctx, "", rlogs)
 		}
 	}
 	for _, g := range groups {
@@ -148,9 +167,34 @@ func (p *logProcessor) group(
 	groups[key] = group
 }
 
+func (p *logProcessor) recordNonRoutedResourceLogs(ctx context.Context, routingKey string, rlogs plog.ResourceLogs) {
+	logRecordsCount := 0
+	sl := rlogs.ScopeLogs()
+	for j := 0; j < sl.Len(); j++ {
+		logRecordsCount += sl.At(j).LogRecords().Len()
+	}
+
+	p.nonRoutedLogRecordsCounter.Add(
+		ctx,
+		int64(logRecordsCount),
+		metric.WithAttributes(
+			attribute.String("routing_key", routingKey),
+		),
+	)
+}
+
 func (p *logProcessor) routeForContext(ctx context.Context, l plog.Logs) error {
 	value := p.extractor.extractFromContext(ctx)
 	exporters := p.router.getExporters(value)
+	if value == "" { // "" is a  key for default exporters
+		p.nonRoutedLogRecordsCounter.Add(
+			ctx,
+			int64(l.LogRecordCount()),
+			metric.WithAttributes(
+				attribute.String("routing_key", p.extractor.fromAttr),
+			),
+		)
+	}
 
 	var errs error
 	for _, e := range exporters {

processor/routingprocessor/logs_test.go

@@ -23,7 +23,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/plog"
-	"go.uber.org/zap"
 	"google.golang.org/grpc/metadata"
 )
 
@@ -38,7 +37,8 @@ func TestLogProcessorCapabilities(t *testing.T) {
 	}
 
 	// test
-	p := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, config)
+	p, err := newLogProcessor(noopTelemetrySettings, config)
+	require.NoError(t, err)
 	require.NotNil(t, p)
 
 	// verify
@@ -56,7 +56,7 @@ func TestLogs_RoutingWorks_Context(t *testing.T) {
 		},
 	})
 
-	exp := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, &Config{
+	exp, err := newLogProcessor(noopTelemetrySettings, &Config{
 		FromAttribute:    "X-Tenant",
 		AttributeSource:  contextAttributeSource,
 		DefaultExporters: []string{"otlp"},
@@ -67,6 +67,8 @@ func TestLogs_RoutingWorks_Context(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
+
 	require.NoError(t, exp.Start(context.Background(), host))
 
 	l := plog.NewLogs()
@@ -115,7 +117,7 @@ func TestLogs_RoutingWorks_ResourceAttribute(t *testing.T) {
 		},
 	})
 
-	exp := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, &Config{
+	exp, err := newLogProcessor(noopTelemetrySettings, &Config{
 		FromAttribute:    "X-Tenant",
 		AttributeSource:  resourceAttributeSource,
 		DefaultExporters: []string{"otlp"},
@@ -126,6 +128,8 @@ func TestLogs_RoutingWorks_ResourceAttribute(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
+
 	require.NoError(t, exp.Start(context.Background(), host))
 
 	t.Run("non default route is properly used", func(t *testing.T) {
@@ -168,7 +172,7 @@ func TestLogs_RoutingWorks_ResourceAttribute_DropsRoutingAttribute(t *testing.T)
 		},
 	})
 
-	exp := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, &Config{
+	exp, err := newLogProcessor(noopTelemetrySettings, &Config{
 		AttributeSource:              resourceAttributeSource,
 		FromAttribute:                "X-Tenant",
 		DropRoutingResourceAttribute: true,
@@ -180,6 +184,8 @@ func TestLogs_RoutingWorks_ResourceAttribute_DropsRoutingAttribute(t *testing.T)
 			},
 		},
 	})
+	require.NoError(t, err)
+
 	require.NoError(t, exp.Start(context.Background(), host))
 
 	l := plog.NewLogs()
@@ -210,7 +216,7 @@ func TestLogs_AreCorrectlySplitPerResourceAttributeRouting(t *testing.T) {
 		},
 	})
 
-	exp := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, &Config{
+	exp, err := newLogProcessor(noopTelemetrySettings, &Config{
 		FromAttribute:    "X-Tenant",
 		AttributeSource:  resourceAttributeSource,
 		DefaultExporters: []string{"otlp"},
@@ -221,6 +227,7 @@ func TestLogs_AreCorrectlySplitPerResourceAttributeRouting(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
 
 	l := plog.NewLogs()
 
@@ -264,7 +271,7 @@ func TestLogsAreCorrectlySplitPerResourceAttributeWithOTTL(t *testing.T) {
 		},
 	})
 
-	exp := newLogProcessor(component.TelemetrySettings{Logger: zap.NewNop()}, &Config{
+	exp, err := newLogProcessor(noopTelemetrySettings, &Config{
 		DefaultExporters: []string{"otlp"},
 		Table: []RoutingTableItem{
 			{
@@ -277,6 +284,7 @@ func TestLogsAreCorrectlySplitPerResourceAttributeWithOTTL(t *testing.T) {
 			},
 		},
 	})
+	require.NoError(t, err)
 
 	require.NoError(t, exp.Start(context.Background(), host))