AppInsightsProvider.cs

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

#if NETCOREAPP
using System.Threading.Channels;
#endif

#if DEBUG
using Microsoft.Testing.Platform;
#endif
using Microsoft.Testing.Platform.Configurations;
using Microsoft.Testing.Platform.Helpers;
using Microsoft.Testing.Platform.Logging;
#if !NETCOREAPP
using Microsoft.Testing.Platform.Messages;
#endif
using Microsoft.Testing.Platform.Services;
using Microsoft.Testing.Platform.Telemetry;

namespace Microsoft.Testing.Extensions.Telemetry;

/// <summary>
/// Allows to log telemetry events via AppInsights.
/// </summary>
internal sealed partial class AppInsightsProvider :
    ITelemetryCollector,
#if NETCOREAPP
    IAsyncDisposable,
#endif
    IDisposable
{
    // Note: We're currently using the same environment variable as dotnet CLI.
    public static readonly string SessionIdEnvVar = "TESTINGPLATFORM_APPINSIGHTS_SESSIONID";

    // Allows us to correlate events produced from the same process.
    // Not calling this ProcessId, because it has a different meaning.
    private static readonly string CurrentReporterId = Guid.NewGuid().ToString();
    private readonly string _currentSessionId;
    private readonly bool _isCi;
    private readonly IEnvironment _environment;
    private readonly ITestApplicationCancellationTokenSource _testApplicationCancellationTokenSource;
    private readonly IClock _clock;
    private readonly ITelemetryInformation _telemetryInformation;
    private readonly ITelemetryClientFactory _telemetryClientFactory;
    private readonly bool _isDevelopmentRepository;
    private readonly ILogger<AppInsightsProvider> _logger;
    private readonly Task _telemetryTask;
    private readonly CancellationTokenSource _flushTimeoutOrStop = new();
#if NETCOREAPP
    private readonly Channel<(string EventName, IDictionary<string, object> ParamsMap)> _payloads;
#else
    private readonly SingleConsumerUnboundedChannel<(string EventName, IDictionary<string, object> ParamsMap)> _payloads;
#endif
#if DEBUG
    // Telemetry properties that are allowed to contain unhashed information.
    private static readonly HashSet<string> KnownUnhashedProperties =
    [
        TelemetryProperties.VersionPropertyName,
        TelemetryProperties.ReporterIdPropertyName,
        TelemetryProperties.SessionId,
        TelemetryProperties.HostProperties.TestingPlatformVersionPropertyName,
        TelemetryProperties.HostProperties.FrameworkDescriptionPropertyName,
        TelemetryProperties.HostProperties.OSDescriptionPropertyName,
        TelemetryProperties.HostProperties.RuntimeIdentifierPropertyName,
        TelemetryProperties.HostProperties.ApplicationModePropertyName,
        TelemetryProperties.HostProperties.ExitCodePropertyName,
        TelemetryProperties.HostProperties.ExtensionsPropertyName,

        // MSTest session telemetry (see MSTestTelemetryDataCollector). These carry aggregated
        // counts, anonymized SHA256 type hashes inside JSON envelopes, and well-known enum/string
        // values - none of them contain unhashed user-identifying data.
        "mstest.config_source",
        "mstest.attribute_usage",
        "mstest.custom_test_method_types",
        "mstest.custom_test_class_types",
        "mstest.assertion_usage",
        "mstest.setting.parallelization_scope",
    ];
#endif

    private ITelemetryClient? _client;
    private bool _isDisposed;

    public AppInsightsProvider(
        IEnvironment environment,
        ITestApplicationCancellationTokenSource testApplicationCancellationTokenSource,
        ITask task,
        ILoggerFactory loggerFactory,
        IClock clock,
        IConfiguration configuration,
        ITelemetryInformation telemetryInformation,
        ITelemetryClientFactory telemetryClientFactory,
        string sessionId)
    {
        _ = bool.TryParse(configuration[PlatformConfigurationConstants.PlatformTelemetryIsDevelopmentRepository], out _isDevelopmentRepository);
        _isCi = new CIEnvironmentDetector(environment).IsCIEnvironment();
        _environment = environment;
        _currentSessionId = sessionId;
        _testApplicationCancellationTokenSource = testApplicationCancellationTokenSource;
        _clock = clock;
        _telemetryInformation = telemetryInformation;
        _telemetryClientFactory = telemetryClientFactory;

#if NETCOREAPP
        _payloads = Channel.CreateUnbounded<(string EventName, IDictionary<string, object> ParamsMap)>(new UnboundedChannelOptions
        {
            // We process only 1 data at a time
            SingleReader = true,

            // We don't know how many threads will call the Log method
            SingleWriter = false,

            // We want to unlink the caller from the consumer
            AllowSynchronousContinuations = false,
        });

#else
        _payloads = new SingleConsumerUnboundedChannel<(string EventName, IDictionary<string, object> ParamsMap)>();
#endif

        _telemetryTask = task.Run(IngestLoopAsync, _testApplicationCancellationTokenSource.CancellationToken);
        _logger = loggerFactory.CreateLogger<AppInsightsProvider>();
    }

    // Initialize the telemetry client and start ingesting events.
    private async Task IngestLoopAsync()
    {
        if (_testApplicationCancellationTokenSource.CancellationToken.IsCancellationRequested)
        {
            return;
        }

        try
        {
            _client = _telemetryClientFactory.Create(_currentSessionId, _environment.OsVersion);
        }
        catch (Exception e)
        {
            _client = null;

            await _logger.LogErrorAsync("Failed to initialize telemetry client", e).ConfigureAwait(false);
            return;
        }

        DateTimeOffset? lastLoggedError = null;
        _testApplicationCancellationTokenSource.CancellationToken.Register(_flushTimeoutOrStop.Cancel);

        try
        {
#if NETCOREAPP
            while (await _payloads.Reader.WaitToReadAsync(_flushTimeoutOrStop.Token).ConfigureAwait(false))
            {
                {
                    (string eventName, IDictionary<string, object> paramsMap) = await _payloads.Reader.ReadAsync().ConfigureAwait(false);
#else
            while (await _payloads.WaitToReadAsync(_flushTimeoutOrStop.Token).ConfigureAwait(false))
            {
                while (_payloads.TryRead(out (string EventName, IDictionary<string, object> ParamsMap) payload))
                {
                    if (_flushTimeoutOrStop.Token.IsCancellationRequested)
                    {
                        return;
                    }

                    (string eventName, IDictionary<string, object> paramsMap) = payload;
#endif

                    // Add common properties.
                    paramsMap.Add(TelemetryProperties.VersionPropertyName, _telemetryInformation.Version);
                    paramsMap.Add(TelemetryProperties.SessionId, _currentSessionId);
                    paramsMap.Add(TelemetryProperties.ReporterIdPropertyName, CurrentReporterId);
                    paramsMap.Add(TelemetryProperties.IsCIPropertyName, _isCi.AsTelemetryBool());

                    if (_isDevelopmentRepository)
                    {
                        paramsMap.Add(TelemetryProperties.HostProperties.IsDevelopmentRepositoryPropertyName, TelemetryProperties.True);
                    }

                    var metrics = new Dictionary<string, double>();
                    var properties = new Dictionary<string, string>();

                    foreach (KeyValuePair<string, object> pair in paramsMap)
                    {
                        switch (pair.Value)
                        {
                            // Metrics:
                            case double value:
                                metrics.Add(pair.Key, value);
                                break;
                            case DateTimeOffset value:
                                metrics.Add(pair.Key, ToUnixTimeNanoseconds(value));
                                break;

                            // Properties:
#if DEBUG
                            case string value:
                                AssertHashed(pair.Key, value);
                                properties.Add(pair.Key, value);
                                break;
#endif
                            case bool value:
                                properties.Add(pair.Key, value.AsTelemetryBool());
                                break;
                            default:
                                properties.Add(pair.Key, pair.Value?.ToString() ?? string.Empty);
                                break;
                        }
                    }

                    if (_logger.IsEnabled(LogLevel.Trace))
                    {
                        StringBuilder builder = new();
                        builder.AppendLine(CultureInfo.InvariantCulture, $"Send telemetry event: {eventName}");
                        foreach (KeyValuePair<string, string> kvp in properties)
                        {
                            builder.AppendLine(CultureInfo.InvariantCulture, $"    {kvp.Key}: {kvp.Value}");
                        }

                        foreach (KeyValuePair<string, double> kvp in metrics)
                        {
                            builder.AppendLine(CultureInfo.InvariantCulture, $"    {kvp.Key}: {kvp.Value.ToString("f", CultureInfo.InvariantCulture)}");
                        }

                        await _logger.LogTraceAsync(builder.ToString()).ConfigureAwait(false);
                    }

                    try
                    {
                        _client.TrackEvent(eventName, properties, metrics);
                    }
                    catch (Exception ex)
                    {
                        // If we have a lot of issues with the network we could have a lot of logs here.
                        // We log one error every 3 seconds.
                        // We could do better back-pressure.
                        if (_logger.IsEnabled(LogLevel.Error) && (!lastLoggedError.HasValue || (_clock.UtcNow - lastLoggedError.Value).TotalSeconds > 3))
                        {
                            await _logger.LogErrorAsync("Error during telemetry report.", ex).ConfigureAwait(false);
                            lastLoggedError = _clock.UtcNow;
                        }
                    }
                }
            }
        }
        catch (OperationCanceledException)
        {
            // This is expected when the test application is shutting down or if flush timeout.
        }
    }

    private static double ToUnixTimeNanoseconds(DateTimeOffset value) =>

        // The magic number is DateTimeOffset.UnixEpoch.Ticks in newer TFMs.
        // We multiply by 100 because Ticks are 100 ns, and we want to report ns.
        (value.UtcTicks - 621355968000000000L) * 100;

#if DEBUG
    private static void AssertHashed(string key, string value)
    {
        if (value is TelemetryProperties.True or TelemetryProperties.False)
        {
            return;
        }

        // Full qualification of Regex to avoid adding conditional 'using' on top of the file.
        if (value.Length == 64 && GetValidHashPattern().IsMatch(value))
        {
            return;
        }

        if (KnownUnhashedProperties.Contains(key))
        {
            return;
        }

        RoslynDebug.Assert(false, $"Telemetry entry '{key}' contains an unhashed string value '{value}'. Strings need to be hashed using {nameof(Sha256Hasher)}.{nameof(Sha256Hasher.HashWithNormalizedCasing)}(), or white-listed.");
    }

#if NET7_0_OR_GREATER
    [GeneratedRegex("[a-f0-9]{64}")]
    private static partial Regex GetValidHashPattern();
#else
    private static Regex GetValidHashPattern()
        => new("[a-f0-9]{64}");
#endif
#endif

    public
#if NETCOREAPP
        async
#endif
        Task LogEventAsync(string eventName, IDictionary<string, object> paramsMap, CancellationToken cancellationToken)
    {
#if NETCOREAPP
        await _payloads.Writer.WriteAsync((eventName, paramsMap), cancellationToken).ConfigureAwait(false);
#else
        if (cancellationToken.IsCancellationRequested)
        {
            return Task.FromCanceled(cancellationToken);
        }

        _payloads.Write((eventName, paramsMap));
        return Task.CompletedTask;
#endif
    }

    // Adding dispose on graceful shutdown per https://github.com/microsoft/ApplicationInsights-dotnet/issues/1152#issuecomment-518742922
    public void Dispose()
    {
#if NETCOREAPP
        _payloads.Writer.Complete();
#else
        _payloads.Complete();
#endif
        if (!_isDisposed)
        {
            int flushForSeconds = 3;
            if (!_telemetryTask.Wait(TimeSpan.FromSeconds(flushForSeconds)))
            {
                _flushTimeoutOrStop.Cancel();
                _logger.LogWarning($"Telemetry task didn't flush after '{flushForSeconds}', some payload could be lost");
            }

            _isDisposed = true;
        }
    }

#if NETCOREAPP
    public async ValueTask DisposeAsync()
    {
        _payloads.Writer.Complete();
        if (!_isDisposed)
        {
            int flushForSeconds = 3;
            try
            {
                await _telemetryTask.TimeoutAfterAsync(TimeSpan.FromSeconds(flushForSeconds)).ConfigureAwait(false);
            }
            catch (TimeoutException)
            {
                await _flushTimeoutOrStop.CancelAsync().ConfigureAwait(false);
                await _logger.LogWarningAsync($"Telemetry task didn't flush after '{flushForSeconds}', some payload could be lost").ConfigureAwait(false);
            }

            _isDisposed = true;
        }
    }
#endif
}