fix(r2-04): scripted-alarm alerts emit uses PrimaryGatePolicy (consistency with 03/S4)
This commit is contained in:
@@ -13,7 +13,7 @@
|
|||||||
{ "id": "T9", "subject": "PrimaryGatePolicy pure class + truth-table tests + otopcua.redundancy.primary_gate_denied counter (03/S4)", "status": "completed", "blockedBy": [] },
|
{ "id": "T9", "subject": "PrimaryGatePolicy pure class + truth-table tests + otopcua.redundancy.primary_gate_denied counter (03/S4)", "status": "completed", "blockedBy": [] },
|
||||||
{ "id": "T10", "subject": "Wire DriverHostActor gates (:977/:1039/:1095) through PrimaryGatePolicy with member-count provider + S5 log-once", "status": "completed", "blockedBy": ["T9"] },
|
{ "id": "T10", "subject": "Wire DriverHostActor gates (:977/:1039/:1095) through PrimaryGatePolicy with member-count provider + S5 log-once", "status": "completed", "blockedBy": ["T9"] },
|
||||||
{ "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "completed", "blockedBy": ["T10"] },
|
{ "id": "T11", "subject": "DriverHostActorPrimaryGateTests: deny-on-unknown multi-node, allow single-node, ack/emit variants, meter tags", "status": "completed", "blockedBy": ["T10"] },
|
||||||
{ "id": "T12", "subject": "ScriptedAlarmHostActor alerts-emit gate adopts PrimaryGatePolicy (consistency extension)", "status": "pending", "blockedBy": ["T9"] },
|
{ "id": "T12", "subject": "ScriptedAlarmHostActor alerts-emit gate adopts PrimaryGatePolicy (consistency extension)", "status": "completed", "blockedBy": ["T9"] },
|
||||||
{ "id": "T13", "subject": "In-process 2-node delivery-path test (TwoNodeClusterHarness): DPS-delivered snapshot drives the gate", "status": "pending", "blockedBy": ["T11"] },
|
{ "id": "T13", "subject": "In-process 2-node delivery-path test (TwoNodeClusterHarness): DPS-delivered snapshot drives the gate", "status": "pending", "blockedBy": ["T11"] },
|
||||||
{ "id": "T14", "subject": "docs/Redundancy.md: gate policy table, boot-window client experience, new meters, Galaxy fail-closed semantics", "status": "pending", "blockedBy": [] },
|
{ "id": "T14", "subject": "docs/Redundancy.md: gate policy table, boot-window client experience, new meters, Galaxy fail-closed semantics", "status": "pending", "blockedBy": [] },
|
||||||
{ "id": "T15", "subject": "LIVE GATE: 2-node docker-dev rig verify (ServiceLevel 240/100, boot-window write rejected+reverted, single /alerts row)", "status": "pending", "blockedBy": ["T10", "T11", "T12", "T13"] }
|
{ "id": "T15", "subject": "LIVE GATE: 2-node docker-dev rig verify (ServiceLevel 240/100, boot-window write rejected+reverted, single /alerts row)", "status": "pending", "blockedBy": ["T10", "T11", "T12", "T13"] }
|
||||||
|
|||||||
@@ -3,11 +3,13 @@ using Akka.Cluster.Tools.PublishSubscribe;
|
|||||||
using Akka.Event;
|
using Akka.Event;
|
||||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
|
||||||
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
using ZB.MOM.WW.OtOpcUa.Commons.Messages.Redundancy;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
|
||||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||||
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms;
|
||||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||||
|
using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
|
||||||
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
|
||||||
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
|
||||||
|
|
||||||
@@ -104,10 +106,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
|||||||
|
|
||||||
/// <summary>Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
|
/// <summary>Cached local <see cref="RedundancyRole"/> from the latest <see cref="RedundancyStateChanged"/>
|
||||||
/// snapshot (null = unknown until the first snapshot arrives, or no <see cref="_localNode"/> wired). The
|
/// snapshot (null = unknown until the first snapshot arrives, or no <see cref="_localNode"/> wired). The
|
||||||
/// cluster-wide <c>alerts</c> publish in <see cref="OnEngineEmission"/> is gated on this: only the Primary
|
/// cluster-wide <c>alerts</c> publish in <see cref="OnEngineEmission"/> resolves this through
|
||||||
/// publishes (default-emit while unknown so single-node deploys + the boot window never drop transitions).</summary>
|
/// <see cref="Drivers.PrimaryGatePolicy"/>: only the Primary publishes; an unknown role default-emits on a
|
||||||
|
/// single-driver cluster but default-DENIES on a multi-driver one, so the dual-primary boot window can't
|
||||||
|
/// historize duplicate AVEVA alarm rows (archreview 03/S4).</summary>
|
||||||
private RedundancyRole? _localRole;
|
private RedundancyRole? _localRole;
|
||||||
|
|
||||||
|
/// <summary>Test seam (archreview 03/S4): overrides the count of Up <c>driver</c>-role cluster members the
|
||||||
|
/// alerts-emit gate reads while the role is unknown. Null ⇒ read the live cluster state.</summary>
|
||||||
|
private readonly Func<int>? _driverMemberCountProvider;
|
||||||
|
|
||||||
/// <summary>Monotonic load generation, bumped on every <see cref="OnApply"/>. The continuation that
|
/// <summary>Monotonic load generation, bumped on every <see cref="OnApply"/>. The continuation that
|
||||||
/// pipes back an <see cref="AlarmsLoaded"/> captures the generation it was started under; a stale
|
/// pipes back an <see cref="AlarmsLoaded"/> captures the generation it was started under; a stale
|
||||||
/// completion (an earlier generation arriving after a newer apply) is discarded in
|
/// completion (an earlier generation arriving after a newer apply) is discarded in
|
||||||
@@ -136,8 +144,9 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
|||||||
IActorRef? mux,
|
IActorRef? mux,
|
||||||
DependencyMuxTagUpstreamSource upstream,
|
DependencyMuxTagUpstreamSource upstream,
|
||||||
ScriptedAlarmEngine engine,
|
ScriptedAlarmEngine engine,
|
||||||
NodeId? localNode = null) =>
|
NodeId? localNode = null,
|
||||||
Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode));
|
Func<int>? driverMemberCountProvider = null) =>
|
||||||
|
Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider));
|
||||||
|
|
||||||
/// <summary>Initializes a new instance of the <see cref="ScriptedAlarmHostActor"/> class.</summary>
|
/// <summary>Initializes a new instance of the <see cref="ScriptedAlarmHostActor"/> class.</summary>
|
||||||
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
|
/// <param name="publishActor">The OPC UA publish actor emissions are bridged to.</param>
|
||||||
@@ -147,12 +156,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
|||||||
/// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
|
/// <param name="localNode">The local cluster node id, used to read this node's <see cref="RedundancyRole"/>
|
||||||
/// from the <c>redundancy-state</c> topic so only the Primary publishes the cluster-wide <c>alerts</c>
|
/// from the <c>redundancy-state</c> topic so only the Primary publishes the cluster-wide <c>alerts</c>
|
||||||
/// transition. Null leaves the role unknown ⇒ default-emit (single-node deploys + tests).</param>
|
/// transition. Null leaves the role unknown ⇒ default-emit (single-node deploys + tests).</param>
|
||||||
|
/// <param name="driverMemberCountProvider">Test seam (archreview 03/S4): overrides the count of Up
|
||||||
|
/// <c>driver</c>-role cluster members the alerts-emit gate reads while the role is unknown. Null reads
|
||||||
|
/// the live cluster state (0 on a non-cluster ActorRefProvider ⇒ single-node default-emit).</param>
|
||||||
public ScriptedAlarmHostActor(
|
public ScriptedAlarmHostActor(
|
||||||
IActorRef publishActor,
|
IActorRef publishActor,
|
||||||
IActorRef? mux,
|
IActorRef? mux,
|
||||||
DependencyMuxTagUpstreamSource upstream,
|
DependencyMuxTagUpstreamSource upstream,
|
||||||
ScriptedAlarmEngine engine,
|
ScriptedAlarmEngine engine,
|
||||||
NodeId? localNode = null)
|
NodeId? localNode = null,
|
||||||
|
Func<int>? driverMemberCountProvider = null)
|
||||||
{
|
{
|
||||||
ArgumentNullException.ThrowIfNull(publishActor);
|
ArgumentNullException.ThrowIfNull(publishActor);
|
||||||
ArgumentNullException.ThrowIfNull(upstream);
|
ArgumentNullException.ThrowIfNull(upstream);
|
||||||
@@ -162,6 +175,7 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
|||||||
_upstream = upstream;
|
_upstream = upstream;
|
||||||
_engine = engine;
|
_engine = engine;
|
||||||
_localNode = localNode;
|
_localNode = localNode;
|
||||||
|
_driverMemberCountProvider = driverMemberCountProvider;
|
||||||
|
|
||||||
// OnEvent fires on the engine's worker thread. NEVER touch Context / actor state here —
|
// OnEvent fires on the engine's worker thread. NEVER touch Context / actor state here —
|
||||||
// marshal onto the actor thread via the thread-safe Self.Tell. Keep the handler in a field
|
// marshal onto the actor thread via the thread-safe Self.Tell. Keep the handler in a field
|
||||||
@@ -307,18 +321,45 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor
|
|||||||
// suppress only the sink write. This publish (and the live `/alerts` fan-out) is NOT gated on it.
|
// suppress only the sink write. This publish (and the live `/alerts` fan-out) is NOT gated on it.
|
||||||
HistorizeToAveva: e.HistorizeToAveva);
|
HistorizeToAveva: e.HistorizeToAveva);
|
||||||
|
|
||||||
// Warm-standby dedup: only the Primary (driver-role leader) publishes the cluster-wide
|
// Warm-standby dedup: only the Primary publishes the cluster-wide transition (archreview 03/S4 —
|
||||||
// transition. Default-emit until told we are Secondary/Detached so single-node deploys + the
|
// same PrimaryGatePolicy as DriverHostActor). A KNOWN Secondary/Detached is denied; an UNKNOWN role
|
||||||
// boot window never drop transitions. The OPC UA node write above + inbound command processing
|
// default-emits on a single-driver cluster but default-DENIES on a multi-driver one (so the
|
||||||
// stay ungated (the secondary keeps its address space + engine state warm for failover).
|
// dual-primary boot window can't historize duplicate AVEVA alarm rows). The OPC UA node write above +
|
||||||
if (_localRole is RedundancyRole.Secondary or RedundancyRole.Detached)
|
// inbound command processing stay ungated (the secondary keeps its state warm for failover).
|
||||||
|
if (!PrimaryGatePolicy.ShouldServiceAsPrimary(_localRole, DriverMemberCount()))
|
||||||
{
|
{
|
||||||
|
OtOpcUaTelemetry.PrimaryGateDenied.Add(1,
|
||||||
|
new KeyValuePair<string, object?>("site", "alarm-emit"),
|
||||||
|
new KeyValuePair<string, object?>("reason", _localRole switch
|
||||||
|
{
|
||||||
|
RedundancyRole.Secondary => "secondary",
|
||||||
|
RedundancyRole.Detached => "detached",
|
||||||
|
_ => "role-unknown",
|
||||||
|
}));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
_mediator.Tell(new Publish(AlertsTopic, evt));
|
_mediator.Tell(new Publish(AlertsTopic, evt));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>Count of Up cluster members carrying the <c>driver</c> role, for the alerts-emit gate. Uses
|
||||||
|
/// the injected test seam when present; otherwise reads the live cluster state guarded by try/catch so a
|
||||||
|
/// non-cluster ActorRefProvider yields 0 ⇒ the single-node default-emit posture (archreview 03/S4).</summary>
|
||||||
|
/// <returns>The number of Up driver-role members, or 0 when the cluster extension is unavailable.</returns>
|
||||||
|
private int DriverMemberCount()
|
||||||
|
{
|
||||||
|
if (_driverMemberCountProvider is not null) return _driverMemberCountProvider();
|
||||||
|
try
|
||||||
|
{
|
||||||
|
return Akka.Cluster.Cluster.Get(Context.System).State.Members
|
||||||
|
.Count(m => m.Status == Akka.Cluster.MemberStatus.Up && m.Roles.Contains(ServiceCollectionExtensions.DriverRole));
|
||||||
|
}
|
||||||
|
catch (Exception)
|
||||||
|
{
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Caches this node's <see cref="RedundancyRole"/> from a cluster redundancy snapshot so
|
/// <summary>Caches this node's <see cref="RedundancyRole"/> from a cluster redundancy snapshot so
|
||||||
/// <see cref="OnEngineEmission"/> can gate the cluster-wide <c>alerts</c> publish to the Primary. A
|
/// <see cref="OnEngineEmission"/> can gate the cluster-wide <c>alerts</c> publish to the Primary. A
|
||||||
/// snapshot that doesn't mention <see cref="_localNode"/> (or no local node wired) leaves the cached
|
/// snapshot that doesn't mention <see cref="_localNode"/> (or no local node wired) leaves the cached
|
||||||
|
|||||||
+52
@@ -519,6 +519,58 @@ public sealed class ScriptedAlarmHostActorTests : RuntimeActorTestBase
|
|||||||
evt.TransitionKind.ShouldBe("Activated");
|
evt.TransitionKind.ShouldBe("Activated");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>Boot-window multi-driver suppression (archreview 03/S4): role unknown + a real driver peer
|
||||||
|
/// (member count 2) ⇒ the alerts publish is DENIED (default-deny closes the dual-primary window that would
|
||||||
|
/// historize duplicate AVEVA rows), while the OPC UA node write stays ungated.</summary>
|
||||||
|
[Fact]
|
||||||
|
public void Unknown_role_multi_driver_suppresses_alerts_but_still_writes_opcua()
|
||||||
|
{
|
||||||
|
var publish = CreateTestProbe();
|
||||||
|
var mux = CreateTestProbe();
|
||||||
|
var alerts = CreateTestProbe();
|
||||||
|
SubscribeToAlerts(alerts);
|
||||||
|
|
||||||
|
var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 2);
|
||||||
|
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
|
||||||
|
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
|
||||||
|
|
||||||
|
// No snapshot sent ⇒ role unknown; count 2 ⇒ default-DENY.
|
||||||
|
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
|
||||||
|
|
||||||
|
publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
|
||||||
|
alerts.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Boot-window single-driver emit (archreview 03/S4): role unknown + no driver peer (member
|
||||||
|
/// count 1) ⇒ the alerts publish proceeds (single-node posture preserved).</summary>
|
||||||
|
[Fact]
|
||||||
|
public void Unknown_role_single_driver_publishes_alerts()
|
||||||
|
{
|
||||||
|
var publish = CreateTestProbe();
|
||||||
|
var mux = CreateTestProbe();
|
||||||
|
var alerts = CreateTestProbe();
|
||||||
|
SubscribeToAlerts(alerts);
|
||||||
|
|
||||||
|
var (host, _) = SpawnWithMemberCount(publish, mux, LocalNode, driverMemberCount: 1);
|
||||||
|
host.Tell(new ScriptedAlarmHostActor.ApplyScriptedAlarms(new[] { Plan(severity: 800) }));
|
||||||
|
mux.ExpectMsg<DependencyMuxActor.RegisterInterest>(Timeout);
|
||||||
|
|
||||||
|
host.Tell(new VirtualTagActor.DependencyValueChanged("M.T", 99, DateTime.UtcNow));
|
||||||
|
|
||||||
|
publish.FishForMessage<OpcUaPublishActor.AlarmStateUpdate>(m => m.State.Active, Timeout);
|
||||||
|
alerts.ExpectMsg<AlarmTransitionEvent>(Timeout).AlarmId.ShouldBe("alm-1");
|
||||||
|
}
|
||||||
|
|
||||||
|
private (IActorRef Host, DependencyMuxTagUpstreamSource Upstream) SpawnWithMemberCount(
|
||||||
|
TestProbe publish, TestProbe mux, NodeId localNode, int driverMemberCount)
|
||||||
|
{
|
||||||
|
var upstream = new DependencyMuxTagUpstreamSource();
|
||||||
|
var engine = BuildEngine(upstream);
|
||||||
|
var host = Sys.ActorOf(ScriptedAlarmHostActor.Props(
|
||||||
|
publish.Ref, mux.Ref, upstream, engine, localNode, driverMemberCountProvider: () => driverMemberCount));
|
||||||
|
return (host, upstream);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Inbound command ungated by role (T1): the alerts-publish gate must NOT affect inbound
|
/// <summary>Inbound command ungated by role (T1): the alerts-publish gate must NOT affect inbound
|
||||||
/// command processing. Under a Secondary role, an AlarmCommand("Acknowledge") for an owned, active
|
/// command processing. Under a Secondary role, an AlarmCommand("Acknowledge") for an owned, active
|
||||||
/// alarm still drives the engine — observed via the resulting AlarmStateUpdate(Acknowledged=true)
|
/// alarm still drives the engine — observed via the resulting AlarmStateUpdate(Acknowledged=true)
|
||||||
|
|||||||
Reference in New Issue
Block a user