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 Instance_configuration_reads_the_setup_and_reports_what_changed()
    {
        await using var session = await AgentSession.OpenAsync(sql);
        var collector = new InstanceConfigurationCollector();
        var first = await session.CollectAsync<InstanceConfigurationSnapshot>(collector);
        await sql.ExecuteAsSaAsync("EXEC sp_configure 'show advanced options', 1; RECONFIGURE; EXEC sp_configure 'cost threshold for parallelism', 7; RECONFIGURE;");
        try
        {
            var second = await session.CollectAsync(collector, first);

            Assert.Empty(first.Changes);
            Assert.False(string.IsNullOrEmpty(first.Collation));
            Assert.NotEmpty(first.TempdbFiles);
            Assert.True(first.ServicesAvailable); // sys.dm_server_services with VIEW SERVER STATE
            Assert.Contains(first.Items, i => i.Area == ConfigurationFingerprint.SpConfigure && i.Name == "cost threshold for parallelism" && i.Value == "5");
            var change = Assert.Single(second.Changes, c => c.Name == "cost threshold for parallelism");
            Assert.Equal(("5", "7"), (change.OldValue, change.NewValue));
            Assert.Contains(second.RecentChanges, c => c.Name == "cost threshold for parallelism");
        }
        finally
        {
            await sql.ExecuteAsSaAsync("EXEC sp_configure 'cost threshold for parallelism', 5; RECONFIGURE; EXEC sp_configure 'show advanced options', 0; RECONFIGURE;");
        }
    }

    [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");
        // SQL Server 2022 has sys.dm_os_volume_stats: every file knows its drive (a mount point on Linux).
        Assert.All(io1.Snapshot.Files, f => Assert.False(string.IsNullOrWhiteSpace(f.Volume)));
        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);
        Assert.Equal("Linux", cpu.HostPlatform); // test containers run SQL Server on Linux
        Assert.False(cpu.OtherProcessesMeasurable);
    }

    [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);
    }

    [Fact]
    public async Task Query_store_reports_plans_and_failed_forced_plans()
    {
        // Force a plan that uses an index, then drop the index: forcing fails with NO_INDEX.
        // The statement lives in a procedure so it is not auto-parameterized and can be found by object. The data is
        // selective enough that the optimizer seeks on the index by itself.
        await sql.ExecuteAsSaAsync($"""
            USE [{SqlServerFixture.TestDatabase}];
            IF OBJECT_ID(N'dbo.QsProbe') IS NULL
            BEGIN
                CREATE TABLE dbo.QsProbe (Id int IDENTITY PRIMARY KEY, Code int NOT NULL, Filler char(200) NOT NULL DEFAULT 'x');
                INSERT dbo.QsProbe (Code) SELECT TOP (50000) ABS(CHECKSUM(NEWID())) % 20000 FROM sys.all_objects a CROSS JOIN sys.all_objects b;
            END;
            IF INDEXPROPERTY(OBJECT_ID(N'dbo.QsProbe'), N'IX_QsProbe_Code', 'IndexID') IS NULL
                CREATE INDEX IX_QsProbe_Code ON dbo.QsProbe (Code);
            """);
        await sql.ExecuteAsSaAsync($"USE [{SqlServerFixture.TestDatabase}]; EXEC (N'CREATE OR ALTER PROCEDURE dbo.QsProbeProc @code int AS SELECT Id, Filler FROM dbo.QsProbe WHERE Code = @code;');");
        await sql.ExecuteAsSaAsync($"USE [{SqlServerFixture.TestDatabase}]; EXEC dbo.QsProbeProc 1; EXEC dbo.QsProbeProc 2; EXEC dbo.QsProbeProc 3; EXEC dbo.QsProbeProc 4; EXEC dbo.QsProbeProc 5; EXEC sys.sp_query_store_flush_db;");

        await sql.ExecuteAsSaAsync($"""
            USE [{SqlServerFixture.TestDatabase}];
            DECLARE @q bigint, @p bigint;
            SELECT TOP (1) @q = q.query_id, @p = p.plan_id
            FROM sys.query_store_query q JOIN sys.query_store_plan p ON p.query_id = q.query_id
            WHERE q.object_id = OBJECT_ID(N'dbo.QsProbeProc');
            EXEC sys.sp_query_store_force_plan @q, @p;
            DROP INDEX IX_QsProbe_Code ON dbo.QsProbe;
            """);
        await sql.ExecuteAsSaAsync($"USE [{SqlServerFixture.TestDatabase}]; EXEC dbo.QsProbeProc 42; EXEC sys.sp_query_store_flush_db;");

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

        var db = Assert.Single(snapshot.Databases, d => d.Name == SqlServerFixture.TestDatabase);
        Assert.Null(db.Error);
        var probe = Assert.Single(db.Plans, p => p.IsRecent && p.ObjectName == "dbo.QsProbeProc" && p.Executions >= 5);
        Assert.True(probe.IsForced);
        var failed = Assert.Single(db.FailedForcedPlans);
        Assert.Equal("NO_INDEX", failed.Reason);
    }

    [Fact]
    public async Task Plan_analysis_finds_implicit_conversion_udf_and_missing_join_predicate()
    {
        await sql.ExecuteAsSaAsync($"""
            USE [{SqlServerFixture.TestDatabase}];
            IF OBJECT_ID(N'dbo.PlanProbe') IS NULL
            BEGIN
                CREATE TABLE dbo.PlanProbe (Id int IDENTITY PRIMARY KEY, Code varchar(20) NOT NULL, Price int NOT NULL);
                INSERT dbo.PlanProbe (Code, Price) SELECT TOP (20000) CONCAT('C', ROW_NUMBER() OVER (ORDER BY (SELECT 1))), 10 FROM sys.all_objects a CROSS JOIN sys.all_objects b;
                CREATE INDEX IX_PlanProbe_Code ON dbo.PlanProbe (Code);
            END;
            EXEC (N'CREATE OR ALTER FUNCTION dbo.fn_PlanProbePrice (@p int) RETURNS int AS BEGIN RETURN @p * DATEPART(ms, GETDATE()); END;');
            """);
        // Run in the database (not after USE in a master batch) so the plans belong to it, and put the tags inside the
        // statements: the statement text ends at the statement, not at a trailing comment.
        await using (var connection = new Microsoft.Data.SqlClient.SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase)))
        {
            await connection.OpenAsync();
            // Enough executions to rank among the 50 heaviest statements even while the rest of the suite and SQL Agent run.
            for (var i = 0; i < 60; i++)
            {
                await using var command = connection.CreateCommand();
                command.CommandText = """
                    EXEC sys.sp_executesql N'SELECT /* plan-probe-convert */ Id FROM dbo.PlanProbe WHERE Code = @code', N'@code nvarchar(20)', N'C42';
                    SELECT /* plan-probe-udf */ SUM(dbo.fn_PlanProbePrice(Price)) FROM dbo.PlanProbe;
                    SELECT /* plan-probe-cross */ a.Id, b.Id FROM dbo.PlanProbe a, dbo.CustomerData b WHERE a.Id < 3;
                    """;
                await command.ExecuteNonQueryAsync();
            }
        }

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

        AnalyzedStatement Probe(string tag) => Assert.Single(snapshot.Statements, s => s.StatementText?.Contains(tag, StringComparison.Ordinal) == true);
        Assert.NotEmpty(Probe("plan-probe-convert").Warnings.SeekPreventingConversions);
        Assert.Contains(Probe("plan-probe-udf").Warnings.ScalarUdfs, f => f.Contains("fn_PlanProbePrice", StringComparison.Ordinal));
        Assert.True(Probe("plan-probe-cross").Warnings.NoJoinPredicate);
        Assert.Equal(SqlServerFixture.TestDatabase, Probe("plan-probe-udf").DatabaseName);
        Assert.NotNull(Probe("plan-probe-udf").TotalGrantKb); // 2017 CU3+ and 2022
    }

    [Fact]
    public async Task Indexes_include_usage_missing_suggestions_duplicates_heaps_and_statistics()
    {
        await sql.ExecuteAsSaAsync($"""
            USE [{SqlServerFixture.TestDatabase}];
            IF OBJECT_ID(N'dbo.IdxProbe') IS NULL
            BEGIN
                CREATE TABLE dbo.IdxProbe (Id int IDENTITY CONSTRAINT PK_IdxProbe PRIMARY KEY, CustomerId int NOT NULL, Status int NOT NULL, Note varchar(10) NOT NULL);
                INSERT dbo.IdxProbe (CustomerId, Status, Note) SELECT TOP (20000) ABS(CHECKSUM(NEWID())) % 5000, 1, 'x' FROM sys.all_objects a CROSS JOIN sys.all_objects b;
                CREATE INDEX IX_IdxProbe_Status ON dbo.IdxProbe (Status);
                CREATE INDEX IX_IdxProbe_Status_Copy ON dbo.IdxProbe (Status);
                CREATE TABLE dbo.HeapProbe (Id int NOT NULL, Payload varchar(4000) NULL);
                INSERT dbo.HeapProbe (Id, Payload) SELECT TOP (2000) ROW_NUMBER() OVER (ORDER BY (SELECT 1)), 'a' FROM sys.all_objects;
                UPDATE dbo.HeapProbe SET Payload = REPLICATE('b', 3000);
            END;
            """);
        await using (var connection = new Microsoft.Data.SqlClient.SqlConnection(sql.DatabaseConnectionString(SqlServerFixture.TestDatabase)))
        {
            await connection.OpenAsync();
            await using var command = connection.CreateCommand();
            command.CommandText = """
                DECLARE @i int = 0, @n int;
                WHILE @i < 20 BEGIN SELECT @n = COUNT(*) FROM dbo.IdxProbe WHERE CustomerId = @i; SET @i += 1; END;
                SELECT @n = COUNT(*) FROM dbo.HeapProbe WHERE Payload LIKE 'b%';
                """;
            await command.ExecuteNonQueryAsync();
        }

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

        var db = Assert.Single(snapshot.Databases, d => d.Name == SqlServerFixture.TestDatabase);
        Assert.Null(db.Error);
        var pk = Assert.Single(db.Indexes, i => i.Name == "PK_IdxProbe");
        Assert.True(pk.IsPrimaryKey);
        Assert.Equal(["Id"], pk.KeyColumns);
        Assert.Equal(20000, pk.Rows);
        Assert.True(pk.UserScans > 0);
        Assert.Equal(2, db.Indexes.Count(i => i.Table == "IdxProbe" && i.KeyColumns.SequenceEqual(["Status"])));
        Assert.Contains(db.MissingIndexes, m => m.Table.Contains("IdxProbe", StringComparison.Ordinal) && m.EqualityColumns == "[CustomerId]");
        var heap = Assert.Single(db.Indexes, i => i.Table == "HeapProbe");
        Assert.Equal("HEAP", heap.Type);
        Assert.True(heap.ForwardedFetches > 0);
        Assert.Contains(db.Statistics, s => s.Table == "IdxProbe" && s.Name == "PK_IdxProbe");
    }

    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();
    }
}