Update processor replacement logic for native histograms (#3923)

* Update processor replacement logic for native histograms

* Spell correctly

* Sort results for testing

* Drop duplicate Store and update notes

* Ensure no changes when no changes

* Mvoe the overrides desire to the processor config

modules/generator/config.go

@@ -116,6 +116,11 @@ func (cfg *ProcessorConfig) copyWithOverrides(o metricsGeneratorOverrides, userI
 		copyCfg.LocalBlocks.CompleteBlockTimeout = timeout
 	}
 
+	if histograms := o.MetricsGeneratorGenerateNativeHistograms(userID); histograms != "" {
+		copyCfg.ServiceGraphs.HistogramOverride = histograms
+		copyCfg.SpanMetrics.HistogramOverride = histograms
+	}
+
 	copyCfg.SpanMetrics.DimensionMappings = o.MetricsGeneratorProcessorSpanMetricsDimensionMappings(userID)
 
 	copyCfg.SpanMetrics.EnableTargetInfo = o.MetricsGeneratorProcessorSpanMetricsEnableTargetInfo(userID)

modules/generator/instance_test.go

@@ -283,6 +283,64 @@ func Test_instance_updateProcessors(t *testing.T) {
 
 		assert.Equal(t, expectedProcessors, actualProcessors)
 	})
+
+	t.Run("replace span-metrics and servicegraphs processors when histograms impementation changes", func(t *testing.T) {
+		overrides.nativeHistograms = "native"
+		overrides.processors = map[string]struct{}{
+			servicegraphs.Name: {},
+			spanmetrics.Name:   {},
+		}
+		err := instance.updateProcessors()
+		assert.NoError(t, err)
+
+		assertHistogramsReload := func(t *testing.T) {
+			desiredProcessors := instance.overrides.MetricsGeneratorProcessors(instance.instanceID)
+			desiredCfg, copyErr := instance.cfg.Processor.copyWithOverrides(instance.overrides, instance.instanceID)
+			assert.NoError(t, copyErr)
+			toAdd, toRemove, toReplace, diffErr := instance.diffProcessors(desiredProcessors, desiredCfg)
+			assert.NoError(t, diffErr)
+			assert.Empty(t, toAdd)
+			assert.Empty(t, toRemove)
+
+			sort.Strings(toReplace)
+			assert.Equal(t, []string{servicegraphs.Name, spanmetrics.Name}, toReplace)
+		}
+
+		assertHistogramsNoChange := func(t *testing.T) {
+			desiredProcessors := instance.overrides.MetricsGeneratorProcessors(instance.instanceID)
+			desiredCfg, copyErr := instance.cfg.Processor.copyWithOverrides(instance.overrides, instance.instanceID)
+			assert.NoError(t, copyErr)
+			toAdd, toRemove, toReplace, diffErr := instance.diffProcessors(desiredProcessors, desiredCfg)
+			assert.NoError(t, diffErr)
+			assert.Empty(t, toAdd)
+			assert.Empty(t, toRemove)
+			assert.Empty(t, toReplace)
+		}
+
+		// Downgrade to classic
+		overrides.nativeHistograms = "classic"
+		assertHistogramsReload(t)
+
+		err = instance.updateProcessors()
+		assert.NoError(t, err)
+		assertHistogramsNoChange(t)
+
+		// Upgrade to both native and classic
+		overrides.nativeHistograms = "both"
+		assertHistogramsReload(t)
+
+		err = instance.updateProcessors()
+		assert.NoError(t, err)
+		assertHistogramsNoChange(t)
+
+		// Upgrade back to native
+		overrides.nativeHistograms = "native"
+		assertHistogramsReload(t)
+
+		err = instance.updateProcessors()
+		assert.NoError(t, err)
+		assertHistogramsNoChange(t)
+	})
 }
 
 type noopStorage struct{}

modules/generator/overrides_test.go

@@ -32,6 +32,7 @@ type mockOverrides struct {
 	dedicatedColumns                                   backend.DedicatedColumns
 	maxBytesPerTrace                                   int
 	unsafeQueryHints                                   bool
+	nativeHistograms                                   string
 }
 
 var _ metricsGeneratorOverrides = (*mockOverrides)(nil)
