plumb selector to reservoir func

dashpole · Oct 15, 2024 · eefbce2 · eefbce2
1 parent 7acaf00
commit eefbce2
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/sdk/metric/exemplar.go b/sdk/metric/exemplar.go
@@ -18,8 +18,7 @@ type ExemplarReservoirProviderSelector func(Aggregation) exemplar.ReservoirProvi
 
 // reservoirFunc returns the appropriately configured exemplar reservoir
 // creation func based on the passed InstrumentKind and filter configuration.
-func reservoirFunc[N int64 | float64](agg Aggregation, filter exemplar.Filter) func(attribute.Set) aggregate.FilteredExemplarReservoir[N] {
-	provider := DefaultExemplarReservoirProviderSelector(agg)
+func reservoirFunc[N int64 | float64](provider exemplar.ReservoirProvider, filter exemplar.Filter) func(attribute.Set) aggregate.FilteredExemplarReservoir[N] {
 	return func(attrs attribute.Set) aggregate.FilteredExemplarReservoir[N] {
 		return aggregate.NewFilteredExemplarReservoir[N](filter, provider(attrs))
 	}

diff --git a/sdk/metric/pipeline.go b/sdk/metric/pipeline.go
@@ -332,6 +332,9 @@ func (i *inserter[N]) cachedAggregator(scope instrumentation.Scope, kind Instrum
 		// The view explicitly requested the default aggregation.
 		stream.Aggregation = DefaultAggregationSelector(kind)
 	}
+	if stream.ExemplarReservoirProviderSelector == nil {
+		stream.ExemplarReservoirProviderSelector = DefaultExemplarReservoirProviderSelector
+	}
 
 	if err := isAggregatorCompatible(kind, stream.Aggregation); err != nil {
 		return nil, 0, fmt.Errorf(
@@ -352,7 +355,7 @@ func (i *inserter[N]) cachedAggregator(scope instrumentation.Scope, kind Instrum
 	cv := i.aggregators.Lookup(normID, func() aggVal[N] {
 		b := aggregate.Builder[N]{
 			Temporality:   i.pipeline.reader.temporality(kind),
-			ReservoirFunc: reservoirFunc[N](stream.Aggregation, i.pipeline.exemplarFilter),
+			ReservoirFunc: reservoirFunc[N](stream.ExemplarReservoirProviderSelector(stream.Aggregation), i.pipeline.exemplarFilter),
 		}
 		b.Filter = stream.AttributeFilter
 		// A value less than or equal to zero will disable the aggregation