Merge feat/667-sqlserver-retention: SQL Server historian retention/purge (#667)
This commit is contained in:
Joseph Doherty
+174
@@ -0,0 +1,174 @@
|
||||
using Microsoft.Data.SqlClient;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.IntegrationTests;
|
||||
|
||||
[Collection(SqlHistorianCollection.Name)]
|
||||
public sealed class RetentionTests(SqlHistorianFixture fx)
|
||||
{
|
||||
private static readonly DateTime Now = new(2026, 10, 11, 12, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
private sealed class FixedTime(DateTime utc) : TimeProvider
|
||||
{
|
||||
public override DateTimeOffset GetUtcNow() => new(utc, TimeSpan.Zero);
|
||||
}
|
||||
|
||||
private (SqlHistorianConnection C, SqlServerSchemaBootstrapper B, SqlServerHistorianOptions O, string S) Make()
|
||||
{
|
||||
var s = SqlHistorianFixture.UniqueSchema();
|
||||
var o = new SqlServerHistorianOptions { ConnectionString = fx.ConnectionString, Schema = s };
|
||||
var c = new SqlHistorianConnection(o);
|
||||
return (c, new SqlServerSchemaBootstrapper(c, o), o, s);
|
||||
}
|
||||
|
||||
private async Task ExecAsync(string sql)
|
||||
{
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = sql;
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
private async Task<long> CountAsync(string sql)
|
||||
{
|
||||
await using var c = new SqlConnection(fx.ConnectionString);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = sql;
|
||||
return Convert.ToInt64(await cmd.ExecuteScalarAsync());
|
||||
}
|
||||
|
||||
/// <summary>3 tags x (5 old + 2 new) samples, plus 4 old + 1 new events. Retention 30 days at <see cref="Now"/>.</summary>
|
||||
private async Task SeedAsync(SqlServerSchemaBootstrapper b, string s)
|
||||
{
|
||||
await b.EnsureAsync(CancellationToken.None);
|
||||
var old = Now.AddDays(-40).ToString("yyyy-MM-ddTHH:mm:ss");
|
||||
var fresh = Now.AddDays(-1).ToString("yyyy-MM-ddTHH:mm:ss");
|
||||
var sql = new System.Text.StringBuilder();
|
||||
for (var t = 1; t <= 3; t++)
|
||||
{
|
||||
sql.Append($"INSERT [{s}].Tag (Name, DataType, CreatedUtc, UpdatedUtc) VALUES (N'tag{t}', 1, SYSUTCDATETIME(), SYSUTCDATETIME());");
|
||||
}
|
||||
sql.Append($"DECLARE @i int = 0; WHILE @i < 5 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{old}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;");
|
||||
sql.Append($"SET @i = 0; WHILE @i < 2 BEGIN INSERT [{s}].Sample (TagId, SourceTimeUtc, ValueKind, NumValue, StatusCode, ServerTimeUtc) SELECT TagId, DATEADD(minute, @i, '{fresh}'), 1, 1.0, 0, SYSUTCDATETIME() FROM [{s}].Tag; SET @i += 1; END;");
|
||||
sql.Append($"SET @i = 0; WHILE @i < 4 BEGIN INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES (CONCAT('old', @i), 's', 'a', 'n', 't', 1, 'Activated', DATEADD(minute, @i, '{old}'), SYSUTCDATETIME()); SET @i += 1; END;");
|
||||
sql.Append($"INSERT [{s}].Event (EventId, SourceName, AlarmId, AlarmName, AlarmTypeName, Severity, EventKind, EventTimeUtc, ReceivedTimeUtc) VALUES ('new0', 's', 'a', 'n', 't', 1, 'Activated', '{fresh}', SYSUTCDATETIME());");
|
||||
await ExecAsync(sql.ToString());
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(4000)]
|
||||
[InlineData(1)] // forces the delete loop to iterate once per row
|
||||
public async Task Purge_removes_only_rows_older_than_the_window(int batch)
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, batch, time: new FixedTime(Now));
|
||||
var results = await purger.PurgeAsync(CancellationToken.None);
|
||||
|
||||
results.ShouldBe([new RetentionPurgeResult("Sample", 15), new RetentionPurgeResult("Event", 4)]);
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6);
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample WHERE SourceTimeUtc < '{Now.AddDays(-30):yyyy-MM-ddTHH:mm:ss}'")).ShouldBe(0);
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Event")).ShouldBe(1);
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Tag")).ShouldBe(3);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_second_purge_is_a_no_op()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
|
||||
await purger.PurgeAsync(CancellationToken.None);
|
||||
var again = await purger.PurgeAsync(CancellationToken.None);
|
||||
again.ShouldBe([new RetentionPurgeResult("Sample", 0), new RetentionPurgeResult("Event", 0)]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_purger_skips_while_another_holds_the_applock()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
|
||||
// Hold the same session applock the purger takes, on a separate connection.
|
||||
await using var holder = new SqlConnection(fx.ConnectionString);
|
||||
await holder.OpenAsync();
|
||||
await using (var cmd = holder.CreateCommand())
|
||||
{
|
||||
cmd.CommandText = $"EXEC sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0;";
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
|
||||
(await purger.PurgeAsync(CancellationToken.None)).ShouldBeEmpty();
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(21);
|
||||
|
||||
await using (var cmd = holder.CreateCommand())
|
||||
{
|
||||
cmd.CommandText = $"EXEC sp_releaseapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockOwner = 'Session';";
|
||||
await cmd.ExecuteNonQueryAsync();
|
||||
}
|
||||
(await purger.PurgeAsync(CancellationToken.None)).ShouldNotBeEmpty();
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(6);
|
||||
}
|
||||
|
||||
/// <summary>True when a fresh, non-pooled session can take the retention applock (i.e. nobody holds it).</summary>
|
||||
private async Task<bool> LockIsFreeAsync(string s)
|
||||
{
|
||||
var cs = new SqlConnectionStringBuilder(fx.ConnectionString) { Pooling = false }.ConnectionString;
|
||||
await using var c = new SqlConnection(cs);
|
||||
await c.OpenAsync();
|
||||
await using var cmd = c.CreateCommand();
|
||||
cmd.CommandText = $"DECLARE @r int; EXEC @r = sp_getapplock @Resource = N'OtOpcUa.Historian.Retention.{s}', @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;";
|
||||
return Convert.ToInt32(await cmd.ExecuteScalarAsync()) >= 0;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task The_applock_is_released_after_a_purge()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 4000, time: new FixedTime(Now));
|
||||
await purger.PurgeAsync(CancellationToken.None);
|
||||
// A pooled-session re-entry would hide a missing release; a separate non-pooled session cannot.
|
||||
(await LockIsFreeAsync(s)).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Cancelling_between_batches_stops_the_purge_and_still_releases_the_applock()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
using var cts = new CancellationTokenSource();
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
|
||||
var batches = 0;
|
||||
purger.AfterBatch = () => { if (++batches == 2) cts.Cancel(); };
|
||||
|
||||
await Should.ThrowAsync<OperationCanceledException>(() => purger.PurgeAsync(cts.Token));
|
||||
|
||||
batches.ShouldBe(2);
|
||||
(await CountAsync($"SELECT COUNT(*) FROM [{s}].Sample")).ShouldBe(19); // 21 - two batches of 1
|
||||
(await LockIsFreeAsync(s)).ShouldBeTrue();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_throwing_purge_releases_the_applock()
|
||||
{
|
||||
fx.SkipIfUnavailable();
|
||||
var (c, b, _, s) = Make();
|
||||
await SeedAsync(b, s);
|
||||
var purger = new SqlServerHistorianRetention(c, b, TimeSpan.FromHours(1), 30, 1, time: new FixedTime(Now));
|
||||
purger.AfterBatch = () => throw new InvalidOperationException("boom");
|
||||
await Should.ThrowAsync<InvalidOperationException>(() => purger.PurgeAsync(CancellationToken.None));
|
||||
(await LockIsFreeAsync(s)).ShouldBeTrue();
|
||||
}
|
||||
}
|
||||
+62
@@ -0,0 +1,62 @@
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.Tests;
|
||||
|
||||
public sealed class SqlServerHistorianRetentionOptionsTests
|
||||
{
|
||||
private static SqlServerHistorianOptions Valid() => new() { ConnectionString = "Server=x" };
|
||||
|
||||
[Fact]
|
||||
public void Retention_is_off_by_default_with_sane_defaults()
|
||||
{
|
||||
var o = Valid();
|
||||
o.RetentionDays.ShouldBeNull();
|
||||
o.PurgeInterval.ShouldBe(TimeSpan.FromHours(1));
|
||||
o.PurgeBatchSize.ShouldBe(4000);
|
||||
o.Validate().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(0)]
|
||||
[InlineData(-5)]
|
||||
public void Non_positive_RetentionDays_is_rejected(int days)
|
||||
{
|
||||
var o = Valid();
|
||||
o.RetentionDays = days;
|
||||
o.Validate().ShouldContain(e => e.Contains("RetentionDays"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Positive_RetentionDays_is_valid()
|
||||
{
|
||||
var o = Valid();
|
||||
o.RetentionDays = 30;
|
||||
o.Validate().ShouldBeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Bad_interval_and_batch_are_rejected()
|
||||
{
|
||||
var o = Valid();
|
||||
o.PurgeInterval = TimeSpan.Zero;
|
||||
o.PurgeBatchSize = 0;
|
||||
var errors = o.Validate();
|
||||
errors.ShouldContain(e => e.Contains("PurgeInterval"));
|
||||
errors.ShouldContain(e => e.Contains("PurgeBatchSize"));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Provider_creates_retention_only_when_RetentionDays_set()
|
||||
{
|
||||
var o = Valid();
|
||||
new SqlServerHistorianProvider(o).CreateRetention(new EmptyServices()).ShouldBeNull();
|
||||
o.RetentionDays = 7;
|
||||
new SqlServerHistorianProvider(o).CreateRetention(new EmptyServices()).ShouldNotBeNull();
|
||||
}
|
||||
|
||||
private sealed class EmptyServices : IServiceProvider
|
||||
{
|
||||
public object? GetService(Type serviceType) => null;
|
||||
}
|
||||
}
|
||||
+95
@@ -0,0 +1,95 @@
|
||||
using System.Diagnostics.Metrics;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Historian;
|
||||
|
||||
public sealed class HistorianRetentionServiceTests
|
||||
{
|
||||
private sealed class FakeRetention(Func<Task<IReadOnlyList<RetentionPurgeResult>>> run) : IHistorianRetention
|
||||
{
|
||||
public int Calls;
|
||||
public TimeSpan Interval => TimeSpan.FromHours(1);
|
||||
public Task<IReadOnlyList<RetentionPurgeResult>> PurgeAsync(CancellationToken ct)
|
||||
{
|
||||
Interlocked.Increment(ref Calls);
|
||||
return run();
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class FakeRole(bool record) : IHistorizationRoleView
|
||||
{
|
||||
public bool ShouldRecordHistorizedValues => record;
|
||||
public void PublishHistorizationRole(bool shouldRecordHistorizedValues) { }
|
||||
}
|
||||
|
||||
private static Task<IReadOnlyList<RetentionPurgeResult>> Rows(long sample, long ev) =>
|
||||
Task.FromResult<IReadOnlyList<RetentionPurgeResult>>(
|
||||
[new RetentionPurgeResult("Sample", sample), new RetentionPurgeResult("Event", ev)]);
|
||||
|
||||
[Fact]
|
||||
public async Task Runs_when_role_is_unknown_and_when_recording()
|
||||
{
|
||||
var r1 = new FakeRetention(() => Rows(0, 0));
|
||||
(await new HistorianRetentionService(r1, null).TickAsync(CancellationToken.None)).ShouldBeTrue();
|
||||
var r2 = new FakeRetention(() => Rows(0, 0));
|
||||
(await new HistorianRetentionService(r2, new FakeRole(true)).TickAsync(CancellationToken.None)).ShouldBeTrue();
|
||||
r1.Calls.ShouldBe(1);
|
||||
r2.Calls.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Skips_when_not_the_recording_node()
|
||||
{
|
||||
var r = new FakeRetention(() => Rows(1, 1));
|
||||
(await new HistorianRetentionService(r, new FakeRole(false)).TickAsync(CancellationToken.None)).ShouldBeFalse();
|
||||
r.Calls.ShouldBe(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Null_retention_is_inert()
|
||||
{
|
||||
(await new HistorianRetentionService(null, null).TickAsync(CancellationToken.None)).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task A_throwing_purge_does_not_escape()
|
||||
{
|
||||
var r = new FakeRetention(() => throw new InvalidOperationException("boom"));
|
||||
(await new HistorianRetentionService(r, null).TickAsync(CancellationToken.None)).ShouldBeFalse();
|
||||
r.Calls.ShouldBe(1);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Emits_purged_counter_per_table_and_skips_zero_rows()
|
||||
{
|
||||
var marker = Guid.NewGuid().ToString("N");
|
||||
var seen = new List<(string Table, long Value)>();
|
||||
using var listener = new MeterListener();
|
||||
listener.InstrumentPublished = (inst, l) =>
|
||||
{
|
||||
if (inst.Meter.Name == OtOpcUaTelemetry.MeterName && inst.Name == HistorianRetentionService.PurgedCounterName)
|
||||
l.EnableMeasurementEvents(inst);
|
||||
};
|
||||
listener.SetMeasurementEventCallback<long>((_, value, tags, _) =>
|
||||
{
|
||||
string? table = null; var mine = false;
|
||||
foreach (var t in tags)
|
||||
{
|
||||
if (t.Key == "table") table = (string?)t.Value;
|
||||
if (t.Key == "test" && (string?)t.Value == marker) mine = true;
|
||||
}
|
||||
if (mine) lock (seen) seen.Add((table!, value));
|
||||
});
|
||||
listener.Start();
|
||||
|
||||
var r = new FakeRetention(() => Rows(7, 0));
|
||||
var svc = new HistorianRetentionService(r, null, null, new KeyValuePair<string, object?>("test", marker));
|
||||
await svc.TickAsync(CancellationToken.None);
|
||||
|
||||
lock (seen) seen.ShouldBe([("Sample", 7L)]);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user