feat(historian): SQL Server retention purge (RetentionDays, Primary-gated, applock) (#667)

2026-10-11 08:21:30 -04:00
parent db6a53036b
commit 924eff8d3c
13 changed files with 602 additions and 2 deletions
@@ -22,6 +22,21 @@ public sealed partial class SqlServerHistorianOptions
    /// </summary>
    public bool AutoCreateSchema { get; set; } = true;

+    /// <summary>
+    ///     Days of history to keep, measured on source time. <c>null</c> (default) keeps everything. When set,
+    ///     a driver node deletes older <c>hist.Sample</c>/<c>hist.Event</c> rows every <see cref="PurgeInterval"/>.
+    /// </summary>
+    public int? RetentionDays { get; set; }
+
+    /// <summary>How often the retention purge runs. Must be positive.</summary>
+    public TimeSpan PurgeInterval { get; set; } = TimeSpan.FromHours(1);
+
+    /// <summary>
+    ///     Rows per <c>DELETE TOP(n)</c> batch. Default 4000 stays under SQL Server's 5000-lock escalation
+    ///     threshold so a purge never escalates to a table lock.
+    /// </summary>
+    public int PurgeBatchSize { get; set; } = 4000;
+
    [GeneratedRegex("^[A-Za-z_][A-Za-z0-9_]{0,127}$")]
    private static partial Regex IdentifierRegex();

@@ -41,6 +56,12 @@ public sealed partial class SqlServerHistorianOptions
            errors.Add("ServerHistorian:SqlServer:Schema must be a SQL identifier matching ^[A-Za-z_][A-Za-z0-9_]{0,127}$.");
        if (CommandTimeout <= TimeSpan.Zero)
            errors.Add("ServerHistorian:SqlServer:CommandTimeout must be greater than zero.");
+        if (RetentionDays is < 1)
+            errors.Add("ServerHistorian:SqlServer:RetentionDays must be at least 1 when set (omit it to keep history forever).");
+        if (PurgeInterval <= TimeSpan.Zero)
+            errors.Add("ServerHistorian:SqlServer:PurgeInterval must be greater than zero.");
+        if (PurgeBatchSize < 1)
+            errors.Add("ServerHistorian:SqlServer:PurgeBatchSize must be at least 1.");
        return errors;
    }
 }
@@ -96,5 +96,14 @@ public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
        return new SqlServerHistorianDataUpdater(s.Conn, s.Catalog);
    }

+    /// <inheritdoc />
+    public IHistorianRetention? CreateRetention(IServiceProvider sp)
+    {
+        if (_options.RetentionDays is not { } days) return null;
+        var s = Get(sp);
+        return new SqlServerHistorianRetention(s.Conn, s.Bootstrap, _options.PurgeInterval, days,
+            _options.PurgeBatchSize, Loggers(sp).CreateLogger<SqlServerHistorianRetention>());
+    }
+
    private sealed record Shared(SqlHistorianConnection Conn, SqlServerSchemaBootstrapper Bootstrap, SqlServerTagCatalog Catalog);
 }
