[sdk/metric] Exemplar timestamps can come after end time

[dashpole]

Forked from #8077 (review). I'd like to make sure I capture the whole problem and context so we can pick a path forward.

This blocks #8077.

Specification Requirements

The OpenTelemetry specification defines two related timing rules for metric collection and exemplars:

1. Collection End Timestamp (TimeUnixNano):
   From the OpenTelemetry Metrics SDK Specification - MetricReader:

   "The ending timestamp (i.e. TimeUnixNano) MUST always be equal to time the metric data point took effect, which is equal to when MetricReader.Collect was invoked. These rules apply to all metrics, not just those whose point kinds includes an aggregation temporality field."

2. Exemplar Interval Containment:
   From the OpenTelemetry Metrics SDK Specification - ExemplarReservoir:

   "The 'collect' method MUST return accumulated Exemplars. Exemplars are expected to abide by the AggregationTemporality of any metric point they are recorded with. In other words, Exemplars reported against a metric data point SHOULD have occurred within the start/stop timestamps of that point. SDKs are free to decide whether 'collect' should also reset internal storage for delta temporal aggregation collection, or use a more optimal implementation."

These two requirements are hard to satisfy at the same time when measurements are lock-free:

• If TimeUnixNano is set to the time Collect is called, any measurement that happens while Collect is running has a timestamp t_meas > t_collect = t_end.
• If that measurement (or its exemplar) gets picked up in the current collection cycle before the reader finishes swapping/reading data, its timestamp will be after the data point's end time.
• The only way to strictly satisfy both requirements is to block concurrent measurements and exemplar offers while Collect runs, which hurts measurement performance.

---

Problem Description & Current SDK State

In the Go SDK (go.opentelemetry.io/otel/sdk/metric), collection and measurements run concurrently without global locks. This leads to a few edge cases across cumulative and delta metrics:

