sqlinsight-pro/tests/SQLInsight.IntegrationTests/CollectorTests.cs

using Microsoft.Data.SqlClient;
using SQLInsight.Collectors;
using SQLInsight.Collectors.Concurrency;
using SQLInsight.Collectors.Databases;
using SQLInsight.Collectors.Instance;
using SQLInsight.Collectors.Performance;
using SQLInsight.Core.Data;

namespace SQLInsight.IntegrationTests;

/// <summary>
/// Every collector against real SQL Server, as the least-privilege agent login.
/// </summary>
[Collection(SqlServerTestGroup.Name)]
[Trait("Category", "Integration")]
public class CollectorTests(SqlServerFixture sql)
{
    [Fact]
    public async Task Every_built_in_collector_runs_with_least_privilege()
    {
        await using var session = await AgentSession.OpenAsync(sql);

        foreach (var collector in BuiltInCollectors.Create())
        {
            var result = await session.CollectAsync<object>(collector);
            Assert.True(collector.DataType.IsInstanceOfType(result), $"{collector.Name} returned {result.GetType().Name}");
        }
    }

    [Fact]
    public async Task Server_reports_cpu_memory_and_start_time()
    {
        await using var session = await AgentSession.OpenAsync(sql);

        var server = await session.CollectAsync<ServerSnapshot>(new ServerCollector());

        Assert.True(server.LogicalCpuCount > 0);
        Assert.True(server.PhysicalMemoryMb > 0);
        Assert.True(server.VisibleOnlineSchedulers > 0);
        Assert.True(server.SqlServerStartTime < DateTime.Now.AddMinutes(1));
    }

    [Fact]
    public async Task Configuration_contains_well_known_options()
    {
        await using var session = await AgentSession.OpenAsync(sql);

        var config = await session.CollectAsync<ConfigurationSnapshot>(new ConfigurationCollector());

        Assert.NotNull(config.Find("max server memory (MB)"));
        Assert.Equal(5, config.Find("cost threshold for parallelism")!.ValueInUse);
    }

    [Fact]
    public async Task Databases_include_settings_files_backups_and_query_store()
    {
        await sql.ExecuteAsSaAsync($"BACKUP DATABASE [{SqlServerFixture.TestDatabase}] TO DISK = N'/tmp/qs.bak' WITH INIT, COPY_ONLY");
        await using var session = await AgentSession.OpenAsync(sql);

        var snapshot = await session.CollectAsync<DatabaseSnapshot>(new DatabaseCollector());

        var master = Assert.Single(snapshot.Databases, d => d.Name == "master");
        Assert.True(master.IsSystemDatabase);
        Assert.DoesNotContain(snapshot.Databases, d => d.Name == "tempdb");

        var db = Assert.Single(snapshot.Databases, d => d.Name == SqlServerFixture.TestDatabase);
        Assert.Equal("ONLINE", db.State);
        Assert.Equal("CHECKSUM", db.PageVerify);
        Assert.NotNull(db.LastFullBackup);
        Assert.Contains(db.Files, f => f.Type == "ROWS");
        Assert.Contains(db.Files, f => f.Type == "LOG");
        Assert.True(db.VirtualLogFileCount > 0);
        Assert.NotNull(db.QueryStore);
        Assert.Equal("READ_WRITE", db.QueryStore.ActualState);
        Assert.NotNull(db.LastGoodCheckDb); // 1900-01-01: CHECKDB never ran
    }

    [Fact]
    public async Task Wait_stats_produce_a_delta_on_the_second_collection()
    {
        await using var session = await AgentSession.OpenAsync(sql);
        var collector = new WaitStatsCollector();

        var first = await session.CollectAsync<CumulativeSample<WaitStatsSnapshot, WaitStatsDelta>>(collector);
        await sql.ExecuteAsSaAsync("WAITFOR DELAY '00:00:01'; SELECT COUNT(*) FROM sys.objects a CROSS JOIN sys.objects b;");
        var second = await session.CollectAsync(collector, first);

        Assert.Null(first.Delta);
        Assert.NotEmpty(first.Snapshot.Waits);
        Assert.NotNull(second.Delta);
        Assert.DoesNotContain(second.Delta.Waits, w => WaitCatalog.IsBenign(w.WaitType));
    }

    [Fact]
    public async Task File_io_and_counters_produce_deltas()
    {
        await using var session = await AgentSession.OpenAsync(sql);

        var io1 = await session.CollectAsync<CumulativeSample<FileIoSnapshot, FileIoDelta>>(new FileIoCollector());
        var io2 = await session.CollectAsync(new FileIoCollector(), io1);
        var pc1 = await session.CollectAsync<CumulativeSample<PerformanceCounterSnapshot, PerformanceCounterRates>>(new PerformanceCounterCollector());
        await Task.Delay(1100);
        var pc2 = await session.CollectAsync(new PerformanceCounterCollector(), pc1);

        Assert.Contains(io1.Snapshot.Files, f => f.DatabaseName == "master" && f.FileType == "LOG");
        Assert.NotNull(io2.Delta);
        Assert.NotNull(pc1.Snapshot.PageLifeExpectancySeconds);
        Assert.NotNull(pc1.Snapshot.MemoryGrantsPending);
        Assert.NotNull(pc2.Delta?.BatchRequestsPerSecond);
    }

