NCBC-4160: AppTelemetry reports significantly lower metrics than it should

Motivation
----------
- Histogram bins are not thread-safe. Multiple threads can read `currentValue` simultaneously, both compute count + 1, and both write back the same value, so one increment is silently lost.
- Race condition in TryExportMetricsAndReset: An increment thread obtains a reference to metricSetA via GetOrAdd (inside lock), then the export thread takes a snapshot and clears the dictionary.
The increment thread thenwrites to metricSetA, but it's already been snapshotted and will be discarded after export, so the increment is permanently lost.
Or, a new entry metricSetB is added to the dictionary AFTER the snapshot ToArray() but BEFORE Clear(). It ends up in neither the snapshot nor the post-clear dictionary.
- WebSocket send failure permanently loses exported metrics
- TryExportMetricsAndReset clears the dictionary before the WebSocket send. If SendAsync fails (connection drop, timeout), those metrics are lost.
- Backoff never resets after a successful connection (so it grows to 1h backoff each time)

Changes
-------
- `AppTelemetryHistogramBins.cs`: Added a lock around the read-modify-write in IncrementCountAndSum() to prevent concurrent threads from overwriting each other's updates to bin counts and sums.
- `AppTelemetryCollector.cs`: Replaced the snapshot then clear pattern with an atomic dictionary swap under _metricsLock.
The old dictionary is swapped out and exported while new increments go to the fresh dictionary, removing the window where metrics could be lost between snapshot and clear.
Also fixed Disable() to use the same swap pattern under the metrics lock.
- `WebSocketClientHandler.cs`: Added a _pendingMetrics cache. If SendAsync fails after export, the serialized metrics are saved and retried on the next telemetry request instead of being lost.
- `WebSocketClientHandler.cs`: Added _attempt = 0 after a successful WebSocket connection opens, so the backoff resets to 100ms instead of staying at the max (up to 1h) after a connection terminates.

Change-Id: I1a96ddbd0c908b1ec90a20868aed264ca499b607
Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/241592
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: David Kelly <davidmichaelkelly@gmail.com>

Author: emilienbev

src/Couchbase/Core/Diagnostics/Metrics/AppTelemetry/AppTelemetryCollector.cs

@@ -22,8 +22,9 @@ internal class AppTelemetryCollector : IAppTelemetryCollector
     private readonly object _metricsLock = new();
     private readonly Uri? _endpoint;
     private CancellationTokenSource? _webSocketTokenSource;
+    private ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet> _metricSets = new();
 
-    public ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet> MetricSets { get; } = new();
+    internal ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet> MetricSets => _metricSets;
 
     public AppTelemetryCollector()
     {
@@ -75,7 +76,10 @@ public void Disable()
         {
             _enabled = false;
             _webSocketTokenSource?.Cancel();
-            MetricSets.Clear();
+        }
+        lock (_metricsLock)
+        {
+            _metricSets = new ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet>();
         }
     }
 
