feat(dcl): ServiceLevel-aware OPC UA failover for primary/backup pairs (#138)
Prerequisite for OtOpcUa Warm redundancy (lmxopcua#717): a Warm standby keeps a
healthy session but serves BadOutOfService at a standby ServiceLevel (90/50, boot 1),
so a connection that lands there sat on Bad quality forever.
- OPC UA connections with a backup poll Server_ServiceLevel (i=2267) on the connected
endpoint (IServiceLevelReadable, implemented by OpcUaDataConnection; never raises
Disconnected). A level < 200 sustained for the degraded window (default 10 s), or
the first sample after any (re)connect, triggers a probe of the other endpoint
(short-lived session: connect, read, dispose). Switch only when it reports a
Healthy (>= 200) higher level, and not within the minimum dwell (default 60 s) of
the previous ServiceLevel switch. Both degraded, other unreachable or level
unknown: stay put. 250/240 (today's OtOpcUa) never starts a streak.
- The switch reuses the failover path: SwitchToOtherEndpoint() now shared by all
three triggers, then BecomeReconnecting(ReconnectCause.ServiceLevelFailover) -
zero delay, no unstable accounting, ReSubscribeAll on the new endpoint.
- Config: OpcUaEndpointConfig.ServiceLevelFailover {enabled, degradedWindowSeconds,
minDwellSeconds}, read from the PRIMARY endpoint's JSON only; null = defaults
(enabled, 10, 60). Inside the encrypted protocol JSON, so no schema change or
migration. Central UI: fields in the Backup card next to Failover Retry Count.
Global: ScadaBridge:DataConnection:ServiceLevelPollInterval (2 s) and
ServiceLevelProbeTimeout (15 s).
- Tests: actor tests per trigger/non-trigger with positive controls, an unsolicited
promotion RefreshStart/RefreshEnd test at client and actor level, adapter tests,
and a real in-process OPC UA server pair end to end. All 15 mutants killed.
- Docs: Component-DataConnectionLayer, components/DataConnectionLayer, Component-CLI.
Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
+110
@@ -0,0 +1,110 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138 / OtOpcUa Warm design §5 "Alarms are compatible", at the actor: the promotion
|
||||
/// refresh OtOpcUa sends on its own (an unsolicited RefreshStart / retained conditions / RefreshEnd
|
||||
/// on an EXISTING alarm feed, already framed by <c>RealOpcUaClient</c> as SnapshotStart / Snapshot /
|
||||
/// SnapshotComplete under a new epoch — see <c>RealOpcUaPromotionRefreshTests</c>) goes through the
|
||||
/// existing snapshot handling: it is forwarded in order to the subscriber, and it RECONCILES the
|
||||
/// connection's alarm cache — a condition the promoted node no longer retains is gone from what a
|
||||
/// later subscriber is replayed.
|
||||
/// </summary>
|
||||
public class DataConnectionPromotionRefreshTests : TestKit
|
||||
{
|
||||
private const string Source = "Tank01";
|
||||
|
||||
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
||||
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
|
||||
private readonly DataConnectionOptions _options = new()
|
||||
{
|
||||
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
||||
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
|
||||
WriteTimeout = TimeSpan.FromSeconds(5),
|
||||
};
|
||||
|
||||
private static NativeAlarmTransition Boundary(AlarmTransitionKind kind, Guid epoch) => new(
|
||||
"", Source, "", kind,
|
||||
new AlarmConditionState(false, true, null, AlarmShelveState.Unshelved, false, 0),
|
||||
"", "", "", "", "", null, DateTimeOffset.UtcNow, "", "",
|
||||
SnapshotSourceReference: Source, SnapshotId: epoch);
|
||||
|
||||
private static NativeAlarmTransition Condition(
|
||||
string condition, AlarmTransitionKind kind, Guid? epoch, bool active = true, bool acked = false) => new(
|
||||
$"{Source}.{condition}", Source, "AlarmConditionType", kind,
|
||||
new AlarmConditionState(active, acked, null, AlarmShelveState.Unshelved, false, 500),
|
||||
"", "", "msg", "", "", null, DateTimeOffset.UtcNow, "", "",
|
||||
SnapshotSourceReference: epoch is null ? null : Source, SnapshotId: epoch);
|
||||
|
||||
/// <summary>
|
||||
/// Subscribe-time snapshot {Hi, Lo}; then the unsolicited promotion snapshot {Hi} (Lo cleared
|
||||
/// while the node was Standby). The subscriber sees the promotion frame in order, and a second
|
||||
/// subscriber that joins afterwards is replayed Hi only.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void UnsolicitedPromotionSnapshot_IsForwardedInOrder_AndReconcilesTheCache()
|
||||
{
|
||||
AlarmTransitionCallback? cb = null;
|
||||
var adapter = Substitute.For<IDataConnection, IAlarmSubscribableConnection>();
|
||||
adapter.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.CompletedTask);
|
||||
((IAlarmSubscribableConnection)adapter)
|
||||
.SubscribeAlarmsAsync(Arg.Any<string>(), Arg.Any<string?>(),
|
||||
Arg.Do<AlarmTransitionCallback>(c => cb = c), Arg.Any<CancellationToken>())
|
||||
.Returns(Task.FromResult("alarm-sub-1"));
|
||||
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
||||
"conn", adapter, _options, _health, _factory, "OpcUa")));
|
||||
|
||||
var first = CreateTestProbe();
|
||||
actor.Tell(new SubscribeAlarmsRequest("c1", "inst1", "conn", Source, null, DateTimeOffset.UtcNow), first);
|
||||
first.ExpectMsg<SubscribeAlarmsResponse>(m => m.Success);
|
||||
Assert.NotNull(cb);
|
||||
|
||||
// Subscribe-time refresh on the then-Active node.
|
||||
var subscribeEpoch = Guid.NewGuid();
|
||||
cb!(Boundary(AlarmTransitionKind.SnapshotStart, subscribeEpoch));
|
||||
cb(Condition("Hi", AlarmTransitionKind.Snapshot, subscribeEpoch));
|
||||
cb(Condition("Lo", AlarmTransitionKind.Snapshot, subscribeEpoch));
|
||||
cb(Boundary(AlarmTransitionKind.SnapshotComplete, subscribeEpoch));
|
||||
first.ReceiveN(4, TimeSpan.FromSeconds(3));
|
||||
|
||||
// Promotion on the other node: an unsolicited refresh on this existing feed.
|
||||
var promotionEpoch = Guid.NewGuid();
|
||||
cb(Boundary(AlarmTransitionKind.SnapshotStart, promotionEpoch));
|
||||
cb(Condition("Hi", AlarmTransitionKind.Snapshot, promotionEpoch));
|
||||
cb(Boundary(AlarmTransitionKind.SnapshotComplete, promotionEpoch));
|
||||
|
||||
var frame = first.ReceiveN(3, TimeSpan.FromSeconds(3)).Cast<NativeAlarmTransitionUpdate>().ToList();
|
||||
Assert.Equal(
|
||||
[AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete],
|
||||
frame.Select(u => u.Transition.Kind));
|
||||
Assert.All(frame, u => Assert.Equal(promotionEpoch, u.Transition.SnapshotId));
|
||||
Assert.Equal($"{Source}.Hi", frame[1].Transition.SourceReference);
|
||||
|
||||
// A live event after the promotion is still routed live.
|
||||
cb(Condition("Hi", AlarmTransitionKind.Acknowledge, epoch: null, acked: true));
|
||||
first.ExpectMsg<NativeAlarmTransitionUpdate>(u => u.Transition.Kind == AlarmTransitionKind.Acknowledge);
|
||||
|
||||
// A subscriber joining now is replayed the reconciled state: Hi only, Lo is gone.
|
||||
var second = CreateTestProbe();
|
||||
actor.Tell(new SubscribeAlarmsRequest("c2", "inst2", "conn", Source, null, DateTimeOffset.UtcNow), second);
|
||||
var replay = second.ReceiveN(4, TimeSpan.FromSeconds(3));
|
||||
var transitions = replay.OfType<NativeAlarmTransitionUpdate>().Select(u => u.Transition).ToList();
|
||||
Assert.Single(replay.OfType<SubscribeAlarmsResponse>(), r => r.Success);
|
||||
Assert.Equal(
|
||||
[AlarmTransitionKind.SnapshotStart, AlarmTransitionKind.Snapshot, AlarmTransitionKind.SnapshotComplete],
|
||||
transitions.Select(t => t.Kind));
|
||||
Assert.Equal([$"{Source}.Hi"], transitions.Where(t => t.Kind == AlarmTransitionKind.Snapshot).Select(t => t.SourceReference));
|
||||
second.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
|
||||
}
|
||||
}
|
||||
+377
@@ -0,0 +1,377 @@
|
||||
using System.Diagnostics;
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using NSubstitute;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138 — ServiceLevel-aware failover for a primary/backup OPC UA connection, the
|
||||
/// ScadaBridge prerequisite for OtOpcUa Warm redundancy (lmxopcua#717). Under Warm the standby keeps
|
||||
/// a healthy session but serves <c>BadOutOfService</c> data at a standby ServiceLevel (90 / 50 /
|
||||
/// boot 1); the Active node is 200 or more (250 Primary / 240 Secondary today).
|
||||
///
|
||||
/// <para>
|
||||
/// Each trigger has a matching non-trigger case, and every "no switch" assertion is paired with a
|
||||
/// positive control proving the evaluation it relies on actually ran (the ServiceLevel was read, or
|
||||
/// the other endpoint was probed) — an inert feature would otherwise pass them all.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public class DataConnectionServiceLevelFailoverTests : TestKit
|
||||
{
|
||||
private const string Primary = "opc.tcp://otopcua-a:4840";
|
||||
private const string Backup = "opc.tcp://otopcua-b:4840";
|
||||
private const string Tag = "nsu=https://zb.com/otopcua/uns;s=Line1.Pump.Speed";
|
||||
|
||||
private readonly ServiceLevelRig _rig = new();
|
||||
private readonly ISiteHealthCollector _health = Substitute.For<ISiteHealthCollector>();
|
||||
private readonly IDataConnectionFactory _factory = Substitute.For<IDataConnectionFactory>();
|
||||
private readonly DataConnectionOptions _options = new()
|
||||
{
|
||||
ReconnectInterval = TimeSpan.FromMilliseconds(100),
|
||||
TagResolutionRetryInterval = TimeSpan.FromMilliseconds(200),
|
||||
WriteTimeout = TimeSpan.FromSeconds(5),
|
||||
SeedReadMaxAttempts = 1,
|
||||
SeedReadRetryDelay = TimeSpan.FromMilliseconds(10),
|
||||
ServiceLevelPollInterval = TimeSpan.FromMilliseconds(50),
|
||||
ServiceLevelProbeTimeout = TimeSpan.FromSeconds(2),
|
||||
};
|
||||
|
||||
/// <summary>Wires the factory to build a rig-bound fake for whichever endpoint it is asked for.</summary>
|
||||
public DataConnectionServiceLevelFailoverTests()
|
||||
: base(@"akka.loglevel = DEBUG")
|
||||
{
|
||||
_factory.Create(Arg.Any<string>(), Arg.Any<IDictionary<string, string>>())
|
||||
.Returns(ci => new FakeServiceLevelConnection(_rig, ((IDictionary<string, string>)ci[1])["Endpoint"]));
|
||||
}
|
||||
|
||||
private static Dictionary<string, string> Config(
|
||||
string endpoint, bool? enabled = null, int? windowSeconds = null, int? dwellSeconds = null)
|
||||
{
|
||||
var config = new Dictionary<string, string> { ["Endpoint"] = endpoint };
|
||||
if (enabled is { } e) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverEnabledKey] = e.ToString();
|
||||
if (windowSeconds is { } w) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverWindowKey] = w.ToString();
|
||||
if (dwellSeconds is { } d) config[OpcUaEndpointConfigSerializer.ServiceLevelFailoverDwellKey] = d.ToString();
|
||||
return config;
|
||||
}
|
||||
|
||||
private IActorRef Start(
|
||||
string name, IDictionary<string, string> primaryConfig, IDictionary<string, string>? backupConfig,
|
||||
IDataConnection? initialAdapter = null)
|
||||
{
|
||||
var adapter = initialAdapter ?? new FakeServiceLevelConnection(_rig, Primary);
|
||||
var actor = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
||||
name, adapter, _options, _health, _factory, "OpcUa", primaryConfig, backupConfig, 3, null)), name);
|
||||
actor.Tell(new SubscribeTagsRequest("c1", "inst1", name, [Tag], DateTimeOffset.UtcNow));
|
||||
ExpectMsg<SubscribeTagsResponse>(TimeSpan.FromSeconds(5));
|
||||
return actor;
|
||||
}
|
||||
|
||||
private string ActiveEndpoint(IActorRef actor) =>
|
||||
actor.Ask<DataConnectionHealthReport>(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(3))
|
||||
.GetAwaiter().GetResult().ActiveEndpoint;
|
||||
|
||||
private bool LiveOn(string endpoint) => _rig.Subscribes.Contains($"{endpoint}|{Tag}");
|
||||
|
||||
/// <summary>
|
||||
/// Warm promotion seen from a connection on the old Active: it was Healthy (250), then the
|
||||
/// node becomes the Standby (90) while the other node is promoted (250). The switch waits for
|
||||
/// the degraded window, then lands on the backup, where every subscription is re-issued.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Sustained_Degraded_FailsOver_AfterTheWindow_AndResubscribesOnTheOtherEndpoint()
|
||||
{
|
||||
_rig.SetLevel(Primary, 250);
|
||||
_rig.SetLevel(Backup, 240);
|
||||
var actor = Start("sl-window", Config(Primary, windowSeconds: 2, dwellSeconds: 0), Config(Backup));
|
||||
AwaitCondition(() => _rig.ReadCount(Primary) >= 3, TimeSpan.FromSeconds(5));
|
||||
|
||||
var degradedAt = Stopwatch.StartNew();
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
|
||||
// Not before the window: half-way through, still on the primary and nothing probed.
|
||||
Thread.Sleep(1000);
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
Assert.Equal(0, _rig.ConnectCount(Backup));
|
||||
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
Assert.True(degradedAt.Elapsed >= TimeSpan.FromSeconds(1.9),
|
||||
$"switched after {degradedAt.Elapsed.TotalSeconds:F2}s, before the 2s window");
|
||||
Assert.Equal("Backup", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Connect-time preference: the connection starts on a primary that is the Warm standby (90)
|
||||
/// while the backup is Active (250). It leaves at once — no 30 s window — because the first
|
||||
/// sample of a session is evaluated immediately.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Connect_OnDegradedEndpoint_PrefersTheHealthierOther_WithoutWaitingTheWindow()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var sw = Stopwatch.StartNew();
|
||||
var actor = Start("sl-connect", Config(Primary, windowSeconds: 30), Config(Backup));
|
||||
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
Assert.True(sw.Elapsed < TimeSpan.FromSeconds(10), $"took {sw.Elapsed.TotalSeconds:F1}s");
|
||||
Assert.Equal("Backup", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>Booting node (ServiceLevel 1) on connect, Active other → leave it too.</summary>
|
||||
[Fact]
|
||||
public void Connect_OnBootingEndpoint_ServiceLevel1_PrefersTheActiveOther()
|
||||
{
|
||||
_rig.SetLevel(Primary, 1);
|
||||
_rig.SetLevel(Backup, 240);
|
||||
var actor = Start("sl-boot", Config(Primary, windowSeconds: 30), Config(Backup));
|
||||
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("Backup", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Today's OtOpcUa pair: both nodes Active-band, 250/240 in either order. Nothing is ever
|
||||
/// probed and nothing switches — a 240 Secondary never triggers. Positive control: the
|
||||
/// connected endpoint's ServiceLevel WAS read many times across more than a full window.
|
||||
/// </summary>
|
||||
/// <param name="primaryLevel">The connected (primary) endpoint's ServiceLevel.</param>
|
||||
/// <param name="backupLevel">The backup endpoint's ServiceLevel.</param>
|
||||
[Theory]
|
||||
[InlineData(240, 250)]
|
||||
[InlineData(250, 240)]
|
||||
[InlineData(200, 250)]
|
||||
public void ActiveBandPair_NeverSwitches_NorProbes(byte primaryLevel, byte backupLevel)
|
||||
{
|
||||
_rig.SetLevel(Primary, primaryLevel);
|
||||
_rig.SetLevel(Backup, backupLevel);
|
||||
var actor = Start($"sl-today-{primaryLevel}-{backupLevel}", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
|
||||
Thread.Sleep(2000);
|
||||
Assert.True(_rig.ReadCount(Primary) >= 10, $"only {_rig.ReadCount(Primary)} reads — the watch did not run");
|
||||
Assert.Equal(0, _rig.ConnectCount(Backup));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Both endpoints degraded (Warm standby 90 vs standby-degraded 50, either way round): stay
|
||||
/// put. Positive control: the other endpoint WAS probed, so the verdict, not a dead watch,
|
||||
/// kept the connection where it is.
|
||||
/// </summary>
|
||||
/// <param name="primaryLevel">The connected (primary) endpoint's ServiceLevel.</param>
|
||||
/// <param name="backupLevel">The backup endpoint's ServiceLevel.</param>
|
||||
[Theory]
|
||||
[InlineData(50, 90)]
|
||||
[InlineData(90, 50)]
|
||||
[InlineData(90, 199)]
|
||||
public void BothDegraded_StaysPut_AfterProbingTheOther(byte primaryLevel, byte backupLevel)
|
||||
{
|
||||
_rig.SetLevel(Primary, primaryLevel);
|
||||
_rig.SetLevel(Backup, backupLevel);
|
||||
var actor = Start($"sl-both-{primaryLevel}-{backupLevel}", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
|
||||
AwaitCondition(() => _rig.ReadCount(Backup) >= 2, TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
Assert.False(LiveOn(Backup));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The other endpoint is down: stay on the degraded one (it still answers). Positive control:
|
||||
/// sessions to the backup were attempted.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void OtherEndpointUnreachable_StaysPut()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
_rig.SetReachable(Backup, false);
|
||||
var actor = Start("sl-unreachable", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
|
||||
AwaitCondition(() => _rig.ConnectCount(Backup) >= 2, TimeSpan.FromSeconds(5));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
Assert.False(LiveOn(Backup));
|
||||
|
||||
// ...and once it comes back, the next window moves the connection.
|
||||
_rig.SetReachable(Backup, true);
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An unreadable ServiceLevel (Bad status, server without the node) is unknown, never
|
||||
/// Degraded: no probe, no switch. Positive control: the reads happened.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void UnknownServiceLevel_IsNotDegraded()
|
||||
{
|
||||
_rig.SetLevel(Primary, (byte?)null);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-unknown", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
|
||||
Thread.Sleep(2000);
|
||||
Assert.True(_rig.ReadCount(Primary) >= 10);
|
||||
Assert.Equal(0, _rig.ConnectCount(Backup));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Hysteresis: a level that keeps dipping below 200 but recovers between samples never forms
|
||||
/// an unbroken window, so it never switches. Positive control: Degraded samples were seen.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void FlickeringLevel_NeverCompletesTheWindow()
|
||||
{
|
||||
var reads = 0;
|
||||
// Healthy for the first samples, so the connect-time check is not what is exercised; then
|
||||
// three Degraded samples out of every four.
|
||||
_rig.SetLevel(Primary, () =>
|
||||
{
|
||||
var n = Interlocked.Increment(ref reads);
|
||||
return n <= 3 || n % 4 == 0 ? (byte)250 : (byte)90;
|
||||
});
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-flicker", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
|
||||
Thread.Sleep(2500);
|
||||
Assert.True(Volatile.Read(ref reads) >= 20);
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
Assert.False(LiveOn(Backup));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Anti-flap: right after a ServiceLevel switch the roles flip again. The connection does not
|
||||
/// bounce straight back — not even through the connect-time check — until the dwell has passed;
|
||||
/// then it does.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Dwell_BlocksAnImmediateSwitchBack_ThenAllowsIt()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-dwell", Config(Primary, windowSeconds: 1, dwellSeconds: 8), Config(Backup));
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(10));
|
||||
var switchedAt = Stopwatch.StartNew();
|
||||
|
||||
// Roles flip back at once. The dwell runs from the switch decision, which comes before the
|
||||
// observed re-subscribe on the backup by a reconnect + subscribe round trip, so the
|
||||
// "blocked" window checked here (3 s) leaves wide slack inside the 8 s dwell under load.
|
||||
_rig.SetLevel(Primary, 250);
|
||||
_rig.SetLevel(Backup, 90);
|
||||
var primaryConnectsAfterSwitch = _rig.ConnectCount(Primary);
|
||||
|
||||
Thread.Sleep(3000);
|
||||
Assert.Equal("Backup", ActiveEndpoint(actor));
|
||||
Assert.Equal(primaryConnectsAfterSwitch, _rig.ConnectCount(Primary)); // not even probed
|
||||
|
||||
// Positive control: once the dwell is over, it does switch back.
|
||||
AwaitCondition(() => ActiveEndpoint(actor) == "Primary", TimeSpan.FromSeconds(20));
|
||||
Assert.True(switchedAt.Elapsed >= TimeSpan.FromSeconds(5),
|
||||
$"switched back after {switchedAt.Elapsed.TotalSeconds:F2}s, well inside the 8s dwell");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// <c>ServiceLevelFailover.Enabled=False</c> on the primary: the ServiceLevel is never even
|
||||
/// read, whatever it is.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void Disabled_NeverReadsTheServiceLevel()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-disabled", Config(Primary, enabled: false, windowSeconds: 1), Config(Backup));
|
||||
|
||||
Thread.Sleep(1500);
|
||||
Assert.Equal(0, _rig.ReadCount(Primary));
|
||||
Assert.Equal(0, _rig.ConnectCount(Backup));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>Only the PRIMARY's copy of the setting counts: disabling it on the backup changes nothing.</summary>
|
||||
[Fact]
|
||||
public void BackupCopyOfTheSetting_IsIgnored()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
Start("sl-backup-copy", Config(Primary, windowSeconds: 1), Config(Backup, enabled: false));
|
||||
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
}
|
||||
|
||||
/// <summary>A connection without a backup has nowhere to go: the ServiceLevel is never read.</summary>
|
||||
[Fact]
|
||||
public void NoBackup_NeverReadsTheServiceLevel()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
var actor = Start("sl-nobackup", Config(Primary, windowSeconds: 1), backupConfig: null);
|
||||
|
||||
Thread.Sleep(1500);
|
||||
Assert.True(LiveOn(Primary)); // positive control: the session is up and subscribed
|
||||
Assert.Equal(0, _rig.ReadCount(Primary));
|
||||
Assert.Equal("Primary (no backup)", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Symmetry: sitting on the backup (after a connectivity failover) when the backup becomes the
|
||||
/// Warm standby moves the connection back to the primary.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void OnBackup_DegradedBackup_FailsBackToAHealthyPrimary()
|
||||
{
|
||||
_rig.SetLevel(Primary, 90);
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-symmetric", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup));
|
||||
AwaitCondition(() => LiveOn(Backup), TimeSpan.FromSeconds(5));
|
||||
var primarySubscribes = _rig.Subscribes.Count(s => s.StartsWith(Primary + "|", StringComparison.Ordinal));
|
||||
|
||||
_rig.SetLevel(Primary, 250);
|
||||
_rig.SetLevel(Backup, 90);
|
||||
|
||||
AwaitCondition(() => _rig.Subscribes.Count(s => s.StartsWith(Primary + "|", StringComparison.Ordinal)) > primarySubscribes,
|
||||
TimeSpan.FromSeconds(6));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An adapter without <see cref="IServiceLevelReadable"/> (MxGateway, or any plain adapter) is
|
||||
/// never watched and never probed, even with a backup and the feature enabled.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void AdapterWithoutTheCapability_IsNeverWatched()
|
||||
{
|
||||
var plain = Substitute.For<IDataConnection>();
|
||||
plain.ConnectAsync(Arg.Any<IDictionary<string, string>>(), Arg.Any<CancellationToken>()).Returns(Task.CompletedTask);
|
||||
plain.SubscribeAsync(Arg.Any<string>(), Arg.Any<SubscriptionCallback>(), Arg.Any<CancellationToken>())
|
||||
.Returns("sub-1");
|
||||
plain.ReadAsync(Arg.Any<string>(), Arg.Any<CancellationToken>())
|
||||
.Returns(new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null));
|
||||
_rig.SetLevel(Backup, 250);
|
||||
var actor = Start("sl-plain", Config(Primary, windowSeconds: 1, dwellSeconds: 0), Config(Backup), plain);
|
||||
|
||||
Thread.Sleep(1500);
|
||||
Assert.Equal(0, _rig.ConnectCount(Backup));
|
||||
Assert.Equal("Primary", ActiveEndpoint(actor));
|
||||
}
|
||||
|
||||
/// <summary>The policy parser: defaults when absent, and clamps nonsense to safe values.</summary>
|
||||
[Fact]
|
||||
public void Policy_DefaultsAndClamps()
|
||||
{
|
||||
var defaults = ServiceLevelFailoverPolicy.From(new Dictionary<string, string> { ["Endpoint"] = Primary });
|
||||
Assert.True(defaults.Enabled);
|
||||
Assert.Equal(TimeSpan.FromSeconds(10), defaults.DegradedWindow);
|
||||
Assert.Equal(TimeSpan.FromSeconds(60), defaults.MinDwell);
|
||||
|
||||
var clamped = ServiceLevelFailoverPolicy.From(Config(Primary, enabled: false, windowSeconds: 0, dwellSeconds: -5));
|
||||
Assert.False(clamped.Enabled);
|
||||
Assert.Equal(TimeSpan.FromSeconds(1), clamped.DegradedWindow);
|
||||
Assert.Equal(TimeSpan.Zero, clamped.MinDwell);
|
||||
|
||||
Assert.True(ServiceLevelFailoverPolicy.From(null).Enabled);
|
||||
}
|
||||
}
|
||||
+119
@@ -0,0 +1,119 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// The two OPC UA servers of a redundant pair as the ScadaBridge#138 tests see them: a
|
||||
/// ServiceLevel per endpoint (changeable at any time, null = unreadable), whether each endpoint
|
||||
/// accepts a session, and a record of what the actor did to them.
|
||||
/// </summary>
|
||||
internal sealed class ServiceLevelRig
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, Func<byte?>> _levels = new();
|
||||
private readonly ConcurrentDictionary<string, bool> _unreachable = new();
|
||||
|
||||
/// <summary>Every endpoint a session was opened on, in order — live sessions and probes alike.</summary>
|
||||
public ConcurrentQueue<string> Connects { get; } = new();
|
||||
|
||||
/// <summary>Every tag subscribed, as "endpoint|tag" — only a live session receives these.</summary>
|
||||
public ConcurrentQueue<string> Subscribes { get; } = new();
|
||||
|
||||
/// <summary>Number of ServiceLevel reads, per endpoint.</summary>
|
||||
public ConcurrentDictionary<string, int> Reads { get; } = new();
|
||||
|
||||
/// <summary>Sets a fixed ServiceLevel for <paramref name="endpoint"/>.</summary>
|
||||
public void SetLevel(string endpoint, byte? level) => _levels[endpoint] = () => level;
|
||||
|
||||
/// <summary>Sets a ServiceLevel source evaluated on every read of <paramref name="endpoint"/>.</summary>
|
||||
public void SetLevel(string endpoint, Func<byte?> level) => _levels[endpoint] = level;
|
||||
|
||||
/// <summary>Makes <paramref name="endpoint"/> refuse (or accept again) new sessions.</summary>
|
||||
public void SetReachable(string endpoint, bool reachable) => _unreachable[endpoint] = !reachable;
|
||||
|
||||
public bool IsReachable(string endpoint) => !_unreachable.GetValueOrDefault(endpoint);
|
||||
|
||||
public byte? Read(string endpoint)
|
||||
{
|
||||
Reads.AddOrUpdate(endpoint, 1, (_, n) => n + 1);
|
||||
return _levels.TryGetValue(endpoint, out var level) ? level() : null;
|
||||
}
|
||||
|
||||
public int ConnectCount(string endpoint) => Connects.Count(e => e == endpoint);
|
||||
|
||||
public int ReadCount(string endpoint) => Reads.GetValueOrDefault(endpoint);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// An <see cref="IDataConnection"/> with the <see cref="IServiceLevelReadable"/> capability, bound to
|
||||
/// one endpoint of a <see cref="ServiceLevelRig"/>. Subscriptions and reads always succeed.
|
||||
/// </summary>
|
||||
internal sealed class FakeServiceLevelConnection : IDataConnection, IServiceLevelReadable
|
||||
{
|
||||
private readonly ServiceLevelRig _rig;
|
||||
private volatile bool _connected;
|
||||
|
||||
public FakeServiceLevelConnection(ServiceLevelRig rig, string endpoint)
|
||||
{
|
||||
_rig = rig;
|
||||
Endpoint = endpoint;
|
||||
}
|
||||
|
||||
public string Endpoint { get; }
|
||||
|
||||
public ConnectionHealth Status => _connected ? ConnectionHealth.Connected : ConnectionHealth.Disconnected;
|
||||
|
||||
public event Action? Disconnected;
|
||||
|
||||
public Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_rig.Connects.Enqueue(Endpoint);
|
||||
if (!_rig.IsReachable(Endpoint))
|
||||
return Task.FromException(new InvalidOperationException($"{Endpoint} refused the session"));
|
||||
_connected = true;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DisconnectAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
_connected = false;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task<byte?> ReadServiceLevelAsync(CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(_connected ? _rig.Read(Endpoint) : null);
|
||||
|
||||
public Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_rig.Subscribes.Enqueue($"{Endpoint}|{tagPath}");
|
||||
return Task.FromResult($"{Endpoint}:{tagPath}");
|
||||
}
|
||||
|
||||
public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) => Task.CompletedTask;
|
||||
|
||||
public Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null));
|
||||
|
||||
public Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult<IReadOnlyDictionary<string, ReadResult>>(tagPaths.ToDictionary(
|
||||
t => t, _ => new ReadResult(true, new TagValue(1, QualityCode.Good, DateTimeOffset.UtcNow), null)));
|
||||
|
||||
public Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult(new WriteResult(true, null));
|
||||
|
||||
public Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default) =>
|
||||
Task.FromResult<IReadOnlyDictionary<string, WriteResult>>(values.ToDictionary(v => v.Key, _ => new WriteResult(true, null)));
|
||||
|
||||
public void RaiseDisconnected()
|
||||
{
|
||||
_connected = false;
|
||||
Disconnected?.Invoke();
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync()
|
||||
{
|
||||
_connected = false;
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
+211
@@ -0,0 +1,211 @@
|
||||
using Akka.Actor;
|
||||
using Akka.TestKit.Xunit2;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
using NSubstitute;
|
||||
using NSubstitute.ExceptionExtensions;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
|
||||
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138 at the OPC UA adapter: <see cref="OpcUaDataConnection.ReadServiceLevelAsync"/>
|
||||
/// against a substitute client (status / failure mapping) and against real in-process OPC UA
|
||||
/// servers (the actual <c>i=2267</c> node), plus the whole failover through the real
|
||||
/// <see cref="DataConnectionFactory"/> against a real server pair.
|
||||
/// </summary>
|
||||
public sealed class OpcUaServiceLevelTests : TestKit
|
||||
{
|
||||
private readonly string _clientPki =
|
||||
Path.Combine(Path.GetTempPath(), "sb138-client", Guid.NewGuid().ToString("N"));
|
||||
|
||||
/// <summary>Initializes the test kit.</summary>
|
||||
public OpcUaServiceLevelTests() : base(@"akka.loglevel = DEBUG") { }
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
base.Dispose(disposing);
|
||||
try { Directory.Delete(_clientPki, recursive: true); } catch { /* best-effort */ }
|
||||
}
|
||||
|
||||
private OpcUaGlobalOptions ClientOptions() => new()
|
||||
{
|
||||
ApplicationName = "SB138Client",
|
||||
ApplicationUri = "urn:localhost:SB138Client",
|
||||
OwnCertificateStorePath = Path.Combine(_clientPki, "own"),
|
||||
TrustedPeerStorePath = Path.Combine(_clientPki, "trusted"),
|
||||
TrustedIssuerStorePath = Path.Combine(_clientPki, "issuers"),
|
||||
RejectedCertificateStorePath = Path.Combine(_clientPki, "rejected"),
|
||||
AutoCreateApplicationCertificate = true,
|
||||
};
|
||||
|
||||
private static IDictionary<string, string> Flat(string url, OpcUaServiceLevelFailoverConfig? sl = null) =>
|
||||
OpcUaEndpointConfigSerializer.ToFlatDict(new OpcUaEndpointConfig
|
||||
{
|
||||
EndpointUrl = url,
|
||||
SecurityMode = OpcUaSecurityMode.None,
|
||||
ServiceLevelFailover = sl,
|
||||
});
|
||||
|
||||
private static (OpcUaDataConnection Adapter, IOpcUaClient Client) WithSubstituteClient()
|
||||
{
|
||||
var client = Substitute.For<IOpcUaClient>();
|
||||
client.IsConnected.Returns(true);
|
||||
var factory = Substitute.For<IOpcUaClientFactory>();
|
||||
factory.Create().Returns(client);
|
||||
return (new OpcUaDataConnection(factory, NullLogger<OpcUaDataConnection>.Instance), client);
|
||||
}
|
||||
|
||||
// ── Status / failure mapping (substitute client) ──
|
||||
|
||||
/// <summary>The value comes from <c>i=2267</c>; a Good read of a byte is that byte.</summary>
|
||||
[Fact]
|
||||
public async Task Read_GoodByte_IsTheLevel_FromTheServiceLevelNode()
|
||||
{
|
||||
var (adapter, client) = WithSubstituteClient();
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.Returns(((object?)(byte)90, DateTime.UtcNow, 0u));
|
||||
await adapter.ConnectAsync(new Dictionary<string, string>());
|
||||
|
||||
Assert.Equal((byte)90, await adapter.ReadServiceLevelAsync());
|
||||
}
|
||||
|
||||
/// <summary>A Bad status, or a value that is not a byte-ranged integer, is unknown (null).</summary>
|
||||
[Fact]
|
||||
public async Task Read_BadStatusOrNonByteValue_IsUnknown()
|
||||
{
|
||||
var (adapter, client) = WithSubstituteClient();
|
||||
await adapter.ConnectAsync(new Dictionary<string, string>());
|
||||
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.Returns(((object?)(byte)250, DateTime.UtcNow, 0x80340000u)); // BadNodeIdUnknown
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.Returns(((object?)"250", DateTime.UtcNow, 0u));
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.Returns(((object?)300, DateTime.UtcNow, 0u));
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.Returns(((object?)240, DateTime.UtcNow, 0u)); // Int32 in range → accepted
|
||||
Assert.Equal((byte)240, await adapter.ReadServiceLevelAsync());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A failed read is unknown and — unlike <see cref="OpcUaDataConnection.ReadAsync"/> — never
|
||||
/// raises <c>Disconnected</c>: the poll must not drive the reconnect machine.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task Read_Failure_IsUnknown_AndNeverRaisesDisconnected()
|
||||
{
|
||||
var (adapter, client) = WithSubstituteClient();
|
||||
client.ReadValueAsync("i=2267", Arg.Any<CancellationToken>())
|
||||
.ThrowsAsync(new InvalidOperationException("session dropped"));
|
||||
var disconnected = 0;
|
||||
adapter.Disconnected += () => disconnected++;
|
||||
await adapter.ConnectAsync(new Dictionary<string, string>());
|
||||
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
Assert.Equal(0, disconnected);
|
||||
|
||||
// Positive control: the ordinary read path DOES raise it for the same failure.
|
||||
client.ReadValueAsync("ns=2;s=x", Arg.Any<CancellationToken>())
|
||||
.ThrowsAsync(new InvalidOperationException("session dropped"));
|
||||
await Assert.ThrowsAsync<InvalidOperationException>(() => adapter.ReadAsync("ns=2;s=x"));
|
||||
Assert.Equal(1, disconnected);
|
||||
}
|
||||
|
||||
/// <summary>Before connect (no client) the level is unknown, and no read is attempted.</summary>
|
||||
[Fact]
|
||||
public async Task Read_NotConnected_IsUnknown()
|
||||
{
|
||||
var (adapter, client) = WithSubstituteClient();
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
|
||||
await adapter.ConnectAsync(new Dictionary<string, string>());
|
||||
client.IsConnected.Returns(false);
|
||||
Assert.Null(await adapter.ReadServiceLevelAsync());
|
||||
await client.DidNotReceive().ReadValueAsync(Arg.Any<string>(), Arg.Any<CancellationToken>());
|
||||
}
|
||||
|
||||
// ── Real OPC UA servers ──
|
||||
|
||||
/// <summary>The real adapter reads the real server's <c>Server_ServiceLevel</c> as it changes.</summary>
|
||||
[Fact]
|
||||
public async Task RealServer_ReadServiceLevel_TracksTheServersValue()
|
||||
{
|
||||
await using var server = await SecureOpcUaTestServer.StartAsync();
|
||||
await using var adapter = new OpcUaDataConnection(
|
||||
new RealOpcUaClientFactory(ClientOptions()), NullLogger<OpcUaDataConnection>.Instance);
|
||||
await adapter.ConnectAsync(Flat(server.Url));
|
||||
|
||||
foreach (var level in new byte[] { 250, 240, 90, 50, 1 })
|
||||
{
|
||||
server.SetServiceLevel(level);
|
||||
Assert.Equal(level, await adapter.ReadServiceLevelAsync());
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// End to end on two real servers through the production factory: the connection starts on a
|
||||
/// primary that is the Warm standby (90) while the backup is Active (250), and ends up on the
|
||||
/// backup. Then the mirror case on a fresh connection — today's 250/240 pair — stays on the
|
||||
/// primary.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public async Task RealServerPair_StandbyPrimary_FailsOverToTheActiveBackup_ActivePair_StaysPut()
|
||||
{
|
||||
await using var a = await SecureOpcUaTestServer.StartAsync();
|
||||
await using var b = await SecureOpcUaTestServer.StartAsync();
|
||||
var factory = new DataConnectionFactory(NullLoggerFactory.Instance, Options.Create(ClientOptions()));
|
||||
var options = new DataConnectionOptions
|
||||
{
|
||||
ReconnectInterval = TimeSpan.FromMilliseconds(200),
|
||||
ServiceLevelPollInterval = TimeSpan.FromMilliseconds(100),
|
||||
ServiceLevelProbeTimeout = TimeSpan.FromSeconds(20),
|
||||
};
|
||||
var sl = new OpcUaServiceLevelFailoverConfig { DegradedWindowSeconds = 1, MinDwellSeconds = 0 };
|
||||
|
||||
async Task<string> Endpoint(IActorRef actor) =>
|
||||
(await actor.Ask<DataConnectionHealthReport>(new DataConnectionActor.GetHealthReport(), TimeSpan.FromSeconds(5)))
|
||||
.ActiveEndpoint;
|
||||
|
||||
// Warm: A is the standby, B the Active.
|
||||
a.SetServiceLevel(90);
|
||||
b.SetServiceLevel(250);
|
||||
var health = Substitute.For<ISiteHealthCollector>();
|
||||
var primary = Flat(a.Url, sl);
|
||||
var warm = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
||||
"warm", factory.Create("OpcUa", primary), options, health, factory, "OpcUa",
|
||||
primary, Flat(b.Url), 3, null)));
|
||||
await AwaitConditionAsync(async () => await Endpoint(warm) == "Backup", TimeSpan.FromSeconds(30));
|
||||
AwaitCondition(() => health.ReceivedCalls().Any(c =>
|
||||
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint)
|
||||
&& c.GetArguments() is ["warm", "Connected to backup"]), TimeSpan.FromSeconds(30));
|
||||
Sys.Stop(warm);
|
||||
|
||||
// Today: A is the Secondary (240), B the Primary (250) — both Active band.
|
||||
a.SetServiceLevel(240);
|
||||
var today = Sys.ActorOf(Props.Create(() => new DataConnectionActor(
|
||||
"today", factory.Create("OpcUa", primary), options, health, factory, "OpcUa",
|
||||
primary, Flat(b.Url), 3, null)));
|
||||
AwaitCondition(() => health.ReceivedCalls().Any(c =>
|
||||
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint)
|
||||
&& c.GetArguments() is ["today", "Connected to primary"]), TimeSpan.FromSeconds(30));
|
||||
await Task.Delay(TimeSpan.FromSeconds(2.5)); // > window, ~25 polls
|
||||
Assert.Equal("Primary", await Endpoint(today));
|
||||
Assert.DoesNotContain(health.ReceivedCalls(), c =>
|
||||
c.GetMethodInfo().Name == nameof(ISiteHealthCollector.UpdateConnectionEndpoint)
|
||||
&& c.GetArguments() is ["today", "Connected to backup"]);
|
||||
}
|
||||
}
|
||||
+112
@@ -0,0 +1,112 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Reflection;
|
||||
using Opc.Ua;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138 / OtOpcUa Warm design §4.5 + §5 "Alarms are compatible": when a Warm standby
|
||||
/// is promoted, OtOpcUa runs a ConditionRefresh on every live event item ON ITS OWN — the client
|
||||
/// never asked. The existing items therefore receive an unsolicited RefreshStart, the retained
|
||||
/// conditions, RefreshEnd, then new events. ScadaBridge must treat that exactly like the refresh it
|
||||
/// triggers itself on subscribe: one framed snapshot with its own epoch, and live derivation
|
||||
/// before and after it.
|
||||
/// </summary>
|
||||
public class RealOpcUaPromotionRefreshTests
|
||||
{
|
||||
private const string Handle = "h1";
|
||||
private const string Source = "ns=2;s=Tank01";
|
||||
|
||||
private static readonly MethodInfo Handler =
|
||||
typeof(RealOpcUaClient).GetMethod("HandleAlarmEvent", BindingFlags.Instance | BindingFlags.NonPublic)!;
|
||||
|
||||
/// <summary>What <c>CreateAlarmSubscriptionAsync</c> sets up for a live handle: its per-condition state map.</summary>
|
||||
private static void SeedSubscribedHandle(RealOpcUaClient client)
|
||||
{
|
||||
var field = typeof(RealOpcUaClient).GetField("_alarmLastState", BindingFlags.Instance | BindingFlags.NonPublic)!;
|
||||
var map = (ConcurrentDictionary<string, Dictionary<string, (bool Active, bool Acked)>>)field.GetValue(client)!;
|
||||
map[Handle] = new Dictionary<string, (bool Active, bool Acked)>(StringComparer.Ordinal);
|
||||
}
|
||||
|
||||
private static void Boundary(RealOpcUaClient client, NodeId type, List<NativeAlarmTransition> sink) =>
|
||||
Handler.Invoke(client, [Handle, Source,
|
||||
new EventFieldList { EventFields = new Variant[] { Variant.From(type), Variant.Null, Variant.Null, Variant.Null, Variant.Null, Variant.Null } },
|
||||
(Action<NativeAlarmTransition>)sink.Add]);
|
||||
|
||||
private static void Condition(RealOpcUaClient client, string condition, bool active, bool acked, List<NativeAlarmTransition> sink)
|
||||
{
|
||||
var fields = new Variant[]
|
||||
{
|
||||
Variant.From(ObjectTypeIds.AlarmConditionType), // 0 EventType
|
||||
Variant.From(new NodeId("Tank01", 2)), // 1 SourceNode
|
||||
Variant.From("Tank01"), // 2 SourceName
|
||||
Variant.From(DateTime.UtcNow), // 3 Time
|
||||
Variant.From(new LocalizedText("msg")), // 4 Message
|
||||
Variant.From((ushort)500), // 5 Severity
|
||||
Variant.From(active), // 6 ActiveState/Id
|
||||
Variant.From(acked), // 7 AckedState/Id
|
||||
Variant.Null, // 8 ConfirmedState/Id
|
||||
Variant.From(false), // 9 SuppressedOrShelved
|
||||
Variant.Null, // 10 ShelvingState
|
||||
Variant.From(condition), // 11 ConditionName
|
||||
};
|
||||
Handler.Invoke(client, [Handle, Source, new EventFieldList { EventFields = fields },
|
||||
(Action<NativeAlarmTransition>)sink.Add]);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Subscribe-time refresh, live traffic, then the UNSOLICITED promotion refresh on the same
|
||||
/// item, then live traffic again. The promotion replay is framed SnapshotStart → Snapshot… →
|
||||
/// SnapshotComplete under a fresh epoch (distinct from the subscribe-time one), and live events
|
||||
/// after it are derived live again — the replay does not leave the item stuck in Snapshot mode.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void UnsolicitedPromotionRefresh_IsFramedAsItsOwnSnapshot_AndLiveDerivationResumes()
|
||||
{
|
||||
var client = new RealOpcUaClient();
|
||||
SeedSubscribedHandle(client);
|
||||
var sink = new List<NativeAlarmTransition>();
|
||||
|
||||
// Subscribe-time refresh (the one ScadaBridge triggers): Hi is active.
|
||||
Boundary(client, ObjectTypeIds.RefreshStartEventType, sink);
|
||||
Condition(client, "Hi", active: true, acked: false, sink);
|
||||
Boundary(client, ObjectTypeIds.RefreshEndEventType, sink);
|
||||
|
||||
// Live while the old Active serves: Lo raises.
|
||||
Condition(client, "Lo", active: true, acked: false, sink);
|
||||
var liveBefore = sink[^1];
|
||||
|
||||
// Promotion on the other node: an unsolicited refresh on the existing item.
|
||||
Boundary(client, ObjectTypeIds.RefreshStartEventType, sink);
|
||||
Condition(client, "Hi", active: true, acked: false, sink);
|
||||
Condition(client, "Lo", active: true, acked: false, sink);
|
||||
Boundary(client, ObjectTypeIds.RefreshEndEventType, sink);
|
||||
|
||||
// Live after promotion: Lo is acknowledged.
|
||||
Condition(client, "Lo", active: true, acked: true, sink);
|
||||
var liveAfter = sink[^1];
|
||||
|
||||
Assert.Equal(
|
||||
["SnapshotStart", "Snapshot", "SnapshotComplete", liveBefore.Kind.ToString(),
|
||||
"SnapshotStart", "Snapshot", "Snapshot", "SnapshotComplete", liveAfter.Kind.ToString()],
|
||||
sink.Select(t => t.Kind.ToString()));
|
||||
|
||||
// Live kinds really are live (not Snapshot) on both sides of the promotion refresh.
|
||||
Assert.Equal(AlarmTransitionKind.Raise, liveBefore.Kind);
|
||||
Assert.Equal(AlarmTransitionKind.Acknowledge, liveAfter.Kind);
|
||||
Assert.Null(liveBefore.SnapshotId);
|
||||
Assert.Null(liveAfter.SnapshotId);
|
||||
|
||||
// The promotion replay has its own epoch, shared by its boundaries and its conditions.
|
||||
var subscribeEpoch = sink[0].SnapshotId;
|
||||
var promotionEpoch = sink[4].SnapshotId;
|
||||
Assert.NotNull(subscribeEpoch);
|
||||
Assert.NotNull(promotionEpoch);
|
||||
Assert.NotEqual(subscribeEpoch, promotionEpoch);
|
||||
Assert.All(sink.Skip(4).Take(4), t => Assert.Equal(promotionEpoch, t.SnapshotId));
|
||||
Assert.All(sink.Skip(4).Take(4), t => Assert.Equal(Source, t.SnapshotSourceReference));
|
||||
}
|
||||
}
|
||||
+15
-3
@@ -18,11 +18,14 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters;
|
||||
internal sealed class SecureOpcUaTestServer : IAsyncDisposable
|
||||
{
|
||||
private readonly ApplicationInstance _app;
|
||||
private readonly StandardServer _server;
|
||||
private readonly string _pkiRoot;
|
||||
|
||||
private SecureOpcUaTestServer(ApplicationInstance app, string pkiRoot, string url, X509Certificate2 certificate)
|
||||
private SecureOpcUaTestServer(
|
||||
ApplicationInstance app, StandardServer server, string pkiRoot, string url, X509Certificate2 certificate)
|
||||
{
|
||||
_app = app;
|
||||
_server = server;
|
||||
_pkiRoot = pkiRoot;
|
||||
Url = url;
|
||||
Certificate = certificate;
|
||||
@@ -110,16 +113,25 @@ internal sealed class SecureOpcUaTestServer : IAsyncDisposable
|
||||
ApplicationType = ApplicationType.Server,
|
||||
};
|
||||
await app.CheckApplicationInstanceCertificatesAsync(silent: true, lifeTimeInMonths: 12);
|
||||
await app.StartAsync(new TestServer(telemetry, apiKey));
|
||||
var server = new TestServer(telemetry, apiKey);
|
||||
await app.StartAsync(server);
|
||||
|
||||
// SDK 2: the loaded application certificate lives in the certificate manager, not on the identifier.
|
||||
using var entry = config.CertificateManager.AcquireApplicationCertificateByType(
|
||||
ObjectTypeIds.RsaSha256ApplicationCertificateType)
|
||||
?? throw new InvalidOperationException("The test server has no application certificate.");
|
||||
var cert = X509CertificateLoader.LoadCertificate(entry.Certificate.RawData);
|
||||
return new SecureOpcUaTestServer(app, pkiRoot, url, cert);
|
||||
return new SecureOpcUaTestServer(app, server, pkiRoot, url, cert);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sets the server's <c>Server_ServiceLevel</c> (<c>i=2267</c>) — what a redundant server
|
||||
/// publishes for its role (ScadaBridge#138: 250/240 Active, 90/50 Warm standby, 1 booting).
|
||||
/// </summary>
|
||||
/// <param name="level">The ServiceLevel to publish.</param>
|
||||
public void SetServiceLevel(byte level) =>
|
||||
_server.CurrentInstance.ServerObject.ServiceLevel!.Value = level;
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user