sqlinsight-pro/src/SQLInsight.Agent/Monitoring/MonitoringService.cs

using Microsoft.Extensions.Options;
using SQLInsight.Collectors;
using SQLInsight.Connectivity;
using SQLInsight.Notifications;
using SQLInsight.Repository;
using SQLInsight.Rules.Engine;

namespace SQLInsight.Agent.Monitoring;

/// <summary>
/// The agent's main loop: migrates the repository, then checks every instance on each tick and runs maintenance hourly.
/// The instances are read from the configuration (file and SQLInsight Web) on every tick, so changes need no restart.
/// </summary>
public sealed partial class MonitoringService(
    IOptions<AgentOptions> options,
    ISqlConnectionFactory connectionFactory,
    RuleEngine ruleEngine,
    InsightRepository repository,
    TimeProvider timeProvider,
    ILoggerFactory loggerFactory,
    NotificationDispatcher notifications,
    IOptions<NotificationOptions> notificationOptions,
    Hub.HubOutbox hubOutbox,
    Core.CustomRules.CustomRuleCatalog customRules,
    Core.Data.IGroupMemberResolver groupResolver,
    Configuration.InstanceConfigurationStore store,
    Configuration.RepositoryReadiness readiness) : BackgroundService
{
    private readonly ILogger _logger = loggerFactory.CreateLogger<MonitoringService>();
    private readonly Dictionary<string, string> _credentialErrors = new(StringComparer.OrdinalIgnoreCase);
    private bool _configurationLoaded;
    private bool _warnedNoInstances;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var settings = options.Value;
        await MigrateWithRetryAsync(stoppingToken).ConfigureAwait(false);
        await SyncConfigFileAsync(stoppingToken).ConfigureAwait(false);
        readiness.MarkReady();

        var intervals = new Dictionary<string, TimeSpan>(settings.CollectorIntervals, StringComparer.OrdinalIgnoreCase);
        var monitors = new Dictionary<string, (InstanceMonitor Monitor, string Fingerprint)>(StringComparer.OrdinalIgnoreCase);
        using var throttle = new SemaphoreSlim(Math.Max(1, settings.MaxParallelInstances));
        var running = new Dictionary<string, Task>(StringComparer.OrdinalIgnoreCase);
        var nextMaintenance = timeProvider.GetUtcNow();
        using var timer = new PeriodicTimer(settings.Tick, timeProvider);

        do
        {
            var now = timeProvider.GetUtcNow();

            // Instances added, paused, removed or changed in SQLInsight Web take effect here, without a restart.
            await RefreshMonitorsAsync(monitors, intervals, settings, now, stoppingToken).ConfigureAwait(false);
            foreach (var (name, (monitor, _)) in monitors)
            {
                // A slow instance keeps its own cycle; it never delays the others. A rebuilt monitor waits for the old cycle.
                if (running.TryGetValue(name, out var task) && !task.IsCompleted || !monitor.IsDue(now))
                {
                    continue;
                }

                running[name] = RunThrottledAsync(monitor, now, throttle, stoppingToken);
            }

            foreach (var done in running.Where(r => r.Value.IsCompleted && !monitors.ContainsKey(r.Key)).Select(r => r.Key).ToList())
            {
                running.Remove(done);
            }

            if (now >= nextMaintenance)
            {
                nextMaintenance = now.AddHours(1);
                await RunMaintenanceAsync(settings, now, stoppingToken).ConfigureAwait(false);
            }
        }
        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false));

        await Task.WhenAll(running.Values).ConfigureAwait(false);
    }

    private InstanceMonitor CreateMonitor(MonitoredInstance instance, AgentOptions settings, IReadOnlyDictionary<string, TimeSpan> intervals) =>
        new(instance, BuiltInCollectors.Create(customRules, new CollectorSettings(groupResolver, settings.Privileges.MaxSysadminAccounts, settings.TopSql.ActivityThresholdPercent, settings.ErrorLog.LoginFailureBurst,
                settings.Activity.TransactionMinutes, settings.Activity.TransactionLogMegabytes, settings.Activity.RequestMinutes,
                settings.Tempdb.VersionStoreMegabytes, settings.Tempdb.SessionMegabytes, settings.DynamicsAx.ToThresholds())), intervals, connectionFactory, ruleEngine, repository,
            loggerFactory.CreateLogger($"SQLInsight.Instance.{instance.Name}"), notifications, notificationOptions.Value.WebUrl, hubOutbox);

    /// <summary>
    /// Brings the monitors in step with the configuration (the configuration file and SQLInsight Web). When the
    /// repository cannot be read, the monitors keep running as they are; before the first successful read the
    /// configuration file alone is used, as before 1.3.
    /// </summary>
    private async Task RefreshMonitorsAsync(Dictionary<string, (InstanceMonitor Monitor, string Fingerprint)> monitors, IReadOnlyDictionary<string, TimeSpan> intervals,
        AgentOptions settings, DateTimeOffset now, CancellationToken cancellationToken)
    {
        IReadOnlyList<Configuration.EffectiveInstance> configured;
        try
        {
            configured = await store.LoadAsync(cancellationToken).ConfigureAwait(false);
            _configurationLoaded = true;
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            LogConfigurationFailed(_logger, ex);
            if (_configurationLoaded)
            {
                return;
            }

            configured = store.FromConfigFileOnly();
        }

        var active = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
        foreach (var instance in configured)
        {
            var name = instance.Instance.Name;
            if (instance.Paused)
            {
                continue;
            }

            if (instance.CredentialError is { } error)
            {
                // Shown on the overview like any other connection problem, once per change.
                if (!_credentialErrors.TryGetValue(name, out var reported) || reported != error)
                {
                    _credentialErrors[name] = error;
                    await RecordQuietlyAsync(name, "Adgangskoden kan ikke bruges: " + error, now, cancellationToken).ConfigureAwait(false);
                }

                continue;
            }

            _credentialErrors.Remove(name);
            active.Add(name);
            if (!monitors.TryGetValue(name, out var current) || current.Fingerprint != instance.Fingerprint)
            {
                current = (CreateMonitor(instance.Instance, settings, intervals), instance.Fingerprint);
                monitors[name] = current;
                LogMonitorStarted(_logger, name);
            }

            current.Monitor.Databases = instance.Databases;
        }

        foreach (var gone in monitors.Keys.Where(n => !active.Contains(n)).ToList())
        {
            monitors.Remove(gone);
            LogMonitorStopped(_logger, gone);
        }

        if (monitors.Count == 0 && !_warnedNoInstances)
        {
            LogNoInstances(_logger);
        }

        _warnedNoInstances = monitors.Count == 0;
    }

    private async Task RecordQuietlyAsync(string name, string error, DateTimeOffset now, CancellationToken cancellationToken)
    {
        try
        {
            await repository.RecordInstanceErrorAsync(name, error, now, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            LogConfigurationFailed(_logger, ex);
        }
    }

    private async Task SyncConfigFileAsync(CancellationToken cancellationToken)
    {
        try
        {
            await store.SyncConfigFileAsync(timeProvider.GetUtcNow(), cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            // The instances in the file are monitored anyway (InstanceConfigurationStore.LoadAsync adds them).
            LogConfigurationFailed(_logger, ex);
        }
    }

    private async Task RunThrottledAsync(InstanceMonitor monitor, DateTimeOffset now, SemaphoreSlim throttle, CancellationToken cancellationToken)
    {
        await throttle.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            await monitor.RunCycleAsync(now, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            // Typically the repository is unreachable. Try again next time.
            LogCycleFailed(_logger, ex, monitor.Name);
        }
        finally
        {
            throttle.Release();
        }
    }

    private async Task MigrateWithRetryAsync(CancellationToken cancellationToken)
    {
        for (var attempt = 1; ; attempt++)
        {
            try
            {
                var applied = await repository.MigrateAsync(cancellationToken).ConfigureAwait(false);
                LogMigrated(_logger, applied);
                return;
            }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                LogMigrationFailed(_logger, ex, attempt);
                await Task.Delay(TimeSpan.FromSeconds(Math.Min(60, 5 * attempt)), timeProvider, cancellationToken).ConfigureAwait(false);
            }
        }
    }

    private async Task RunMaintenanceAsync(AgentOptions settings, DateTimeOffset now, CancellationToken cancellationToken)
    {
        try
        {
            await repository.RunMaintenanceAsync(settings.Retention, now, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            LogMaintenanceFailed(_logger, ex);
        }
    }

    [LoggerMessage(Level = LogLevel.Warning, Message = "No instances configured under 'SQLInsight:Instances'.")]
    private static partial void LogNoInstances(ILogger logger);

    [LoggerMessage(Level = LogLevel.Information, Message = "Repository ready ({Applied} migrations applied).")]
    private static partial void LogMigrated(ILogger logger, int applied);

    [LoggerMessage(Level = LogLevel.Error, Message = "Repository migration failed (attempt {Attempt}); retrying.")]
    private static partial void LogMigrationFailed(ILogger logger, Exception exception, int attempt);

    [LoggerMessage(Level = LogLevel.Error, Message = "{Instance}: cycle failed")]
    private static partial void LogCycleFailed(ILogger logger, Exception exception, string instance);

    [LoggerMessage(Level = LogLevel.Information, Message = "{Instance}: monitoring started")]
    private static partial void LogMonitorStarted(ILogger logger, string instance);

    [LoggerMessage(Level = LogLevel.Information, Message = "{Instance}: monitoring stopped (paused, removed or no longer configured)")]
    private static partial void LogMonitorStopped(ILogger logger, string instance);

    [LoggerMessage(Level = LogLevel.Warning, Message = "The instance configuration could not be read from the repository.")]
    private static partial void LogConfigurationFailed(ILogger logger, Exception exception);

    [LoggerMessage(Level = LogLevel.Error, Message = "Repository maintenance failed.")]
    private static partial void LogMaintenanceFailed(ILogger logger, Exception exception);
}