Joseph Doherty
7f2108dcf2
`ActAsPrimaryHost` was accepted by config and by the AdminUI form and did nothing but
log a warning. It now delivers the whole three-part contract or none of it: a retained
`ONLINE` on `spBv1.0/STATE/{hostId}`, a matching retained `OFFLINE` armed as the MQTT
Last Will, and an explicit `OFFLINE` on a clean stand-down.
**The claim is a dedicated connection, not a message.** A Will can only be registered in
the CONNECT packet — it cannot be added to a live session, changed, or withdrawn. So a
claim that is acquired and released at runtime cannot ride the long-lived ingest session.
`SparkplugPrimaryHost` opens its own connection (client id `{clientId}-spbhost`, forced
clean-session) when the node gains Primary and closes it when the node loses Primary; the
connection's lifetime *is* the claim, so every leg follows from its lifecycle. Ingest is
untouched by a role change — no dropped subscriptions, no lost birth cache, no data gap at
the Secondary->Primary transition.
**Primary-gated** (owner decision, recorded on the issue). Both pair nodes subscribe
independently, so an ungated claim means both claim the same host id — and the sharp
failure is not the two fighting over the retained STATE, it is that the SECONDARY's death
fires its Will and publishes OFFLINE while the Primary is healthy, stopping every edge node
in the group. Registering the Will only on whichever node is Primary at connect time fails
the other way: a node that later gains Primary would publish ONLINE with no Will behind it.
New seam: `IDriverPrimaryRoleSource` (`Core.Abstractions`) carrying the *data-plane* verdict
(`PrimaryGatePolicy.ShouldServiceAsPrimary`), not the alarm-drain verdict — an unknown role
with a driver peer present must deny. Registered unconditionally by `AddOtOpcUaRuntime`,
deliberately NOT alongside `IRedundancyRoleView`, which `AddAlarmHistorian` registers only
when that section is enabled: piggybacking would leave a pair with no alarm historian
falling back to "claim unconditionally". Polled on a 1s timer rather than evented — a
process singleton raising events at per-instance drivers outlives every driver it holds.
Accepted trade: a brief edge-visible OFFLINE window during failover. That is a genuine
state transition, not a glitch to hide.
Live-gated 5/5 against the Mosquitto fixture, including the Will leg — `AbandonForTests`
drops the socket with no DISCONNECT, because a clean disconnect makes the broker discard
the Will by protocol rule, so a test driven through `StopAsync` would observe the
stand-down OFFLINE and pass against an implementation that never armed a Will. Falsified:
removing the Will turns that leg and only that leg red.
Also corrects two stale claims found en route: the AdminUI form still said MQTT tags are
read-only (#508 shipped), and the driver doc still said nothing consumes `IRediscoverable`.
Claude-Session: https://claude.ai/code/session_015p7wGqy3YpZNCpDzTpGMKo
552 lines
36 KiB
C#
552 lines
36 KiB
C#
using System.Collections.Immutable;
|
|
using Akka.Actor;
|
|
using Akka.Cluster.Tools.Client;
|
|
using Akka.Hosting;
|
|
using Microsoft.EntityFrameworkCore;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.DependencyInjection.Extensions;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.OtOpcUa.Cluster;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Communication;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.Interfaces;
|
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Configuration;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
|
using ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian;
|
|
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
|
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
|
using ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Health;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
|
|
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
|
using ZB.MOM.WW.LocalDb;
|
|
|
|
namespace ZB.MOM.WW.OtOpcUa.Runtime;
|
|
|
|
public static class ServiceCollectionExtensions
|
|
{
|
|
public const string DriverRole = RoleParser.Driver;
|
|
|
|
public const string DriverHostActorName = "driver-host";
|
|
public const string DbHealthProbeActorName = "db-health";
|
|
public const string HistorianAdapterActorName = "historian-adapter";
|
|
public const string DependencyMuxActorName = "dependency-mux";
|
|
public const string OpcUaPublishActorName = "opcua-publish";
|
|
public const string PeerProbeSupervisorName = "peer-probe-supervisor";
|
|
public const string ContinuousHistorizationRecorderActorName = "continuous-historization-recorder";
|
|
public const string CentralClusterClientName = "central-cluster-client";
|
|
|
|
/// <summary>
|
|
/// Registers shared runtime services. Currently binds <see cref="IAlarmHistorianSink"/>
|
|
/// to <see cref="NullAlarmHistorianSink"/> as the default; production deployments
|
|
/// override this with <c>LocalDbStoreAndForwardSink</c> wrapping the HistorianGateway alarm writer.
|
|
/// Call this BEFORE <c>AddAkka</c>.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddOtOpcUaRuntime(this IServiceCollection services)
|
|
{
|
|
services.TryAddSingleton<IAlarmHistorianSink>(NullAlarmHistorianSink.Instance);
|
|
services.TryAddSingleton<IHistorianDataSource>(NullHistorianDataSource.Instance);
|
|
// Historian tag provisioning. Null default (no-op) so the AddressSpaceApplier resolves a real
|
|
// IHistorianProvisioning only when AddHistorianProvisioning registers the gateway-backed one
|
|
// (ServerHistorian:Enabled). TryAddSingleton so the gateway registration wins last.
|
|
services.TryAddSingleton<IHistorianProvisioning>(NullHistorianProvisioning.Instance);
|
|
// VirtualTag historization sink. Null default — the durable AVEVA sink is infra-gated (there is
|
|
// no live-data historian write RPC). TryAddSingleton so a deployment that bound a real
|
|
// IHistoryWriter earlier wins.
|
|
services.TryAddSingleton<IHistoryWriter>(NullHistoryWriter.Instance);
|
|
services.TryAddSingleton<IDriverFactory>(NullDriverFactory.Instance);
|
|
services.TryAddSingleton<IOpcUaAddressSpaceSink>(NullOpcUaAddressSpaceSink.Instance);
|
|
services.TryAddSingleton<IServiceLevelPublisher>(NullServiceLevelPublisher.Instance);
|
|
services.TryAddSingleton<IDriverHealthPublisher, AkkaDriverHealthPublisher>();
|
|
// Per-cluster mesh Phase 5: the node-local live-telemetry fan-out hub. Feeds this node's OWN
|
|
// telemetry (never DPS) to the Phase-5 gRPC streaming service central dials in on. AddOtOpcUaRuntime
|
|
// runs only inside Program.cs's hasDriver block, so this lands on driver-role nodes only.
|
|
services.TryAddSingleton<ITelemetryLocalHub, TelemetryLocalHub>();
|
|
|
|
// Gitea #511: the Primary data-plane verdict, published for drivers that assert a standing
|
|
// exclusive identity on a shared external system (today the MQTT driver's Sparkplug host id).
|
|
//
|
|
// Registered HERE, unconditionally on every driver-role node, and NOT alongside the alarm
|
|
// drain's IRedundancyRoleView — which AddAlarmHistorian registers only when that section is
|
|
// enabled. Piggybacking on it would leave a redundant pair with no alarm historian configured
|
|
// with no primary-role source at all, and the driver's documented no-source fallback is
|
|
// "claim" — so both nodes of that pair would claim the same host id and their Wills would
|
|
// fight over the retained STATE. Exactly the failure #511 exists to prevent, reached by a
|
|
// registration accident in an unrelated section.
|
|
//
|
|
// Both interfaces are forwarded to ONE RedundancyRoleView instance, so a node that has both
|
|
// publishes two verdicts into the same object rather than into two that disagree.
|
|
services.TryAddSingleton<RedundancyRoleView>();
|
|
services.TryAddSingleton<IDriverPrimaryRoleSource>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated durable alarm-historian sink. When the <c>AlarmHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers a <see cref="LocalDbStoreAndForwardSink"/> (draining via the
|
|
/// <paramref name="writerFactory"/>-supplied writer) as the <see cref="IAlarmHistorianSink"/>,
|
|
/// overriding the <see cref="NullAlarmHistorianSink"/> default. Otherwise a no-op (Null stays).
|
|
/// The writer is injected so the durable downstream (the HistorianGateway alarm writer) can be
|
|
/// supplied by the Host, which is the only project that references it.
|
|
/// </summary>
|
|
/// <remarks>
|
|
/// <para>
|
|
/// The queue lives in the node's consolidated <see cref="ILocalDb"/>, which replicates it
|
|
/// to the redundant pair peer — so undelivered alarm history survives losing the node
|
|
/// holding it. That is also why the drain is gated on <see cref="IRedundancyRoleView"/>:
|
|
/// the Secondary holds a full replica of the Primary's queue, and an ungated drain there
|
|
/// would re-deliver every event. Both services are resolved from the provider, so a
|
|
/// deployment that enables this section without registering LocalDb fails loudly at
|
|
/// resolution rather than silently buffering to nowhere.
|
|
/// </para>
|
|
/// </remarks>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>AlarmHistorian</c> section.</param>
|
|
/// <param name="writerFactory">
|
|
/// Factory the Host supplies to build the concrete <see cref="IAlarmHistorianWriter"/>
|
|
/// (the HistorianGateway alarm writer) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddAlarmHistorian(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<AlarmHistorianOptions, IServiceProvider, IAlarmHistorianWriter> writerFactory)
|
|
{
|
|
var opts = configuration.GetSection(AlarmHistorianOptions.SectionName).Get<AlarmHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
foreach (var warning in opts.Validate())
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>().Warning("Historian config: {HistorianConfigWarning}", warning);
|
|
|
|
// The view is a plain singleton so it exists even on nodes whose DriverHostActor never
|
|
// publishes to it — an unpublished view reads as "no peer, drain", which is the correct
|
|
// posture for a deployment that runs no redundancy at all. Forwarded to the same instance
|
|
// AddOtOpcUaRuntime registered for IDriverPrimaryRoleSource: one object, two verdicts.
|
|
services.TryAddSingleton<RedundancyRoleView>();
|
|
services.TryAddSingleton<IRedundancyRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
|
|
|
// THE GATE ONLY APPLIES TO A REPLICATED QUEUE. Standing down is only ever safe because some
|
|
// other node holds the same rows and will send them instead — and the only node that does is
|
|
// this node's LocalDb replication peer. Without replication configured, these rows exist here
|
|
// and nowhere else, so deferring to anyone means they are never delivered by anyone.
|
|
//
|
|
// This is not hypothetical. The redundancy role is a CLUSTER-WIDE election
|
|
// (RedundancyStateActor keys on Akka's RoleLeader("driver")), while the queue is PAIR-LOCAL.
|
|
// On the docker-dev rig the elected driver Primary is a central node — which carries the
|
|
// driver Akka role, replicates nobody's LocalDb, and does not even run the alarm historian —
|
|
// so every site node dutifully suspended its drain in favour of a node that could not
|
|
// possibly deliver its events. Scoping the gate to "is my queue actually shared?" is what
|
|
// keeps the two scopes from disagreeing.
|
|
// BOTH halves of a pair share the queue, but only one of them dials: the initiator sets
|
|
// Replication:PeerAddress, its partner only sets SyncListenPort and waits. Testing the dial
|
|
// side alone would leave the listening half permanently ungated — one drainer per pair by
|
|
// accident rather than by role, and the wrong one whenever the roles swap.
|
|
var replicated =
|
|
!string.IsNullOrWhiteSpace(configuration["LocalDb:Replication:PeerAddress"])
|
|
|| !string.IsNullOrWhiteSpace(configuration["LocalDb:SyncListenPort"]);
|
|
|
|
if (!replicated)
|
|
{
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>().Information(
|
|
"Alarm historian: LocalDb replication is not configured, so this node's queue is not "
|
|
+ "shared with any peer and the Primary drain gate does not apply — this node always "
|
|
+ "drains its own alarm queue.");
|
|
}
|
|
|
|
services.AddSingleton<IAlarmHistorianSink>(sp =>
|
|
{
|
|
// LocalDbStoreAndForwardSink takes a Serilog ILogger (not Microsoft.Extensions.Logging).
|
|
// Resolve it off the host's configured static logger so the drain worker's WARN/INFO
|
|
// lines land in the same sinks as the rest of the process.
|
|
var roleView = sp.GetRequiredService<IRedundancyRoleView>();
|
|
var sink = new LocalDbStoreAndForwardSink(
|
|
sp.GetRequiredService<ILocalDb>(),
|
|
writerFactory(opts, sp),
|
|
Serilog.Log.Logger.ForContext<LocalDbStoreAndForwardSink>(),
|
|
batchSize: opts.BatchSize,
|
|
capacity: opts.Capacity,
|
|
deadLetterRetention: TimeSpan.FromDays(opts.DeadLetterRetentionDays),
|
|
maxAttempts: opts.MaxAttempts,
|
|
drainGate: () => !replicated || roleView.ShouldDrainAlarmHistory);
|
|
sink.StartDrainLoop(TimeSpan.FromSeconds(opts.DrainIntervalSeconds));
|
|
return sink;
|
|
});
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated server-side HistoryRead backend. When the <c>ServerHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers the <paramref name="dataSourceFactory"/>-supplied
|
|
/// <see cref="IHistorianDataSource"/> (the read-only HistorianGateway-backed data source) overriding
|
|
/// the <see cref="NullHistorianDataSource"/> default from <see cref="AddOtOpcUaRuntime"/>. Otherwise
|
|
/// a no-op (the Null default stays and the node manager's HistoryRead returns
|
|
/// <c>GoodNoData</c>-empty). The data source is injected so the gateway-backed client can be supplied
|
|
/// by the Host, which is the only project that references the driver.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>ServerHistorian</c> section.</param>
|
|
/// <param name="dataSourceFactory">
|
|
/// Factory the Host supplies to build the concrete read <see cref="IHistorianDataSource"/>
|
|
/// (the gateway-backed data source) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddServerHistorian(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<ServerHistorianOptions, IServiceProvider, IHistorianDataSource> dataSourceFactory)
|
|
{
|
|
var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get<ServerHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
foreach (var warning in opts.Validate())
|
|
Serilog.Log.Logger.ForContext<IHistorianDataSource>().Warning("ServerHistorian config: {ServerHistorianConfigWarning}", warning);
|
|
|
|
// Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime.
|
|
services.AddSingleton<IHistorianDataSource>(sp => dataSourceFactory(opts, sp));
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Config-gated historian tag provisioning. When the <c>ServerHistorian</c> section has
|
|
/// <c>Enabled=true</c>, registers the <paramref name="provisioningFactory"/>-supplied
|
|
/// <see cref="IHistorianProvisioning"/> (the gateway-backed <c>GatewayTagProvisioner</c> that calls
|
|
/// the gateway's <c>EnsureTags</c>) overriding the <see cref="NullHistorianProvisioning"/> default from
|
|
/// <see cref="AddOtOpcUaRuntime"/>. Otherwise a no-op (the Null default stays and deploying historized
|
|
/// tags provisions nothing). The provisioner is consumed by the <c>AddressSpaceApplier</c>, which fires
|
|
/// a non-blocking <see cref="IHistorianProvisioning.EnsureTagsAsync"/> for added historized tags on
|
|
/// every deploy. Gated on the <b>same</b> <c>ServerHistorian:Enabled</c> flag as the read path
|
|
/// (<see cref="AddServerHistorian"/>) — provisioning targets the same single gateway. The provisioner
|
|
/// is injected so the gateway-backed client can be supplied by the Host, which is the only project that
|
|
/// references the driver.
|
|
/// </summary>
|
|
/// <param name="services">The service collection to register with.</param>
|
|
/// <param name="configuration">The configuration carrying the <c>ServerHistorian</c> section.</param>
|
|
/// <param name="provisioningFactory">
|
|
/// Factory the Host supplies to build the concrete <see cref="IHistorianProvisioning"/>
|
|
/// (the gateway-backed provisioner) from the bound options + the resolving provider.
|
|
/// </param>
|
|
/// <returns>The same <paramref name="services"/> instance for chaining.</returns>
|
|
public static IServiceCollection AddHistorianProvisioning(
|
|
this IServiceCollection services,
|
|
IConfiguration configuration,
|
|
Func<ServerHistorianOptions, IServiceProvider, IHistorianProvisioning> provisioningFactory)
|
|
{
|
|
var opts = configuration.GetSection(ServerHistorianOptions.SectionName).Get<ServerHistorianOptions>();
|
|
if (opts is not { Enabled: true }) return services; // leave the Null default from AddOtOpcUaRuntime
|
|
|
|
// Last-registration-wins over the TryAddSingleton Null default seeded by AddOtOpcUaRuntime.
|
|
services.AddSingleton<IHistorianProvisioning>(sp => provisioningFactory(opts, sp));
|
|
return services;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Spawns the per-node driver-role actors on the host's <see cref="ActorSystem"/>:
|
|
/// <see cref="DriverHostActor"/> (one per node), <see cref="DbHealthProbeActor"/>
|
|
/// (consumed by the health endpoint + redundancy calc), and
|
|
/// <see cref="HistorianAdapterActor"/> wrapping the registered <see cref="IAlarmHistorianSink"/>.
|
|
///
|
|
/// Mirror of <c>WithOtOpcUaControlPlaneSingletons</c> for the driver role. Both must
|
|
/// be registered on the same <see cref="AkkaConfigurationBuilder"/> as the cluster
|
|
/// bootstrap so the actors share the host's ActorSystem.
|
|
///
|
|
/// Wire from the fused Host's Program.cs when the node carries the <c>driver</c> role:
|
|
/// <code>
|
|
/// services.AddOtOpcUaRuntime();
|
|
/// services.AddAkka("otopcua", (ab, sp) => { ab.WithOtOpcUaClusterBootstrap(sp); if (hasDriver) ab.WithOtOpcUaRuntimeActors(); });
|
|
/// </code>
|
|
/// </summary>
|
|
/// <param name="builder">The Akka configuration builder.</param>
|
|
/// <returns>The same <paramref name="builder"/> instance for chaining.</returns>
|
|
public static AkkaConfigurationBuilder WithOtOpcUaRuntimeActors(this AkkaConfigurationBuilder builder)
|
|
{
|
|
// Production cluster HOCON (akka.conf) carries this dispatcher block, but consumers that
|
|
// bootstrap their own HOCON (e.g. ServiceCollectionExtensionsTests) wouldn't pick it up
|
|
// — OpcUaPublishActor.Props pins itself to opcua-synchronized-dispatcher and Akka throws
|
|
// ConfigurationException if it doesn't exist. Prepend a fallback so the runtime extension
|
|
// is self-contained.
|
|
builder.AddHocon(@"
|
|
opcua-synchronized-dispatcher {
|
|
type = ""PinnedDispatcher""
|
|
executor = ""thread-pool-executor""
|
|
throughput = 1
|
|
}
|
|
", HoconAddMode.Prepend);
|
|
|
|
builder.WithActors((system, registry, resolver) =>
|
|
{
|
|
var dbFactory = resolver.GetService<IDbContextFactory<OtOpcUaConfigDbContext>>();
|
|
var roleInfo = resolver.GetService<IClusterRoleInfo>();
|
|
// Fallback to Null* if AddOtOpcUaRuntime wasn't called (e.g., test harnesses).
|
|
var historianSink = resolver.GetService<IAlarmHistorianSink>() ?? NullAlarmHistorianSink.Instance;
|
|
var driverFactory = resolver.GetService<IDriverFactory>() ?? NullDriverFactory.Instance;
|
|
// Phase 6.1 resilience-invoker factory — bound by the fused Host (which references Core +
|
|
// Polly); pass-through on nodes/harnesses that didn't register it. Resolved through the
|
|
// Core.Abstractions seam so this Polly-free assembly stays that way.
|
|
var invokerFactory = resolver.GetService<IDriverCapabilityInvokerFactory>()
|
|
?? NullDriverCapabilityInvokerFactory.Instance;
|
|
var addressSpaceSink = resolver.GetService<IOpcUaAddressSpaceSink>() ?? NullOpcUaAddressSpaceSink.Instance;
|
|
var serviceLevel = resolver.GetService<IServiceLevelPublisher>() ?? NullServiceLevelPublisher.Instance;
|
|
var loggerFactory = resolver.GetService<ILoggerFactory>() ?? NullLoggerFactory.Instance;
|
|
var healthPublisher = resolver.GetService<IDriverHealthPublisher>() ?? NullDriverHealthPublisher.Instance;
|
|
// Node-local deployment-artifact cache. Registered by the Host's AddOtOpcUaLocalDb on
|
|
// driver-role nodes only; deliberately left null elsewhere (admin-only graphs, test
|
|
// harnesses) rather than given a null-object, so DriverHostActor skips caching outright
|
|
// instead of pretending to cache into a sink that drops everything.
|
|
var deploymentArtifactCache = resolver.GetService<IDeploymentArtifactCache>();
|
|
// Per-cluster mesh Phase 4: scripted-alarm condition-state store. Registered by the Host's
|
|
// AddOtOpcUaLocalDb on driver-role nodes (the replicated LocalDb store), so condition state
|
|
// persists with no ConfigDb; null elsewhere (admin-only graphs, test harnesses) → the actor
|
|
// skips spawning the alarm host (the ConfigDb-backed EF fallback was retired in Phase 4 Task 9).
|
|
var alarmStateStore = resolver.GetService<IAlarmStateStore>();
|
|
// Per-cluster mesh Phase 3: ConfigSource:Mode selects where this driver reads config.
|
|
// FetchAndCache pulls the artifact from central over gRPC and reads only the LocalDb cache;
|
|
// Direct (default) reads central SQL as before. The fetcher is resolved only in FetchAndCache
|
|
// mode (registered by the Host under hasDriver); its absence there is a misconfiguration the
|
|
// actor fails the apply on rather than fetching nothing forever.
|
|
var configSourceOptions =
|
|
resolver.GetService<IOptions<ConfigSourceOptions>>()?.Value ?? new ConfigSourceOptions();
|
|
var fetchAndCacheMode = string.Equals(
|
|
configSourceOptions.Mode, ConfigSourceOptions.ModeFetchAndCache, StringComparison.OrdinalIgnoreCase);
|
|
var artifactFetcher = fetchAndCacheMode ? resolver.GetService<IDeploymentArtifactFetcher>() : null;
|
|
// Where this actor publishes its Primary-gate verdict for the alarm store-and-forward
|
|
// drain, which runs on a timer and so cannot read RedundancyStateChanged itself.
|
|
// Registered by AddAlarmHistorian; absent when no durable sink is configured, in which
|
|
// case there is nothing downstream to inform.
|
|
var redundancyRoleView = resolver.GetService<IRedundancyRoleView>();
|
|
// Host of this node's LocalDb replication partner: the ONLY node that holds a copy of this
|
|
// node's alarm queue, and so the only node it may stand down in favour of. Null when this
|
|
// node dials nobody, which correctly means "never stand down".
|
|
var replicationPeerHost =
|
|
Uri.TryCreate(
|
|
resolver.GetService<IConfiguration>()?["LocalDb:Replication:PeerAddress"],
|
|
UriKind.Absolute,
|
|
out var peerUri)
|
|
? peerUri.Host
|
|
: null;
|
|
// Root script logger backs the ScriptedAlarm host's engine + script logging. Registered in
|
|
// Host DI inside the hasDriver block; may be absent in some role configs / test harnesses,
|
|
// in which case the DriverHostActor gracefully skips spawning the ScriptedAlarm host.
|
|
var scriptRootLogger = resolver.GetService<ScriptRootLogger>();
|
|
// Production evaluator is the Host's RoslynVirtualTagEvaluator (registered as
|
|
// IVirtualTagEvaluator); fall back to the null evaluator for test harnesses that don't
|
|
// register one (VirtualTagActor children then evaluate to nothing).
|
|
var virtualTagEvaluator = resolver.GetService<IVirtualTagEvaluator>();
|
|
if (virtualTagEvaluator is null)
|
|
{
|
|
loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions")
|
|
.LogWarning("IVirtualTagEvaluator not registered; Equipment VirtualTags will evaluate to NoChange (no live values). Expected only in test harnesses — driver-role nodes should register RoslynVirtualTagEvaluator.");
|
|
virtualTagEvaluator = NullVirtualTagEvaluator.Instance;
|
|
}
|
|
|
|
// VirtualTag historization sink threaded to the spawned VirtualTagHostActor. Null default
|
|
// (durable AVEVA sink is infra-gated); a deployment binding a real IHistoryWriter overrides.
|
|
var historyWriter = resolver.GetService<IHistoryWriter>() ?? NullHistoryWriter.Instance;
|
|
|
|
// Per-cluster mesh Phase 4: a driver-only node has NO ConfigDb (LocalDb is its config
|
|
// store), so dbFactory is null and there is nothing to probe. Skip spawning the probe
|
|
// entirely — spawning it with a null factory would NRE on the first SELECT 1 — and do not
|
|
// register its marker key. OpcUaPublishActor then runs in dbLess mode (DbReachable retired,
|
|
// treated constant-true) so a healthy DB-less node still publishes full ServiceLevel.
|
|
IActorRef? dbHealth = null;
|
|
if (dbFactory is not null)
|
|
{
|
|
dbHealth = system.ActorOf(
|
|
DbHealthProbeActor.Props(dbFactory),
|
|
DbHealthProbeActorName);
|
|
registry.Register<DbHealthProbeActorKey>(dbHealth);
|
|
}
|
|
|
|
// Dependency mux must be spawned before DriverHostActor so the host can forward
|
|
// AttributeValuePublished into it from the very first driver spawn.
|
|
var mux = system.ActorOf(DependencyMuxActor.Props(), DependencyMuxActorName);
|
|
registry.Register<DependencyMuxActorKey>(mux);
|
|
|
|
// Per-cluster mesh Phase 2: the node end of the central↔node command boundary. Spawned
|
|
// BEFORE DriverHostActor because it is the host's ackRouter under ClusterClient mode.
|
|
// It has no dependency of its own on the host — inbound commands travel via the node's
|
|
// EventStream — so this ordering costs nothing.
|
|
var meshOptions = resolver.GetService<IOptions<MeshTransportOptions>>().Value;
|
|
var useClusterClient = string.Equals(
|
|
meshOptions.Mode, MeshTransportOptions.ModeClusterClient, StringComparison.OrdinalIgnoreCase);
|
|
|
|
IActorRef? centralClient = null;
|
|
if (useClusterClient)
|
|
{
|
|
// Contacts come from appsettings, not the database — the deliberate asymmetry:
|
|
// central's node set changes as operators add and retire nodes, but central's own
|
|
// address is part of the deployment. MeshTransportOptionsValidator has already
|
|
// rejected an empty or malformed list at boot, so this cannot silently produce a
|
|
// client with no contacts.
|
|
var contacts = meshOptions.CentralContactPoints
|
|
.Select(cp => ActorPath.Parse($"{cp}/system/receptionist"))
|
|
.ToImmutableHashSet();
|
|
centralClient = system.ActorOf(
|
|
ClusterClient.Props(ClusterClientSettings.Create(system).WithInitialContacts(contacts)),
|
|
CentralClusterClientName);
|
|
}
|
|
|
|
var nodeComm = system.ActorOf(
|
|
NodeCommunicationActor.Props(centralClient),
|
|
MeshPaths.NodeCommunicationName);
|
|
|
|
// Registered with the receptionist in BOTH modes: under Dps nothing dials it, and an
|
|
// idle registration costs nothing — whereas registering only under ClusterClient would
|
|
// make flipping the flag require a node restart before central could reach it.
|
|
ClusterClientReceptionist.Get(system).RegisterService(nodeComm);
|
|
registry.Register<NodeCommunicationActorKey>(nodeComm);
|
|
|
|
// Continuous-historization recorder — gated on ContinuousHistorization:Enabled AND the
|
|
// gateway-backed IHistorianValueWriter + the durable IHistorizationOutbox being registered
|
|
// (the Host registers both ONLY when historization is enabled and the ServerHistorian gateway
|
|
// is configured). The recorder taps the dependency mux's value fan-out, so it is spawned after
|
|
// (and fed) the same `mux` ref the DriverHostActor uses. It is spawned BEFORE the applier so
|
|
// the applier's historized-ref subscription sink can wrap this recorder's IActorRef and feed it
|
|
// the add/remove delta of historized refs on every deploy (closing the T18 ref-feed gap).
|
|
IActorRef? continuousRecorder = null;
|
|
var continuousOptions = resolver.GetService<ContinuousHistorizationOptions>();
|
|
if (continuousOptions is { Enabled: true })
|
|
{
|
|
var valueWriter = resolver.GetService<IHistorianValueWriter>();
|
|
var outbox = resolver.GetService<IHistorizationOutbox>();
|
|
if (valueWriter is not null && outbox is not null)
|
|
{
|
|
// Initial ref set is EMPTY: the deployed address space (and thus the historized-ref
|
|
// set) is built later at deploy time, not here. The applier's per-deploy add/remove
|
|
// feed populates the recorder's interest from that point on.
|
|
continuousRecorder = system.ActorOf(
|
|
ContinuousHistorizationRecorder.Props(
|
|
dependencyMux: mux,
|
|
writer: valueWriter,
|
|
outbox: outbox,
|
|
historizedRefs: Array.Empty<string>(),
|
|
drainBatchSize: continuousOptions.DrainBatchSize,
|
|
drainInterval: TimeSpan.FromSeconds(continuousOptions.DrainIntervalSeconds),
|
|
minBackoff: TimeSpan.FromSeconds(continuousOptions.MinBackoffSeconds),
|
|
maxBackoff: TimeSpan.FromSeconds(continuousOptions.MaxBackoffSeconds)),
|
|
ContinuousHistorizationRecorderActorName);
|
|
registry.Register<ContinuousHistorizationRecorderKey>(continuousRecorder);
|
|
}
|
|
else
|
|
{
|
|
loggerFactory.CreateLogger("ZB.MOM.WW.OtOpcUa.Runtime.ServiceCollectionExtensions")
|
|
.LogWarning("ContinuousHistorization is enabled but IHistorianValueWriter and/or IHistorizationOutbox are not registered; the recorder will not be spawned. Expected only in misconfigured deployments or test harnesses.");
|
|
}
|
|
}
|
|
|
|
// Historized-ref subscription sink fed by the applier on every deploy. When the recorder was
|
|
// spawned, an adapter wraps its IActorRef (a non-blocking Tell of the add/remove delta);
|
|
// otherwise the Null no-op sink, so the applier behaves identically when historization is off.
|
|
IHistorizedTagSubscriptionSink historizedSubscriptions = continuousRecorder is not null
|
|
? new ActorHistorizedTagSubscriptionSink(continuousRecorder)
|
|
: NullHistorizedTagSubscriptionSink.Instance;
|
|
|
|
// Historian tag provisioner fed to the applier so deploying historized tags auto-ensures
|
|
// them in the historian (EnsureTags). The Host registers the gateway-backed provisioner via
|
|
// AddHistorianProvisioning when ServerHistorian:Enabled; otherwise this resolves the no-op
|
|
// NullHistorianProvisioning default seeded by AddOtOpcUaRuntime (so the applier's hook is inert).
|
|
var provisioning = resolver.GetService<IHistorianProvisioning>();
|
|
|
|
// OPC UA publish actor — pinned dispatcher, owns the address-space side of the
|
|
// pipeline. AddressSpaceApplier is constructed here so the actor + applier share the
|
|
// same sink reference (when DeferredAddressSpaceSink swaps later, both see it).
|
|
var applier = new AddressSpaceApplier(
|
|
addressSpaceSink,
|
|
loggerFactory.CreateLogger<AddressSpaceApplier>(),
|
|
provisioning: provisioning,
|
|
historizedSubscriptions: historizedSubscriptions);
|
|
var publishActor = system.ActorOf(
|
|
OpcUaPublishActor.Props(
|
|
sink: addressSpaceSink,
|
|
serviceLevel: serviceLevel,
|
|
localNode: roleInfo.LocalNode,
|
|
dbFactory: dbFactory,
|
|
applier: applier,
|
|
dbHealthProbe: dbHealth,
|
|
dbLess: dbFactory is null),
|
|
OpcUaPublishActorName);
|
|
registry.Register<OpcUaPublishActorKey>(publishActor);
|
|
|
|
// Per-node peer-probe supervisor — keeps one OPC UA TCP probe per OTHER non-Detached
|
|
// driver node, so every node is continuously probed by all its peers. OpcUaPublishActor
|
|
// consumes the resulting probe verdicts to learn this node's own reachability.
|
|
var peerProbes = system.ActorOf(
|
|
PeerProbeSupervisor.Props(roleInfo.LocalNode),
|
|
PeerProbeSupervisorName);
|
|
registry.Register<PeerProbeSupervisorKey>(peerProbes);
|
|
|
|
var driverHost = system.ActorOf(
|
|
// ackRouter: under ClusterClient mode the node comm actor relays ApplyAcks across
|
|
// the boundary; under Dps it stays null and the host publishes on the
|
|
// deployment-acks topic exactly as before.
|
|
// dbFactory is nullable (Phase 4): a driver-only node passes null here — ConfigDb is
|
|
// gated on hasAdmin and driver-only nodes are forced to FetchAndCache. DriverHostActor
|
|
// guards every ConfigDb-touching path on it (UpsertNodeDeploymentState no-ops; central
|
|
// persists acks from the ApplyAck) and serves scripted-alarm state from the LocalDb
|
|
// alarmStateStore (there is no ConfigDb-backed EF fallback; the host skips the alarm
|
|
// host when no store was wired).
|
|
DriverHostActor.Props(dbFactory, roleInfo.LocalNode,
|
|
ackRouter: useClusterClient ? nodeComm : null,
|
|
driverFactory: driverFactory, localRoles: roleInfo.LocalRoles,
|
|
dependencyMux: mux,
|
|
opcUaPublishActor: publishActor,
|
|
healthPublisher: healthPublisher,
|
|
virtualTagEvaluator: virtualTagEvaluator,
|
|
historyWriter: historyWriter,
|
|
loggerFactory: loggerFactory,
|
|
scriptRootLogger: scriptRootLogger,
|
|
invokerFactory: invokerFactory,
|
|
deploymentArtifactCache: deploymentArtifactCache,
|
|
redundancyRoleView: redundancyRoleView,
|
|
replicationPeerHost: replicationPeerHost,
|
|
fetchAndCacheMode: fetchAndCacheMode,
|
|
artifactFetcher: artifactFetcher,
|
|
alarmStateStore: alarmStateStore,
|
|
// Phase 5: the node-local live-telemetry hub (registered in AddOtOpcUaRuntime). The host
|
|
// emits its Primary-gated native-alarm transitions into it and threads it into the
|
|
// VirtualTag + ScriptedAlarm hosts it spawns.
|
|
telemetryHub: resolver.GetService<ITelemetryLocalHub>(),
|
|
// Gitea #511: where this actor publishes its Primary DATA-PLANE verdict for drivers
|
|
// that claim an exclusive external identity. Registered unconditionally by
|
|
// AddOtOpcUaRuntime, unlike redundancyRoleView above — see the comment there.
|
|
primaryRoleSource: resolver.GetService<IDriverPrimaryRoleSource>()),
|
|
DriverHostActorName);
|
|
registry.Register<DriverHostActorKey>(driverHost);
|
|
|
|
var historian = system.ActorOf(
|
|
HistorianAdapterActor.Props(historianSink, roleInfo.LocalNode),
|
|
HistorianAdapterActorName);
|
|
registry.Register<HistorianAdapterActorKey>(historian);
|
|
});
|
|
|
|
return builder;
|
|
}
|
|
}
|
|
|
|
/// <summary>Marker key types used by <c>Akka.Hosting</c> to resolve runtime actors from the registry.</summary>
|
|
public sealed class DriverHostActorKey { }
|
|
public sealed class DbHealthProbeActorKey { }
|
|
public sealed class HistorianAdapterActorKey { }
|
|
public sealed class DependencyMuxActorKey { }
|
|
public sealed class OpcUaPublishActorKey { }
|
|
public sealed class NodeCommunicationActorKey { }
|
|
|
|
/// <summary>Marker key for the per-node ContinuousHistorizationRecorder (spawned only when
|
|
/// <c>ContinuousHistorization:Enabled=true</c> and the gateway value-writer + outbox are registered).</summary>
|
|
public sealed class ContinuousHistorizationRecorderKey { }
|
|
|
|
/// <summary>Marker key for the per-node PeerProbeSupervisor.</summary>
|
|
public sealed class PeerProbeSupervisorKey { }
|