Delta exporters will Export only those points which received new update

src/OpenTelemetry/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+* Metrics SDK will not provide inactive Metrics to delta exporter.
+  ([#2629](https://github.com/open-telemetry/opentelemetry-dotnet/pull/2629))
+
 * Histogram bounds are validated when added to a View.
   ([#2573](https://github.com/open-telemetry/opentelemetry-dotnet/pull/2573))
 

src/OpenTelemetry/Metrics/AggregatorStore.cs

@@ -37,11 +37,13 @@ internal sealed class AggregatorStore
         private readonly AggregationTemporality temporality;
         private readonly bool outputDelta;
         private readonly MetricPoint[] metricPoints;
+        private readonly int[] currentMetricPointBatch;
         private readonly AggregationType aggType;
         private readonly double[] histogramBounds;
         private readonly UpdateLongDelegate updateLongCallback;
         private readonly UpdateDoubleDelegate updateDoubleCallback;
         private int metricPointIndex = 0;
+        private int batchSize = 0;
         private bool zeroTagMetricPointInitialized;
         private DateTimeOffset startTimeExclusive;
         private DateTimeOffset endTimeInclusive;
@@ -53,6 +55,7 @@ internal AggregatorStore(
             string[] tagKeysInteresting = null)
         {
             this.metricPoints = new MetricPoint[MaxMetricPoints];
+            this.currentMetricPointBatch = new int[MaxMetricPoints];
             this.aggType = aggType;
             this.temporality = temporality;
             this.outputDelta = temporality == AggregationTemporality.Delta ? true : false;
@@ -92,37 +95,63 @@ internal void Update(double value, ReadOnlySpan<KeyValuePair<string, object>> ta
             this.updateDoubleCallback(value, tags);
         }
 
-        internal void SnapShot()
+        internal int SnapShot()
         {
+            this.batchSize = 0;
             var indexSnapShot = Math.Min(this.metricPointIndex, MaxMetricPoints - 1);
+            if (this.temporality == AggregationTemporality.Delta)
+            {
+                this.SnapShotDelta(indexSnapShot);
+            }
+            else
+            {
+                this.SnapShotCumulative(indexSnapShot);
+            }
 
+            this.endTimeInclusive = DateTimeOffset.UtcNow;
+            return this.batchSize;
+        }
+
+        internal void SnapShotDelta(int indexSnapShot)
+        {
             for (int i = 0; i <= indexSnapShot; i++)
             {
                 ref var metricPoint = ref this.metricPoints[i];
-                if (metricPoint.StartTime == default)
+                if (metricPoint.MetricPointStatus == MetricPointStatus.NoCollectPending)
                 {
                     continue;
                 }
 
                 metricPoint.TakeSnapShot(this.outputDelta);
+                this.currentMetricPointBatch[this.batchSize] = i;
+                this.batchSize++;
             }
 
-            if (this.temporality == AggregationTemporality.Delta)
+            if (this.endTimeInclusive != default)
+            {
+                this.startTimeExclusive = this.endTimeInclusive;
+            }
+        }
+
+        internal void SnapShotCumulative(int indexSnapShot)
+        {
+            for (int i = 0; i <= indexSnapShot; i++)
             {
-                if (this.endTimeInclusive != default)
+                ref var metricPoint = ref this.metricPoints[i];
+                if (metricPoint.StartTime == default)
                 {
-                    this.startTimeExclusive = this.endTimeInclusive;
+                    continue;
                 }
-            }
 
-            DateTimeOffset dt = DateTimeOffset.UtcNow;
-            this.endTimeInclusive = dt;
+                metricPoint.TakeSnapShot(this.outputDelta);
+                this.currentMetricPointBatch[this.batchSize] = i;
+                this.batchSize++;
+            }
         }
 
         internal BatchMetricPoint GetMetricPoints()
         {
-            var indexSnapShot = Math.Min(this.metricPointIndex, MaxMetricPoints - 1);
-            return new BatchMetricPoint(this.metricPoints, indexSnapShot + 1, this.startTimeExclusive, this.endTimeInclusive);
+            return new BatchMetricPoint(this.metricPoints, this.currentMetricPointBatch, this.batchSize, this.startTimeExclusive, this.endTimeInclusive);
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]

src/OpenTelemetry/Metrics/BatchMetricPoint.cs

@@ -16,25 +16,25 @@
 
 using System;
 using System.Collections;
-using System.Diagnostics;
 using OpenTelemetry.Internal;
 
 namespace OpenTelemetry.Metrics
 {
     public readonly struct BatchMetricPoint : IDisposable
     {
         private readonly MetricPoint[] metricsPoints;
+        private readonly int[] metricPointsToProcess;
         private readonly long targetCount;
         private readonly DateTimeOffset start;
         private readonly DateTimeOffset end;
 
-        internal BatchMetricPoint(MetricPoint[] metricsPoints, int maxSize, DateTimeOffset start, DateTimeOffset end)
+        internal BatchMetricPoint(MetricPoint[] metricsPoints, int[] metricPointsToProcess, long targetCount, DateTimeOffset start, DateTimeOffset end)
         {
-            Debug.Assert(maxSize > 0, $"{nameof(maxSize)} should be a positive number.");
             Guard.Null(metricsPoints, nameof(metricsPoints));
 
             this.metricsPoints = metricsPoints;
-            this.targetCount = maxSize;
+            this.metricPointsToProcess = metricPointsToProcess;
+            this.targetCount = targetCount;
             this.start = start;
             this.end = end;
         }
@@ -50,7 +50,7 @@ public void Dispose()
         /// <returns><see cref="Enumerator"/>.</returns>
         public Enumerator GetEnumerator()
         {
-            return new Enumerator(this.metricsPoints, this.targetCount, this.start, this.end);
+            return new Enumerator(this.metricsPoints, this.metricPointsToProcess, this.targetCount, this.start, this.end);
         }
 
         /// <summary>
@@ -59,14 +59,16 @@ public Enumerator GetEnumerator()
         public struct Enumerator : IEnumerator
         {
             private readonly MetricPoint[] metricsPoints;
+            private readonly int[] metricPointsToProcess;
             private readonly DateTimeOffset start;
             private readonly DateTimeOffset end;
             private long targetCount;
             private long index;
 
-            internal Enumerator(MetricPoint[] metricsPoints, long targetCount, DateTimeOffset start, DateTimeOffset end)
+            internal Enumerator(MetricPoint[] metricsPoints, int[] metricPointsToProcess, long targetCount, DateTimeOffset start, DateTimeOffset end)
             {
                 this.metricsPoints = metricsPoints;
+                this.metricPointsToProcess = metricPointsToProcess;
                 this.targetCount = targetCount;
                 this.index = -1;
                 this.start = start;
@@ -77,7 +79,7 @@ public ref MetricPoint Current
             {
                 get
                 {
-                    return ref this.metricsPoints[this.index];
+                    return ref this.metricsPoints[this.metricPointsToProcess[this.index]];
                 }
             }
 
@@ -93,12 +95,7 @@ public bool MoveNext()
             {
                 while (++this.index < this.targetCount)
                 {
-                    ref var metricPoint = ref this.metricsPoints[this.index];
-                    if (metricPoint.StartTime == default)
-                    {
-                        continue;
-                    }
-
+                    ref var metricPoint = ref this.metricsPoints[this.metricPointsToProcess[this.index]];
                     metricPoint.StartTime = this.start;
                     metricPoint.EndTime = this.end;
                     return true;

src/OpenTelemetry/Metrics/MeterProviderSdk.cs

@@ -443,19 +443,23 @@ internal Batch<Metric> Collect()
                     for (int i = 0; i < target; i++)
                     {
                         var metric = this.metrics[i];
+                        int metricPointSize = 0;
                         if (metric != null)
                         {
                             if (metric.InstrumentDisposed)
                             {
-                                metric.SnapShot();
+                                metricPointSize = metric.SnapShot();
                                 this.metrics[i] = null;
                             }
                             else
                             {
-                                metric.SnapShot();
+                                metricPointSize = metric.SnapShot();
                             }
 
-                            this.metricsCurrentBatch[metricCountCurrentBatch++] = metric;
+                            if (metricPointSize > 0)
+                            {
+                                this.metricsCurrentBatch[metricCountCurrentBatch++] = metric;
+                            }
                         }
                     }
 

src/OpenTelemetry/Metrics/Metric.cs

@@ -138,9 +138,9 @@ internal void UpdateDouble(double value, ReadOnlySpan<KeyValuePair<string, objec
             this.aggStore.Update(value, tags);
         }
 
-        internal void SnapShot()
+        internal int SnapShot()
         {
-            this.aggStore.SnapShot();
+            return this.aggStore.SnapShot();
         }
     }
 }

src/OpenTelemetry/Metrics/MetricPoint.cs

@@ -46,6 +46,7 @@ internal MetricPoint(
             this.DoubleValue = default;
             this.doubleVal = default;
             this.lastDoubleSum = default;
+            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
 
             if (this.AggType == AggregationType.Histogram)
             {
@@ -86,6 +87,8 @@ internal MetricPoint(
 
         public double[] ExplicitBounds { get; internal set; }
 
+        internal MetricPointStatus MetricPointStatus { get; private set; }
+
         private readonly AggregationType AggType { get; }
 
         internal void Update(long number)
@@ -117,6 +120,19 @@ internal void Update(long number)
                         break;
                     }
             }
+
+            // There is a race with Snapshot:
+            // Update() updates the value
+            // Snapshot snapshots the value
+            // Snapshot sets status to NoCollectPending
+            // Update sets status to CollectPending -- this is not right as the Snapshot
+            // already included the updated value.
+            // In the absence of any new Update call until next Snapshot,
+            // this results in exporting an Update even though
+            // it had no update.
+            // TODO: For Delta, this can be mitigated
+            // by ignoring Zero points
+            this.MetricPointStatus = MetricPointStatus.CollectPending;
         }
 
         internal void Update(double number)
@@ -180,6 +196,19 @@ internal void Update(double number)
                         break;
                     }
             }
+
+            // There is a race with Snapshot:
+            // Update() updates the value
+            // Snapshot snapshots the value
+            // Snapshot sets status to NoCollectPending
+            // Update sets status to CollectPending -- this is not right as the Snapshot
+            // already included the updated value.
+            // In the absence of any new Update call until next Snapshot,
+            // this results in exporting an Update even though
+            // it had no update.
+            // TODO: For Delta, this can be mitigated
+            // by ignoring Zero points
+            this.MetricPointStatus = MetricPointStatus.CollectPending;
         }
 
         internal void TakeSnapShot(bool outputDelta)
@@ -194,6 +223,14 @@ internal void TakeSnapShot(bool outputDelta)
                             long initValue = Interlocked.Read(ref this.longVal);
                             this.LongValue = initValue - this.lastLongSum;
                             this.lastLongSum = initValue;
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+
+                            // Check again if value got updated, if yes reset status.
+                            // This ensures no Updates get Lost.
+                            if (initValue != Interlocked.Read(ref this.longVal))
+                            {
+                                this.MetricPointStatus = MetricPointStatus.CollectPending;
+                            }
                         }
                         else
                         {
@@ -216,6 +253,14 @@ internal void TakeSnapShot(bool outputDelta)
                             double initValue = Interlocked.CompareExchange(ref this.doubleVal, 0.0, double.NegativeInfinity);
                             this.DoubleValue = initValue - this.lastDoubleSum;
                             this.lastDoubleSum = initValue;
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+
+                            // Check again if value got updated, if yes reset status.
+                            // This ensures no Updates get Lost.
+                            if (initValue != Interlocked.CompareExchange(ref this.doubleVal, 0.0, double.NegativeInfinity))
+                            {
+                                this.MetricPointStatus = MetricPointStatus.CollectPending;
+                            }
                         }
                         else
                         {
@@ -233,6 +278,15 @@ internal void TakeSnapShot(bool outputDelta)
                 case AggregationType.LongGauge:
                     {
                         this.LongValue = Interlocked.Read(ref this.longVal);
+                        this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+
+                        // Check again if value got updated, if yes reset status.
+                        // This ensures no Updates get Lost.
+                        if (this.LongValue != Interlocked.Read(ref this.longVal))
+                        {
+                            this.MetricPointStatus = MetricPointStatus.CollectPending;
+                        }
+
                         break;
                     }
 
@@ -244,6 +298,15 @@ internal void TakeSnapShot(bool outputDelta)
                         // the exchange (to 0.0) will never occur,
                         // but we get the original value atomically.
                         this.DoubleValue = Interlocked.CompareExchange(ref this.doubleVal, 0.0, double.NegativeInfinity);
+                        this.MetricPointStatus = MetricPointStatus.NoCollectPending;
+
+                        // Check again if value got updated, if yes reset status.
+                        // This ensures no Updates get Lost.
+                        if (this.DoubleValue != Interlocked.CompareExchange(ref this.doubleVal, 0.0, double.NegativeInfinity))
+                        {
+                            this.MetricPointStatus = MetricPointStatus.CollectPending;
+                        }
+
                         break;
                     }
 
@@ -267,6 +330,8 @@ internal void TakeSnapShot(bool outputDelta)
                                     this.bucketCounts[i] = 0;
                                 }
                             }
+
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
                         }
 
                         break;
@@ -283,6 +348,8 @@ internal void TakeSnapShot(bool outputDelta)
                                 this.longVal = 0;
                                 this.doubleVal = 0;
                             }
+
+                            this.MetricPointStatus = MetricPointStatus.NoCollectPending;
                         }
 
                         break;

src/OpenTelemetry/Metrics/MetricPointStatus.cs

@@ -0,0 +1,33 @@
+// <copyright file="MetricPointStatus.cs" company="OpenTelemetry Authors">
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+// </copyright>
+
+namespace OpenTelemetry.Metrics
+{
+    internal enum MetricPointStatus
+    {
+        /// <summary>
+        /// This status is applied to <see cref="MetricPoint"/>s with status <see cref="CollectPending"/> after a Collect.
+        /// If an update occurs, status will be moved to <see cref="CollectPending"/>.
+        /// </summary>
+        NoCollectPending,
+
+        /// <summary>
+        /// The <see cref="MetricPoint"/> has been updated since the previous Collect cycle.
+        /// Collect will move it to <see cref="NoCollectPending"/>.
+        /// </summary>
+        CollectPending,
+    }
+}