    [Fact]
    public async Task Cpu_history_comes_from_the_scheduler_monitor()
    {
        await using var session = await AgentSession.OpenAsync(sql);

        var cpu = await session.CollectAsync<CpuUtilization>(new CpuCollector());

        // The ring buffer gets its first record about a minute after startup.
        Assert.All(cpu.Samples, s => Assert.InRange(s.SqlServerPercent + s.IdlePercent, 0, 100));
        Assert.Equal(cpu.Samples.OrderBy(s => s.Time), cpu.Samples);
    }

    [Fact]
    public async Task Blocking_finds_sleeping_head_blocker_with_open_transaction()
    {
        await using var blocker = new SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase));
        await blocker.OpenAsync();
        await using (var begin = blocker.CreateCommand())
        {
            begin.CommandText = "BEGIN TRAN; UPDATE dbo.CustomerData SET Secret = N'låst' WHERE Id = 1;";
            await begin.ExecuteNonQueryAsync();
        }

        await using var victim = new SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase));
        await victim.OpenAsync();
        await using var blockedCommand = victim.CreateCommand();
        blockedCommand.CommandText = "SELECT Secret FROM dbo.CustomerData WHERE Id = 1;";
        blockedCommand.CommandTimeout = 30;
        var blockedQuery = blockedCommand.ExecuteScalarAsync();
        await Task.Delay(1500);

        await using var session = await AgentSession.OpenAsync(sql);
        var snapshot = await session.CollectAsync<BlockingSnapshot>(new BlockingCollector());

        await using (var rollback = blocker.CreateCommand())
        {
            rollback.CommandText = "ROLLBACK;";
            await rollback.ExecuteNonQueryAsync();
        }

        await blockedQuery;

        var head = Assert.Single(snapshot.HeadBlockers);
        Assert.Equal("sleeping", head.Status);
        Assert.Equal(1, head.OpenTransactionCount);
        Assert.Contains("UPDATE dbo.CustomerData", head.SqlText, StringComparison.Ordinal);
        var blocked = Assert.Single(snapshot.BlockedBy(head.SessionId));
        Assert.StartsWith("LCK_M_S", blocked.WaitType, StringComparison.Ordinal);
        Assert.Equal(SqlServerFixture.TestDatabase, blocked.DatabaseName);
        // Simple statements are auto-parameterized: SELECT [Secret] FROM [dbo].[CustomerData] WHERE [Id]=@1.
        Assert.Contains("CustomerData", blocked.SqlText, StringComparison.Ordinal);
    }

    [Fact]
    public async Task Deadlocks_are_read_from_system_health_and_continue_from_last_position()
    {
        await using var session = await AgentSession.OpenAsync(sql);
        var collector = new DeadlockCollector();
        var before = await session.CollectAsync<DeadlockSnapshot>(collector);

        await CauseDeadlockAsync();
        var after = await WaitForDeadlockAsync(session, collector, before);

        var deadlock = Assert.Single(after.NewDeadlocks);
        Assert.Equal(2, deadlock.Processes.Count);
        Assert.Single(deadlock.Processes, p => p.IsVictim);
        Assert.All(deadlock.Processes, p => Assert.Contains("UPDATE dbo.CustomerData", p.InputBuffer, StringComparison.Ordinal));
        Assert.StartsWith("<deadlock", deadlock.GraphXml, StringComparison.Ordinal);
        Assert.NotNull(after.LastFileName);

        var again = await session.CollectAsync(collector, after);
        Assert.Empty(again.NewDeadlocks);
    }

    private static async Task<DeadlockSnapshot> WaitForDeadlockAsync(AgentSession session, DeadlockCollector collector, DeadlockSnapshot previous)
    {
        // system_health dispatches events to its targets with up to 120 seconds latency.
        for (var attempt = 0; attempt < 180; attempt++)
        {
            var snapshot = await session.CollectAsync(collector, previous);
            if (snapshot.NewDeadlocks.Count > 0)
            {
                return snapshot;
            }

            await Task.Delay(1000);
        }

        throw new TimeoutException("The deadlock did not appear in system_health.");
    }

    private async Task CauseDeadlockAsync()
    {
        await using var a = new SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase));
        await using var b = new SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase));
        await a.OpenAsync();
        await b.OpenAsync();

        await ExecuteAsync(a, "BEGIN TRAN; UPDATE dbo.CustomerData SET Secret = N'a' WHERE Id = 1;");
        await ExecuteAsync(b, "BEGIN TRAN; UPDATE dbo.CustomerData SET Secret = N'b' WHERE Id = 2;");

        var first = ExecuteAsync(a, "UPDATE dbo.CustomerData SET Secret = N'a' WHERE Id = 2;");
        await Task.Delay(500);
        var second = ExecuteAsync(b, "UPDATE dbo.CustomerData SET Secret = N'b' WHERE Id = 1;");

        var results = await Task.WhenAll(Capture(first), Capture(second));
        Assert.Single(results, e => e is SqlException { Number: 1205 });

        await ExecuteAsync(a, "IF @@TRANCOUNT > 0 ROLLBACK;");
        await ExecuteAsync(b, "IF @@TRANCOUNT > 0 ROLLBACK;");
    }

    private static async Task<Exception?> Capture(Task task)
    {
        try
        {
            await task;
            return null;
        }
        catch (SqlException ex)
        {
            return ex;
        }
    }

    private static async Task ExecuteAsync(SqlConnection connection, string sql)
    {
        await using var command = connection.CreateCommand();
        command.CommandText = sql;
        command.CommandTimeout = 30;
        await command.ExecuteNonQueryAsync();
    }
}