Add invalid_utf8 to reasons spans could be rejected

CHANGELOG.md

@@ -50,6 +50,7 @@
 * [ENHANCEMENT] Add a max flush attempts and metric to the metrics generator [#4254](https://github.com/grafana/tempo/pull/4254) (@joe-elliott)
 * [ENHANCEMENT] Collection of query-frontend changes to reduce allocs. [#4242]https://github.com/grafana/tempo/pull/4242 (@joe-elliott)
 * [ENHANCEMENT] Added `insecure-skip-verify` option in tempo-cli to skip SSL certificate validation when connecting to the S3 backend. [#44236](https://github.com/grafana/tempo/pull/4259) (@faridtmammadov)
+* [ENHANCEMENT] Add `invalid_utf8` to reasons spanmetrics will discard spans. [#4293](https://github.com/grafana/tempo/pull/4293) (@zalegrala)
 * [BUGFIX] Replace hedged requests roundtrips total with a counter. [#4063](https://github.com/grafana/tempo/pull/4063) [#4078](https://github.com/grafana/tempo/pull/4078) (@galalen)
 * [BUGFIX] Metrics generators: Correctly drop from the ring before stopping ingestion to reduce drops during a rollout. [#4101](https://github.com/grafana/tempo/pull/4101) (@joe-elliott)
 * [BUGFIX] Correctly handle 400 Bad Request and 404 Not Found in gRPC streaming [#4144](https://github.com/grafana/tempo/pull/4144) (@mapno)

docs/sources/tempo/troubleshooting/metrics-generator.md

@@ -57,6 +57,8 @@ If spans are regularly exceeding this value you may want to consider reviewing y
 Note that increasing this value allows the generator to consume more spans, but does reduce the accuracy of metrics because spans farther
 away from "now" are included.
 
+Spans could also be discarded if the attributes are not valid UTF-8 characters when those attributes are converted to metric labels.
+
 ### Max active series
 
 The generator protects itself and your remote-write target by having a maximum number of series the generator produces.

integration/e2e/metrics_generator_test.go

@@ -130,6 +130,24 @@ func TestMetricsGenerator(t *testing.T) {
 	})
 	require.NoError(t, err)
 
+	// also send one with an invalid label value
+	err = c.EmitBatch(context.Background(), &thrift.Batch{
+		Process: &thrift.Process{ServiceName: "app"},
+		Spans: []*thrift.Span{
+			{
+				TraceIdLow:    traceIDLow,
+				TraceIdHigh:   traceIDHigh,
+				SpanId:        r.Int63(),
+				ParentSpanId:  parentSpanID,
+				OperationName: "\xff\xff",
+				StartTime:     time.Now().UnixMicro(),
+				Duration:      int64(2 * time.Second / time.Microsecond),
+				Tags:          []*thrift.Tag{{Key: "span.kind", VStr: stringPtr("server")}},
+			},
+		},
+	})
+	require.NoError(t, err)
+
 	// Fetch metrics from Prometheus once they are received
 	var metricFamilies map[string]*io_prometheus_client.MetricFamily
 	for {
@@ -191,8 +209,8 @@ func TestMetricsGenerator(t *testing.T) {
 	assert.Equal(t, 1.0, sumValues(metricFamilies, "traces_spanmetrics_latency_sum", lbls))
 
 	// Verify metrics
-	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(4), "tempo_metrics_generator_spans_received_total"))
-	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(2), "tempo_metrics_generator_spans_discarded_total"))
+	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(5), "tempo_metrics_generator_spans_received_total"))
+	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(3), "tempo_metrics_generator_spans_discarded_total"))
 	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(25), "tempo_metrics_generator_registry_active_series"))
 	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(1000), "tempo_metrics_generator_registry_max_active_series"))
 	assert.NoError(t, tempoMetricsGenerator.WaitSumMetrics(e2e.Equals(25), "tempo_metrics_generator_registry_series_added_total"))

modules/generator/instance.go

