Joseph Doherty
ab5b958d47
WithOtOpcUaRuntimeActors no longer builds a ClusterClient from MeshTransport:CentralContactPoints, and NodeCommunicationActor.Props takes only the gRPC IApplyAckSender. The ackRouter gate narrows from `useClusterClient || useGrpc` to `useGrpc`, and the CentralClusterClientName constant is gone. NodeCommunicationActor's no-transport warning now names MeshTransport:CentralGrpcEndpoints and MeshTransport:Mode only — the old text pointed an operator at a key that no longer exists. Ripples in this step: the two ClusterClient-shaped facts in NodeCommunicationActorTests (Send-not-SendToAll, and sender-wins-over-a-client) are deleted — the first has no subject left, the second guarded a both-transports-present state that is now unconstructible. The Runtime ServiceCollectionExtensions fact keeps its ackRouter pin and drops its "no ClusterClient actor exists" half for the same reason.
616 lines
41 KiB
C#
616 lines
41 KiB
C#
using System.Collections.Immutable;
|
|
using Akka.Actor;
|
|
using Akka.Cluster.Tools.Client;
|
|
using Akka.Event;
|
|
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.Authorization;
|
|
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";
|
|
|
|
/// <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>());
|
|
|
|
// Gitea #537: the continuous-historization capture verdict, registered here — unconditionally on
|
|
// every driver-role node — for the same reason as the line above, and NOT alongside the alarm
|
|
// drain's IRedundancyRoleView. AddAlarmHistorian registers that one only when its own section is
|
|
// enabled, so piggybacking would leave a redundant pair that historizes values but not alarms
|
|
// with no gate at all, and both nodes would write every sample twice with nothing to say so.
|
|
services.TryAddSingleton<IHistorizationRoleView>(sp => sp.GetRequiredService<RedundancyRoleView>());
|
|
|
|
// The node→central ack transport under MeshTransport:Mode = Grpc (mesh gRPC command plane,
|
|
// Task 1B.2). Registered unconditionally but resolved only by the Grpc arm of
|
|
// WithOtOpcUaRuntimeActors — the mode is not knowable here (this runs before configuration is
|
|
// read into options), and a singleton nobody resolves is never constructed, so a Dps node
|
|
// opens no channel.
|
|
//
|
|
// Registered in the CONTAINER rather than new-ed up beside the actor so its per-endpoint
|
|
// GrpcChannel cache is disposed at host shutdown; an actor-scoped instance would leak the
|
|
// channels on every restart.
|
|
//
|
|
// Built by a factory, not by type: MS.DI does not honour optional constructor parameters, so
|
|
// the clientFactory test seam would make type-based activation fail outright. The logger is
|
|
// resolved leniently for the same reason DriverHostActor's is — a harness that skipped logging
|
|
// should not fail to build the graph.
|
|
services.TryAddSingleton<IApplyAckSender>(sp => new GrpcApplyAckSender(
|
|
sp.GetRequiredService<IOptions<MeshTransportOptions>>(),
|
|
sp.GetRequiredService<IOptions<AkkaClusterOptions>>(),
|
|
sp.GetService<ILoggerFactory>()?.CreateLogger<GrpcApplyAckSender>()
|
|
?? NullLogger<GrpcApplyAckSender>.Instance));
|
|
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 Grpc 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 useGrpc = string.Equals(
|
|
meshOptions.Mode, MeshTransportOptions.ModeGrpc, StringComparison.OrdinalIgnoreCase);
|
|
|
|
// The node→central ack transport. Built ONLY under Grpc: under Dps the ack goes out on
|
|
// the deployment-acks topic from DriverHostActor and never reaches this actor at all.
|
|
IApplyAckSender? ackSender = null;
|
|
if (useGrpc)
|
|
{
|
|
ackSender = resolver.GetService<IApplyAckSender>();
|
|
if (ackSender is null)
|
|
{
|
|
// Reachable only in a graph that skipped AddOtOpcUaRuntime (a test harness), never
|
|
// in the Host. Said here rather than left to the actor's per-ack drop warning,
|
|
// because the cause is a missing registration and the symptom would otherwise
|
|
// appear once per deployment with no hint of why.
|
|
system.Log.Warning(
|
|
"MeshTransport:Mode is {Mode} but no {Sender} is registered — every ApplyAck "
|
|
+ "this node produces will be dropped. AddOtOpcUaRuntime registers it.",
|
|
meshOptions.Mode, nameof(IApplyAckSender));
|
|
}
|
|
}
|
|
|
|
var nodeComm = system.ActorOf(
|
|
NodeCommunicationActor.Props(ackSender),
|
|
MeshPaths.NodeCommunicationName);
|
|
|
|
// Registered with the receptionist in ALL modes: under Dps and Grpc 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),
|
|
// Gitea #537: read the verdict PER VALUE off the shared role view rather
|
|
// than capturing a bool, so a promotion resumes capture on the next sample.
|
|
// The view is registered unconditionally on driver nodes; a null one (test
|
|
// graphs) captures unconditionally, which is the right posture there.
|
|
captureGate: resolver.GetService<IHistorizationRoleView>() is { } histRoleView
|
|
? () => histRoleView.ShouldRecordHistorizedValues
|
|
: null,
|
|
maxWriteAttempts: continuousOptions.MaxWriteAttempts),
|
|
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);
|
|
// #520 stage 4 — the sink that receives each applied artifact's API keys + ACL grants, so OPC
|
|
// UA session activation can authenticate. Optional: GetService (not GetRequiredService), because
|
|
// a node with no driver role never registers it and test harnesses build this graph without it;
|
|
// the actor treats null as "nothing consumes authorization inputs" and skips the decode.
|
|
var dataPlaneAuthority = resolver.GetService<IDataPlaneAuthoritySink>();
|
|
var publishActor = system.ActorOf(
|
|
OpcUaPublishActor.Props(
|
|
sink: addressSpaceSink,
|
|
serviceLevel: serviceLevel,
|
|
localNode: roleInfo.LocalNode,
|
|
dbFactory: dbFactory,
|
|
applier: applier,
|
|
dbHealthProbe: dbHealth,
|
|
dbLess: dbFactory is null,
|
|
dataPlaneAuthority: dataPlaneAuthority),
|
|
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);
|
|
|
|
// Gitea #537: "does this snapshot node record the same historized refs as me?" — true only
|
|
// for a partner in this node's own per-cluster mesh, which is exactly the set that serves
|
|
// this node's address space. A node carrying NO cluster-scoped role (the legacy,
|
|
// pre-Phase-6 shape) cannot establish that, so the predicate stays null and this node never
|
|
// stands down: on a fleet-wide election the elected Primary is usually running a DIFFERENT
|
|
// cluster's configuration, and deferring to it would silently end this cluster's
|
|
// historization altogether — the same scope mismatch that once suspended the alarm drain on
|
|
// every node at once.
|
|
Func<Commons.Types.NodeId, bool>? historizationScopePeer =
|
|
roleInfo.ClusterRole is { } meshRole
|
|
? nodeId => roleInfo.MembersWithRole(meshRole).Contains(nodeId)
|
|
: null;
|
|
|
|
var driverHost = system.ActorOf(
|
|
// ackRouter: under Grpc the node comm actor relays ApplyAcks across the mesh
|
|
// boundary through the gRPC sender it was built with; under Dps it stays null and
|
|
// the host publishes on the deployment-acks topic exactly as before. Routed to DPS
|
|
// instead, a Grpc node's ack would never reach the sender at all and every
|
|
// deployment to it would time out while it applied perfectly.
|
|
// 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: useGrpc ? 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>(),
|
|
// Gitea #537: where this actor publishes the continuous-historization CAPTURE
|
|
// verdict, and how it decides which peer is worth standing down for.
|
|
historizationRoleView: resolver.GetService<IHistorizationRoleView>(),
|
|
sharesHistorizationScopeWithLocal: historizationScopePeer),
|
|
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 { }
|