Files
lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianProvider.cs
T
Joseph Doherty 0658d244d0 feat(historian): #674 HistoryUpdate posture — future-skew bound, Primary gate, SqlServer v1 accepted, QuestDB TTL pin
- HistoryUpdate items with SourceTimestamp > now + ServerHistorian:HistoryUpdateMaxFutureSkew
  (default 1 day, <=0 disables) => BadOutOfRange in the node manager (every backend, never sent);
  clock injected via OtOpcUaNodeManager.Clock (host TimeProvider).
- HistoryUpdate is Primary-only: off-Primary => BadInvalidState per node, after authorization,
  same IDriverPrimaryRoleSource verdict as the native-ack gate; HistoryRead ungated.
- SqlServer: base schema requirement back to v1; SupportsModifiedTrail (v>=2). On v1 with
  AutoCreateSchema=false: writes/reads/events/retention work, modified reads + HistoryUpdate =>
  BadHistoryOperationUnsupported with nothing sent, retention skips SampleModified, one Warning per boot.
- QuestDB integration test pinning wall-clock TTL (far-future row evicts only partitions older
  than the TTL by the wall clock), with a positive control.
- Docs: Historian.md, Configuration.md, security.md, CLAUDE.md (incl. QuestDB retry duplicates,
  #674.4 anonymous-readable trail with ACL off, #674.5 not implemented).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-11 11:18:44 -04:00

116 lines
4.9 KiB
C#

using ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
using System.Collections.Frozen;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer;
/// <summary>
/// <see cref="IHistorianBackendProvider"/> for the SQL Server historian. Supports every
/// <see cref="HistorianValueKind"/> and HistoryUpdate. All seams share ONE connection helper, ONE schema
/// bootstrapper and ONE tag catalog, so the schema is ensured once and tag ids are cached once.
/// </summary>
/// <remarks>
/// The shared state is built on the first <c>Create*</c> call, not in the constructor: the constructor does no
/// validation and no I/O, so resolving the provider for a backend nobody consumes never fails on its options
/// (they are validated, consumer-gated, by the host's <c>ServerHistorianOptionsValidator</c>). The first
/// <c>Create*</c> call's service provider supplies the logger factory the bootstrapper logs through.
/// </remarks>
public sealed class SqlServerHistorianProvider : IHistorianBackendProvider
{
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().ToFrozenSet();
private readonly SqlServerHistorianOptions _options;
private readonly object _initLock = new();
private Shared? _shared;
/// <summary>Creates the provider over the <c>ServerHistorian:SqlServer</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:SqlServer</c> options.</param>
/// <param name="maxTimeWeightedRawRows">The shared <c>ServerHistorian:MaxTimeWeightedRawRows</c> cap (&lt;= 0 = unbounded).</param>
public SqlServerHistorianProvider(SqlServerHistorianOptions options,
int maxTimeWeightedRawRows = TimeWeightedAggregator.DefaultMaxWindowRows)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
_maxTimeWeightedRawRows = maxTimeWeightedRawRows;
}
private readonly int _maxTimeWeightedRawRows;
/// <inheritdoc />
public string Name => "SqlServer";
/// <inheritdoc />
public IReadOnlySet<HistorianValueKind> SupportedKinds => Kinds;
private static ILoggerFactory Loggers(IServiceProvider sp) =>
(sp.GetService(typeof(ILoggerFactory)) as ILoggerFactory) ?? NullLoggerFactory.Instance;
private Shared Get(IServiceProvider sp)
{
if (Volatile.Read(ref _shared) is { } built) return built;
lock (_initLock)
{
if (_shared is { } raced) return raced;
var conn = new SqlHistorianConnection(_options);
var bootstrap = new SqlServerSchemaBootstrapper(conn, _options,
Loggers(sp).CreateLogger<SqlServerSchemaBootstrapper>());
var shared = new Shared(conn, bootstrap, new SqlServerTagCatalog(conn, bootstrap));
Volatile.Write(ref _shared, shared);
return shared;
}
}
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<SqlServerHistorianDataSource>(), maxTimeWeightedRawRows: _maxTimeWeightedRawRows);
}
/// <inheritdoc />
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianValueWriter(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<SqlServerHistorianValueWriter>());
}
/// <inheritdoc />
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerAlarmHistorianWriter(s.Conn, s.Bootstrap,
Loggers(sp).CreateLogger<SqlServerAlarmHistorianWriter>());
}
/// <inheritdoc />
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerTagProvisioner(s.Conn, s.Bootstrap, Loggers(sp).CreateLogger<SqlServerTagProvisioner>());
}
/// <inheritdoc />
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp)
{
var s = Get(sp);
return new SqlServerHistorianDataUpdater(s.Conn, s.Bootstrap, 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);
}