@@ -57,7 +58,7 @@ func (m *mockOverrides) MetricsGeneratorDisableCollection(string) bool {
 }
 
 func (m *mockOverrides) MetricsGeneratorGenerateNativeHistograms(string) string {
-	return ""
+	return m.nativeHistograms
 }
 
 func (m *mockOverrides) MetricsGenerationTraceIDLabelName(string) string {

modules/generator/processor/servicegraphs/config.go

@@ -23,6 +23,9 @@ type Config struct {
 	// Buckets for latency histogram in seconds.
 	HistogramBuckets []float64 `yaml:"histogram_buckets"`
 
+	// The histogram implementation to select.
+	HistogramOverride string `yaml:"-"`
+
 	// Additional dimensions (labels) to be added to the metric along with the default ones.
 	// If client and server spans have the same attribute and EnableClientServerPrefix is not enabled,
 	// behaviour is undetermined (either value could get used)

modules/generator/processor/servicegraphs/servicegraphs.go

@@ -118,9 +118,9 @@ func New(cfg Config, tenant string, registry registry.Registry, logger log.Logge
 
 		serviceGraphRequestTotal:                           registry.NewCounter(metricRequestTotal),
 		serviceGraphRequestFailedTotal:                     registry.NewCounter(metricRequestFailedTotal),
-		serviceGraphRequestServerSecondsHistogram:          registry.NewHistogram(metricRequestServerSeconds, cfg.HistogramBuckets),
-		serviceGraphRequestClientSecondsHistogram:          registry.NewHistogram(metricRequestClientSeconds, cfg.HistogramBuckets),
-		serviceGraphRequestMessagingSystemSecondsHistogram: registry.NewHistogram(metricRequestMessagingSystemSeconds, cfg.HistogramBuckets),
+		serviceGraphRequestServerSecondsHistogram:          registry.NewHistogram(metricRequestServerSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
+		serviceGraphRequestClientSecondsHistogram:          registry.NewHistogram(metricRequestClientSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
+		serviceGraphRequestMessagingSystemSecondsHistogram: registry.NewHistogram(metricRequestMessagingSystemSeconds, cfg.HistogramBuckets, cfg.HistogramOverride),
 
 		metricDroppedSpans: metricDroppedSpans.WithLabelValues(tenant),
 		metricTotalEdges:   metricTotalEdges.WithLabelValues(tenant),

modules/generator/processor/spanmetrics/config.go

@@ -24,15 +24,22 @@ const (
 type Config struct {
 	// Buckets for latency histogram in seconds.
 	HistogramBuckets []float64 `yaml:"histogram_buckets"`
+
+	// The histogram implementation to select.
+	HistogramOverride string `yaml:"-"`
+
 	// Intrinsic dimensions (labels) added to the metric, that are generated from fixed span
 	// data. The dimensions service, span_name, span_kind, status_code, job and instance are enabled by
 	// default, whereas the dimension status_message must be enabled explicitly.
 	IntrinsicDimensions IntrinsicDimensions `yaml:"intrinsic_dimensions"`
+
 	// Additional dimensions (labels) to be added to the metric. The dimensions are generated
 	// from span attributes and are created along with the intrinsic dimensions.
 	Dimensions []string `yaml:"dimensions"`
+
 	// Dimension label mapping to allow the user to rename attributes in their metrics
 	DimensionMappings []sharedconfig.DimensionMappings `yaml:"dimension_mappings"`
+
 	// Enable target_info as a metrics
 	EnableTargetInfo bool `yaml:"enable_target_info"`
 

modules/generator/processor/spanmetrics/spanmetrics.go

@@ -80,7 +80,7 @@ func New(cfg Config, registry registry.Registry, spanDiscardCounter prometheus.C
 	}
 
 	if cfg.Subprocessors[Latency] {
-		p.spanMetricsDurationSeconds = registry.NewHistogram(metricDurationSeconds, cfg.HistogramBuckets)
+		p.spanMetricsDurationSeconds = registry.NewHistogram(metricDurationSeconds, cfg.HistogramBuckets, cfg.HistogramOverride)
 	}
 	if cfg.Subprocessors[Count] {
 		p.spanMetricsCallsTotal = registry.NewCounter(metricCallsTotal)

modules/generator/registry/interface.go

@@ -4,7 +4,7 @@ package registry
 type Registry interface {
 	NewLabelValueCombo(labels []string, values []string) *LabelValueCombo
 	NewCounter(name string) Counter
-	NewHistogram(name string, buckets []float64) Histogram
+	NewHistogram(name string, buckets []float64, histogramOverride string) Histogram
 	NewGauge(name string) Gauge
 }
 

modules/generator/registry/native_histogram.go

@@ -35,10 +35,10 @@ type nativeHistogram struct {
 
 	traceIDLabelName string
 
-	// Returns "native" or "both", to include classic histograms.  We wrap the
-	// ovverrides with this func so that we can update the implementation after
-	// the nativeHistogram has been created.
-	modeFunc func() string
+	// Can be "native", classic", "both" to determine which histograms to
+	// generate.  A diff in the configured value on the processors will cause a
+	// reload of the process, and a new instance of the histogram to be created.
+	histogramOverride string
 }
 
 type nativeHistogramSeries struct {
@@ -54,7 +54,7 @@ var (
 	_ metric    = (*nativeHistogram)(nil)
 )
 
-func newNativeHistogram(name string, buckets []float64, onAddSeries func(uint32) bool, onRemoveSeries func(count uint32), traceIDLabelName string, histogramsModeFunc func() string) *nativeHistogram {
+func newNativeHistogram(name string, buckets []float64, onAddSeries func(uint32) bool, onRemoveSeries func(count uint32), traceIDLabelName string, histogramOverride string) *nativeHistogram {
 	if onAddSeries == nil {
 		onAddSeries = func(uint32) bool {
 			return true
@@ -64,24 +64,22 @@ func newNativeHistogram(name string, buckets []float64, onAddSeries func(uint32)
 		onRemoveSeries = func(uint32) {}
 	}
 
-	if histogramsModeFunc == nil {
-		histogramsModeFunc = func() string {
-			return "native"
-		}
+	if histogramOverride == "" {
+		histogramOverride = "native"
 	}
 
 	if traceIDLabelName == "" {
 		traceIDLabelName = "traceID"
 	}
 
 	return &nativeHistogram{
-		metricName:       name,
-		series:           make(map[uint64]*nativeHistogramSeries),
-		onAddSerie:       onAddSeries,
-		onRemoveSerie:    onRemoveSeries,
-		traceIDLabelName: traceIDLabelName,
-		buckets:          buckets,
-		modeFunc:         histogramsModeFunc,
+		metricName:        name,
+		series:            make(map[uint64]*nativeHistogramSeries),
+		onAddSerie:        onAddSeries,
+		onRemoveSerie:     onRemoveSeries,
+		traceIDLabelName:  traceIDLabelName,
+		buckets:           buckets,
+		histogramOverride: histogramOverride,
 	}
 }
 
@@ -176,18 +174,8 @@ func (h *nativeHistogram) collectMetrics(appender storage.Appender, timeMs int64
 		// the same, and so Reset() call below can be used to clear the exemplars.
 		s.histogram = encodedMetric.GetHistogram()
 
-		// NOTE: A subtle point here is that the new histogram implementation is
-		// chosen in the registry.  This means that if the user overrides change
-		// from "both" to "classic", we will still be using the new histograms
-		// implementation. This is because the registry is not recreated when the
-		// overrides change. This is a tradeoff to avoid recreating the registry
-		// for every change in the overrides, but it does mean that even if user
-		// changes to "classic" histograms, we will still be using the new
-		// histograms implementation and want to avoid generating native
-		// histograms.
-
 		// If we are in "both" or "classic" mode, also emit classic histograms.
-		if overrides.HasClassicHistograms(h.modeFunc()) {
+		if overrides.HasClassicHistograms(h.histogramOverride) {
 			classicSeries, classicErr := h.classicHistograms(appender, lb, timeMs, s)
 			if classicErr != nil {
 				return activeSeries, classicErr
@@ -196,7 +184,7 @@ func (h *nativeHistogram) collectMetrics(appender storage.Appender, timeMs int64
 		}
 
 		// If we are in "both" or "native" mode, also emit native histograms.
-		if overrides.HasNativeHistograms(h.modeFunc()) {
+		if overrides.HasNativeHistograms(h.histogramOverride) {
 			nativeErr := h.nativeHistograms(appender, lb, timeMs, s)
 			if nativeErr != nil {
 				return activeSeries, nativeErr

modules/generator/registry/native_histogram_test.go

@@ -20,11 +20,7 @@ func Test_ObserveWithExemplar_duplicate(t *testing.T) {
 		return true
 	}
 
-	f := func() string {
-		return "both"
-	}
-
-	h := newNativeHistogram("my_histogram", []float64{0.1, 0.2}, onAdd, nil, "trace_id", f)
+	h := newNativeHistogram("my_histogram", []float64{0.1, 0.2}, onAdd, nil, "trace_id", "both")
 
 	lv := newLabelValueCombo([]string{"label"}, []string{"value-1"})
 
@@ -59,7 +55,6 @@ func Test_Histograms(t *testing.T) {
 		collections collections
 		// native histogram does not support all features yet
 		skipNativeHistogram bool
-		modeFunc            func() string
 	}{
 		{
 			name:    "single collection single observation",
@@ -459,20 +454,13 @@ func Test_Histograms(t *testing.T) {
 					t.SkipNow()
 				}
 
-				modeFunc := tc.modeFunc
-				if modeFunc == nil {
-					modeFunc = func() string {
-						return "both"
-					}
-				}
-
 				var seriesAdded int
 				onAdd := func(count uint32) bool {
 					seriesAdded += int(count)
 					return true
 				}
 
-				h := newNativeHistogram("test_histogram", tc.buckets, onAdd, nil, "trace_id", modeFunc)
+				h := newNativeHistogram("test_histogram", tc.buckets, onAdd, nil, "trace_id", "both")
 				testHistogram(t, h, tc.collections)
 			})
 		})

modules/generator/registry/registry.go

@@ -148,19 +148,13 @@ func (r *ManagedRegistry) NewCounter(name string) Counter {
 	return c
 }
 
-func (r *ManagedRegistry) NewHistogram(name string, buckets []float64) (h Histogram) {
+func (r *ManagedRegistry) NewHistogram(name string, buckets []float64, histogramOverride string) (h Histogram) {
 	traceIDLabelName := r.overrides.MetricsGenerationTraceIDLabelName(r.tenant)
 
-	histograms := r.overrides.MetricsGeneratorGenerateNativeHistograms(r.tenant)
-
-	histogramsModeFunc := func() string {
-		return r.overrides.MetricsGeneratorGenerateNativeHistograms(r.tenant)
-	}
-
-	// Temporary switch: use the old implementation when native histograms are
-	// disabled, eventually the new implementation can handle all cases
-	if overrides.HasNativeHistograms(histograms) {
-		h = newNativeHistogram(name, buckets, r.onAddMetricSeries, r.onRemoveMetricSeries, traceIDLabelName, histogramsModeFunc)
+	// TODO: Temporary switch: use the old implementation when native histograms
+	// are disabled, eventually the new implementation can handle all cases
+	if overrides.HasNativeHistograms(histogramOverride) {
+		h = newNativeHistogram(name, buckets, r.onAddMetricSeries, r.onRemoveMetricSeries, traceIDLabelName, histogramOverride)
 	} else {
 		h = newHistogram(name, buckets, r.onAddMetricSeries, r.onRemoveMetricSeries, traceIDLabelName)
 	}

modules/generator/registry/registry_test.go

@@ -79,7 +79,7 @@ func TestManagedRegistry_histogram(t *testing.T) {
 	registry := New(&Config{}, &mockOverrides{}, "test", appender, log.NewNopLogger())
 	defer registry.Close()
 
-	histogram := registry.NewHistogram("histogram", []float64{1.0, 2.0})
+	histogram := registry.NewHistogram("histogram", []float64{1.0, 2.0}, "classic")
 
 	histogram.ObserveWithExemplar(newLabelValueCombo([]string{"label"}, []string{"value-1"}), 1.0, "", 1.0)
 
@@ -228,7 +228,7 @@ func TestManagedRegistry_maxLabelNameLength(t *testing.T) {
 	defer registry.Close()
 
 	counter := registry.NewCounter("counter")
-	histogram := registry.NewHistogram("histogram", []float64{1.0})
+	histogram := registry.NewHistogram("histogram", []float64{1.0}, "classic")
 
 	counter.Inc(registry.NewLabelValueCombo([]string{"very_lengthy_label"}, []string{"very_length_value"}), 1.0)
 	histogram.ObserveWithExemplar(registry.NewLabelValueCombo([]string{"another_very_lengthy_label"}, []string{"another_very_lengthy_value"}), 1.0, "", 1.0)
@@ -292,12 +292,17 @@ func TestHistogramOverridesConfig(t *testing.T) {
 		t.Run(c.name, func(t *testing.T) {
 			appender := &capturingAppender{}
 			overrides := &mockOverrides{
-				generateNativeHistograms: c.nativeHistogramMode,
+
+				// TODO: Review this test.  Since we no longer switch on the overrides,
+				// this is only testing the New call returns the correct implementation
+				// based on the received string.  We might have coverage elsewhere.
+
+				// generateNativeHistograms: c.nativeHistogramMode,
 			}
 			registry := New(&Config{}, overrides, "test", appender, log.NewNopLogger())
 			defer registry.Close()
 
-			tt := registry.NewHistogram("histogram", []float64{1.0, 2.0})
+			tt := registry.NewHistogram("histogram", []float64{1.0, 2.0}, c.nativeHistogramMode)
 			require.IsType(t, c.typeOfHistogram, tt)
 		})
 	}

modules/generator/registry/test.go

@@ -42,13 +42,14 @@ func (t *TestRegistry) NewLabelValueCombo(labels []string, values []string) *Lab
 	return newLabelValueCombo(labels, values)
 }
 
-func (t *TestRegistry) NewHistogram(name string, buckets []float64) Histogram {
+func (t *TestRegistry) NewHistogram(name string, buckets []float64, histogramOverrides string) Histogram {
 	return &testHistogram{
-		nameSum:    name + "_sum",
-		nameCount:  name + "_count",
-		nameBucket: name + "_bucket",
-		buckets:    buckets,
-		registry:   t,
+		nameSum:            name + "_sum",
+		nameCount:          name + "_count",
+		nameBucket:         name + "_bucket",
+		buckets:            buckets,
+		registry:           t,
+		histogramOverrides: histogramOverrides,
 	}
 }
 
@@ -164,11 +165,12 @@ func (t *testGauge) removeStaleSeries(int64) {
 }
 
 type testHistogram struct {
-	nameSum    string
-	nameCount  string
-	nameBucket string
-	buckets    []float64
-	registry   *TestRegistry
+	nameSum            string
+	nameCount          string
+	nameBucket         string
+	buckets            []float64
+	registry           *TestRegistry
+	histogramOverrides string
 }
 
 var (

modules/generator/registry/test_test.go

@@ -28,7 +28,7 @@ func TestTestRegistry_counter(t *testing.T) {
 func TestTestRegistry_histogram(t *testing.T) {
 	testRegistry := NewTestRegistry()
 
-	histogram := testRegistry.NewHistogram("histogram", []float64{1.0, 2.0})
+	histogram := testRegistry.NewHistogram("histogram", []float64{1.0, 2.0}, "classic")
 
 	labelValues := newLabelValueCombo([]string{"foo", "bar"}, []string{"foo-value", "bar-value"})
 	histogram.ObserveWithExemplar(labelValues, 1.0, "", 1.0)