PoC: allow reservoirs to defer recording the timestamp until storage

sdk/metric/exemplar/fixed_size_reservoir.go

@@ -37,6 +37,7 @@ var _ Reservoir = &FixedSizeReservoir{}
 // additional measurement with a decreasing probability.
 type FixedSizeReservoir struct {
 	reservoir.ConcurrentSafe
+	reservoir.DeferTimestamp
 	*storage
 	mu sync.Mutex
 

sdk/metric/exemplar/histogram_reservoir.go

@@ -41,6 +41,7 @@ var _ Reservoir = &HistogramReservoir{}
 // define by bounds.
 type HistogramReservoir struct {
 	reservoir.ConcurrentSafe
+	reservoir.DeferTimestamp
 	*storage
 
 	// bounds are bucket bounds in ascending order.

sdk/metric/exemplar/storage.go

@@ -26,6 +26,9 @@ func newStorage(n int) *storage {
 }
 
 func (r *storage) store(ctx context.Context, idx int, ts time.Time, v Value, droppedAttr []attribute.KeyValue) {
+	if ts.IsZero() {
+		ts = time.Now()
+	}
 	r.measurements[idx].mux.Lock()
 	defer r.measurements[idx].mux.Unlock()
 	r.measurements[idx].FilteredAttributes = droppedAttr

sdk/metric/internal/aggregate/filtered_reservoir.go

@@ -36,6 +36,10 @@ type filteredExemplarReservoir[N int64 | float64] struct {
 	// reservoir.ConcurrentSafe in order to improve performance.
 	reservoirMux   sync.Mutex
 	concurrentSafe bool
+	// reservoirs can indicate that they would like to record the exemplars'
+	// timestamp by embedding reservoir.DeferTimestamp. The
+	// FilteredExemplarReservoir will pass a zero timestamp in this case.
+	deferTimestamp bool
 }
 
 // NewFilteredExemplarReservoir creates a [FilteredExemplarReservoir] which only offers values
@@ -45,17 +49,22 @@ func NewFilteredExemplarReservoir[N int64 | float64](
 	r exemplar.Reservoir,
 ) FilteredExemplarReservoir[N] {
 	_, concurrentSafe := r.(reservoir.ConcurrentSafe)
+	_, deferTimestamp := r.(reservoir.DeferTimestamp)
 	return &filteredExemplarReservoir[N]{
 		filter:         f,
 		reservoir:      r,
 		concurrentSafe: concurrentSafe,
+		deferTimestamp: deferTimestamp,
 	}
 }
 
 func (f *filteredExemplarReservoir[N]) Offer(ctx context.Context, val N, attr []attribute.KeyValue) {
 	if f.filter(ctx) {
 		// only record the current time if we are sampling this measurement.
-		ts := time.Now()
+		var ts time.Time
+		if !f.deferTimestamp {
+			ts = time.Now()
+		}
 		if !f.concurrentSafe {
 			f.reservoirMux.Lock()
 			defer f.reservoirMux.Unlock()

sdk/metric/internal/reservoir/deferred_timestamp.go

@@ -0,0 +1,10 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package reservoir // import "go.opentelemetry.io/otel/sdk/metric/internal/reservoir"
+
+// DeferTimestamp is an interface that can be embedded in an
+// exemplar.Reservoir to indicate to the SDK that it would like to take control
+// over measuring the current timestamp for performance reasons. The SDK will
+// provide a zero timestamp to reservoirs that embed this interface.
+type DeferTimestamp interface{ deferTimestamp() }