[HTTP] High cardinality metrics to observable

src/libraries/System.Net.Http/src/System/Net/Http/Metrics/MetricsHandler.cs

@@ -1,6 +1,7 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
@@ -10,10 +11,125 @@
 
 namespace System.Net.Http.Metrics
 {
+    /// <summary>
+    /// Represents a unique combination of tags for tracking active requests.
+    /// </summary>
+    internal readonly struct ActiveRequestsTagKey : IEquatable<ActiveRequestsTagKey>
+    {
+        public readonly string? Scheme;
+        public readonly string? Host;
+        public readonly int Port;
+        public readonly string Method;
+        private readonly int _hashCode;
+
+        public ActiveRequestsTagKey(string? scheme, string? host, int port, string method)
+        {
+            Scheme = scheme;
+            Host = host;
+            Port = port;
+            Method = method;
+            _hashCode = HashCode.Combine(scheme, host, port, method);
+        }
+
+        public bool Equals(ActiveRequestsTagKey other) =>
+            Scheme == other.Scheme &&
+            Host == other.Host &&
+            Port == other.Port &&
+            Method == other.Method;
+
+        public override bool Equals(object? obj) => obj is ActiveRequestsTagKey other && Equals(other);
+
+        public override int GetHashCode() => _hashCode;
+
+        public TagList ToTagList()
+        {
+            TagList tags = default;
+            if (Scheme is not null)
+            {
+                tags.Add("url.scheme", Scheme);
+                tags.Add("server.address", Host);
+                tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(Port));
+            }
+            tags.Add("http.request.method", Method);
+            return tags;
+        }
+
+        public override string ToString() =>
+            $"{Method}{(Scheme is not null ? $" {Scheme}://{Host}:{Port}" : "")}";
+    }
+
+    /// <summary>
+    /// Thread-safe tracker for active request counts by tag combination.
+    /// </summary>
+    internal sealed class ActiveRequestsTracker
+    {
+        private readonly ConcurrentDictionary<ActiveRequestsTagKey, long> _counts = new();
+
+        /// <summary>
+        /// Increments the count for the specified tag combination.
+        /// </summary>
+        public void Increment(in ActiveRequestsTagKey key)
+        {
+            _counts.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1);
+        }
+
+        /// <summary>
+        /// Decrements the count for the specified tag combination.
+        /// Removes the entry if the count reaches zero.
+        /// </summary>
+        public void Decrement(in ActiveRequestsTagKey key)
+        {
+            // We need to atomically decrement and remove if zero.
+            // Use a spin loop with TryGetValue/TryUpdate/TryRemove to handle this safely.
+            while (true)
+            {
+                if (!_counts.TryGetValue(key, out long currentValue))
+                {
+                    // Key doesn't exist, nothing to decrement.
+                    // This shouldn't happen in normal operation but we handle it gracefully.
+                    Debug.Fail($"Decrement for non-existing request {key}");
+                    return;
+                }
+
+                if (currentValue <= 1)
+                {
+                    // Try to remove the entry since it will become zero.
+                    // Use the overload that checks the current value to ensure atomicity.
+                    if (_counts.TryRemove(new KeyValuePair<ActiveRequestsTagKey, long>(key, currentValue)))
+                    {
+                        return;
+                    }
+                    // Another thread modified the value, retry.
+                }
+                else
+                {
+                    // Try to decrement the value.
+                    if (_counts.TryUpdate(key, currentValue - 1, currentValue))
+                    {
+                        return;
+                    }
+                    // Another thread modified the value, retry.
+                }
+            }
+        }
+
+        /// <summary>
+        /// Returns measurements for all tag combinations with non-zero counts.
+        /// </summary>
+        public IEnumerable<Measurement<long>> GetMeasurements()
+        {
+            foreach (KeyValuePair<ActiveRequestsTagKey, long> entry in _counts)
+            {
+                yield return new Measurement<long>(entry.Value, entry.Key.ToTagList());
+            }
+        }
+    }
+
     internal sealed class MetricsHandler : HttpMessageHandlerStage
     {
         private readonly HttpMessageHandler _innerHandler;
-        private readonly UpDownCounter<long> _activeRequests;
+        private readonly ActiveRequestsTracker _activeRequestsTracker = new();
+        private readonly ObservableUpDownCounter<long> _activeRequests;
         private readonly Histogram<double> _requestsDuration;
         private readonly IWebProxy? _proxy;
 
@@ -27,8 +143,9 @@ public MetricsHandler(HttpMessageHandler innerHandler, IMeterFactory? meterFacto
             meter = meterFactory?.Create("System.Net.Http") ?? SharedMeter.Instance;
 
             // Meter has a cache for the instruments it owns
-            _activeRequests = meter.CreateUpDownCounter<long>(
+            _activeRequests = meter.CreateObservableUpDownCounter<long>(
                 "http.client.active_requests",
+                observeValues: _activeRequestsTracker.GetMeasurements,
                 unit: "{request}",
                 description: "Number of outbound HTTP requests that are currently active on the client.");
             _requestsDuration = meter.CreateHistogram<double>(
@@ -94,27 +211,25 @@ protected override void Dispose(bool disposing)
 
             if (recordCurrentRequests)
             {
-                TagList tags = InitializeCommonTags(request);
-                _activeRequests.Add(1, tags);
+                _activeRequestsTracker.Increment(CreateActiveRequestsTagKey(request));
             }
 
             return (startTimestamp, recordCurrentRequests);
         }
 
         private void RequestStop(HttpRequestMessage request, HttpResponseMessage? response, Exception? exception, long startTimestamp, bool recordCurrentRequests)
         {
-            TagList tags = InitializeCommonTags(request);
-
             if (recordCurrentRequests)
             {
-                _activeRequests.Add(-1, tags);
+                _activeRequestsTracker.Decrement(CreateActiveRequestsTagKey(request));
             }
 
             if (!_requestsDuration.Enabled)
             {
                 return;
             }
 
+            TagList tags = InitializeCommonTags(request);
             if (response is not null)
             {
                 tags.Add("http.response.status_code", DiagnosticsHelper.GetBoxedInt32((int)response.StatusCode));
@@ -154,6 +269,24 @@ private TagList InitializeCommonTags(HttpRequestMessage request)
             return tags;
         }
 
+        private ActiveRequestsTagKey CreateActiveRequestsTagKey(HttpRequestMessage request)
+        {
+            string? scheme = null;
+            string? host = null;
+            int port = 0;
+
+            if (request.RequestUri is Uri requestUri && requestUri.IsAbsoluteUri)
+            {
+                scheme = requestUri.Scheme;
+                host = DiagnosticsHelper.GetServerAddress(request, _proxy);
+                port = requestUri.Port;
+            }
+
+            string method = (string)DiagnosticsHelper.GetMethodTag(request.Method, out _).Value!;
+
+            return new ActiveRequestsTagKey(scheme, host, port, method);
+        }
+
         private sealed class SharedMeter : Meter
         {
             public static Meter Instance { get; } = new SharedMeter();

src/libraries/System.Net.Http/src/System/Net/Http/SocketsHttpHandler/Metrics/ConnectionMetrics.cs

@@ -10,11 +10,11 @@ internal sealed class ConnectionMetrics
     {
         private readonly SocketsHttpHandlerMetrics _metrics;
         private readonly bool _openConnectionsEnabled;
-        private readonly object _protocolVersionTag;
-        private readonly object _schemeTag;
-        private readonly object _hostTag;
-        private readonly object _portTag;
-        private readonly object? _peerAddressTag;
+        private readonly string _protocolVersionTag;
+        private readonly string _schemeTag;
+        private readonly string _hostTag;
+        private readonly int _portTag;
+        private readonly string? _peerAddressTag;
         private bool _currentlyIdle;
 
         public ConnectionMetrics(SocketsHttpHandlerMetrics metrics, string protocolVersion, string scheme, string host, int port, string? peerAddress)
@@ -24,7 +24,7 @@ public ConnectionMetrics(SocketsHttpHandlerMetrics metrics, string protocolVersi
             _protocolVersionTag = protocolVersion;
             _schemeTag = scheme;
             _hostTag = host;
-            _portTag = DiagnosticsHelper.GetBoxedInt32(port);
+            _portTag = port;
             _peerAddressTag = peerAddress;
         }
 
@@ -36,7 +36,7 @@ private TagList GetTags()
             tags.Add("network.protocol.version", _protocolVersionTag);
             tags.Add("url.scheme", _schemeTag);
             tags.Add("server.address", _hostTag);
-            tags.Add("server.port", _portTag);
+            tags.Add("server.port", DiagnosticsHelper.GetBoxedInt32(_portTag));
 
             if (_peerAddressTag is not null)
             {
@@ -46,45 +46,47 @@ private TagList GetTags()
             return tags;
         }
 
-        private static KeyValuePair<string, object?> GetStateTag(bool idle) => new KeyValuePair<string, object?>("http.connection.state", idle ? "idle" : "active");
+        private OpenConnectionsTagKey CreateTagKey(bool idle) =>
+            new OpenConnectionsTagKey(_protocolVersionTag, _schemeTag, _hostTag, _portTag, idle, _peerAddressTag);
 
         public void ConnectionEstablished()
         {
             if (_openConnectionsEnabled)
             {
-                _currentlyIdle = true;
-                TagList tags = GetTags();
-                tags.Add(GetStateTag(idle: true));
-                _metrics.OpenConnections.Add(1, tags);
+                lock (this)
+                {
+                    _currentlyIdle = true;
+                    _metrics.OpenConnectionsTracker.Increment(CreateTagKey(idle: true));
+                }
             }
         }
 
         public void ConnectionClosed(long durationMs)
         {
-            TagList tags = GetTags();
-
             if (_metrics.ConnectionDuration.Enabled)
             {
-                _metrics.ConnectionDuration.Record(durationMs / 1000d, tags);
+                _metrics.ConnectionDuration.Record(durationMs / 1000d, GetTags());
             }
 
             if (_openConnectionsEnabled)
             {
-                tags.Add(GetStateTag(idle: _currentlyIdle));
-                _metrics.OpenConnections.Add(-1, tags);
+                lock (this)
+                {
+                    _metrics.OpenConnectionsTracker.Decrement(CreateTagKey(idle: _currentlyIdle));
+                }
             }
         }
 
         public void IdleStateChanged(bool idle)
         {
             if (_openConnectionsEnabled && _currentlyIdle != idle)
             {
-                _currentlyIdle = idle;
-                TagList tags = GetTags();
-                tags.Add(GetStateTag(idle: !idle));
-                _metrics.OpenConnections.Add(-1, tags);
-                tags[tags.Count - 1] = GetStateTag(idle: idle);
-                _metrics.OpenConnections.Add(1, tags);
+                lock (this)
+                {
+                    _currentlyIdle = idle;
+                    _metrics.OpenConnectionsTracker.Decrement(CreateTagKey(idle: !idle));
+                    _metrics.OpenConnectionsTracker.Increment(CreateTagKey(idle: idle));
+                }
             }
         }
     }