@@ -61,6 +61,7 @@ var (
 const (
 	reasonOutsideTimeRangeSlack = "outside_metrics_ingestion_slack"
 	reasonSpanMetricsFiltered   = "span_metrics_filtered"
+	reasonInvalidUTF8           = "invalid_utf8"
 )
 
 type instance struct {
@@ -290,7 +291,8 @@ func (i *instance) addProcessor(processorName string, cfg ProcessorConfig) error
 	switch processorName {
 	case spanmetrics.Name:
 		filteredSpansCounter := metricSpansDiscarded.WithLabelValues(i.instanceID, reasonSpanMetricsFiltered)
-		newProcessor, err = spanmetrics.New(cfg.SpanMetrics, i.registry, filteredSpansCounter)
+		invalidUTF8Counter := metricSpansDiscarded.WithLabelValues(i.instanceID, reasonInvalidUTF8)
+		newProcessor, err = spanmetrics.New(cfg.SpanMetrics, i.registry, filteredSpansCounter, invalidUTF8Counter)
 		if err != nil {
 			return err
 		}

modules/generator/processor/spanmetrics/spanmetrics.go

@@ -2,7 +2,9 @@ package spanmetrics
 
 import (
 	"context"
+	"fmt"
 	"time"
+	"unicode/utf8"
 
 	"github.com/prometheus/prometheus/util/strutil"
 	"go.opentelemetry.io/otel"
@@ -40,12 +42,13 @@ type Processor struct {
 
 	filter               *spanfilter.SpanFilter
 	filteredSpansCounter prometheus.Counter
+	invalidUTF8Counter   prometheus.Counter
 
 	// for testing
 	now func() time.Time
 }
 
-func New(cfg Config, reg registry.Registry, spanDiscardCounter prometheus.Counter) (gen.Processor, error) {
+func New(cfg Config, reg registry.Registry, filteredSpansCounter, invalidUTF8Counter prometheus.Counter) (gen.Processor, error) {
 	labels := make([]string, 0, 4+len(cfg.Dimensions))
 
 	if cfg.IntrinsicDimensions.Service {
@@ -72,13 +75,19 @@ func New(cfg Config, reg registry.Registry, spanDiscardCounter prometheus.Counte
 		labels = append(labels, sanitizeLabelNameWithCollisions(m.Name))
 	}
 
+	err := validateLabelValues(labels)
+	if err != nil {
+		return nil, err
+	}
+
 	p := &Processor{
 		Cfg:                   cfg,
 		registry:              reg,
 		spanMetricsTargetInfo: reg.NewGauge(targetInfo),
 		now:                   time.Now,
 		labels:                labels,
-		filteredSpansCounter:  spanDiscardCounter,
+		filteredSpansCounter:  filteredSpansCounter,
+		invalidUTF8Counter:    invalidUTF8Counter,
 	}
 
 	if cfg.Subprocessors[Latency] {
@@ -96,7 +105,6 @@ func New(cfg Config, reg registry.Registry, spanDiscardCounter prometheus.Counte
 		return nil, err
 	}
 
-	p.filteredSpansCounter = spanDiscardCounter
 	p.filter = filter
 	return p, nil
 }
@@ -203,6 +211,12 @@ func (p *Processor) aggregateMetricsForSpan(svcName string, jobName string, inst
 
 	spanMultiplier := processor_util.GetSpanMultiplier(p.Cfg.SpanMultiplierKey, span, rs)
 
+	err := validateLabelValues(labelValues)
+	if err != nil {
+		p.invalidUTF8Counter.Inc()
+		return
+	}
+
 	registryLabelValues := p.registry.NewLabelValueCombo(labels, labelValues)
 
 	if p.Cfg.Subprocessors[Count] {
@@ -259,6 +273,16 @@ func sanitizeLabelNameWithCollisions(name string) string {
 	return sanitized
 }
 
+func validateLabelValues(v []string) error {
+	for _, value := range v {
+		if !utf8.ValidString(value) {
+			return fmt.Errorf("invalid utf8 string: %s", value)
+		}
+	}
+
+	return nil
+}
+
 func isIntrinsicDimension(name string) bool {
 	return processor_util.Contains(name, []string{dimJob, dimSpanName, dimSpanKind, dimStatusCode, dimStatusMessage, dimInstance})
 }