1. Measurements and exemplars can happen after the reported end timestamp:

   • In all aggregators (deltaSum, deltaHistogram, cumulativeHistogram, lastvalue, etc., as well as the exponential histogram in #8077), t := now() is taken at the start of collect() / ComputeAggregation(), before calling swapHotAndWait() or reading data.
   • Any measurement that starts after t := now() but before swapHotAndWait() finishes is written to the active hot buffer and swapped into the cold buffer for the current cycle.
   • This means the measurement data is included in a data point with end timestamp Time = t, even though the measurement happened after t.
   • Also, the exemplar records its timestamp at Offer() time (t_meas > t), so the exported data point ends up with exemplar.Time > dataPoint.Time.

2. Cumulative aggregations can export exemplars for measurements not yet in the data point:

   • In double-buffered cumulative aggregations (like cumulativeHistogram and doubleBufferedCumulativePoint in #8077), hot/cold delta counters are swapped per-series while the series shares a single persistent FilteredExemplarReservoir.
   • If a measurement happens after swapHotAndWait() but before collectExemplars(&dp.Exemplars, val.res.Collect) runs:
     1. The measurement goes into the new hot buffer (so its count/sum/buckets will only show up in the next collection cycle).
     2. The measurement calls val.res.Offer(...), updating the shared reservoir right away.
     3. val.res.Collect picks up the new exemplar in the current collection cycle.
   • This produces a data point with cumulative counter values from before the measurement, but with an exemplar from the new measurement (e.g. Count = 1, but exemplar value = 2).

3. Cumulative sums can split measurements and exemplars:

   • For sums, sv.n.add(value) and sv.res.Offer(...) run sequentially without locking.
   • If collection runs between add() and Offer(), the measurement value is included in the current sum, but the exemplar is offered after collection and is pushed to the next collection interval (or dropped).

---

Prior Discussions

In #7427, we explicitly accepted that measurements and exemplars could be split across collection intervals to keep the measurement path fast:

// It is possible for collection to race with measurement and observe the
// exemplar in the batch of metrics after the add() for cumulative sums.
// This is an accepted tradeoff to avoid locking during measurement.

Prior to #8077 (review), we haven't discussed cases where exemplar timestamps (or measurement times) come after the interval's end timestamp, or where cumulative instruments export exemplars for measurements that aren't included in the data point yet.

---

Possible Solutions

Option 1: Accept out-of-interval timestamps during races (No code changes / Document behavior)

• Idea: Accept that in rare race conditions around collection boundaries, measurements and exemplar timestamps can fall slightly after the data point's Time (or split across intervals).
• Why this works:
  • The specification uses SHOULD for exemplar interval containment ("Exemplars reported against a metric data point SHOULD have occurred within the start/stop timestamps of that point"), which gives us room to make this tradeoff for performance.
  • Because t := now() is captured at the start of collection, the measurements themselves can also physically happen after t. Adding complex reservoir synchronization doesn't change the fact that measurements happening during collection occur after t.
• Downside: Callers and backends have to tolerate exemplar timestamps that are slightly ahead of data point end times.

Option 2: Double-buffered exemplar reservoirs + MergeableReservoir interface

• Idea:
  • Double-buffer the exemplar reservoirs alongside counter buffers ([2]FilteredExemplarReservoir).
  • Swapping active buffers with swapHotAndWait() swaps both the measurements and the exemplar reservoir together.
  • For cumulative metrics, each collection cycle collects from the cold delta reservoir and merges it into a persistent cumulative reservoir using an optional MergeableReservoir interface:

    type MergeableReservoir[N int64 | float64] interface {
        FilteredExemplarReservoir[N]
        Merge(other FilteredExemplarReservoir[N])
    }

  • In-tree reservoirs implement Merge(); third-party reservoirs fallback gracefully.
• Pros: Fixes the issue where a measurement in cycle $k+1$ has its exemplar exported in cycle $k$.
• Cons: Adds extra memory and state per series. It also doesn't prevent t_meas > t_end when measurements race with the start of Collect.

Option 3: Discard or clamp exemplars outside [StartTime, Time] during collection

• Idea: In collectExemplars, drop or clamp any exemplar where exemplar.Time > dataPoint.Time (or < dataPoint.StartTime).
• Cons: Since the measurements themselves can happen after t during collection, discarding exemplars would drop valid exemplars for measurements that are included in the point. This reduces sample count and biases distributions near collection boundaries. Clamping timestamps distorts the real time of the trace/span.

Option 4: Set the end time at the end of collection for cumulative instruments

• Idea: During Collect, leave the end time unset for all data points on an instrument until after all datapoints have been collected. Only at that point do we set the end time, which ensures it is always after all measurements and exemplars that are included in the distribution.
• Cons: We would have different behavior for cumulatives and deltas. If we did this for deltas, we would also be moving the start time for the next interval backwards, and could then have an exemplar that is earlier than the start time. This is also a bit further from being equal to when MetricReader.Collect was invoked.

---

I'm working on a prototype of the MergeableReservoir so we can better evaluate it.

[dashpole]

exemplar.time > dataPoint.time on cumulative datapoints is possible (and accepted) in most multi-threaded SDKs (Java, Python, C++).

Only .NET avoids this, specifically because it timestamps the end time at the end of collection.

---

Language Comparison

Language	Is exemplar.time > dataPoint.time possible on cumulative points?	Why / Mechanism
Java	YES	Collection captures collectTime at the start of produce(). Cumulative storage iterates over persistent handles without locking against recorders. An offerLongMeasurement() occurring mid-collection records recordTime = clock.now() > collectTime, which is then returned in a datapoint stamped with collectTime.
Python	YES	collection_start_nanos is captured at the beginning of the collection pass. _sample_exemplar() runs outside the aggregation lock. A concurrent measurement offers an exemplar with time_ns() > collection_start_nanos, which is included in the datapoint whose time_unix_nano is collection_start_nanos.
C++	YES	collection_ts is captured at the beginning of Meter::Collect(). Reservoir cells record system_clock::now() on measurement offer. Exemplars collected during the pass retain their measurement timestamp (> collection_ts).
.NET	NO	EndTimeInclusive is set to DateTimeOffset.UtcNow after all series and exemplar reservoirs have been snapshotted and collected in AggregatorStore.Snapshot(). Therefore, exemplar.Timestamp <= dataPoint.EndTime.
JS	NO	Single-threaded execution prevents measurement and collection races.

---

The Anatomy of the Race in Java, Python, and C++

1. At $t_0$: Collection starts. The SDK captures $T_{\text{collect}} = t_0$ (to use as dataPoint.Time).
2. At $t_1$ ($t_1 > t_0$): While collection is in progress (iterating through meters/instruments), a worker thread records a measurement.
   • The measurement offers to the reservoir, recording $T_{\text{meas}} = t_1$.
3. At $t_2$ ($t_2 \ge t_1$): Collection reaches this instrument's reservoir and calls CollectAndReset().
4. Exported Point:
   • dataPoint.Time $= t_0$
   • exemplar.Time $= t_1$
   • Result: exemplar.Time > dataPoint.Time (by $t_1 - t_0$).

---

Based on the above, my recommendation would be either:

• Accept out-of-interval timestamps during races (aligning with Java, Python, and C++), or
• Set the end time at the end of collection for cumulative instruments (aligning with .NET).