fix: observable instruments should not export stale series (#5950)

   Per the OpenTelemetry spec, MetricReader.Collect MUST only receive data
   points with measurements recorded since the previous collection for
   asynchronous instruments. The SDK was re-exporting the last known value
   indefinitely when an observable callback stopped reporting a series.

   Fixes #5950

Author: EliaSaSe

src/OpenTelemetry/CHANGELOG.md

@@ -6,6 +6,12 @@ Notes](../../RELEASENOTES.md).
 
 ## Unreleased
 
+* Fixed observable instruments (ObservableCounter, ObservableUpDownCounter,
+  ObservableGauge) continuing to export stale data points after a callback
+  stops reporting a series. Per the spec, only measurements recorded since
+  the previous collection should be exported for asynchronous instruments.
+  ([#5950](https://github.com/open-telemetry/opentelemetry-dotnet/issues/5950))
+
 ## 1.15.0
 
 Released 2026-Jan-21

src/OpenTelemetry/Metrics/AggregatorStore.cs

@@ -20,6 +20,7 @@ internal sealed class AggregatorStore
     internal readonly HashSet<string>? TagKeysInteresting;
 #endif
     internal readonly bool OutputDelta;
+    internal readonly bool IsAsynchronous;
     internal readonly int NumberOfMetricPoints;
     internal readonly ConcurrentDictionary<Tags, LookupData>? TagsToMetricPointIndexDictionaryDelta;
     internal readonly Func<ExemplarReservoir?>? ExemplarReservoirFactory;
@@ -74,6 +75,7 @@ internal AggregatorStore(
         this.currentMetricPointBatch = new int[this.NumberOfMetricPoints];
         this.aggType = aggType;
         this.OutputDelta = temporality == AggregationTemporality.Delta;
+        this.IsAsynchronous = metricStreamIdentity.IsAsynchronous;
         this.histogramExplicitBounds = new(metricStreamIdentity.HistogramBucketBounds ?? FindDefaultHistogramBounds(in metricStreamIdentity));
         this.exponentialHistogramMaxSize = metricStreamIdentity.ExponentialHistogramMaxSize;
         this.exponentialHistogramMaxScale = metricStreamIdentity.ExponentialHistogramMaxScale;
@@ -288,6 +290,14 @@ internal void SnapshotCumulative(int indexSnapshot)
                 continue;
             }
 
+            // For asynchronous instruments, only export points observed this cycle.
+            // Synchronous instruments carry forward all cumulative values.
+            if (this.IsAsynchronous
+                && metricPoint.MetricPointStatus == MetricPointStatus.NoCollectPending)
+            {
+                continue;
+            }
+
             this.TakeMetricPointSnapshot(ref metricPoint, outputDelta: false);
 
             this.currentMetricPointBatch[this.batchSize] = i;

src/OpenTelemetry/Metrics/MetricPoint/MetricPoint.cs

@@ -637,6 +637,13 @@ internal void TakeSnapshot(bool outputDelta)
                     else
                     {
                         this.snapshotValue.AsLong = Interlocked.Read(ref this.runningValue.AsLong);
+
+                        // For asynchronous instruments, reset status so that points
+                        // not reported in the next callback are treated as stale.
+                        if (this.aggregatorStore.IsAsynchronous)
+                        {
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+                        }
                     }
 
                     break;
@@ -662,6 +669,13 @@ internal void TakeSnapshot(bool outputDelta)
                     else
                     {
                         this.snapshotValue.AsDouble = InterlockedHelper.Read(ref this.runningValue.AsDouble);
+
+                        // For asynchronous instruments, reset status so that points
+                        // not reported in the next callback are treated as stale.
+                        if (this.aggregatorStore.IsAsynchronous)
+                        {
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+                        }
                     }
 
                     break;

src/OpenTelemetry/Metrics/MetricStreamIdentity.cs

@@ -129,6 +129,26 @@ public MetricStreamIdentity(Instrument instrument, MetricStreamConfiguration? me
         || this.InstrumentType == typeof(Histogram<float>)
         || this.InstrumentType == typeof(Histogram<double>);
 
+    public bool IsAsynchronous =>
+        this.InstrumentType == typeof(ObservableCounter<long>)
+        || this.InstrumentType == typeof(ObservableCounter<int>)
+        || this.InstrumentType == typeof(ObservableCounter<short>)
+        || this.InstrumentType == typeof(ObservableCounter<byte>)
+        || this.InstrumentType == typeof(ObservableCounter<float>)
+        || this.InstrumentType == typeof(ObservableCounter<double>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<long>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<int>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<short>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<byte>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<float>)
+        || this.InstrumentType == typeof(ObservableUpDownCounter<double>)
+        || this.InstrumentType == typeof(ObservableGauge<long>)
+        || this.InstrumentType == typeof(ObservableGauge<int>)
+        || this.InstrumentType == typeof(ObservableGauge<short>)
+        || this.InstrumentType == typeof(ObservableGauge<byte>)
+        || this.InstrumentType == typeof(ObservableGauge<float>)
+        || this.InstrumentType == typeof(ObservableGauge<double>);
+
     public static bool operator ==(MetricStreamIdentity metricIdentity1, MetricStreamIdentity metricIdentity2) => metricIdentity1.Equals(metricIdentity2);
 
     public static bool operator !=(MetricStreamIdentity metricIdentity1, MetricStreamIdentity metricIdentity2) => !metricIdentity1.Equals(metricIdentity2);

test/OpenTelemetry.Tests/Metrics/MetricApiTests.cs

@@ -1062,13 +1062,11 @@ public void ObservableUpDownCounterAggregationTest(bool exportDelta)
         Assert.Equal(30, sumReceived);
     }
 
-    [Theory(Skip = "Known issue. See https://github.com/open-telemetry/opentelemetry-dotnet/issues/5950")]
+    [Theory]
     [InlineData(MetricReaderTemporalityPreference.Delta)]
     [InlineData(MetricReaderTemporalityPreference.Cumulative)]
     public void ObservableUpDownCounterReportsActiveMeasurementsOnlyTest(MetricReaderTemporalityPreference temporality)
     {
-        // dotnet test --filter "FullyQualifiedName~ObservableUpDownCounterReportsActiveMeasurementsOnlyTest" --framework net10.0
-        // Testing
         // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#metricreader
         // For asynchronous instruments with Delta or Cumulative aggregation
         // temporality, MetricReader.Collect MUST only receive data points with
@@ -1169,6 +1167,258 @@ public void ObservableUpDownCounterReportsActiveMeasurementsOnlyTest(MetricReade
         Assert.Empty(exportedItems);
     }
 
+    [Theory]
+    [InlineData(MetricReaderTemporalityPreference.Delta)]
+    [InlineData(MetricReaderTemporalityPreference.Cumulative)]
+    public void ObservableGaugeReportsActiveMeasurementsOnlyTest(MetricReaderTemporalityPreference temporality)
+    {
+        // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#metricreader
+        // For asynchronous instruments with Delta or Cumulative aggregation
+        // temporality, MetricReader.Collect MUST only receive data points with
+        // measurements recorded since the previous collection. These rules
+        // apply to all metrics, not just those whose point kinds includes an
+        // aggregation temporality field.
+
+        var exportedItems = new List<Metric>();
+        var tags1 = new List<KeyValuePair<string, object?>>
+        {
+            new("key", "value1"),
+        };
+
+        var tags2 = new List<KeyValuePair<string, object?>>
+        {
+            new("key", "value2"),
+        };
+
+        int callbackInvocationCount = 0;
+
+        using var meter = new Meter($"{Utils.GetCurrentMethodName()}.{temporality}");
+        var gauge = meter.CreateObservableGauge(
+            "observable-gauge",
+            () =>
+            {
+                callbackInvocationCount++;
+                if (callbackInvocationCount == 1)
+                {
+                    return new List<Measurement<long>>
+                    {
+                        new(10L, tags1),
+                        new(20L, tags2),
+                    };
+                }
+                else if (callbackInvocationCount == 2)
+                {
+                    return new List<Measurement<long>>
+                    {
+                        new(30L, tags1),
+                    };
+                }
+                else
+                {
+                    return new List<Measurement<long>>();
+                }
+            });
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .AddInMemoryExporter(exportedItems, metricReaderOptions =>
+            {
+                metricReaderOptions.TemporalityPreference = temporality;
+            }));
+
+        // Export 1: Should get both time series
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Single(exportedItems);
+        var metric = exportedItems[0];
+        Assert.Equal("observable-gauge", metric.Name);
+        List<MetricPoint> metricPoints = [];
+        foreach (ref readonly var mp in metric.GetMetricPoints())
+        {
+            metricPoints.Add(mp);
+        }
+
+        Assert.Equal(2, metricPoints.Count);
+
+        var metricPoint1 = metricPoints[0];
+        Assert.Equal(10, metricPoint1.GetGaugeLastValueLong());
+        ValidateMetricPointTags(tags1, metricPoint1.Tags);
+
+        var metricPoint2 = metricPoints[1];
+        Assert.Equal(20, metricPoint2.GetGaugeLastValueLong());
+        ValidateMetricPointTags(tags2, metricPoint2.Tags);
+
+        // Export 2: Should get only tags1
+        exportedItems.Clear();
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Single(exportedItems);
+        metric = exportedItems[0];
+        Assert.Equal("observable-gauge", metric.Name);
+        metricPoints.Clear();
+        foreach (ref readonly var mp in metric.GetMetricPoints())
+        {
+            metricPoints.Add(mp);
+        }
+
+        Assert.Single(metricPoints);
+        metricPoint1 = metricPoints[0];
+        Assert.Equal(30, metricPoint1.GetGaugeLastValueLong());
+        ValidateMetricPointTags(tags1, metricPoint1.Tags);
+
+        // Export 3: Should get nothing
+        exportedItems.Clear();
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Empty(exportedItems);
+    }
+
+    [Theory]
+    [InlineData(MetricReaderTemporalityPreference.Delta)]
+    [InlineData(MetricReaderTemporalityPreference.Cumulative)]
+    public void ObservableCounterReportsActiveMeasurementsOnlyTest(MetricReaderTemporalityPreference temporality)
+    {
+        // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#metricreader
+        // For asynchronous instruments with Delta or Cumulative aggregation
+        // temporality, MetricReader.Collect MUST only receive data points with
+        // measurements recorded since the previous collection. These rules
+        // apply to all metrics, not just those whose point kinds includes an
+        // aggregation temporality field.
+
+        var exportedItems = new List<Metric>();
+        var tags1 = new List<KeyValuePair<string, object?>>
+        {
+            new("key", "value1"),
+        };
+
+        var tags2 = new List<KeyValuePair<string, object?>>
+        {
+            new("key", "value2"),
+        };
+
+        int callbackInvocationCount = 0;
+
+        using var meter = new Meter($"{Utils.GetCurrentMethodName()}.{temporality}");
+        var counter = meter.CreateObservableCounter(
+            "observable-counter",
+            () =>
+            {
+                callbackInvocationCount++;
+                if (callbackInvocationCount == 1)
+                {
+                    return new List<Measurement<long>>
+                    {
+                        new(10L, tags1),
+                        new(10L, tags2),
+                    };
+                }
+                else if (callbackInvocationCount == 2)
+                {
+                    return new List<Measurement<long>>
+                    {
+                        new(20L, tags1),
+                    };
+                }
+                else
+                {
+                    return new List<Measurement<long>>();
+                }
+            });
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .AddInMemoryExporter(exportedItems, metricReaderOptions =>
+            {
+                metricReaderOptions.TemporalityPreference = temporality;
+            }));
+
+        // Export 1: Should get both time series
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Single(exportedItems);
+        var metric = exportedItems[0];
+        Assert.Equal("observable-counter", metric.Name);
+        List<MetricPoint> metricPoints = [];
+        foreach (ref readonly var mp in metric.GetMetricPoints())
+        {
+            metricPoints.Add(mp);
+        }
+
+        Assert.Equal(2, metricPoints.Count);
+
+        // Export 2: Should get only tags1
+        exportedItems.Clear();
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Single(exportedItems);
+        metric = exportedItems[0];
+        Assert.Equal("observable-counter", metric.Name);
+        metricPoints.Clear();
+        foreach (ref readonly var mp in metric.GetMetricPoints())
+        {
+            metricPoints.Add(mp);
+        }
+
+        Assert.Single(metricPoints);
+
+        // Export 3: Should get nothing
+        exportedItems.Clear();
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Empty(exportedItems);
+    }
+
+    [Theory]
+    [InlineData(MetricReaderTemporalityPreference.Delta)]
+    [InlineData(MetricReaderTemporalityPreference.Cumulative)]
+    public void SynchronousCounterCumulativeCarriesForwardTest(MetricReaderTemporalityPreference temporality)
+    {
+        var exportedItems = new List<Metric>();
+        var tags1 = new List<KeyValuePair<string, object?>>
+        {
+            new("key", "value1"),
+        };
+
+        using var meter = new Meter($"{Utils.GetCurrentMethodName()}.{temporality}");
+        var counter = meter.CreateCounter<long>("sync-counter");
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .AddInMemoryExporter(exportedItems, metricReaderOptions =>
+            {
+                metricReaderOptions.TemporalityPreference = temporality;
+            }));
+
+        // Record a measurement then export
+        counter.Add(10, tags1.ToArray());
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+        Assert.Single(exportedItems);
+        var metric = exportedItems[0];
+        Assert.Equal("sync-counter", metric.Name);
+        List<MetricPoint> metricPoints = [];
+        foreach (ref readonly var mp in metric.GetMetricPoints())
+        {
+            metricPoints.Add(mp);
+        }
+
+        Assert.Single(metricPoints);
+        Assert.Equal(10, metricPoints[0].GetSumLong());
+
+        // Export again without recording new measurements.
+        // For Cumulative temporality, synchronous counters must carry forward the total.
+        // For Delta temporality, they should report 0 delta (no new measurements).
+        exportedItems.Clear();
+        meterProvider.ForceFlush(MaxTimeToAllowForFlush);
+
+        if (temporality == MetricReaderTemporalityPreference.Cumulative)
+        {
+            Assert.Single(exportedItems);
+            metric = exportedItems[0];
+            metricPoints.Clear();
+            foreach (ref readonly var mp in metric.GetMetricPoints())
+            {
+                metricPoints.Add(mp);
+            }
+
+            Assert.Single(metricPoints);
+            Assert.Equal(10, metricPoints[0].GetSumLong());
+        }
+    }
+
     [Theory]
     [InlineData(true)]
     [InlineData(false)]