@@ -125,8 +129,7 @@ public void IncrementHistogram(AppTelemetryRequestType name, TimeSpan? operation
         AppTelemetryMetricSet metricSet;
         lock (_metricsLock)
         {
-            // Ensures we don't add during a Clear
-            metricSet = MetricSets.GetOrAdd(targetKey, _ => new AppTelemetryMetricSet());
+            metricSet = _metricSets.GetOrAdd(targetKey, _ => new AppTelemetryMetricSet());
         }
 
         metricSet.IncrementHistogram(name, operationLatency.Value);
@@ -142,27 +145,25 @@ public void IncrementCounter(AppTelemetryServiceType serviceType, AppTelemetryCo
         AppTelemetryMetricSet metricSet;
         lock (_metricsLock)
         {
-            // Ensures we don't add during a Clear
-            metricSet = MetricSets.GetOrAdd(targetKey, _ => new AppTelemetryMetricSet());
+            metricSet = _metricSets.GetOrAdd(targetKey, _ => new AppTelemetryMetricSet());
         }
         metricSet.IncrementCounter(serviceType, counterType);
     }
 
     public bool TryExportMetricsAndReset(out string metricsString)
     {
         metricsString = string.Empty;
-        if (MetricSets.IsEmpty) return false;
-
-        var sb = new StringBuilder();
-
-        var snapshot = MetricSets.ToArray();
 
+        ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet> oldMetrics;
         lock (_metricsLock)
         {
-            MetricSets.Clear();
+            if (_metricSets.IsEmpty) return false;
+            oldMetrics = _metricSets;
+            _metricSets = new ConcurrentDictionary<NodeAndBucket, AppTelemetryMetricSet>();
         }
 
-        foreach (var exported in snapshot.Select(entry => entry.Value.ExportAllMetrics(entry.Key)))
+        var sb = new StringBuilder();
+        foreach (var exported in oldMetrics.Select(entry => entry.Value.ExportAllMetrics(entry.Key)))
         {
             sb.Append(exported);
         }

src/Couchbase/Core/Diagnostics/Metrics/AppTelemetry/AppTelemetryHistogramBins.cs

@@ -1,13 +1,19 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using Couchbase.Core.Compatibility;
 
 namespace Couchbase.Core.Diagnostics.Metrics.AppTelemetry;
 
 [InterfaceStability(Level.Volatile)]
 internal class AppTelemetryHistogramBins
 {
+#if NET10_0_OR_GREATER
+    private readonly Lock _binLock = new();
+#else
+    private readonly object _binLock = new();
+#endif
     /// <summary>
     /// Count of all operations. This value is the cumulative sum of all smaller bins.
     /// Sum of all operations' durations in seconds.
@@ -32,16 +38,19 @@ public void IncrementCountAndSum(TimeSpan operationLatency)
     {
         var opLatency = operationLatency.TotalMilliseconds;
 
-        // For each bin, if the operation latency is less than or equal to the bin's upper bound,
-        // increment the count and add to the sum
-        foreach (var bin in Bins)
+        lock (_binLock)
         {
-            var le = bin.Keys.First(); // There's only 1 key per bin
+            // For each bin, if the operation latency is less than or equal to the bin's upper bound,
+            // increment the count and add to the sum
+            foreach (var bin in Bins)
+            {
+                var le = bin.Keys.First(); // There's only 1 key per bin
 
-            if (!(opLatency <= le)) continue;
-            var currentValue = bin[le];
-            bin[le] = new KeyValuePair<uint, double>(currentValue.Key + 1, currentValue.Value + opLatency);
-            break; //Break immediately after incrementing the bin, then we'll sum each bin cumulatively in the export method.
+                if (!(opLatency <= le)) continue;
+                var currentValue = bin[le];
+                bin[le] = new KeyValuePair<uint, double>(currentValue.Key + 1, currentValue.Value + opLatency);
+                break; //Break immediately after incrementing the bin, then we'll sum each bin cumulatively in the export method.
+            }
         }
     }
 }

src/Couchbase/Core/Diagnostics/Metrics/AppTelemetry/WebSocketClientHandler.cs

@@ -38,6 +38,7 @@ internal class WebSocketClientHandler : IDisposable
     private readonly IRedactor _redactor;
     private int _attempt = 0;
     private readonly int _clampedExponent = 0;
+    private string? _pendingMetrics;
     private Uri? Endpoint => _appTelemetryCollector.Endpoint(_attempt);
 
     public WebSocketClientHandler(IAppTelemetryCollector appTelemetryCollector)
@@ -69,6 +70,7 @@ public async Task StartAsync(CancellationToken cancellationToken = default)
 
                 if (_webSocket?.State == WebSocketState.Open)
                 {
+                    _attempt = 0;
                     await ReceiveAsync(cancellationToken).ConfigureAwait(false);
                 }
             }
@@ -156,20 +158,42 @@ private async Task HandleMessage(byte[] message, CancellationToken cancellationT
 
     private async Task SendTelemetryAsync(CancellationToken cancellationToken)
     {
-        if (_appTelemetryCollector.TryExportMetricsAndReset(out var metrics))
+        string? metrics;
+
+        if (_pendingMetrics != null)
+        {
+            // Retry previously unsent metrics before exporting new ones.
+            metrics = _pendingMetrics;
+        }
+        else if (_appTelemetryCollector.TryExportMetricsAndReset(out var newMetrics))
+        {
+            metrics = newMetrics;
+        }
+        else
         {
-            var metricsBytes = Encoding.UTF8.GetBytes(metrics);
-            var response = new byte[1 + metricsBytes.Length];
-            response[0] = SuccessOpcode;
-            metricsBytes.CopyTo(response, 1);
+            return;
+        }
+
+        var metricsBytes = Encoding.UTF8.GetBytes(metrics);
+        var response = new byte[1 + metricsBytes.Length];
+        response[0] = SuccessOpcode;
+        metricsBytes.CopyTo(response, 1);
 
-            _logger.LogDebug("Sending AppTelemetry metrics {Metrics}", metrics); //TODO: this is for debugging
+        _logger.LogTrace("Sending AppTelemetry metrics {Metrics}", metrics);
 
+        try
+        {
             await _webSocket!.SendAsync(
                 new ArraySegment<byte>(response),
                 WebSocketMessageType.Binary,
                 true,
                 cancellationToken).ConfigureAwait(false);
+            _pendingMetrics = null;
+        }
+        catch
+        {
+            _pendingMetrics = metrics;
+            throw;
         }
     }