using System.Text.Json;
using Dapper;
using Microsoft.Data.SqlClient;
using SQLInsight.Core.Data;
using SQLInsight.Core.Findings;
using SQLInsight.Core.Instances;
using SQLInsight.Rules.Engine;
namespace SQLInsight.Repository;
///
/// The on-prem repository: what the agent has found and measured. Lives on a SQL Server instance separate from
/// the ones being monitored.
///
public sealed partial class InsightRepository(string connectionString)
{
static InsightRepository()
{
// Dapper sends DateTime as the legacy datetime type (≈3 ms steps). Since SQL Server 2016, comparing
// datetime with datetime2 uses the exact binary value, so .127 no longer equals .1266667. All columns are
// datetime2, so send datetime2.
SqlMapper.AddTypeMap(typeof(DateTime), System.Data.DbType.DateTime2);
SqlMapper.AddTypeMap(typeof(DateTime?), System.Data.DbType.DateTime2);
}
// PascalCase: OPENJSON ... WITH matches property names case-sensitively.
internal static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.General);
public Task MigrateAsync(CancellationToken cancellationToken = default) => SchemaMigrator.MigrateAsync(connectionString, cancellationToken);
private async Task OpenAsync(CancellationToken cancellationToken)
{
var connection = new SqlConnection(connectionString);
await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
return connection;
}
///
/// UTC truncated to whole seconds, matching the datetime2(0) columns. Without truncation SQL Server rounds on
/// insert but compares at full precision, so "insert if not exists" would insert duplicates.
///
internal static DateTime Utc(DateTimeOffset value)
{
var utc = value.UtcDateTime;
return new DateTime(utc.Ticks - (utc.Ticks % TimeSpan.TicksPerSecond), DateTimeKind.Utc);
}
/// UTC truncated to milliseconds, matching datetime2(3) (deadlock times must survive a round trip exactly).
internal static DateTime UtcMilliseconds(DateTimeOffset value)
{
var utc = value.UtcDateTime;
return new DateTime(utc.Ticks - (utc.Ticks % TimeSpan.TicksPerMillisecond), DateTimeKind.Utc);
}
/// Registers a successful contact with an instance and returns its id.
public async Task UpsertInstanceAsync(
string name, InstanceInfo info, string provider, bool usedLegacyFallback, DateTimeOffset seenAt, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(info);
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
return await connection.ExecuteScalarAsync("""
MERGE insight.Instances WITH (HOLDLOCK) AS t
USING (SELECT @name AS Name) AS s ON t.Name = s.Name
WHEN MATCHED THEN UPDATE SET ServerName = @serverName, ProductVersion = @version, ProductLevel = @level, ProductUpdateLevel = @updateLevel,
Edition = @edition, EngineEdition = @engineEdition, ConnectionProvider = @provider, UsedLegacyFallback = @legacy, LastSeenAt = @seenAt
WHEN NOT MATCHED THEN INSERT (Name, ServerName, ProductVersion, ProductLevel, ProductUpdateLevel, Edition, EngineEdition, ConnectionProvider, UsedLegacyFallback, LastSeenAt)
VALUES (@name, @serverName, @version, @level, @updateLevel, @edition, @engineEdition, @provider, @legacy, @seenAt)
OUTPUT inserted.InstanceId;
""", new
{
name,
serverName = info.ServerName,
version = info.Version.ToString(),
level = info.ProductLevel,
updateLevel = info.ProductUpdateLevel,
edition = info.Edition,
engineEdition = (int)info.EngineEdition,
provider,
legacy = usedLegacyFallback,
seenAt = Utc(seenAt),
}).ConfigureAwait(false);
}
/// Records that an instance could not be reached or checked.
public async Task RecordInstanceErrorAsync(string name, string error, DateTimeOffset at, CancellationToken cancellationToken = default)
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await connection.ExecuteAsync("""
MERGE insight.Instances WITH (HOLDLOCK) AS t
USING (SELECT @name AS Name) AS s ON t.Name = s.Name
WHEN MATCHED THEN UPDATE SET LastError = @error, LastErrorAt = @at
WHEN NOT MATCHED THEN INSERT (Name, LastError, LastErrorAt) VALUES (@name, @error, @at);
""", new { name, error, at = Utc(at) }).ConfigureAwait(false);
}
///
/// Stores the outcome of a rule evaluation and tracks findings: new ones start, existing ones are updated, and
/// ones no longer reported by a rule that ran are resolved. A rule that could not run leaves its findings as they were.
///
/// The findings that started and the ones that were resolved, for notifications.
public async Task SaveRuleResultsAsync(int instanceId, IReadOnlyList results, DateTimeOffset evaluatedAt, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(results);
var now = Utc(evaluatedAt);
var opened = new List<(long, Finding)>();
var resolved = new List();
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).ConfigureAwait(false);
foreach (var result in results)
{
await connection.ExecuteAsync("""
MERGE insight.RuleStates AS t
USING (SELECT @instanceId AS InstanceId, @ruleId AS RuleId) AS s ON t.InstanceId = s.InstanceId AND t.RuleId = s.RuleId
WHEN MATCHED THEN UPDATE SET RuleTitle = @title, Outcome = @outcome, Reason = @reason, EvaluatedAt = @now
WHEN NOT MATCHED THEN INSERT (InstanceId, RuleId, RuleTitle, Outcome, Reason, EvaluatedAt) VALUES (@instanceId, @ruleId, @title, @outcome, @reason, @now);
""", new { instanceId, ruleId = result.RuleId, title = result.RuleTitle, outcome = (byte)result.Outcome, reason = result.Reason, now }, transaction).ConfigureAwait(false);
if (result.Outcome is not (RuleOutcome.Passed or RuleOutcome.Findings))
{
continue;
}
var current = result.Findings.Select(f => f.Subject ?? string.Empty).ToList();
resolved.AddRange(await connection.QueryAsync("""
UPDATE insight.Findings SET ResolvedAt = @now
OUTPUT inserted.FindingId, inserted.RuleId, inserted.Subject, inserted.Title, inserted.Severity
WHERE InstanceId = @instanceId AND RuleId = @ruleId AND ResolvedAt IS NULL
AND Subject NOT IN (SELECT value FROM OPENJSON(@current));
""", new { instanceId, ruleId = result.RuleId, now, current = JsonSerializer.Serialize(current, Json) }, transaction).ConfigureAwait(false));
foreach (var finding in result.Findings)
{
var newId = await connection.QuerySingleOrDefaultAsync("""
UPDATE insight.Findings
SET Severity = @severity, Title = @title, Explanation = @explanation, Recommendation = @recommendation,
EvidenceJson = @evidence, ReferencesJson = @references, FixScript = @fixScript, LastSeenAt = @now
WHERE InstanceId = @instanceId AND RuleId = @ruleId AND Subject = @subject AND ResolvedAt IS NULL;
IF @@ROWCOUNT = 0
INSERT insight.Findings (InstanceId, RuleId, Subject, Severity, Title, Explanation, Recommendation, EvidenceJson, ReferencesJson, FixScript, FirstSeenAt, LastSeenAt)
OUTPUT inserted.FindingId
VALUES (@instanceId, @ruleId, @subject, @severity, @title, @explanation, @recommendation, @evidence, @references, @fixScript, @now, @now);
""", new
{
instanceId,
ruleId = finding.RuleId,
subject = finding.Subject ?? string.Empty,
severity = (byte)finding.Severity,
title = finding.Title,
explanation = finding.Explanation,
recommendation = finding.Recommendation,
evidence = JsonSerializer.Serialize(finding.Evidence, Json),
references = JsonSerializer.Serialize(finding.References, Json),
fixScript = finding.FixScript,
now,
}, transaction).ConfigureAwait(false);
if (newId is { } id)
{
opened.Add((id, finding));
}
}
}
await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
return new FindingChanges(opened, resolved);
}
///
/// Stores collected data: time series for trends, events (blocking, deadlocks) and the latest snapshot of
/// everything else for display.
///
public async Task SaveCollectedAsync(int instanceId, string kind, object data, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(data);
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
switch (data)
{
case CumulativeSample { Delta: { } waits }:
await SaveWaitsAsync(connection, instanceId, waits).ConfigureAwait(false);
return;
case CumulativeSample:
return;
case CumulativeSample { Delta: { } io }:
await SaveFileIoAsync(connection, instanceId, io).ConfigureAwait(false);
return;
case CumulativeSample:
return;
case CumulativeSample counters:
await SaveCountersAsync(connection, instanceId, counters).ConfigureAwait(false);
return;
case CpuUtilization cpu:
await SaveCpuAsync(connection, instanceId, cpu).ConfigureAwait(false);
return;
case BlockingSnapshot blocking:
await SaveBlockingAsync(connection, instanceId, blocking).ConfigureAwait(false);
return;
case DeadlockSnapshot deadlocks:
await SaveDeadlocksAsync(connection, instanceId, deadlocks).ConfigureAwait(false);
return;
default:
await connection.ExecuteAsync("""
MERGE insight.Snapshots AS t
USING (SELECT @instanceId AS InstanceId, @kind AS Kind) AS s ON t.InstanceId = s.InstanceId AND t.Kind = s.Kind
WHEN MATCHED THEN UPDATE SET CapturedAt = SYSUTCDATETIME(), Json = @json
WHEN NOT MATCHED THEN INSERT (InstanceId, Kind, CapturedAt, Json) VALUES (@instanceId, @kind, SYSUTCDATETIME(), @json);
""", new { instanceId, kind, json = JsonSerializer.Serialize(data, data.GetType(), Json) }).ConfigureAwait(false);
return;
}
}
private static Task SaveWaitsAsync(SqlConnection connection, int instanceId, WaitStatsDelta delta) =>
connection.ExecuteAsync("""
INSERT insight.WaitSamples (InstanceId, SampledAt, WaitType, IntervalSeconds, WaitingTasks, WaitTimeMs, SignalWaitTimeMs)
SELECT @instanceId, @at, WaitType, @interval, WaitingTasks, WaitTimeMs, SignalWaitTimeMs
FROM OPENJSON(@json) WITH (WaitType nvarchar(60), WaitingTasks bigint, WaitTimeMs bigint, SignalWaitTimeMs bigint) j
WHERE NOT EXISTS (SELECT 1 FROM insight.WaitSamples w WHERE w.InstanceId = @instanceId AND w.SampledAt = @at AND w.WaitType = j.WaitType);
""", new
{
instanceId,
at = Utc(delta.To),
interval = (int)delta.Interval.TotalSeconds,
json = JsonSerializer.Serialize(delta.Waits, Json),
});
private static Task SaveFileIoAsync(SqlConnection connection, int instanceId, FileIoDelta delta) =>
connection.ExecuteAsync("""
INSERT insight.FileIoSamples (InstanceId, SampledAt, DatabaseName, FileId, FileType, Reads, ReadStallMs, Writes, WriteStallMs, BytesRead, BytesWritten)
SELECT @instanceId, @at, DatabaseName, FileId, FileType, Reads, ReadStallMs, Writes, WriteStallMs, BytesRead, BytesWritten
FROM OPENJSON(@json) WITH (DatabaseName nvarchar(128), FileId int, FileType varchar(10), Reads bigint, ReadStallMs bigint,
Writes bigint, WriteStallMs bigint, BytesRead bigint, BytesWritten bigint) j
WHERE NOT EXISTS (SELECT 1 FROM insight.FileIoSamples f WHERE f.InstanceId = @instanceId AND f.SampledAt = @at AND f.DatabaseName = j.DatabaseName AND f.FileId = j.FileId);
""", new { instanceId, at = Utc(delta.To), json = JsonSerializer.Serialize(delta.Files, Json) });
private static Task InsertMetricsAsync(SqlConnection connection, int instanceId, IEnumerable<(string Metric, DateTimeOffset At, double Value)> metrics) =>
connection.ExecuteAsync("""
INSERT insight.MetricSamples (InstanceId, Metric, SampledAt, Value)
SELECT @instanceId, Metric, SampledAt, Value
FROM OPENJSON(@json) WITH (Metric varchar(60), SampledAt datetime2(0), Value float) j
WHERE NOT EXISTS (SELECT 1 FROM insight.MetricSamples m WHERE m.InstanceId = @instanceId AND m.Metric = j.Metric AND m.SampledAt = j.SampledAt);
""", new
{
instanceId,
json = JsonSerializer.Serialize(metrics.Select(m => new { m.Metric, SampledAt = m.At.UtcDateTime.ToString("yyyy-MM-ddTHH:mm:ss", System.Globalization.CultureInfo.InvariantCulture), m.Value }), Json),
});
private static Task SaveCountersAsync(SqlConnection connection, int instanceId, CumulativeSample counters)
{
var at = counters.Snapshot.CapturedAt;
var metrics = new List<(string, DateTimeOffset, double)>();
void Add(string metric, double? value)
{
if (value is { } v)
{
metrics.Add((metric, at, v));
}
}
Add(Metrics.PageLifeExpectancy, counters.Snapshot.PageLifeExpectancySeconds);
Add(Metrics.MemoryGrantsPending, counters.Snapshot.MemoryGrantsPending);
Add(Metrics.UserConnections, counters.Snapshot.UserConnections);
Add(Metrics.ProcessesBlocked, counters.Snapshot.ProcessesBlocked);
Add(Metrics.BatchRequestsPerSecond, counters.Delta?.BatchRequestsPerSecond);
Add(Metrics.CompilationsPerSecond, counters.Delta?.CompilationsPerSecond);
return InsertMetricsAsync(connection, instanceId, metrics);
}
private static Task SaveCpuAsync(SqlConnection connection, int instanceId, CpuUtilization cpu) =>
InsertMetricsAsync(connection, instanceId, cpu.Samples.SelectMany(s =>
cpu.OtherProcessesMeasurable
? new[] { (Metrics.CpuSqlServer, s.Time, (double)s.SqlServerPercent), (Metrics.CpuOther, s.Time, (double)s.OtherProcessesPercent) }
: new[] { (Metrics.CpuSqlServer, s.Time, (double)s.SqlServerPercent) }));
/// Blocking shorter than this is normal and not stored.
public const long BlockingThresholdMs = 5000;
private static async Task SaveBlockingAsync(SqlConnection connection, int instanceId, BlockingSnapshot blocking)
{
foreach (var head in blocking.HeadBlockers)
{
var chain = Chain(blocking, head.SessionId);
var maxWait = chain.Max(s => s.WaitTimeMs ?? 0);
if (maxWait < BlockingThresholdMs)
{
continue;
}
await connection.ExecuteAsync("""
IF NOT EXISTS (SELECT 1 FROM insight.BlockingEvents WHERE InstanceId = @instanceId AND SampledAt = @at AND HeadSessionId = @head)
INSERT insight.BlockingEvents (InstanceId, SampledAt, HeadSessionId, BlockedCount, MaxWaitMs, SessionsJson)
VALUES (@instanceId, @at, @head, @count, @maxWait, @json);
""", new
{
instanceId,
at = Utc(blocking.CapturedAt),
head = head.SessionId,
count = chain.Count,
maxWait,
json = JsonSerializer.Serialize(chain.Prepend(head), Json),
}).ConfigureAwait(false);
}
}
private static List Chain(BlockingSnapshot snapshot, int head)
{
var result = new List();
var queue = new Queue([head]);
var seen = new HashSet { head };
while (queue.Count > 0)
{
foreach (var s in snapshot.BlockedBy(queue.Dequeue()).Where(s => seen.Add(s.SessionId)))
{
result.Add(s);
queue.Enqueue(s.SessionId);
}
}
return result;
}
private static async Task SaveDeadlocksAsync(SqlConnection connection, int instanceId, DeadlockSnapshot snapshot)
{
foreach (var deadlock in snapshot.NewDeadlocks)
{
await connection.ExecuteAsync("""
IF NOT EXISTS (SELECT 1 FROM insight.Deadlocks WHERE InstanceId = @instanceId AND OccurredAt = @at AND GraphXml = @graph)
INSERT insight.Deadlocks (InstanceId, OccurredAt, ProcessesJson, GraphXml)
VALUES (@instanceId, @at, @processes, @graph);
""", new
{
instanceId,
at = UtcMilliseconds(deadlock.Time),
processes = JsonSerializer.Serialize(deadlock.Processes, Json),
graph = deadlock.GraphXml,
}).ConfigureAwait(false);
}
}
}
public sealed partial class InsightRepository
{
///
/// The newest stored deadlock for an instance, so a restarted agent continues after it instead of reporting
/// deadlocks it has already stored.
///
public async Task GetLatestDeadlockTimeAsync(int instanceId, CancellationToken cancellationToken = default)
{
await using var connection = await OpenAsync(cancellationToken).ConfigureAwait(false);
var latest = await connection.ExecuteScalarAsync(
"SELECT MAX(OccurredAt) FROM insight.Deadlocks WHERE InstanceId = @instanceId;", new { instanceId }).ConfigureAwait(false);
return latest is { } value ? new DateTimeOffset(DateTime.SpecifyKind(value, DateTimeKind.Utc)) : null;
}
}
public sealed record FindingChanges(IReadOnlyList<(long FindingId, Finding Finding)> Opened, IReadOnlyList Resolved)
{
public static readonly FindingChanges None = new([], []);
}
public sealed record ResolvedFinding
{
public long FindingId { get; init; }
public string RuleId { get; init; } = string.Empty;
public string Subject { get; init; } = string.Empty;
public string Title { get; init; } = string.Empty;
public byte Severity { get; init; }
}
public static class Metrics
{
public const string CpuSqlServer = "cpu.sqlserver";
public const string CpuOther = "cpu.other";
public const string PageLifeExpectancy = "memory.ple";
public const string MemoryGrantsPending = "memory.grants_pending";
public const string UserConnections = "connections.user";
public const string ProcessesBlocked = "blocking.processes";
public const string BatchRequestsPerSecond = "workload.batch_requests";
public const string CompilationsPerSecond = "workload.compilations";
}