feat(historian): QuestDb provider, registry + consumer-gated validator, compose env

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-11 02:23:21 -04:00
parent 18e85265f9
commit 13d628a07f
8 changed files with 351 additions and 0 deletions
+6
View File
@@ -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.
@@ -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;
/// <summary>
/// <see cref="IHistorianBackendProvider"/> for the QuestDB historian. Supports every
/// <see cref="HistorianValueKind"/> 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.
/// </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 shared pieces log through.
/// </remarks>
public sealed class QuestDbHistorianProvider : IHistorianBackendProvider
{
private static readonly FrozenSet<HistorianValueKind> Kinds =
Enum.GetValues<HistorianValueKind>().ToFrozenSet();
private readonly QuestDbHistorianOptions _options;
private readonly object _initLock = new();
private Shared? _shared;
/// <summary>Creates the provider over the <c>ServerHistorian:QuestDb</c> options (validated elsewhere).</summary>
/// <param name="options">The <c>ServerHistorian:QuestDb</c> options.</param>
public QuestDbHistorianProvider(QuestDbHistorianOptions options)
{
_options = options ?? throw new ArgumentNullException(nameof(options));
}
/// <inheritdoc />
public string Name => "QuestDb";
/// <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 loggers = Loggers(sp);
var conn = new QuestDbConnection(_options);
var bootstrap = new QuestDbSchemaBootstrapper(conn, _options, loggers.CreateLogger<QuestDbSchemaBootstrapper>());
var wal = new QuestDbWalVisibility(conn, _options.VisibilityTimeout, loggers.CreateLogger<QuestDbWalVisibility>());
var shared = new Shared(conn, bootstrap, new QuestDbTagCatalog(conn, bootstrap, wal), wal);
Volatile.Write(ref _shared, shared);
return shared;
}
}
/// <inheritdoc />
public IHistorianDataSource CreateDataSource(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataSource(s.Conn, s.Bootstrap, s.Catalog,
Loggers(sp).CreateLogger<QuestDbHistorianDataSource>());
}
/// <inheritdoc />
public IHistorianValueWriter CreateValueWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianValueWriter(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianValueWriter>());
}
/// <inheritdoc />
public IAlarmHistorianWriter CreateAlarmWriter(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbAlarmHistorianWriter(s.Conn, s.Bootstrap, s.Wal,
Loggers(sp).CreateLogger<QuestDbAlarmHistorianWriter>());
}
/// <inheritdoc />
public IHistorianProvisioning CreateProvisioning(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbTagProvisioner(s.Conn, s.Bootstrap, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbTagProvisioner>());
}
/// <inheritdoc />
/// <remarks>The soft-delete updater: deletes write tombstones (INSERT only, never QuestDB UPDATE/DELETE).</remarks>
public IHistorianDataUpdater? CreateUpdater(IServiceProvider sp)
{
var s = Get(sp);
return new QuestDbHistorianDataUpdater(s.Conn, s.Catalog, s.Wal,
Loggers(sp).CreateLogger<QuestDbHistorianDataUpdater>());
}
private sealed record Shared(
QuestDbConnection Conn, QuestDbSchemaBootstrapper Bootstrap, QuestDbTagCatalog Catalog, QuestDbWalVisibility Wal);
}
@@ -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;
}
/// <summary>Backend name for the QuestDB historian.</summary>
public const string QuestDbBackendName = "QuestDb";
/// <summary>Binds the <c>ServerHistorian:QuestDb</c> section (shared by the registry and the validator).</summary>
public static QuestDbHistorianOptions BindQuestDb(IConfiguration configuration)
{
var o = new QuestDbHistorianOptions();
configuration.GetSection($"{ServerHistorianOptions.SectionName}:QuestDb").Bind(o);
return o;
}
private static readonly IReadOnlyDictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>> Factories =
new Dictionary<string, Func<ServerHistorianOptions, IConfiguration, IHistorianBackendProvider>>(StringComparer.OrdinalIgnoreCase)
{
[ServerHistorianOptions.GatewayBackendName] = (o, _) => new GatewayHistorianProvider(o),
[SqlServerBackendName] = (_, c) => new SqlServerHistorianProvider(BindSqlServer(c)),
[QuestDbBackendName] = (_, c) => new QuestDbHistorianProvider(BindQuestDb(c)),
};
/// <summary>The registered backend names, for validation messages.</summary>
@@ -78,6 +78,13 @@ public sealed class ServerHistorianOptionsValidator : OptionsValidatorBase<Serve
return;
}
if (string.Equals(options.Backend?.Trim(), HistorianBackendRegistry.QuestDbBackendName, StringComparison.OrdinalIgnoreCase))
{
foreach (var error in HistorianBackendRegistry.BindQuestDb(_configuration).Validate())
builder.RequireThat(false, error);
return;
}
if (!options.IsGatewayBackend) return;
var endpointValid = Uri.TryCreate(options.Endpoint, UriKind.Absolute, out var uri)
@@ -81,6 +81,7 @@
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Galaxy\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer\ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb\ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Modbus\ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Mqtt\ZB.MOM.WW.OtOpcUa.Driver.Mqtt.csproj"/>
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.MTConnect\ZB.MOM.WW.OtOpcUa.Driver.MTConnect.csproj"/>
@@ -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<HistorianValueKind>(), 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<QuestDbHistorianDataSource>();
p.CreateValueWriter(sp).ShouldBeOfType<QuestDbHistorianValueWriter>();
p.CreateAlarmWriter(sp).ShouldBeOfType<QuestDbAlarmHistorianWriter>();
p.CreateProvisioning(sp).ShouldBeOfType<QuestDbTagProvisioner>();
p.CreateUpdater(sp).ShouldBeOfType<QuestDbHistorianDataUpdater>();
}
[Fact]
public void Supported_kinds_are_a_frozen_set() =>
Make().SupportedKinds.ShouldBeAssignableTo<System.Collections.Frozen.FrozenSet<HistorianValueKind>>();
/// <summary>
/// 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.
/// </summary>
[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<ArgumentException>(() => 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<ArgumentNullException>(() => 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;
}
}
@@ -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<string, string?> { ["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());
}
/// <summary>QuestDb with a connection string passes; the gateway Endpoint/ApiKey/UseTls checks do not fire.</summary>
[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 ?? []));
}
/// <summary>QuestDb with no connection string fails, naming the key, and never asks for gateway settings.</summary>
[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"));
}
/// <summary>Every <c>QuestDbHistorianOptions.Validate()</c> error is reported, each naming its key.</summary>
[Fact]
public void QuestDb_backend_reports_every_options_error_naming_the_key()
{
var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["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);
}
/// <summary>Unconsumed QuestDb section (nothing enabled) is not validated.</summary>
[Fact]
public void QuestDb_backend_not_consumed_passes_without_connection_string() =>
QuestDbSut(null).Validate(null, new ServerHistorianOptions { Backend = "QuestDb" }).Failed.ShouldBeFalse();
/// <summary>
/// An invalid <c>ServerHistorian:QuestDb:TablePrefix</c> must not fail the host when nothing consumes the
/// section: <see cref="HistorianBackendRegistry.Resolve"/> does not throw and the validator does not check it.
/// </summary>
[Fact]
public void QuestDb_invalid_prefix_not_consumed_neither_resolve_nor_validation_fails()
{
var configuration = new ConfigurationBuilder().AddInMemoryCollection(new Dictionary<string, string?>
{
["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();
}
/// <summary>Consumed (here by AlarmHistorian alone), the same invalid prefix is a named validator error.</summary>
[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"));
}
}
@@ -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<string, string?>
{
["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<QuestDbHistorianProvider>().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<QuestDbHistorianProvider>();
HistorianBackendRegistry.KnownBackends.ShouldContain("QuestDb");
p.SupportedKinds.Count.ShouldBe(6);
p.SupportedKinds.ShouldContain(HistorianValueKind.String);
p.SupportedKinds.ShouldContain(HistorianValueKind.Empty);
p.CreateDataSource(sp).ShouldBeOfType<QuestDbHistorianDataSource>();
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_");
}
}