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

using SQLInsight.Collectors;
using SQLInsight.Connectivity;
using SQLInsight.Core.Collection;
using SQLInsight.Core.Instances;
using SQLInsight.Notifications;
using SQLInsight.Repository;
using SQLInsight.Rules.Engine;

namespace SQLInsight.Agent.Monitoring;

/// <summary>
/// Monitors one instance: runs collectors when they are due, keeps the latest data and each collector's previous
/// result (for deltas), evaluates the rules and stores everything. Not thread-safe; one cycle at a time.
/// </summary>
public sealed partial class InstanceMonitor(
    MonitoredInstance instance,
    IReadOnlyList<ICollector> collectors,
    IReadOnlyDictionary<string, TimeSpan> intervalOverrides,
    ISqlConnectionFactory connectionFactory,
    RuleEngine ruleEngine,
    InsightRepository repository,
    ILogger logger,
    NotificationDispatcher? notifications = null,
    Uri? webUrl = null,
    Hub.HubOutbox? hubOutbox = null)
{
    private bool? _wasReachable;

    private readonly CollectedData _data = new();
    private readonly Dictionary<string, object> _previous = [];
    private readonly Dictionary<string, DateTimeOffset> _nextDue = [];
    private DateTimeOffset _nextBaseline = DateTimeOffset.MinValue;
    private (InstanceInfo Info, int InstanceId)? _last;

    /// <summary>Metrics a baseline is kept for. Must match what the repository stores (see <see cref="Metrics"/>).</summary>
    public static readonly string[] BaselineMetrics = [Metrics.CpuSqlServer, Metrics.BatchRequestsPerSecond, Metrics.PageLifeExpectancy];

    public string Name => instance.Name;

    /// <summary>The databases chosen for monitoring; changed by the monitoring service when the choice changes.</summary>
    public Core.Configuration.DatabaseSelection Databases { get; set; } = Core.Configuration.DatabaseSelection.All;

    public TimeSpan IntervalOf(ICollector collector) =>
        intervalOverrides.TryGetValue(collector.Name, out var interval) ? interval : collector.DefaultInterval;

    public bool IsDue(DateTimeOffset now) => collectors.Any(c => !_nextDue.TryGetValue(c.Name, out var due) || due <= now);

    /// <summary>Runs the collectors that are due, then evaluates all rules. Returns the rule results, or null if nothing was due.</summary>
    public async Task<IReadOnlyList<RuleResult>?> RunCycleAsync(DateTimeOffset now, CancellationToken cancellationToken)
    {
        var due = collectors.Where(c => !_nextDue.TryGetValue(c.Name, out var at) || at <= now).ToList();
        if (due.Count == 0)
        {
            return null;
        }

        foreach (var collector in due)
        {
            _nextDue[collector.Name] = now + IntervalOf(collector);
        }

        // Only frequent samples are due: store them without probing the instance again or evaluating the rules.
        if (_last is { } last && _wasReachable == true && due.All(c => c.IsSampling))
        {
            await SampleAsync(due, last, now, cancellationToken).ConfigureAwait(false);
            return null;
        }

        OpenedConnection opened;
        InstanceInfo info;
        int instanceId;
        try
        {
            opened = await connectionFactory.OpenAsync(instance.ToTarget(), cancellationToken).ConfigureAwait(false);
            info = await InstanceProbe.DetectAsync(opened.Connection, cancellationToken).ConfigureAwait(false);
            instanceId = await repository.UpsertInstanceAsync(instance.Name, info, opened.Provider.ToString(), opened.UsedLegacyFallback, now, cancellationToken)
                .ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            LogConnectFailed(logger, ex, instance.Name);
            await repository.RecordInstanceErrorAsync(instance.Name, ex.Message, now, cancellationToken).ConfigureAwait(false);
            if (_wasReachable != false)
            {
                await NotifyAsync([NotificationFactory.Unreachable(instance.Name, ex.Message, webUrl, now)], cancellationToken).ConfigureAwait(false);
            }

            _wasReachable = false;
            return null;
        }

        await using (opened.ConfigureAwait(false))
        {
            if (opened.UsedLegacyFallback)
            {
                LogLegacyFallback(logger, instance.Name);
            }

            _last = (info, instanceId);
            var capabilities = InstanceCapabilities.From(info);
            foreach (var collector in due)
            {
                await CollectAsync(collector, opened, info, capabilities, instanceId, now, cancellationToken).ConfigureAwait(false);
            }

            await RefreshBaselineAsync(instanceId, now, cancellationToken).ConfigureAwait(false);
            var results = ruleEngine.Evaluate(new RuleContext(info, capabilities, now, _data));
            var changes = await repository.SaveRuleResultsAsync(instanceId, results, now, cancellationToken).ConfigureAwait(false);
            var events = new List<Notification>();
            if (_wasReachable == false)
            {
                events.Add(NotificationFactory.Recovered(instance.Name, webUrl, now));
            }

            _wasReachable = true;
            events.AddRange(changes.Opened.Select(o => NotificationFactory.Opened(instance.Name, o.FindingId, o.Finding, webUrl, now)));
            events.AddRange(changes.Resolved.Select(r => NotificationFactory.Resolved(instance.Name, r.FindingId, r.RuleId, r.Title, (Core.Findings.Severity)r.Severity, webUrl, now)));
            await NotifyAsync(events, cancellationToken).ConfigureAwait(false);
            if (hubOutbox is not null)
            {
                await hubOutbox.QueueAsync(instance.Name, instanceId, info, results, changes.Opened.Count + changes.Resolved.Count > 0, now, cancellationToken, Databases).ConfigureAwait(false);
            }
            var findings = 0;
            var notMeasurable = 0;
            foreach (var result in results)
            {
                findings += result.Findings.Count;
                notMeasurable += result.Outcome is RuleOutcome.NotMeasurable or RuleOutcome.Error ? 1 : 0;
            }

            LogCycle(logger, instance.Name, due.Count, findings, notMeasurable);
            return results;
        }
    }

    /// <summary>
    /// Runs sampling collectors with what the last full cycle learned about the instance. A failed connection is left
    /// to the next full cycle to report.
    /// </summary>
    private async Task SampleAsync(List<ICollector> due, (InstanceInfo Info, int InstanceId) last, DateTimeOffset now, CancellationToken cancellationToken)
    {
        OpenedConnection opened;
        try
        {
            opened = await connectionFactory.OpenAsync(instance.ToTarget(), cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            LogSampleConnectFailed(logger, instance.Name, ex.Message);
            return;
        }

        await using (opened.ConfigureAwait(false))
        {
            var capabilities = InstanceCapabilities.From(last.Info);
            foreach (var collector in due)
            {
                await CollectAsync(collector, opened, last.Info, capabilities, last.InstanceId, now, cancellationToken).ConfigureAwait(false);
            }
        }
    }

    [LoggerMessage(Level = LogLevel.Debug, Message = "{Instance}: sampling skipped, could not connect: {Error}")]
    private static partial void LogSampleConnectFailed(ILogger logger, string instance, string error);

    private async Task CollectAsync(
        ICollector collector, OpenedConnection opened, InstanceInfo info, InstanceCapabilities capabilities, int instanceId, DateTimeOffset now, CancellationToken cancellationToken)
    {
        var missing = capabilities.Missing(collector.RequiredCapabilities);
        if (missing.Count > 0)
        {
            if (missing.All(InstanceCapabilities.IsInherent))
            {
                // The version lacks it (Query Store on 2005 …): not applicable, not a gap in the monitoring.
                _data.MarkNotApplicable(collector.DataType, $"Ikke relevant for {info.Release.DisplayName()}: kræver {InstanceCapabilities.Describe(missing)}.");
            }
            else
            {
                _data.MarkUnavailable(collector.DataType, $"Ikke målbar på {info.Release.DisplayName()}: kræver {string.Join(", ", missing)}.");
            }

            return;
        }

        try
        {
            if (!_previous.TryGetValue(collector.Name, out var previous))
            {
                previous = await SeedAsync(collector, instanceId, now, cancellationToken).ConfigureAwait(false);
            }

            var result = await collector.CollectAsync(new CollectorContext(opened.Connection, info, capabilities, now, previous, cancellationToken, Databases)).ConfigureAwait(false);
            _previous[collector.Name] = result;
            _data.Set(collector.DataType, result);
            await repository.SaveCollectedAsync(instanceId, collector.Name, result, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            // Typically a permission the agent lacks on this instance. Rules depending on it say why.
            LogCollectorFailed(logger, ex, instance.Name, collector.Name);
            _data.MarkUnavailable(collector.DataType, $"Collector '{collector.Name}' fejlede: {ex.Message}");
        }
    }

    private async Task NotifyAsync(List<Notification> events, CancellationToken cancellationToken)
    {
        if (notifications is not null && events.Count > 0)
        {
            await notifications.DispatchAsync(events, cancellationToken).ConfigureAwait(false);
        }
    }

    /// <summary>
    /// The baseline for the current hour of the week, from the repository's hourly history. Refreshed at each new hour.
    /// </summary>
    private async Task RefreshBaselineAsync(int instanceId, DateTimeOffset now, CancellationToken cancellationToken)
    {
        if (now < _nextBaseline)
        {
            return;
        }

        var history = await repository.GetHourlyHistoryAsync(instanceId, BaselineMetrics, now, cancellationToken: cancellationToken).ConfigureAwait(false);
        _data.Set(Core.Data.BaselineCalculator.Calculate(history, now));
        var hour = new DateTimeOffset(now.UtcDateTime.Year, now.UtcDateTime.Month, now.UtcDateTime.Day, now.UtcDateTime.Hour, 0, 0, TimeSpan.Zero);
        _nextBaseline = hour.AddHours(1);
    }

    /// <summary>
    /// State a collector would otherwise lose when the agent restarts. Deadlocks continue after the newest one
    /// already stored; the others simply start over (cumulative collectors need one extra sample).
    /// </summary>
    private async Task<object?> SeedAsync(ICollector collector, int instanceId, DateTimeOffset now, CancellationToken cancellationToken)
    {
        // Changes are found against the stored state, so one made while the agent was down is still reported.
        if (collector is Collectors.Instance.InstanceConfigurationCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.InstanceConfigurationSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        if (collector is Collectors.Instance.PrivilegeCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.PrivilegeSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        // Earlier backup durations survive a restart, so a slow backup is still recognised.
        if (collector is Collectors.Instance.LongOperationCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.LongOperationSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        // The error log continues where it stopped, so nothing written while the agent was down is missed.
        if (collector is Collectors.Instance.ErrorLogCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.ErrorLogSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        // The version store trend and heavy sessions continue after a restart.
        if (collector is Collectors.Instance.TempdbCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.TempdbSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        // Index usage is summed across restarts of SQL Server and of the agent.
        if (collector is Collectors.Databases.IndexCollector)
        {
            return await repository.GetSnapshotAsync<Core.Data.IndexSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false);
        }

        // Dynamics AX: each batch job's own run history, the first-seen times of running jobs, the history watermark and the
        // sequences' daily samples come back from the repository, so comparisons and forecasts continue after a restart.
        if (collector is Collectors.DynamicsAx.AxBatchCollector)
        {
            var stored = await repository.GetSnapshotAsync<Core.DynamicsAx.AxBatchSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false)
                ?? new Core.DynamicsAx.AxBatchSnapshot { CapturedAt = now, Thresholds = new(), Databases = [] };
            var history = await repository.GetAxBatchHistoryAsync(instanceId, now.AddDays(-90), cancellationToken).ConfigureAwait(false);
            return stored with
            {
                History = history,
                RecentRunKeys = history.ScopedRunKeys().ToHashSet(StringComparer.Ordinal),
                FirstSeenExecuting = stored.Databases
                    .SelectMany(d => d.Jobs.Where(j => j.FirstSeenExecutingUtc is not null)
                        .Select(j => (Key: Core.DynamicsAx.AxBatchHistory.Key(d.Database, "#" + j.RecId.ToString(System.Globalization.CultureInfo.InvariantCulture)), At: j.FirstSeenExecutingUtc!.Value)))
                    .GroupBy(x => x.Key, StringComparer.OrdinalIgnoreCase)
                    .ToDictionary(g => g.Key, g => g.First().At, StringComparer.OrdinalIgnoreCase),
            };
        }

        if (collector is Collectors.DynamicsAx.AxSequenceCollector sequences)
        {
            var stored = await repository.GetSnapshotAsync<Core.DynamicsAx.AxSequenceSnapshot>(instanceId, collector.Name, cancellationToken).ConfigureAwait(false)
                ?? new Core.DynamicsAx.AxSequenceSnapshot { CapturedAt = now, Thresholds = new(), Databases = [] };
            return stored with { History = await repository.GetAxSequenceHistoryAsync(instanceId, now.AddDays(-sequences.HistoryDays), cancellationToken).ConfigureAwait(false) };
        }

        if (collector is not Collectors.Concurrency.DeadlockCollector)
        {
            return null;
        }

        var latest = await repository.GetLatestDeadlockTimeAsync(instanceId, cancellationToken).ConfigureAwait(false);
        return latest is null ? null : new Core.Data.DeadlockSnapshot { CapturedAt = now, NewDeadlocks = [], LastEventTime = latest };
    }

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

    [LoggerMessage(Level = LogLevel.Warning, Message = "{Instance}: SqlClient could not connect, using the legacy ODBC connection. Encryption may be weaker.")]
    private static partial void LogLegacyFallback(ILogger logger, string instance);

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

    [LoggerMessage(Level = LogLevel.Information, Message = "{Instance}: {Collectors} collectors, {Findings} findings, {NotMeasurable} rules not measurable")]
    private static partial void LogCycle(ILogger logger, string instance, int collectors, int findings, int notMeasurable);
}