Change InMemoryExporter to clear collection before exporting Metrics

docs/metrics/extending-the-sdk/README.md

@@ -27,8 +27,12 @@ not covered by the built-in exporters:
   does not implement any retry logic.
 * Exporters should avoid generating telemetry and causing live-loop, this can be
   done via `OpenTelemetry.SuppressInstrumentationScope`.
-* Exporters receives a batch of `Metric`, and each `Metric`
-  can contain 1 or more `MetricPoint`s.
+* Exporters receive a batch of `Metric`, and each `Metric` can contain one or
+  more `MetricPoint`s. The exporter should perform all actions
+  (e.g. serializing etc.) with the `Metric`s and `MetricsPoint`s in the batch
+  before returning control from `Export`, once the control is returned, the
+  exporter can no longer make any assumptions about the state of the Batch or
+  anything inside it.
 * Exporters should use `Activity.TagObjects` collection instead of
   `Activity.Tags` to obtain the full set of attributes (tags).
 * Exporters should use `ParentProvider.GetResource()` to get the `Resource`

src/OpenTelemetry.Exporter.InMemory/CHANGELOG.md

@@ -9,6 +9,10 @@ Released 2022-Mar-04
 * Adds the ability to configure `MetricReaderOptions` via the
   `AddInMemoryExporter` extension method.
   ([#2931](https://github.com/open-telemetry/opentelemetry-dotnet/pull/2931))
+* Add `InMemoryMetricExporter` which inherits `InMemoryExporter`.
+  This new class will clear the exported collection before exporting `Metric`s.
+  This prevents the same `Metric` from being repeated in the exported collection.
+  ([#2937](https://github.com/open-telemetry/opentelemetry-dotnet/pull/2937))
 
 ## 1.2.0-rc2
 

src/OpenTelemetry.Exporter.InMemory/InMemoryExporter.cs

@@ -16,28 +16,27 @@
 
 using System.Collections.Generic;
 
+using OpenTelemetry.Internal;
+
 namespace OpenTelemetry.Exporter
 {
     public class InMemoryExporter<T> : BaseExporter<T>
         where T : class
     {
-        private readonly ICollection<T> exportedItems;
+        internal readonly ICollection<T> ExportedItems;
 
         public InMemoryExporter(ICollection<T> exportedItems)
         {
-            this.exportedItems = exportedItems;
+            Guard.ThrowIfNull(exportedItems);
+
+            this.ExportedItems = exportedItems;
         }
 
         public override ExportResult Export(in Batch<T> batch)
         {
-            if (this.exportedItems == null)
-            {
-                return ExportResult.Failure;
-            }
-
             foreach (var data in batch)
             {
-                this.exportedItems.Add(data);
+                this.ExportedItems.Add(data);
             }
 
             return ExportResult.Success;

src/OpenTelemetry.Exporter.InMemory/InMemoryExporterMetricsExtensions.cs

@@ -83,7 +83,7 @@ private static MeterProviderBuilder AddInMemoryExporter(
         {
             configureMetricReader?.Invoke(metricReaderOptions);
 
-            var metricExporter = new InMemoryExporter<Metric>(exportedItems);
+            var metricExporter = new InMemoryMetricExporter(exportedItems);
 
             var metricReader = PeriodicExportingMetricReaderHelper.CreatePeriodicExportingMetricReader(
                 metricExporter,

src/OpenTelemetry.Exporter.InMemory/InMemoryMetricExporter.cs

@@ -0,0 +1,41 @@
+// <copyright file="InMemoryMetricExporter.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>
+
+using System.Collections.Generic;
+
+using OpenTelemetry.Metrics;
+
+namespace OpenTelemetry.Exporter
+{
+    internal class InMemoryMetricExporter : InMemoryExporter<Metric>
+    {
+        public InMemoryMetricExporter(ICollection<Metric> exportedItems)
+            : base(exportedItems)
+        {
+        }
+
+        public override ExportResult Export(in Batch<Metric> batch)
+        {
+            // By design, The MetricApi reuses Metrics (MetricReader.metricsCurrentBatch).
+            // This means that exported Metrics will always reflect the latest values.
+            // Here, we clear the exported collection to prevent populating
+            // this with duplicate instances of the same Metric.
+            this.ExportedItems.Clear();
+
+            return base.Export(batch);
+        }
+    }
+}

test/OpenTelemetry.Tests/Metrics/MemoryEfficiencyTests.cs

@@ -46,7 +46,6 @@ public void ExportOnlyWhenPointChanged(AggregationTemporality temporality)
             meterProvider.ForceFlush();
             Assert.Single(exportedItems);
 
-            exportedItems.Clear();
             meterProvider.ForceFlush();
             if (temporality == AggregationTemporality.Cumulative)
             {

test/OpenTelemetry.Tests/Metrics/MetricAPITest.cs

@@ -533,7 +533,6 @@ public void DuplicateInstrumentNamesFromDifferentMetersAreAllowed(AggregationTem
             var anotherCounterSameNameDiffMeter = meter2.CreateCounter<long>("name1");
             anotherCounterSameNameDiffMeter.Add(10);
             counterLong.Add(10);
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Equal(2, exportedItems.Count);
         }
@@ -641,7 +640,6 @@ public void CounterAggregationTest(bool exportDelta)
             long sumReceived = GetLongSum(exportedItems);
             Assert.Equal(20, sumReceived);
 
-            exportedItems.Clear();
             counterLong.Add(10);
             counterLong.Add(10);
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
@@ -655,7 +653,6 @@ public void CounterAggregationTest(bool exportDelta)
                 Assert.Equal(40, sumReceived);
             }
 
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             sumReceived = GetLongSum(exportedItems);
             if (exportDelta)
@@ -667,7 +664,6 @@ public void CounterAggregationTest(bool exportDelta)
                 Assert.Equal(40, sumReceived);
             }
 
-            exportedItems.Clear();
             counterLong.Add(40);
             counterLong.Add(20);
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
@@ -713,7 +709,6 @@ public void ObservableCounterAggregationTest(bool exportDelta)
             long sumReceived = GetLongSum(exportedItems);
             Assert.Equal(10, sumReceived);
 
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             sumReceived = GetLongSum(exportedItems);
             if (exportDelta)
@@ -725,7 +720,6 @@ public void ObservableCounterAggregationTest(bool exportDelta)
                 Assert.Equal(20, sumReceived);
             }
 
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             sumReceived = GetLongSum(exportedItems);
             if (exportDelta)
@@ -803,7 +797,6 @@ public void ObservableCounterWithTagsAggregationTest(bool exportDelta)
             ValidateMetricPointTags(tags3, metricPoint3.Tags);
 
             // Export 2
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Single(exportedItems);
             metric = exportedItems[0];
@@ -954,8 +947,6 @@ public void DimensionsAreOrderInsensitiveWithSortedKeysFirst(bool exportDelta)
             long sumReceived = GetLongSum(exportedItems);
             Assert.Equal(75, sumReceived);
 
-            exportedItems.Clear();
-
             counterLong.Add(5, new("Key2", "Value2"), new("Key1", "Value1"), new("Key3", "Value3"));
             counterLong.Add(5, new("Key2", "Value2"), new("Key1", "Value1"), new("Key3", "Value3"));
             counterLong.Add(10, new("Key2", "Value2"), new("Key3", "Value3"), new("Key1", "Value1"));
@@ -1045,8 +1036,6 @@ public void DimensionsAreOrderInsensitiveWithUnsortedKeysFirst(bool exportDelta)
             long sumReceived = GetLongSum(exportedItems);
             Assert.Equal(75, sumReceived);
 
-            exportedItems.Clear();
-
             counterLong.Add(5, new("Key2", "Value2"), new("Key1", "Value1"), new("Key3", "Value3"));
             counterLong.Add(5, new("Key2", "Value2"), new("Key1", "Value1"), new("Key3", "Value3"));
             counterLong.Add(10, new("Key2", "Value2"), new("Key3", "Value3"), new("Key1", "Value1"));
@@ -1097,29 +1086,25 @@ public void TestInstrumentDisposal(AggregationTemporality temporality)
 
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Equal(2, exportedItems.Count);
-            exportedItems.Clear();
 
             counter1.Add(10, new KeyValuePair<string, object>("key", "value"));
             counter2.Add(10, new KeyValuePair<string, object>("key", "value"));
             meter1.Dispose();
 
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Equal(2, exportedItems.Count);
-            exportedItems.Clear();
 
             counter1.Add(10, new KeyValuePair<string, object>("key", "value"));
             counter2.Add(10, new KeyValuePair<string, object>("key", "value"));
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Single(exportedItems);
-            exportedItems.Clear();
 
             counter1.Add(10, new KeyValuePair<string, object>("key", "value"));
             counter2.Add(10, new KeyValuePair<string, object>("key", "value"));
             meter2.Dispose();
 
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Single(exportedItems);
-            exportedItems.Clear();
 
             counter1.Add(10, new KeyValuePair<string, object>("key", "value"));
             counter2.Add(10, new KeyValuePair<string, object>("key", "value"));
@@ -1172,7 +1157,6 @@ int MetricPointCount()
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Equal(MeterProviderBuilderBase.MaxMetricPointsPerMetricDefault, MetricPointCount());
 
-            exportedItems.Clear();
             counterLong.Add(10);
             for (int i = 0; i < MeterProviderBuilderBase.MaxMetricPointsPerMetricDefault + 1; i++)
             {
@@ -1192,7 +1176,6 @@ int MetricPointCount()
             counterLong.Add(10, new KeyValuePair<string, object>("key", "valueA"));
             counterLong.Add(10, new KeyValuePair<string, object>("key", "valueB"));
             counterLong.Add(10, new KeyValuePair<string, object>("key", "valueC"));
-            exportedItems.Clear();
             meterProvider.ForceFlush(MaxTimeToAllowForFlush);
             Assert.Equal(MeterProviderBuilderBase.MaxMetricPointsPerMetricDefault, MetricPointCount());
         }

test/OpenTelemetry.Tests/Trace/BatchExportActivityProcessorTest.cs

@@ -190,7 +190,7 @@ public void CheckExportForRecordingButNotSampledActivity()
         [Fact]
         public void CheckExportDrainsBatchOnFailure()
         {
-            using var exporter = new InMemoryExporter<Activity>(null);
+            using var exporter = new InMemoryExporter<Activity>(new List<Activity>());
             using var processor = new BatchActivityExportProcessor(
                 exporter,
                 maxQueueSize: 3,