@@ -0,0 +1,118 @@
+using System.Data;
+using Microsoft.Data.SqlClient;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
+
+/// <summary>
+///     Deletes <c>hist.Sample</c> / <c>hist.Event</c> rows older than the retention window, in small
+///     <c>DELETE TOP(n)</c> batches (under the lock-escalation threshold). Samples are purged per tag so each
+///     delete is a seek on the clustered key <c>(TagId, SourceTimeUtc)</c>; events seek on <c>CX_Event_Time</c>.
+///     A session <c>sp_getapplock</c> (timeout 0) serialises purgers that share one database; a loser skips.
+///     To purge another table later, add it to <see cref="PurgeAsync"/>.
+/// </summary>
+internal sealed class SqlServerHistorianRetention : IHistorianRetention
+{
+    private readonly SqlHistorianConnection _conn;
+    private readonly SqlServerSchemaBootstrapper _bootstrap;
+    private readonly int _days;
+    private readonly int _batch;
+    private readonly TimeProvider _time;
+    private readonly ILogger _log;
+
+    public SqlServerHistorianRetention(SqlHistorianConnection conn, SqlServerSchemaBootstrapper bootstrap,
+        TimeSpan interval, int retentionDays, int batchSize, ILogger? log = null, TimeProvider? time = null)
+    {
+        _conn = conn;
+        _bootstrap = bootstrap;
+        Interval = interval;
+        _days = retentionDays;
+        _batch = batchSize;
+        _log = log ?? NullLogger.Instance;
+        _time = time ?? TimeProvider.System;
+    }
+
+    public TimeSpan Interval { get; }
+
+    public async Task<IReadOnlyList<RetentionPurgeResult>> PurgeAsync(CancellationToken ct)
+    {
+        await _bootstrap.EnsureAsync(ct).ConfigureAwait(false);
+        var cutoff = _time.GetUtcNow().UtcDateTime.AddDays(-_days);
+
+        await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false);
+        if (!await TryLockAsync(conn, ct).ConfigureAwait(false))
+        {
+            _log.LogDebug("Historian retention purge skipped: another purger holds the applock.");
+            return [];
+        }
+
+        try
+        {
+            long samples = 0;
+            var tagIds = new List<int>();
+            await using (var cmd = _conn.CreateCommand(conn, $"SELECT TagId FROM {_conn.Q("Tag")}"))
+            await using (var r = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false))
+                while (await r.ReadAsync(ct).ConfigureAwait(false)) tagIds.Add(r.GetInt32(0));
+
+            foreach (var id in tagIds)
+            {
+                samples += await DeleteLoopAsync(conn,
+                    $"DELETE TOP (@n) FROM {_conn.Q("Sample")} WHERE TagId = @id AND SourceTimeUtc < @cutoff",
+                    cutoff, id, ct).ConfigureAwait(false);
+            }
+
+            var events = await DeleteLoopAsync(conn,
+                $"DELETE TOP (@n) FROM {_conn.Q("Event")} WHERE EventTimeUtc < @cutoff",
+                cutoff, null, ct).ConfigureAwait(false);
+
+            return [new RetentionPurgeResult("Sample", samples), new RetentionPurgeResult("Event", events)];
+        }
+        finally
+        {
+            await ReleaseAsync(conn).ConfigureAwait(false);
+        }
+    }
+
+    private async Task<long> DeleteLoopAsync(SqlConnection conn, string sql, DateTime cutoff, int? tagId, CancellationToken ct)
+    {
+        long total = 0;
+        while (true)
+        {
+            ct.ThrowIfCancellationRequested();
+            await using var cmd = _conn.CreateCommand(conn, sql);
+            cmd.Parameters.Add(new SqlParameter("@n", SqlDbType.Int) { Value = _batch });
+            if (tagId is { } id) cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id });
+            SqlHistorianConnection.AddDateTime2(cmd, "@cutoff", cutoff);
+            var n = await cmd.ExecuteNonQueryAsync(ct).ConfigureAwait(false);
+            total += n;
+            if (n < _batch) return total;
+        }
+    }
+
+    private string LockResource => $"OtOpcUa.Historian.Retention.{_conn.Schema}";
+
+    private async Task<bool> TryLockAsync(SqlConnection conn, CancellationToken ct)
+    {
+        await using var cmd = _conn.CreateCommand(conn,
+            "DECLARE @r int; EXEC @r = sp_getapplock @Resource = @res, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT @r;");
+        cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource });
+        return Convert.ToInt32(await cmd.ExecuteScalarAsync(ct).ConfigureAwait(false)) >= 0;
+    }
+
+    private async Task ReleaseAsync(SqlConnection conn)
+    {
+        try
+        {
+            await using var cmd = _conn.CreateCommand(conn,
+                "EXEC sp_releaseapplock @Resource = @res, @LockOwner = 'Session';");
+            cmd.Parameters.Add(new SqlParameter("@res", SqlDbType.NVarChar, 255) { Value = LockResource });
+            await cmd.ExecuteNonQueryAsync(CancellationToken.None).ConfigureAwait(false);
+        }
+        catch (Exception ex)
+        {
+            _log.LogDebug(ex, "Releasing the retention applock failed.");
+        }
+    }
+}
@@ -326,6 +326,14 @@ if (hasDriver)
        builder.Services.AddSingleton<IHistorianValueWriter>(sp => sp.GetRequiredService<IHistorianBackendProvider>().CreateValueWriter(sp));
    }

+    // Historian retention (#667). Only a backend that manages retention in-process returns a purger
+    // (SqlServer with RetentionDays set); otherwise the service is inert. Primary-gated via IHistorizationRoleView.
+    if (serverHistorianOptions.Enabled)
+        builder.Services.AddHostedService(sp => new HistorianRetentionService(
+            sp.GetRequiredService<IHistorianBackendProvider>().CreateRetention(sp),
+            sp.GetService<ZB.MOM.WW.OtOpcUa.Runtime.Redundancy.IHistorizationRoleView>(),
+            sp.GetService<ILogger<HistorianRetentionService>>()));
+
    // Bind every cross-platform driver factory before AddAkka resolves IDriverFactory — replaces
    // the F7-default NullDriverFactory with a real DriverFactoryRegistryAdapter so DriverHostActor
    // can materialise real IDriver instances on deploy.
