[OTEL] Metrics API Support - Collection Implementation (#7511)

## Summary of changes
This PR implements the metrics collection foundation for OpenTelemetry
Metrics API supportKey additions include:

- MetricReader: Component for collecting instrument measurements and
aggregating them
- MetricReaderHandler: High-performance measurement processing with
proper aggregation (Sum, Last Value, Histogram)
- MetricPoint: Thread-safe data structure for metric aggregation with
histogram bucket support
- Testing: Unit test covering all 7 instrument types (4 sync + 3 async)
with exact OTLP value verification

This is the second of a series of 3 PRs to complete the OpenTelemetry
Metrics API support.
1. [[OTEL] Metrics API Support - Configurations & Telemetry
#7420](#7420)
2. [[OTEL] Metrics API Support - Collection Implementation
#7511](#7511) <- You are
here.
3. [[OTEL] Metrics API Support - Metrics Reader & OTLP Exporter
#7514](#7514)


## Reason for change
The [RFC] OpenTelemetry Metrics API support requires implementing
metrics collection that's compatible with the OpenTelemetry data model.
This PR provides the collection foundation that will enable customers to
use .NET's `System.Diagnostics.Metrics` API with Datadog as a drop-in
replacement for the OpenTelemetry SDK.

Business Impact: Enables the collection phase of OpenTelemetry Metrics
API support, laying the groundwork for OTLP export in the next PR.

## Implementation details
✅ Metrics Collection Components:
- MetricReader - RFC-compliant component for initialization and async
collection
- MetricReaderHandler - Measurement processing and aggregation
- MetricPoint - Aggregation data structure with histogram bucket support

✅ Aggregation Mapping:
- Counter/Asynchronous Counter → Sum Aggregation → Sum Metric Point ✅
- UpDownCounter/Asynchronous UpDownCounter → Sum Aggregation → Sum
Metric Point ✅
- Gauge/Asynchronous Gauge → Last Value Aggregation → Gauge Metric Point
✅
- Histogram → Explicit Bucket Histogram Aggregation → Histogram Metric
Point ✅

## Test coverage
Added unit test `MeterListenerCapturesAllMetrics()` that verifies:
- All 7 instrument types (Counter, Histogram, Gauge, UpDownCounter + 3
Async variants)
- Value verification matching OTLP snapshot expectations
Histogram bucket logic verification
- Sync + Async instrument collection
- Tag capture and temporality settings

## Other details
Jira: [APMAPI-1572](https://datadoghq.atlassian.net/browse/APMAPI-1572)

<!-- Fixes #{issue} -->

<!--  ⚠️ Note:

Where possible, please obtain 2 approvals prior to merging. Unless
CODEOWNERS specifies otherwise, for external teams it is typically best
to have one review from a team member, and one review from apm-dotnet.
Trivial changes do not require 2 reviews.

MergeQueue is NOT enabled in this repository. If you have write access
to the repo, the PR has 1-2 approvals (see above), and all of the
required checks have passed, you can use the Squash and Merge button to
merge the PR. If you don't have write access, or you need help, reach
out in the #apm-dotnet channel in Slack.
-->


[APMAPI-1572]:
https://datadoghq.atlassian.net/browse/APMAPI-1572?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Author: link04

tracer/src/Datadog.Trace.Trimming/build/Datadog.Trace.Trimming.xml

@@ -239,6 +239,10 @@
    <assembly fullname="System.Diagnostics.DiagnosticSource">
       <type fullname="System.Diagnostics.DiagnosticListener" />
       <type fullname="System.Diagnostics.DistributedContextPropagator" />
+      <type fullname="System.Diagnostics.Metrics.Instrument" />
+      <type fullname="System.Diagnostics.Metrics.MeasurementCallback`1" />
+      <type fullname="System.Diagnostics.Metrics.Meter" />
+      <type fullname="System.Diagnostics.Metrics.MeterListener" />
    </assembly>
    <assembly fullname="System.Diagnostics.Process">
       <type fullname="System.Diagnostics.Process" />

tracer/src/Datadog.Trace/ClrProfiler/Instrumentation.cs

@@ -378,25 +378,23 @@ internal static void InitializeNoNativeParts(Stopwatch sw = null)
             {
                 if (Tracer.Instance.Settings.OpenTelemetryMetricsEnabled)
                 {
-                    Log.Debug("Initializing OTel Metrics Exporter.");
                     if (Tracer.Instance.Settings.OpenTelemetryMeterNames.Length > 0)
                     {
+                        Log.Debug("Initializing OTel Metrics Exporter.");
                         OTelMetrics.OtlpMetricsExporter.Initialize();
-                    }
-                    else
-                    {
-                        Log.Debug("No meters were found for DD_METRICS_OTEL_METER_NAMES, OTel Metrics Exporter won't be initialized.");
+                        Log.Debug("Initializing OTel Metrics Reader.");
+                        OTelMetrics.MetricReader.Initialize();
                     }
                 }
             }
             catch (Exception ex)
             {
-                Log.Error(ex, "Error initializing OTel Metrics Exporter");
+                Log.Error(ex, "Error initializing OTel Metrics Reader");
             }
 #else
             if (Tracer.Instance.Settings.OpenTelemetryMetricsEnabled)
             {
-                Log.Information("Unable to initialize OTel Metrics collection, this is only available starting with .NET 6.0..");
+                Log.Information("Unable to initialize OTel Metrics collection, this is only available starting with .NET 6.0.");
             }
 #endif
 

tracer/src/Datadog.Trace/OTelMetrics/AggregationTemporality.cs

@@ -0,0 +1,27 @@
+// <copyright file="AggregationTemporality.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#if NET6_0_OR_GREATER
+
+#nullable enable
+
+namespace Datadog.Trace.OTelMetrics;
+
+/// <summary>
+/// Represents the aggregation temporality of a metric.
+/// </summary>
+public enum AggregationTemporality
+{
+    /// <summary>
+    /// Delta temporality, representing changes since the last measurement.
+    /// </summary>
+    Delta = 0,
+
+    /// <summary>
+    /// Cumulative temporality, representing the total value since the start.
+    /// </summary>
+    Cumulative = 1,
+}
+#endif

tracer/src/Datadog.Trace/OTelMetrics/InstrumentType.cs

@@ -0,0 +1,52 @@
+// <copyright file="InstrumentType.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#if NET6_0_OR_GREATER
+
+#nullable enable
+
+namespace Datadog.Trace.OTelMetrics;
+
+/// <summary>
+/// Represents the type of OpenTelemetry instrument
+/// </summary>
+internal enum InstrumentType
+{
+    /// <summary>
+    /// Counter instrument - monotonic, additive
+    /// </summary>
+    Counter = 0,
+
+    /// <summary>
+    /// UpDownCounter instrument - non-monotonic, additive
+    /// </summary>
+    UpDownCounter = 1,
+
+    /// <summary>
+    /// Histogram instrument - statistical distribution
+    /// </summary>
+    Histogram = 2,
+
+    /// <summary>
+    /// Asynchronous Counter instrument - monotonic, additive, callback-based
+    /// </summary>
+    ObservableCounter = 3,
+
+    /// <summary>
+    /// Asynchronous UpDownCounter instrument - non-monotonic, additive, callback-based
+    /// </summary>
+    ObservableUpDownCounter = 4,
+
+    /// <summary>
+    /// Gauge instrument - non-additive, last value
+    /// </summary>
+    Gauge = 5,
+
+    /// <summary>
+    /// Asynchronous Gauge instrument - non-additive, last value, callback-based
+    /// </summary>
+    ObservableGauge = 6
+}
+#endif

tracer/src/Datadog.Trace/OTelMetrics/MetricPoint.cs

@@ -0,0 +1,90 @@
+// <copyright file="MetricPoint.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#if NET6_0_OR_GREATER
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Datadog.Trace.OTelMetrics;
+
+internal class MetricPoint(string instrumentName, string meterName, InstrumentType instrumentType, AggregationTemporality? temporality, Dictionary<string, object?> tags)
+{
+    internal static readonly double[] DefaultHistogramBounds = [0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000];
+    private readonly long[] _runningBucketCounts = instrumentType == InstrumentType.Histogram ? new long[DefaultHistogramBounds.Length + 1] : [];
+    private readonly object _histogramLock = new();
+    private long _runningCountValue;
+    private double _runningDoubleValue;
+    private double _runningMin = double.PositiveInfinity;
+    private double _runningMax = double.NegativeInfinity;
+
+    public string InstrumentName { get; } = instrumentName;
+
+    public string MeterName { get; } = meterName;
+
+    public InstrumentType InstrumentType { get; } = instrumentType;
+
+    public AggregationTemporality? AggregationTemporality { get; } = temporality;
+
+    public Dictionary<string, object?> Tags { get; } = tags;
+
+    internal long RunningCount => _runningCountValue;
+
+    internal double RunningSum => _runningDoubleValue;
+
+    internal double RunningMin => _runningMin;
+
+    internal double RunningMax => _runningMax;
+
+    internal long[] RunningBucketCounts => _runningBucketCounts;
+
+    internal void UpdateCounter(double value)
+    {
+        lock (_histogramLock)
+        {
+            _runningDoubleValue += value;
+        }
+    }
+
+    internal void UpdateGauge(double value)
+    {
+        Interlocked.Exchange(ref _runningDoubleValue, value);
+    }
+
+    internal void UpdateHistogram(double value)
+    {
+        var bucketIndex = FindBucketIndex(value);
+
+        lock (_histogramLock)
+        {
+            unchecked
+            {
+                _runningCountValue++;
+                _runningDoubleValue += value;
+                _runningBucketCounts[bucketIndex]++;
+            }
+
+            _runningMin = Math.Min(_runningMin, value);
+            _runningMax = Math.Max(_runningMax, value);
+        }
+    }
+
+    private static int FindBucketIndex(double value)
+    {
+        for (var i = 0; i < DefaultHistogramBounds.Length; i++)
+        {
+            if (value <= DefaultHistogramBounds[i])
+            {
+                return i;
+            }
+        }
+
+        return DefaultHistogramBounds.Length;
+    }
+}
+#endif

tracer/src/Datadog.Trace/OTelMetrics/MetricReader.cs

@@ -0,0 +1,88 @@
+// <copyright file="MetricReader.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#if NET6_0_OR_GREATER
+
+#nullable enable
+
+using System;
+using System.Threading;
+using Datadog.Trace.Logging;
+
+namespace Datadog.Trace.OTelMetrics;
+
+internal static class MetricReader
+{
+    private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor(typeof(MetricReader));
+
+    private static System.Diagnostics.Metrics.MeterListener? _meterListenerInstance;
+    private static int _initialized;
+    private static int _stopped;
+
+    public static bool IsRunning =>
+        Interlocked.CompareExchange(ref _initialized, 1, 1) == 1 &&
+        Interlocked.CompareExchange(ref _stopped, 0, 0) == 0;
+
+    public static void Initialize()
+    {
+        if (Interlocked.CompareExchange(ref _initialized, 1, 0) == 1)
+        {
+            return;
+        }
+
+        var meterListener = new System.Diagnostics.Metrics.MeterListener();
+
+#if NET6_0 || NET7_0 || NET8_0
+        // Ensures instruments are fully de-registered on Dispose() for 6–8
+        // Static lambda => no captures/allocations
+        meterListener.MeasurementsCompleted = static (_, __) => { };
+#endif
+
+        meterListener.InstrumentPublished = MetricReaderHandler.OnInstrumentPublished;
+
+        meterListener.SetMeasurementEventCallback<byte>(MetricReaderHandler.OnMeasurementRecordedByte);
+        meterListener.SetMeasurementEventCallback<short>(MetricReaderHandler.OnMeasurementRecordedShort);
+        meterListener.SetMeasurementEventCallback<int>(MetricReaderHandler.OnMeasurementRecordedInt);
+        meterListener.SetMeasurementEventCallback<long>(MetricReaderHandler.OnMeasurementRecordedLong);
+        meterListener.SetMeasurementEventCallback<float>(MetricReaderHandler.OnMeasurementRecordedFloat);
+        meterListener.SetMeasurementEventCallback<double>(MetricReaderHandler.OnMeasurementRecordedDouble);
+
+        meterListener.Start();
+
+        Interlocked.Exchange(ref _meterListenerInstance, meterListener);
+        Interlocked.Exchange(ref _stopped, 0);
+
+        Log.Debug("MeterListener initialized successfully.");
+    }
+
+    public static void Stop()
+    {
+        var listener = Interlocked.Exchange(ref _meterListenerInstance, null);
+        if (listener is IDisposable disposableListener)
+        {
+            disposableListener.Dispose();
+            Interlocked.Exchange(ref _stopped, 1);
+            Interlocked.Exchange(ref _initialized, 0);
+            Log.Debug("MeterListener stopped.");
+        }
+    }
+
+    internal static void CollectObservableInstruments()
+    {
+        if (_meterListenerInstance != null)
+        {
+            try
+            {
+                _meterListenerInstance.RecordObservableInstruments();
+            }
+            catch (Exception ex)
+            {
+                Log.Warning(ex, "Error collecting observable instruments.");
+            }
+        }
+    }
+}
+#endif
+