Move AsyncBatchingWorkQueue usage in telemetry to TelemetryLogging le…

…vel (#73287)

* Move AsyncBatchingWorkQueue usage in telemetry to TelemetryLogging level

Doing this as I noticed a large (10x) difference in the number of requestcounter and requestduration events in our dashboard. These events go through the system in slightly different fashions, requestcounter goes through standard telemetry calls on disposal whereas requestduration goes through the aggregated telemetry logging.

Both of these are intended to aggregate multiple calls into a single message, but the cadence at which they send telemetry differs. The requestduration method will flush both on project/VS close and every 30 minutes, whereas the requestcounter method only flushes on project/VS close.

I've noticed that VS shutdown is now more abrupt than previously, often not giving our disposers a chance to send out telemetry. This is why I believe there is such a large discepency in the telemetry numbers for these methods, when they should be the same. This PR allows for the requestcounter messages to also be sent out every 30 minutes in case the disposal codepath isn't executed.

src/Features/LanguageServer/Protocol/Handler/RequestTelemetryLogger.cs

@@ -36,6 +36,8 @@ public RequestTelemetryLogger(string serverTypeName)
         _requestCounters = new();
         _findDocumentResults = new();
         _usedForkedSolutionCounter = new();
+
+        TelemetryLogging.Flushed += OnFlushed;
     }
 
     public void UpdateFindDocumentTelemetryData(bool success, string? workspaceKind)
@@ -92,6 +94,14 @@ public void Dispose()
             return;
         }
 
+        // Flush all telemetry logged through TelemetryLogging
+        TelemetryLogging.Flush();
+
+        TelemetryLogging.Flushed -= OnFlushed;
+    }
+
+    private void OnFlushed(object? sender, EventArgs e)
+    {
         foreach (var kvp in _requestCounters)
         {
             TelemetryLogging.Log(FunctionId.LSP_RequestCounter, KeyValueLogMessage.Create(LogType.Trace, m =>
@@ -124,9 +134,6 @@ public void Dispose()
             }
         }));
 
-        // Flush all telemetry logged through TelemetryLogging
-        TelemetryLogging.Flush();
-
         _requestCounters.Clear();
     }
 

src/VisualStudio/Core/Def/Telemetry/Shared/AggregatingTelemetryLog.cs

@@ -28,7 +28,6 @@ internal sealed class AggregatingTelemetryLog : ITelemetryLog
     private readonly HistogramConfiguration? _histogramConfiguration;
     private readonly string _eventName;
     private readonly FunctionId _functionId;
-    private readonly AggregatingTelemetryLogManager _aggregatingTelemetryLogManager;
     private readonly object _flushLock;
 
     private ImmutableDictionary<string, (IHistogram<long> Histogram, TelemetryEvent TelemetryEvent, object Lock)> _histograms = ImmutableDictionary<string, (IHistogram<long>, TelemetryEvent, object)>.Empty;
@@ -40,7 +39,7 @@ internal sealed class AggregatingTelemetryLog : ITelemetryLog
     /// <param name="functionId">Used to derive meter name</param>
     /// <param name="bucketBoundaries">Optional values indicating bucket boundaries in milliseconds. If not specified, 
     /// all histograms created will use the default histogram configuration</param>