@@ -0,0 +1,95 @@
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Commons.Observability;
+using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+/// <summary>
+///     Timer that drives an <see cref="IHistorianRetention"/> purge. Each tick is gated on
+///     <see cref="IHistorizationRoleView.ShouldRecordHistorizedValues"/> so only the node that is writing
+///     history purges it (an unknown role allows, matching the capture gate). A null retention (backend has
+///     none, or no window configured) makes the service inert. Never throws out of the loop.
+/// </summary>
+public sealed class HistorianRetentionService : BackgroundService
+{
+    /// <summary>Counter of rows purged, tagged <c>table</c>.</summary>
+    public const string PurgedCounterName = "otopcua.historian.retention.purged";
+
+    private static readonly Counter<long> Purged = OtOpcUaTelemetry.Meter.CreateCounter<long>(
+        PurgedCounterName, unit: "{row}", description: "Historian rows deleted by the retention purge.");
+
+    private readonly IHistorianRetention? _retention;
+    private readonly IHistorizationRoleView? _roleView;
+    private readonly ILogger _log;
+    private readonly KeyValuePair<string, object?>[] _extraTags;
+
+    /// <summary>Creates the service.</summary>
+    /// <param name="retention">Backend purge seam; null makes the service inert.</param>
+    /// <param name="roleView">Primary gate; null means always run.</param>
+    /// <param name="log">Logger.</param>
+    /// <param name="extraTags">Extra tags added to every counter increment (tests use a unique one).</param>
+    public HistorianRetentionService(IHistorianRetention? retention, IHistorizationRoleView? roleView,
+        ILogger<HistorianRetentionService>? log = null, params KeyValuePair<string, object?>[] extraTags)
+    {
+        _retention = retention;
+        _roleView = roleView;
+        _log = (ILogger?)log ?? NullLogger.Instance;
+        _extraTags = extraTags;
+    }
+
+    /// <summary>Runs one gated tick. Never throws.</summary>
+    /// <param name="ct">Cancellation.</param>
+    /// <returns>True when a purge pass ran.</returns>
+    public async Task<bool> TickAsync(CancellationToken ct)
+    {
+        if (_retention is null) return false;
+        if (_roleView is { ShouldRecordHistorizedValues: false })
+        {
+            _log.LogDebug("Historian retention purge skipped: this node is not the recording node.");
+            return false;
+        }
+
+        try
+        {
+            var results = await _retention.PurgeAsync(ct).ConfigureAwait(false);
+            foreach (var r in results)
+            {
+                if (r.Rows <= 0) continue;
+                var tags = new TagList { { "table", r.Table } };
+                foreach (var t in _extraTags) tags.Add(t);
+                Purged.Add(r.Rows, tags);
+                _log.LogInformation("Historian retention purged {Rows} rows from {Table}.", r.Rows, r.Table);
+            }
+            return true;
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            return false;
+        }
+        catch (Exception ex)
+        {
+            _log.LogWarning(ex, "Historian retention purge failed; will retry next interval.");
+            return false;
+        }
+    }
+
+    /// <inheritdoc/>
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        if (_retention is null) return;
+        try
+        {
+            using var timer = new PeriodicTimer(_retention.Interval);
+            await TickAsync(stoppingToken).ConfigureAwait(false);
+            while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
+                await TickAsync(stoppingToken).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+        {
+        }
+    }
+}
@@ -39,4 +39,9 @@ public interface IHistorianBackendProvider
    /// <param name="sp">Service provider.</param>
    /// <returns>The updater, or <c>null</c> when the backend does not support HistoryUpdate.</returns>
    IHistorianDataUpdater? CreateUpdater(IServiceProvider sp);
+
+    /// <summary>Creates the retention purge seam, if the backend manages retention in-process.</summary>
+    /// <param name="sp">Service provider.</param>
+    /// <returns>The purger, or <c>null</c> (default) when the backend has no retention or enforces it itself.</returns>
+    IHistorianRetention? CreateRetention(IServiceProvider sp) => null;
 }
@@ -0,0 +1,25 @@
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
+
+/// <summary>Rows removed from one historian table by a single purge pass.</summary>
+/// <param name="Table">Unqualified table name (the <c>table</c> tag on the retention counter).</param>
+/// <param name="Rows">Rows deleted.</param>
+public sealed record RetentionPurgeResult(string Table, long Rows);
+
+/// <summary>
+///     Backend-side retention purge: deletes history older than the backend's configured retention window.
+///     Created by <see cref="IHistorianBackendProvider.CreateRetention"/> and driven on a timer by
+///     <see cref="HistorianRetentionService"/>, which owns scheduling and the Primary gate.
+/// </summary>
+public interface IHistorianRetention
+{
+    /// <summary>How often <see cref="HistorianRetentionService"/> should run a pass.</summary>
+    TimeSpan Interval { get; }
+
+    /// <summary>
+    ///     Runs one purge pass. Returns an empty list when the pass was skipped (another process holds the
+    ///     purge lock). Implementations honour cancellation between batches.
+    /// </summary>
+    /// <param name="ct">Cancellation.</param>
+    /// <returns>Per-table deleted-row counts.</returns>
+    Task<IReadOnlyList<RetentionPurgeResult>> PurgeAsync(CancellationToken ct);
+}