Remove meters from storage list when removed and handle NoRecordedValue flag when removing meters

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/ExperimentalOptions.cs

@@ -17,6 +17,8 @@ internal sealed class ExperimentalOptions
 
     public const string OtlpDiskRetryDirectoryPathEnvVar = "OTEL_DOTNET_EXPERIMENTAL_OTLP_DISK_RETRY_DIRECTORY_PATH";
 
+    public const string EmitNoRecordedValueNeededDataPointsEnvVar = "OTEL_DOTNET_EXPERIMENTAL_OTLP_METRICS_EMIT_NO_RECORDED_VALUE";
+
     public ExperimentalOptions()
         : this(new ConfigurationBuilder().AddEnvironmentVariables().Build())
     {
@@ -29,6 +31,11 @@ public ExperimentalOptions(IConfiguration configuration)
             this.EmitLogEventAttributes = emitLogEventAttributes;
         }
 
+        if (configuration.TryGetBoolValue(OpenTelemetryProtocolExporterEventSource.Log, EmitNoRecordedValueNeededDataPointsEnvVar, out var emitNoRecordedValueNeededDataPoints))
+        {
+            this.EmitNoRecordedValueNeededDataPoints = emitNoRecordedValueNeededDataPoints;
+        }
+
         if (configuration.TryGetStringValue(OtlpRetryEnvVar, out var retryPolicy) && retryPolicy != null)
         {
             if (retryPolicy.Equals("in_memory", StringComparison.OrdinalIgnoreCase))
@@ -78,4 +85,10 @@ public ExperimentalOptions(IConfiguration configuration)
     /// Gets the path on disk where the telemetry will be stored for retries at a later point.
     /// </summary>
     public string? DiskRetryDirectoryPath { get; }
+
+    /// <summary>
+    /// Gets a value indicating whether the NoRecordedValue measurement should be sent when metrics are removed,
+    /// e.g when disposing a Meter.
+    /// </summary>
+    public bool EmitNoRecordedValueNeededDataPoints { get; }
 }

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufSerializer.cs

