using System.Data.Common;
using SQLInsight.Core.Data;
using SQLInsight.Core.Instances;
namespace SQLInsight.Collectors.Instance;
///
/// On a distributor: each distribution agent's latest run (status, latency, message) and replication errors the last
/// day, as SELECTs in the distribution database. Reading it needs SELECT on those three tables (not on the replicated
/// commands); without it the snapshot says why. SQL Server 2005+.
///
public sealed class ReplicationCollector : Collector
{
internal const string DistributionDatabaseSql = "SELECT TOP 1 name FROM sys.databases WHERE is_distributor = 1 AND state = 0";
internal const string AgentsSql = """
SELECT TOP 500 a.name, a.publisher_db, h.runstatus, DATEADD(minute, DATEDIFF(minute, GETDATE(), GETUTCDATE()), h.time), h.delivery_latency,
CAST(h.comments AS nvarchar(500))
FROM dbo.MSdistribution_agents a
OUTER APPLY (SELECT TOP 1 x.runstatus, x.time, x.delivery_latency, x.comments FROM dbo.MSdistribution_history x
WHERE x.agent_id = a.id ORDER BY x.timestamp DESC) h
""";
internal const string ErrorsSql = "SELECT COUNT(*) FROM dbo.MSrepl_errors WHERE time > DATEADD(hour, -24, GETDATE())";
public const string NoRightsText =
"Distributionsdatabasen kan ikke læses med agentens rettigheder. Giv agenten læseadgang til netop agenternes status (ikke de replikerede data): "
+ "USE distribution; CREATE USER [agentens login]; GRANT SELECT ON dbo.MSdistribution_agents TO [agentens login]; "
+ "GRANT SELECT ON dbo.MSdistribution_history TO [agentens login]; GRANT SELECT ON dbo.MSrepl_errors TO [agentens login].";
public override string Name => "replication";
public override TimeSpan DefaultInterval => TimeSpan.FromMinutes(15);
public override IReadOnlyCollection RequiredCapabilities => [Capability.DynamicManagementViews];
protected override async Task CollectAsync(CollectorContext context, ReplicationSnapshot? previous)
{
var ct = context.CancellationToken;
var connection = context.Connection;
var distribution = (await Db.QueryAsync(connection, DistributionDatabaseSql, r => r.String(0), ct).ConfigureAwait(false)).FirstOrDefault();
if (distribution is null)
{
return new ReplicationSnapshot { CapturedAt = context.Now };
}
var original = (await Db.QueryAsync(connection, "SELECT DB_NAME()", r => r.String(0), ct).ConfigureAwait(false))[0];
try
{
var use = "USE " + Db.Identifier(distribution) + ";\n";
var agents = await Db.QueryAsync(connection, use + AgentsSql, r => new DistributionAgent(
r.String(0), r.NullableString(1), r.NullableInt(2),
r.NullableDateTime(3) is { } at ? new DateTimeOffset(DateTime.SpecifyKind(at, DateTimeKind.Utc)) : null,
r.NullableLong(4), r.NullableString(5)), ct).ConfigureAwait(false);
var errors = (await Db.QueryAsync(connection, use + ErrorsSql, r => r.Int(0), ct).ConfigureAwait(false))[0];
return new ReplicationSnapshot { CapturedAt = context.Now, IsDistributor = true, DistributionDatabase = distribution, Agents = agents, ErrorsLastDay = errors };
}
catch (DbException ex)
{
return new ReplicationSnapshot
{
CapturedAt = context.Now, IsDistributor = true, DistributionDatabase = distribution,
// The error message names the agent's login: kept apart from the reason, which rules and the API show as is.
Unavailable = NoRightsText,
UnavailableDetail = ex.Message.Split('\n')[0].Trim(),
};
}
finally
{
await Db.ExecuteAsync(connection, "USE " + Db.Identifier(original), ct).ConfigureAwait(false);
}
}
}