[Metrics] Resolve accumulated TODOs in SDK core

src/OpenTelemetry/CHANGELOG.md

@@ -19,6 +19,11 @@ Notes](../../RELEASENOTES.md).
   once per collection cycle.
   ([#7188](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7188))
 
+* Added exception safety for user-supplied `ExemplarReservoir` implementations.
+  Exceptions thrown from `Offer` are now caught and logged rather than propagating
+  out of `Counter.Add`/`Histogram.Record`.
+  ([#7277](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7277))
+
 * Update `OpenTelemetrySdkEventSource` to support the W3C randomness flag.
   ([#7301](https://github.com/open-telemetry/opentelemetry-dotnet/pull/7301))
 

src/OpenTelemetry/Internal/OpenTelemetrySdkEventSource.cs

@@ -169,6 +169,15 @@ public void MetricViewException(string source, Exception ex)
         }
     }
 
+    [NonEvent]
+    public void ExemplarReservoirException(Exception ex)
+    {
+        if (this.IsEnabled(EventLevel.Verbose, EventKeywords.All))
+        {
+            this.ExemplarReservoirException(ex.ToInvariantString());
+        }
+    }
+
     [Event(4, Message = "Unknown error in SpanProcessor event '{0}': '{1}'.", Level = EventLevel.Error)]
     public void SpanProcessorException(string evnt, string ex)
         => this.WriteEvent(4, evnt, ex);
@@ -312,6 +321,10 @@ public void TracesSamplerArgConfigInvalid(string configValue)
     public void MetricViewException(string source, string ex)
         => this.WriteEvent(56, source, ex);
 
+    [Event(57, Message = "Exception thrown by user-supplied ExemplarReservoir.Offer implementation: '{0}'.", Level = EventLevel.Verbose)]
+    public void ExemplarReservoirException(string ex)
+        => this.WriteEvent(57, ex);
+
     void IConfigurationExtensionsLogger.LogInvalidConfigurationValue(string key, string value)
         => this.InvalidConfigurationValue(key, value);
 

src/OpenTelemetry/Metrics/MeterProviderSdk.cs

@@ -195,7 +195,7 @@ internal static void MeasurementsCompleted(Instrument instrument, object? state)
     {
         if (state is not MetricState metricState)
         {
-            // todo: Log
+            OpenTelemetrySdkEventSource.Log.MeasurementDropped(instrument?.Name ?? "UnknownInstrument", "SDK internal error occurred.", "Contact SDK owners.");
             return;
         }
 

src/OpenTelemetry/Metrics/MetricPoint/MetricPoint.cs

@@ -1041,9 +1041,7 @@ private readonly void UpdateExemplar(long number, ReadOnlySpan<KeyValuePair<stri
     {
         if (offerExemplar)
         {
-            // TODO: A custom implementation of `ExemplarReservoir.Offer` might throw an exception.
-            this.mpComponents!.ExemplarReservoir!.Offer(
-                new ExemplarMeasurement<long>(number, tags));
+            this.OfferExemplarSafe(number, tags);
         }
     }
 
@@ -1052,10 +1050,36 @@ private readonly void UpdateExemplar(double number, ReadOnlySpan<KeyValuePair<st
     {
         if (offerExemplar)
         {
-            // TODO: A custom implementation of `ExemplarReservoir.Offer` might throw an exception.
+            this.OfferExemplarSafe(number, tags, explicitBucketHistogramBucketIndex);
+        }
+    }
+
+    [MethodImpl(MethodImplOptions.NoInlining)]
+    private readonly void OfferExemplarSafe(long number, ReadOnlySpan<KeyValuePair<string, object?>> tags)
+    {
+        try
+        {
+            this.mpComponents!.ExemplarReservoir!.Offer(
+                new ExemplarMeasurement<long>(number, tags));
+        }
+        catch (Exception ex)
+        {
+            OpenTelemetrySdkEventSource.Log.ExemplarReservoirException(ex);
+        }
+    }
+
+    [MethodImpl(MethodImplOptions.NoInlining)]
+    private readonly void OfferExemplarSafe(double number, ReadOnlySpan<KeyValuePair<string, object?>> tags, int explicitBucketHistogramBucketIndex)
+    {
+        try
+        {
             this.mpComponents!.ExemplarReservoir!.Offer(
                 new ExemplarMeasurement<double>(number, tags, explicitBucketHistogramBucketIndex));
         }
+        catch (Exception ex)
+        {
+            OpenTelemetrySdkEventSource.Log.ExemplarReservoirException(ex);
+        }
     }
 
     [MethodImpl(MethodImplOptions.AggressiveInlining)]

src/OpenTelemetry/Metrics/Reader/MetricReader.cs

@@ -18,21 +18,29 @@ public abstract partial class MetricReader : IDisposable
         static (_) => AggregationTemporality.Cumulative;
 
     private static readonly Func<Type, AggregationTemporality> MonotonicDeltaTemporalityPreferenceFunc =
-        static (instrumentType) => instrumentType.GetGenericTypeDefinition() switch
+        static (instrumentType) =>
         {
-            var type when type == typeof(Counter<>) => AggregationTemporality.Delta,
-            var type when type == typeof(ObservableCounter<>) => AggregationTemporality.Delta,
-            var type when type == typeof(Histogram<>) => AggregationTemporality.Delta,
+            return instrumentType.GetGenericTypeDefinition() switch
+            {
+                var type when type == typeof(Counter<>) => AggregationTemporality.Delta,
+                var type when type == typeof(ObservableCounter<>) => AggregationTemporality.Delta,
+                var type when type == typeof(Histogram<>) => AggregationTemporality.Delta,
 
-            // Temporality is not defined for gauges, so this does not really affect anything.
-            var type when type == typeof(ObservableGauge<>) => AggregationTemporality.Delta,
-            var type when type == typeof(Gauge<>) => AggregationTemporality.Delta,
+                // Temporality is not defined for gauges, so this does not really affect anything.
+                var type when type == typeof(ObservableGauge<>) => AggregationTemporality.Delta,
+                var type when type == typeof(Gauge<>) => AggregationTemporality.Delta,
 
-            var type when type == typeof(UpDownCounter<>) => AggregationTemporality.Cumulative,
-            var type when type == typeof(ObservableUpDownCounter<>) => AggregationTemporality.Cumulative,
+                var type when type == typeof(UpDownCounter<>) => AggregationTemporality.Cumulative,
+                var type when type == typeof(ObservableUpDownCounter<>) => AggregationTemporality.Cumulative,
+
+                _ => LogAndDefault(instrumentType),
+            };
 
-            // TODO: Consider logging here because we should not fall through to this case.
-            _ => AggregationTemporality.Delta,
+            static AggregationTemporality LogAndDefault(Type type)
+            {
+                OpenTelemetrySdkEventSource.Log.MetricReaderEvent($"Unexpected instrument type '{type.FullName}' encountered in temporality preference. Defaulting to Delta.");
+                return AggregationTemporality.Delta;
+            }
         };
 
     private static readonly Func<Type, AggregationTemporality> LowMemoryTemporalityPreferenceFunc =

src/OpenTelemetry/Metrics/Reader/MetricReaderExt.cs

@@ -109,9 +109,7 @@ internal virtual List<Metric> AddMetricWithViews(Instrument instrument, List<Met
         var maxCountMetricsToBeCreated = metricStreamConfigs.Count;
 
         // Create list with initial capacity as the max metric count.
-        // Due to duplicate/max limit, we may not end up using them
-        // all, and that memory is wasted until Meter disposed.
-        // TODO: Revisit to see if we need to do metrics.TrimExcess()
+        // Due to duplicate/max limit, we may not end up using them all.
         var metrics = new List<Metric>(maxCountMetricsToBeCreated);
         lock (this.instrumentCreationLock)
         {
@@ -181,9 +179,9 @@ internal virtual List<Metric> AddMetricWithViews(Instrument instrument, List<Met
                     this.CreateOrUpdateMetricStreamRegistration(in metricStreamIdentity);
                 }
             }
-
-            return metrics;
         }
+
+        return metrics;
     }
 
     internal void ApplyParentProviderSettings(

test/OpenTelemetry.Tests/Metrics/MetricExemplarTests.cs

@@ -837,6 +837,99 @@ public void TestExemplarsFilterTags(bool enableTagFiltering)
         }
     }
 
+    [Fact]
+    public void ExemplarReservoirOfferThrowingForLongCounter_ExceptionSwallowedAndMeasurementRecorded()
+    {
+        var exportedItems = new List<Metric>();
+
+        using var meter = new Meter(Utils.GetCurrentMethodName());
+        var counter = meter.CreateCounter<long>("testCounter");
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .SetExemplarFilter(ExemplarFilterType.AlwaysOn)
+            .AddView(
+                counter.Name,
+                new MetricStreamConfiguration
+                {
+                    ExemplarReservoirFactory = () => new ThrowingExemplarReservoir(),
+                })
+            .AddInMemoryExporter(exportedItems));
+
+        counter.Add(10);
+        counter.Add(20);
+        counter.Add(30);
+
+        Assert.True(meterProvider.ForceFlush(MaxTimeToAllowForFlush));
+
+        var metricPoint = GetFirstMetricPoint(exportedItems);
+        Assert.NotNull(metricPoint);
+        Assert.Equal(60L, metricPoint.Value.GetSumLong());
+    }
+
+    [Fact]
+    public void ExemplarReservoirOfferThrowingForDoubleCounter_ExceptionSwallowedAndMeasurementRecorded()
+    {
+        var exportedItems = new List<Metric>();
+
+        using var meter = new Meter(Utils.GetCurrentMethodName());
+        var counter = meter.CreateCounter<double>("testCounter");
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .SetExemplarFilter(ExemplarFilterType.AlwaysOn)
+            .AddView(
+                counter.Name,
+                new MetricStreamConfiguration
+                {
+                    ExemplarReservoirFactory = () => new ThrowingExemplarReservoir(),
+                })
+            .AddInMemoryExporter(exportedItems));
+
+        counter.Add(1.5);
+        counter.Add(2.5);
+        counter.Add(3.0);
+
+        Assert.True(meterProvider.ForceFlush(MaxTimeToAllowForFlush));
+
+        var metricPoint = GetFirstMetricPoint(exportedItems);
+        Assert.NotNull(metricPoint);
+        Assert.Equal(7.0, metricPoint.Value.GetSumDouble());
+    }
+
+    [Fact]
+    public void ExemplarReservoirOfferThrowing_SubsequentMeasurementsAreStillRecorded()
+    {
+        var exportedItems = new List<Metric>();
+
+        using var meter = new Meter(Utils.GetCurrentMethodName());
+        var histogram = meter.CreateHistogram<double>("testHistogram");
+
+        using var container = BuildMeterProvider(out var meterProvider, builder => builder
+            .AddMeter(meter.Name)
+            .SetExemplarFilter(ExemplarFilterType.AlwaysOn)
+            .AddView(
+                histogram.Name,
+                new MetricStreamConfiguration
+                {
+                    ExemplarReservoirFactory = () => new ThrowingExemplarReservoir(),
+                })
+            .AddInMemoryExporter(exportedItems));
+
+        histogram.Record(1);
+        histogram.Record(2);
+        histogram.Record(3);
+        histogram.Record(4);
+        histogram.Record(5);
+
+        Assert.True(meterProvider.ForceFlush(MaxTimeToAllowForFlush));
+
+        var metricPoint = GetFirstMetricPoint(exportedItems);
+        Assert.NotNull(metricPoint);
+        Assert.Equal(5L, metricPoint.Value.GetHistogramCount());
+        Assert.Equal(15.0, metricPoint.Value.GetHistogramSum());
+    }
+
     private static (double Value, bool ExpectTraceId)[] GenerateRandomValues(
         int count,
         bool expectTraceId,
@@ -917,4 +1010,13 @@ public override void Offer(in ExemplarMeasurement<double> measurement)
         public override void Offer(in ExemplarMeasurement<long> measurement)
             => throw new NotSupportedException();
     }
+
+    private sealed class ThrowingExemplarReservoir() : FixedSizeExemplarReservoir(1)
+    {
+        public override void Offer(in ExemplarMeasurement<long> measurement)
+            => throw new InvalidOperationException("Simulated reservoir failure (long).");
+
+        public override void Offer(in ExemplarMeasurement<double> measurement)
+            => throw new InvalidOperationException("Simulated reservoir failure (double).");
+    }
 }