-    public AggregatingTelemetryLog(TelemetrySession session, FunctionId functionId, double[]? bucketBoundaries, AggregatingTelemetryLogManager aggregatingTelemetryLogManager)
+    public AggregatingTelemetryLog(TelemetrySession session, FunctionId functionId, double[]? bucketBoundaries)
     {
         var meterName = TelemetryLogger.GetPropertyName(functionId, "meter");
         var meterProvider = new VSTelemetryMeterProvider();
@@ -49,7 +48,6 @@ public AggregatingTelemetryLog(TelemetrySession session, FunctionId functionId,
         _meter = meterProvider.CreateMeter(meterName, version: MeterVersion);
         _eventName = TelemetryLogger.GetEventName(functionId);
         _functionId = functionId;
-        _aggregatingTelemetryLogManager = aggregatingTelemetryLogManager;
         _flushLock = new();
 
         if (bucketBoundaries != null)
@@ -104,8 +102,6 @@ public void Log(KeyValueLogMessage logMessage)
         {
             histogram.Record(value);
         }
-
-        _aggregatingTelemetryLogManager.EnsureTelemetryWorkQueued();
     }
 
     public IDisposable? LogBlockTime(KeyValueLogMessage logMessage, int minThresholdMs)

src/VisualStudio/Core/Def/Telemetry/Shared/AggregatingTelemetryLogManager.cs

@@ -2,62 +2,36 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.
 
-using System;
 using System.Collections.Immutable;
-using System.Threading;
-using System.Threading.Tasks;
 using Microsoft.CodeAnalysis.Internal.Log;
-using Microsoft.CodeAnalysis.Shared.TestHooks;
 using Microsoft.VisualStudio.Telemetry;
-using Roslyn.Utilities;
 
 namespace Microsoft.CodeAnalysis.Telemetry;
 
 /// <summary>
-/// Manages creation and obtaining aggregated telemetry logs. Also, notifies logs to
-/// send aggregated events every 30 minutes.
+/// Manages creation and obtaining aggregated telemetry logs.
 /// </summary>
 internal sealed class AggregatingTelemetryLogManager
 {
-    private static readonly TimeSpan s_batchedTelemetryCollectionPeriod = TimeSpan.FromMinutes(30);
-
     private readonly TelemetrySession _session;
-    private readonly AsyncBatchingWorkQueue _postTelemetryQueue;
 
     private ImmutableDictionary<FunctionId, AggregatingTelemetryLog> _aggregatingLogs = ImmutableDictionary<FunctionId, AggregatingTelemetryLog>.Empty;
 
-    public AggregatingTelemetryLogManager(TelemetrySession session, IAsynchronousOperationListener asyncListener)
+    public AggregatingTelemetryLogManager(TelemetrySession session)
     {
         _session = session;
-
-        _postTelemetryQueue = new AsyncBatchingWorkQueue(
-            s_batchedTelemetryCollectionPeriod,
-            PostCollectedTelemetryAsync,
-            asyncListener,
-            CancellationToken.None);
     }
 
     public ITelemetryLog? GetLog(FunctionId functionId, double[]? bucketBoundaries)
     {
         if (!_session.IsOptedIn)
             return null;
 
-        return ImmutableInterlocked.GetOrAdd(ref _aggregatingLogs, functionId, functionId => new AggregatingTelemetryLog(_session, functionId, bucketBoundaries, this));
-    }
-
-    public void EnsureTelemetryWorkQueued()
-    {
-        // Ensure PostCollectedTelemetryAsync will get fired after the collection period.
-        _postTelemetryQueue.AddWork();
-    }
-
-    private ValueTask PostCollectedTelemetryAsync(CancellationToken token)
-    {
-        token.ThrowIfCancellationRequested();
-
-        Flush();
-
-        return ValueTaskFactory.CompletedTask;
+        return ImmutableInterlocked.GetOrAdd(
+            ref _aggregatingLogs,
+            functionId,
+            static (functionId, arg) => new AggregatingTelemetryLog(arg._session, functionId, arg.bucketBoundaries),
+            factoryArgument: (_session, bucketBoundaries));
     }
 
     public void Flush()

src/VisualStudio/Core/Def/Telemetry/Shared/TelemetryLogProvider.cs

@@ -17,17 +17,17 @@ internal sealed class TelemetryLogProvider : ITelemetryLogProvider
     private readonly AggregatingTelemetryLogManager _aggregatingTelemetryLogManager;
     private readonly VisualStudioTelemetryLogManager _visualStudioTelemetryLogManager;
 
-    private TelemetryLogProvider(TelemetrySession session, ILogger telemetryLogger, IAsynchronousOperationListener asyncListener)
+    private TelemetryLogProvider(TelemetrySession session, ILogger telemetryLogger)
     {
-        _aggregatingTelemetryLogManager = new AggregatingTelemetryLogManager(session, asyncListener);
+        _aggregatingTelemetryLogManager = new AggregatingTelemetryLogManager(session);
         _visualStudioTelemetryLogManager = new VisualStudioTelemetryLogManager(session, telemetryLogger);
     }
 
     public static TelemetryLogProvider Create(TelemetrySession session, ILogger telemetryLogger, IAsynchronousOperationListener asyncListener)
     {
-        var logProvider = new TelemetryLogProvider(session, telemetryLogger, asyncListener);
+        var logProvider = new TelemetryLogProvider(session, telemetryLogger);
 
-        TelemetryLogging.SetLogProvider(logProvider);
+        TelemetryLogging.SetLogProvider(logProvider, asyncListener);
 
         return logProvider;
     }

src/Workspaces/Core/Portable/Telemetry/TelemetryLogging.cs

@@ -3,26 +3,44 @@
 // See the LICENSE file in the project root for more information.
 
 using System;
+using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.CodeAnalysis.Internal.Log;
+using Microsoft.CodeAnalysis.Shared.TestHooks;
+using Roslyn.Utilities;
 
 namespace Microsoft.CodeAnalysis.Telemetry;
 
 /// <summary>
 /// Provides access to posting telemetry events or adding information
-/// to aggregated telemetry events.
+/// to aggregated telemetry events. Posts pending telemetry at 30
+/// minute intervals.
 /// </summary>
 internal static class TelemetryLogging
 {
     private static ITelemetryLogProvider? s_logProvider;
+    private static AsyncBatchingWorkQueue? s_postTelemetryQueue;
 
     public const string KeyName = "Name";
     public const string KeyValue = "Value";
     public const string KeyLanguageName = "LanguageName";
     public const string KeyMetricName = "MetricName";
 
-    public static void SetLogProvider(ITelemetryLogProvider logProvider)
+    public static event EventHandler<EventArgs>? Flushed;
+
+    public static void SetLogProvider(ITelemetryLogProvider logProvider, IAsynchronousOperationListener asyncListener)
     {
         s_logProvider = logProvider;
+
+        InterlockedOperations.Initialize(ref s_postTelemetryQueue, () =>
+            new AsyncBatchingWorkQueue(
+                TimeSpan.FromMinutes(30),
+                PostCollectedTelemetryAsync,
+                asyncListener,
+                CancellationToken.None));
+
+        // Add the initial item to the queue to ensure later processing.
+        s_postTelemetryQueue?.AddWork();
     }
 
     /// <summary>
@@ -112,5 +130,17 @@ public static void LogAggregated(FunctionId functionId, KeyValueLogMessage logMe
     public static void Flush()
     {
         s_logProvider?.Flush();
+
+        Flushed?.Invoke(null, EventArgs.Empty);
+    }
+
+    private static ValueTask PostCollectedTelemetryAsync(CancellationToken cancellationToken)
+    {
+        Flush();
+
+        // Ensure PostCollectedTelemetryAsync will get fired again after the collection period.
+        s_postTelemetryQueue?.AddWork();
+
+        return ValueTaskFactory.CompletedTask;
     }
 }