diff --git a/docker-dev/docker-compose.yml b/docker-dev/docker-compose.yml index 3d410597..d6e439b8 100644 --- a/docker-dev/docker-compose.yml +++ b/docker-dev/docker-compose.yml @@ -441,6 +441,9 @@ services: # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}" ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}" + # QuestDb = the QuestDB historian; the `questdb` service comes from the docker-dev/docker-compose.questdb.yml overlay + # (add `-f docker-dev/docker-compose.questdb.yml` and OTOPCUA_HISTORIAN_BACKEND=QuestDb). + ServerHistorian__QuestDb__ConnectionString: "${OTOPCUA_HISTORIAN_QUESTDB:-Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading}" # Continuous historization of driver tag values (Gitea #491's live gate). Gated on BOTH # ContinuousHistorization:Enabled AND ServerHistorian:Enabled — with either off no recorder is # spawned, so this stays inert on the default rig. OutboxPath is the FasterLog crash boundary and @@ -574,6 +577,9 @@ services: # docker compose -f docker-dev/docker-compose.yml up -d central-1 central-2 ServerHistorian__Backend: "${OTOPCUA_HISTORIAN_BACKEND:-Gateway}" ServerHistorian__SqlServer__ConnectionString: "${OTOPCUA_HISTORIAN_SQL:-Server=sql,1433;Database=OtOpcUaHistorian;User Id=sa;Password=OtOpcUa!Dev123;TrustServerCertificate=True;Encrypt=False}" + # QuestDb = the QuestDB historian; the `questdb` service comes from the docker-dev/docker-compose.questdb.yml overlay + # (add `-f docker-dev/docker-compose.questdb.yml` and OTOPCUA_HISTORIAN_BACKEND=QuestDb). + ServerHistorian__QuestDb__ConnectionString: "${OTOPCUA_HISTORIAN_QUESTDB:-Host=questdb;Port=8812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading}" # Continuous historization — see central-1 for the full note + the gate recipe. Both centrals # carry it so the pair is symmetric; :9200 round-robins them, and the recorder must exist on # whichever node holds the drivers. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs new file mode 100644 index 00000000..39af8413 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianProvider.cs @@ -0,0 +1,106 @@ +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.QuestDb; + +/// +/// for the QuestDB historian. Supports every +/// and HistoryUpdate (soft delete). All seams share ONE connection helper, ONE +/// schema bootstrapper, ONE tag catalog (R7: the only "known tag" authority) and ONE WAL visibility wait, so the +/// schema is ensured once and known tags are cached once. +/// +/// +/// The shared state is built on the first Create* 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 ServerHistorianOptionsValidator). The first +/// Create* call's service provider supplies the logger factory the shared pieces log through. +/// +public sealed class QuestDbHistorianProvider : IHistorianBackendProvider +{ + private static readonly FrozenSet Kinds = + Enum.GetValues().ToFrozenSet(); + + private readonly QuestDbHistorianOptions _options; + private readonly object _initLock = new(); + private Shared? _shared; + + /// Creates the provider over the ServerHistorian:QuestDb options (validated elsewhere). + /// The ServerHistorian:QuestDb options. + public QuestDbHistorianProvider(QuestDbHistorianOptions options) + { + _options = options ?? throw new ArgumentNullException(nameof(options)); + } + + /// + public string Name => "QuestDb"; + + /// + public IReadOnlySet 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 loggers = Loggers(sp); + var conn = new QuestDbConnection(_options); + var bootstrap = new QuestDbSchemaBootstrapper(conn, _options, loggers.CreateLogger()); + var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger()); + var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal); + Volatile.Write(ref _shared, shared); + return shared; + } + } + + /// + public IHistorianDataSource CreateDataSource(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog, + Loggers(sp).CreateLogger()); + } + + /// + public IHistorianValueWriter CreateValueWriter(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianValueWriter(s.Conn, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbAlarmHistorianWriter(s.Conn, s.Bootstrap, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + public IHistorianProvisioning CreateProvisioning(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbTagProvisioner(s.Conn, s.Bootstrap, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + /// + /// The soft-delete updater: deletes write tombstones (INSERT only, never QuestDB UPDATE/DELETE). + public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp) + { + var s = Get(sp); + return new QuestDbHistorianDataUpdater(s.Conn, s.Catalog, s.Wal, + Loggers(sp).CreateLogger()); + } + + private sealed record Shared( + QuestDbConnection Conn, QuestDbSchemaBootstrapper Bootstrap, QuestDbTagCatalog Catalog, QuestDbWalVisibility Wal); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs index ce68db91..a14e5bd6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/HistorianBackendRegistry.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Options; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; @@ -25,11 +26,23 @@ public static class HistorianBackendRegistry return o; } + /// Backend name for the QuestDB historian. + public const string QuestDbBackendName = "QuestDb"; + + /// Binds the ServerHistorian:QuestDb section (shared by the registry and the validator). + public static QuestDbHistorianOptions BindQuestDb(IConfiguration configuration) + { + var o = new QuestDbHistorianOptions(); + configuration.GetSection($"{ServerHistorianOptions.SectionName}:QuestDb").Bind(o); + return o; + } + private static readonly IReadOnlyDictionary> Factories = new Dictionary>(StringComparer.OrdinalIgnoreCase) { [ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o), [SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)), + [QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)), }; /// The registered backend names, for validation messages. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs index e421747f..ac51be90 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Configuration/ServerHistorianOptionsValidator.cs @@ -78,6 +78,13 @@ public sealed class ServerHistorianOptionsValidator : OptionsValidatorBase + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs new file mode 100644 index 00000000..b5df992b --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests/QuestDbHistorianProviderTests.cs @@ -0,0 +1,90 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.Tests; + +public sealed class QuestDbHistorianProviderTests +{ + private static QuestDbHistorianProvider Make() => new(new QuestDbHistorianOptions + { + ConnectionString = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb;ServerCompatibilityMode=NoTypeLoading", + }); + + [Fact] + public void Name_and_all_six_kinds() + { + var p = Make(); + p.Name.ShouldBe("QuestDb"); + p.SupportedKinds.ShouldBe(Enum.GetValues(), ignoreOrder: true); + p.SupportedKinds.Count.ShouldBe(6); + } + + [Fact] + public void Creates_all_five_seams_without_touching_questdb() + { + var sp = new NullSp(); + var p = Make(); + p.CreateDataSource(sp).ShouldBeOfType(); + p.CreateValueWriter(sp).ShouldBeOfType(); + p.CreateAlarmWriter(sp).ShouldBeOfType(); + p.CreateProvisioning(sp).ShouldBeOfType(); + p.CreateUpdater(sp).ShouldBeOfType(); + } + + [Fact] + public void Supported_kinds_are_a_frozen_set() => + Make().SupportedKinds.ShouldBeAssignableTo>(); + + /// + /// An invalid table prefix must not fail provider construction (the host resolves the provider even when no + /// consumer is enabled); it surfaces only when a seam is actually built. + /// + [Fact] + public void An_invalid_prefix_fails_only_when_a_seam_is_built() + { + var p = new QuestDbHistorianProvider(new QuestDbHistorianOptions { ConnectionString = "Host=x", TablePrefix = "bad;prefix" }); + p.Name.ShouldBe("QuestDb"); + Should.Throw(() => p.CreateDataSource(new NullSp())); + } + + [Fact] + public void Every_seam_shares_one_connection_bootstrapper_catalog_and_wal_wait() + { + var sp = new NullSp(); + var p = Make(); + var a = (QuestDbHistorianDataSource)p.CreateDataSource(sp); + var b = (QuestDbHistorianDataSource)p.CreateDataSource(sp); + var w = p.CreateValueWriter(sp); + var u = p.CreateUpdater(sp)!; + var prov = p.CreateProvisioning(sp); + var alarm = p.CreateAlarmWriter(sp); + + Field(a, "_conn").ShouldBeSameAs(Field(b, "_conn")); + Field(a, "_conn").ShouldBeSameAs(Field(w, "_conn")); + Field(a, "_conn").ShouldBeSameAs(Field(u, "_conn")); + Field(a, "_conn").ShouldBeSameAs(Field(alarm, "_conn")); + Field(a, "_bootstrap").ShouldBeSameAs(Field(b, "_bootstrap")); + Field(a, "_bootstrap").ShouldBeSameAs(Field(prov, "_bootstrap")); + Field(a, "_bootstrap").ShouldBeSameAs(Field(alarm, "_bootstrap")); + Field(a, "_catalog").ShouldBeSameAs(Field(b, "_catalog")); + Field(a, "_catalog").ShouldBeSameAs(Field(w, "_catalog")); + Field(a, "_catalog").ShouldBeSameAs(Field(u, "_tags")); + Field(a, "_catalog").ShouldBeSameAs(Field(prov, "_catalog")); + Field(w, "_wal").ShouldBeSameAs(Field(u, "_wal")); + Field(w, "_wal").ShouldBeSameAs(Field(prov, "_wal")); + Field(w, "_wal").ShouldBeSameAs(Field(alarm, "_wal")); + } + + [Fact] + public void The_constructor_rejects_null_options() => + Should.Throw(() => new QuestDbHistorianProvider(null!)); + + private static object? Field(object o, string name) => + o.GetType().GetField(name, System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(o); + + private sealed class NullSp : IServiceProvider + { + public object? GetService(Type serviceType) => null; + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs index 00945b11..9b378e4a 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/ServerHistorianOptionsValidatorTests.cs @@ -375,4 +375,89 @@ public sealed class ServerHistorianOptionsValidatorTests result.Failed.ShouldBeTrue(); result.Failures.ShouldContain(f => f.Contains("ServerHistorian:SqlServer:Schema")); } + + private static ServerHistorianOptionsValidator QuestDbSut(string? connectionString, string? prefix = null, bool alarmHistorianEnabled = false) + { + var values = new Dictionary { ["AlarmHistorian:Enabled"] = alarmHistorianEnabled ? "true" : "false" }; + if (connectionString is not null) values["ServerHistorian:QuestDb:ConnectionString"] = connectionString; + if (prefix is not null) values["ServerHistorian:QuestDb:TablePrefix"] = prefix; + return new ServerHistorianOptionsValidator(new ConfigurationBuilder().AddInMemoryCollection(values).Build()); + } + + /// QuestDb with a connection string passes; the gateway Endpoint/ApiKey/UseTls checks do not fire. + [Fact] + public void QuestDb_backend_with_connection_string_passes_without_gateway_settings() + { + var result = QuestDbSut("Host=questdb;Port=8812").Validate(null, + new ServerHistorianOptions { Backend = "QuestDb", Enabled = true }); + + result.Failed.ShouldBeFalse(string.Join("; ", result.Failures ?? [])); + } + + /// QuestDb with no connection string fails, naming the key, and never asks for gateway settings. + [Fact] + public void QuestDb_backend_without_connection_string_fails_naming_the_key() + { + var result = QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "questdb", Enabled = true }); + + result.Failed.ShouldBeTrue(); + result.Failures.ShouldContain(f => f.Contains("ServerHistorian:QuestDb:ConnectionString")); + result.Failures.ShouldNotContain(f => f.Contains("ServerHistorian:Endpoint")); + result.Failures.ShouldNotContain(f => f.Contains("ServerHistorian:ApiKey")); + } + + /// Every QuestDbHistorianOptions.Validate() error is reported, each naming its key. + [Fact] + public void QuestDb_backend_reports_every_options_error_naming_the_key() + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["AlarmHistorian:Enabled"] = "false", + ["ServerHistorian:QuestDb:TablePrefix"] = "bad;prefix", + ["ServerHistorian:QuestDb:RetentionDays"] = "0", + ["ServerHistorian:QuestDb:CommandTimeout"] = "00:00:00", + ["ServerHistorian:QuestDb:VisibilityTimeout"] = "-00:00:01", + }).Build(); + var result = new ServerHistorianOptionsValidator(configuration) + .Validate(null, new ServerHistorianOptions { Backend = "QuestDb", Enabled = true }); + + result.Failed.ShouldBeTrue(); + foreach (var key in new[] { "ConnectionString", "TablePrefix", "RetentionDays", "CommandTimeout", "VisibilityTimeout" }) + result.Failures.ShouldContain(f => f.Contains($"ServerHistorian:QuestDb:{key}"), key); + } + + /// Unconsumed QuestDb section (nothing enabled) is not validated. + [Fact] + public void QuestDb_backend_not_consumed_passes_without_connection_string() => + QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "QuestDb" }).Failed.ShouldBeFalse(); + + /// + /// An invalid ServerHistorian:QuestDb:TablePrefix must not fail the host when nothing consumes the + /// section: does not throw and the validator does not check it. + /// + [Fact] + public void QuestDb_invalid_prefix_not_consumed_neither_resolve_nor_validation_fails() + { + var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["AlarmHistorian:Enabled"] = "false", + ["ServerHistorian:QuestDb:ConnectionString"] = "Host=questdb;Port=8812", + ["ServerHistorian:QuestDb:TablePrefix"] = "bad;prefix", + }).Build(); + var options = new ServerHistorianOptions { Backend = "QuestDb" }; + + Should.NotThrow(() => HistorianBackendRegistry.Resolve(options, configuration)); + new ServerHistorianOptionsValidator(configuration).Validate(null, options).Failed.ShouldBeFalse(); + } + + /// Consumed (here by AlarmHistorian alone), the same invalid prefix is a named validator error. + [Fact] + public void QuestDb_invalid_prefix_consumed_is_a_validator_error_naming_the_key() + { + var options = new ServerHistorianOptions { Backend = "QuestDb" }; + var result = QuestDbSut("Host=questdb;Port=8812", "bad;prefix", alarmHistorianEnabled: true).Validate(null, options); + + result.Failed.ShouldBeTrue(); + result.Failures.ShouldContain(f => f.Contains("ServerHistorian:QuestDb:TablePrefix")); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs index 9251df6a..fe66467c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Configuration/HistorianBackendRegistryTests.cs @@ -6,6 +6,7 @@ using Xunit; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway; using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb; using ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer; using ZB.MOM.WW.OtOpcUa.Host.Configuration; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; @@ -86,4 +87,46 @@ public sealed class HistorianBackendRegistryTests holder.Updater.ShouldNotBeNull(); holder.Updater!.GetType().Name.ShouldBe("SqlServerHistorianDataUpdater"); } + + private static IConfiguration QuestDbCfg() => new ConfigurationBuilder().AddInMemoryCollection(new Dictionary + { + ["ServerHistorian:QuestDb:ConnectionString"] = "Host=localhost;Port=18812;Username=admin;Password=quest;Database=qdb", + ["ServerHistorian:QuestDb:TablePrefix"] = "rig_", + }).Build(); + + [Theory] + [InlineData("questdb")] + [InlineData("QUESTDB")] + [InlineData("QuestDb")] + public void QuestDb_backend_resolves_case_insensitively(string name) => + HistorianBackendRegistry.Resolve(new ServerHistorianOptions { Backend = name, Enabled = true }, QuestDbCfg()) + .ShouldBeOfType().Name.ShouldBe("QuestDb"); + + [Fact] + public void QuestDb_backend_resolves_and_builds_the_questdb_seams() + { + var sp = new ServiceCollection().BuildServiceProvider(); + var p = HistorianBackendRegistry.Resolve(new ServerHistorianOptions { Backend = "questdb", Enabled = true }, QuestDbCfg()); + + p.ShouldBeOfType(); + HistorianBackendRegistry.KnownBackends.ShouldContain("QuestDb"); + p.SupportedKinds.Count.ShouldBe(6); + p.SupportedKinds.ShouldContain(HistorianValueKind.String); + p.SupportedKinds.ShouldContain(HistorianValueKind.Empty); + p.CreateDataSource(sp).ShouldBeOfType(); + p.CreateValueWriter(sp).GetType().Name.ShouldBe("QuestDbHistorianValueWriter"); + p.CreateAlarmWriter(sp).GetType().Name.ShouldBe("QuestDbAlarmHistorianWriter"); + p.CreateProvisioning(sp).GetType().Name.ShouldBe("QuestDbTagProvisioner"); + var holder = new HistorianUpdaterHolder(p.CreateUpdater(sp)); + holder.Updater.ShouldNotBeNull(); + holder.Updater!.GetType().Name.ShouldBe("QuestDbHistorianDataUpdater"); + } + + [Fact] + public void BindQuestDb_reads_the_QuestDb_section() + { + var o = HistorianBackendRegistry.BindQuestDb(QuestDbCfg()); + o.ConnectionString.ShouldContain("Port=18812"); + o.TablePrefix.ShouldBe("rig_"); + } }