@@ -166,6 +166,15 @@ internal static int WriteSInt32WithTag(byte[] buffer, int writePosition, int fie
         return writePosition;
     }
 
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    internal static int WriteVarInt32WithTag(byte[] buffer, int writePosition, int fieldNumber, uint value)
+    {
+        writePosition = WriteTag(buffer, writePosition, fieldNumber, ProtobufWireType.VARINT);
+        writePosition = WriteVarInt32(buffer, writePosition, value);
+
+        return writePosition;
+    }
+
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
     internal static int WriteVarInt32(byte[] buffer, int writePosition, uint value)
     {

src/OpenTelemetry.Exporter.OpenTelemetryProtocol/OtlpMetricExporter.cs

@@ -21,6 +21,7 @@ public class OtlpMetricExporter : BaseExporter<Metric>
     private readonly OtlpExporterTransmissionHandler transmissionHandler;
     private readonly int startWritePosition;
 
+    private readonly bool emitNoRecordedValueNeededDataPoints;
     private Resource? resource;
 
     // Initial buffer size set to ~732KB.
@@ -53,6 +54,7 @@ internal OtlpMetricExporter(
 
         this.startWritePosition = exporterOptions!.Protocol == OtlpExportProtocol.Grpc ? GrpcStartWritePosition : 0;
         this.transmissionHandler = transmissionHandler ?? exporterOptions!.GetExportTransmissionHandler(experimentalOptions!, OtlpSignalType.Metrics);
+        this.emitNoRecordedValueNeededDataPoints = experimentalOptions!.EmitNoRecordedValueNeededDataPoints;
     }
 
     internal Resource Resource => this.resource ??= this.ParentProvider.GetResource();
@@ -65,7 +67,7 @@ public override ExportResult Export(in Batch<Metric> metrics)
 
         try
         {
-            int writePosition = ProtobufOtlpMetricSerializer.WriteMetricsData(ref this.buffer, this.startWritePosition, this.Resource, metrics);
+            int writePosition = ProtobufOtlpMetricSerializer.WriteMetricsData(ref this.buffer, this.startWritePosition, this.Resource, metrics, this.emitNoRecordedValueNeededDataPoints);
 
             if (this.startWritePosition == GrpcStartWritePosition)
             {

src/OpenTelemetry/CHANGELOG.md

@@ -6,6 +6,25 @@ Notes](../../RELEASENOTES.md).
 
 ## Unreleased
 
+* Fixed storage exhaustion when disposing and recreating Meters.
+  **Previous Behavior:** Disposing a meter set the associated metric to null
+  in an array of default size 1000 allocated at creation time. Disposing and
+  recreating meters could exhaust the storage available in that list, leading
+  to an inability to collect the data points from newly created meters.
+
+  **New Behavior:** Disposing a meter now marks the associated metrics for
+  deletion and they are cleaned up after the next collection cycle.
+
+  **Limitation:** This means that quickly recreating meters within the same
+  collection cycle will still exhaust the storage limit.
+
+* Added an experimental flag
+  `OTEL_DOTNET_EXPERIMENTAL_OTLP_METRICS_EMIT_NO_RECORDED_VALUE`. When set to
+  `true`, after disposing a meter, a DataPoint with the flag NoRecordedValue
+  will be sent on the next collection cycle for all associated metrics as per
+  the
+  [OTLP data model](https://opentelemetry.io/docs/specs/otel/metrics/data-model/#no-recorded-value).
+
 ## 1.11.0-rc.1
 
 Released 2024-Dec-11

src/OpenTelemetry/Metrics/Metric.cs

@@ -243,6 +243,8 @@ internal Metric(
 
     internal bool Active { get; set; } = true;
 
+    internal bool NoRecordedValueNeeded { get; set; }
+
     /// <summary>
     /// Get the metric points for the metric stream.
     /// </summary>

src/OpenTelemetry/Metrics/MetricPoint/MetricPoint.cs

@@ -316,7 +316,7 @@ public readonly HistogramBuckets GetHistogramBuckets()
     /// </remarks>
     /// <returns><see cref="ExponentialHistogramData"/>.</returns>
     [MethodImpl(MethodImplOptions.AggressiveInlining)]
-    public ExponentialHistogramData GetExponentialHistogramData()
+    public readonly ExponentialHistogramData GetExponentialHistogramData()
     {
         if (this.aggType != AggregationType.Base2ExponentialHistogram &&
             this.aggType != AggregationType.Base2ExponentialHistogramWithMinMax)

src/OpenTelemetry/Metrics/Reader/MetricReaderExt.cs

@@ -197,8 +197,6 @@ private void CreateOrUpdateMetricStreamRegistration(in MetricStreamIdentity metr
     {
         if (!this.metricStreamNames.Add(metricStreamIdentity.MetricStreamName))
         {
-            // TODO: If a metric is deactivated and then reactivated we log the
-            // same warning as if it was a duplicate.
             OpenTelemetrySdkEventSource.Log.DuplicateMetricInstrument(
                 metricStreamIdentity.InstrumentName,
                 metricStreamIdentity.MeterName,
@@ -231,7 +229,35 @@ private Batch<Metric> GetMetricsBatch()
 
                     if (!metric.Active)
                     {
-                        this.RemoveMetric(ref metric);
+                        // Inactive metrics are sent one last time so the remaining data points and
+                        // NoRecordedValue data points can be sent. The Active property might be
+                        // set to false between collection cycles, so a separate property must be
+                        // used to avoid duplicate staleness markers.
+                        metric.NoRecordedValueNeeded = true;
+
+                        lock (this.instrumentCreationLock)
+                        {
+                            OpenTelemetrySdkEventSource.Log.MetricInstrumentRemoved(metric.Name, metric.MeterName);
+
+                            // Note: This is using TryUpdate and NOT TryRemove because there is a
+                            // race condition. If a metric is deactivated and then reactivated in
+                            // the same collection cycle
+                            // instrumentIdentityToMetric[metric.InstrumentIdentity] may already
+                            // point to the new activated metric and not the old deactivated one.
+                            this.instrumentIdentityToMetric.TryUpdate(metric.InstrumentIdentity, null, metric);
+
+                            this.metricStreamNames.Remove(metric.InstrumentIdentity.MetricStreamName);
+
+                            // Defragment metrics list so storage can be reused on future metrics.
+                            for (int j = i + 1; j < target; j++)
+                            {
+                                this.metrics[j - 1] = this.metrics[j];
+                            }
+
+                            this.metrics[target - 1] = null;
+                            this.metricIndex--;
+                            i--;
+                        }
                     }
                 }
             }
@@ -244,29 +270,4 @@ private Batch<Metric> GetMetricsBatch()
             return default;
         }
     }
-
-    private void RemoveMetric(ref Metric? metric)
-    {
-        Debug.Assert(metric != null, "metric was null");
-
-        // TODO: This logic removes the metric. If the same
-        // metric is published again we will create a new metric
-        // for it. If this happens often we will run out of
-        // storage. Instead, should we keep the metric around
-        // and set a new start time + reset its data if it comes
-        // back?
-
-        OpenTelemetrySdkEventSource.Log.MetricInstrumentRemoved(metric!.Name, metric.MeterName);
-
-        // Note: This is using TryUpdate and NOT TryRemove because there is a
-        // race condition. If a metric is deactivated and then reactivated in
-        // the same collection cycle
-        // instrumentIdentityToMetric[metric.InstrumentIdentity] may already
-        // point to the new activated metric and not the old deactivated one.
-        this.instrumentIdentityToMetric.TryUpdate(metric.InstrumentIdentity, null, metric);
-
-        // Note: metric is a reference to the array storage so
-        // this clears the metric out of the array.
-        metric = null;
-    }
 }