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
@@ -119,6 +119,8 @@ if (result.Success)
|
||||
|
||||
When `BackupConnectionDetails` is provided, the actor tracks consecutive failures and unstable disconnects. After `FailoverRetryCount` consecutive failures (or unstable connections shorter than `StableConnectionThreshold`), the actor disposes the current adapter, creates a fresh one with the other endpoint's config via `DataConnectionFactory.Create`, and re-arms the connect timer. Failover is round-robin: primary → backup → primary. There is no auto-failback; the connection stays on whichever endpoint is currently working.
|
||||
|
||||
**ServiceLevel-aware failover (OPC UA, Gitea #138).** A third trigger, for redundant OPC UA pairs whose standby keeps a healthy session but serves Bad data (OtOpcUa Warm, lmxopcua#717). It lives in `Actors/DataConnectionActor.ServiceLevel.cs`. While Connected, and only when the adapter implements `IServiceLevelReadable`, the connection has a backup and the primary config's `ServiceLevelFailover.Enabled` is true, a periodic timer reads `Server_ServiceLevel` (`i=2267`). A level below 200 starts a degraded streak; a Healthy or unreadable sample ends it. After the degraded window (default 10 s) — or at once on the first sample after a (re)connect — the actor probes the other endpoint with a throwaway adapter from `DataConnectionFactory.Create` (connect, read, dispose; bounded by `ServiceLevelProbeTimeout`). It switches only if that endpoint reports ≥ 200 and higher, and not within the minimum dwell (default 60 s) of the previous ServiceLevel switch; otherwise it stays and re-arms the window. The switch is `SwitchToOtherEndpoint()` — the helper the two connectivity triggers now share — followed by `BecomeReconnecting(ReconnectCause.ServiceLevelFailover)` (zero delay, no unstable accounting). Poll and probe results carry the adapter generation, so a result from a session that has ended is dropped. Today's OtOpcUa (250/240) never starts a streak. Full rule set: `docs/requirements/Component-DataConnectionLayer.md` → *ServiceLevel-aware failover*.
|
||||
|
||||
### Write path
|
||||
|
||||
Writes are fire-and-forget from the script's view only in the sense that there is no store-and-forward. The `DataConnectionActor` pipes `_adapter.WriteAsync` back to the original sender:
|
||||
@@ -206,7 +208,7 @@ private void SwapAdapter(IDictionary<string, string> newConfig)
|
||||
}
|
||||
```
|
||||
|
||||
It then re-enters the reconnect cycle immediately (`BecomeReconnecting(afterReconfigure: true)` — zero delay, and the unstable-disconnect/failover accounting suppressed, since the link was dropped on purpose), and `ReSubscribeAll` / `ReSubscribeAllAlarms` re-issue everything from the untouched subscription registries. A reconfigure carrying the configuration already in force is a no-op.
|
||||
It then re-enters the reconnect cycle immediately (`BecomeReconnecting(ReconnectCause.Reconfigured)` — zero delay, and the unstable-disconnect/failover accounting suppressed, since the link was dropped on purpose), and `ReSubscribeAll` / `ReSubscribeAllAlarms` re-issue everything from the untouched subscription registries. A reconfigure carrying the configuration already in force is a no-op.
|
||||
|
||||
### Subscribing and receiving values
|
||||
|
||||
@@ -253,6 +255,8 @@ All settings under `DataConnectionLayer` in `appsettings.json` bind to `DataConn
|
||||
| `TagResolutionRetryInterval` | `00:00:10` | Retry interval for tag paths that could not be resolved on the device. |
|
||||
| `WriteTimeout` | `00:00:30` | Timeout applied to each `WriteAsync` call. A hung write times out and returns an error to the calling script. |
|
||||
| `StableConnectionThreshold` | `00:01:00` | A connection that drops before this duration is counted as an unstable disconnect toward `FailoverRetryCount`. |
|
||||
| `ServiceLevelPollInterval` | `00:00:02` | Read interval for the connected OPC UA endpoint's ServiceLevel when ServiceLevel-aware failover is in force (#138). |
|
||||
| `ServiceLevelProbeTimeout` | `00:00:15` | Budget for one ServiceLevel read and for probing the other endpoint (connect + read). |
|
||||
|
||||
### OPC UA global settings (`OpcUa` section)
|
||||
|
||||
@@ -286,6 +290,9 @@ Stored in the `DataConnection.PrimaryConfiguration` JSON dict, parsed by `OpcUaE
|
||||
| `QueueSize` | `10` | Per-item notification buffer size. |
|
||||
| `SecurityMode` | `None` | `None`, `Sign`, or `SignAndEncrypt`. |
|
||||
| `AutoAcceptUntrustedCerts` | `true` | Accept untrusted server certificates. |
|
||||
| `ServiceLevelFailover.Enabled` | `true` | ServiceLevel-aware failover (#138). Primary endpoint only; needs a backup. JSON `serviceLevelFailover` object; absent = defaults. |
|
||||
| `ServiceLevelFailover.DegradedWindowSeconds` | `10` | Seconds below ServiceLevel 200 before the other endpoint is probed. |
|
||||
| `ServiceLevelFailover.MinDwellSeconds` | `60` | Minimum seconds between ServiceLevel-driven switches. |
|
||||
|
||||
### Per-connection settings (MxGateway)
|
||||
|
||||
@@ -310,6 +317,7 @@ Carried on `CreateConnectionCommand` from the deployment artifact.
|
||||
|-------|---------|-------------|
|
||||
| `BackupConnectionDetails` | `null` | If absent, the connection retries its single endpoint indefinitely. |
|
||||
| `FailoverRetryCount` | `3` | Consecutive failures (or unstable disconnects) before switching endpoints. |
|
||||
| `ServiceLevelFailover` (OPC UA, inside the primary's configuration JSON) | enabled, 10 s, 60 s | ServiceLevel-aware failover (#138); see the OPC UA per-connection table. Lives in the protocol JSON rather than a column, so it needs no schema change and is encrypted at rest with the rest of the configuration. |
|
||||
|
||||
## Dependencies & Interactions
|
||||
|
||||
|
||||
@@ -190,6 +190,8 @@ scadabridge data-connection delete --id <id>
|
||||
scadabridge data-connection move --id <id> --site-id <destination-site-id>
|
||||
```
|
||||
|
||||
For an OPC UA connection with a backup, ServiceLevel-aware failover (Gitea #138) is set inside `--primary-config` as `"serviceLevelFailover": {"enabled": true, "degradedWindowSeconds": 10, "minDwellSeconds": 60}`; omitted means those defaults. See `Component-DataConnectionLayer.md` → *ServiceLevel-aware failover*.
|
||||
|
||||
#### Site-routed OPC UA operations (browse / search / verify-endpoint / certs)
|
||||
|
||||
CLI parity for the site-routed data-connection operations the ManagementActor dispatches
|
||||
|
||||
@@ -137,7 +137,7 @@ flowchart TD
|
||||
```
|
||||
|
||||
- **Round-robin**: primary → backup → primary → backup. No preferred endpoint after first failover — the connection stays on whichever endpoint is working.
|
||||
- **No auto-failback**: The connection remains on the active endpoint until it fails.
|
||||
- **No auto-failback**: The connection remains on the active endpoint until it fails — or, for OPC UA, until its ServiceLevel says it is no longer the server to be on (see *ServiceLevel-aware failover* below).
|
||||
- **Single-endpoint connections** (no backup): Retry indefinitely on the same endpoint, preserving existing behavior.
|
||||
- **Adapter lifecycle on failover**: The actor disposes the current `IDataConnection` adapter and creates a fresh one via `DataConnectionFactory.Create()` with the other endpoint's configuration. Clean slate — no stale state.
|
||||
|
||||
@@ -152,6 +152,34 @@ flowchart TD
|
||||
|
||||
See [`2026-03-22-primary-backup-data-connections-design.md`](../plans/2026-03-22-primary-backup-data-connections-design.md) for the full design.
|
||||
|
||||
### ServiceLevel-aware failover (OPC UA, Gitea #138)
|
||||
|
||||
The failover above is driven only by the link misbehaving. A redundant OPC UA server pair in **Warm** mode (OtOpcUa lmxopcua#717, design `OtOpcUa/docs/plans/2026-11-11-warm-redundancy-717-design.md` §4.3/§5) breaks that assumption: the standby keeps a perfectly healthy session but serves `BadOutOfService` data and advertises a standby ServiceLevel, so a connection that lands on it would sit on Bad quality forever. OPC UA connections with a backup therefore also watch `Server_ServiceLevel` (`i=2267`).
|
||||
|
||||
| Band (OPC UA Part 4 §6.6.2.4.2) | Levels | OtOpcUa |
|
||||
|---|---|---|
|
||||
| Healthy | 200–255 | Active: 250 Primary / 240 Secondary (today and under Warm), 200 Active-stale |
|
||||
| Degraded | 100–199, and anything below 200 here | Active critically degraded 100; Warm standby 90 (healthy) / 50 (degraded) |
|
||||
| NoData / Maintenance | 1 / 0 | Warm booting 1; down / endpoint failed 0 |
|
||||
|
||||
**Rule.** While Connected, the actor reads the connected endpoint's ServiceLevel every `ServiceLevelPollInterval` (2 s). A level **below 200** starts a *degraded streak*; any level of 200 or more, or an unreadable level (Bad status, no such node, read failure), ends it. When the streak has lasted the connection's **degraded window** (default 10 s), the actor probes the **other** endpoint — a short-lived session built from that endpoint's configuration: connect, read `i=2267`, close, bounded by `ServiceLevelProbeTimeout` (15 s). It fails over only when the other endpoint answers with a **Healthy level (≥ 200) higher than the connected one**. Otherwise (other endpoint unreachable, or also below 200) it **stays put** and the window re-arms, so the other endpoint is probed at most once per window.
|
||||
|
||||
- **Connect-time preference.** The first sample after every (re)connect is evaluated at once, without the window: a connection that comes up on a Degraded endpoint while the other is Healthy moves straight over. Because the trigger needs the connected level below 200, a connection to a 240 Secondary is never moved to a 250 Primary.
|
||||
- **Anti-flap.** A ServiceLevel switch is not repeated within the connection's **minimum dwell** (default 60 s), connect-time preference included. Together with the "Healthy and higher" requirement and the unbroken-streak rule (one Healthy sample resets it) this stops a pair that flips roles from bouncing the connection.
|
||||
- **Same switch as the connectivity failover.** The switch flips the active endpoint and swaps the adapter through the same helper as the two connection-driven triggers, resets both failure counters, then re-enters the reconnect cycle immediately with the unstable-disconnect accounting suppressed (it is a deliberate swap, like an in-place reconfigure). `ReSubscribeAll` / `ReSubscribeAllAlarms` then re-issue every subscription on the new endpoint; tags go Bad for the swap exactly as on any failover. Site event log: `Failover from Primary to Backup (ServiceLevel)` (Warning) with the two levels.
|
||||
- **Scope.** Only adapters implementing `IServiceLevelReadable` (the OPC UA adapter) are watched, and only when the connection has a backup and the setting is enabled. MxGateway connections, single-endpoint connections and disabled connections never read a ServiceLevel. The ServiceLevel read never raises the adapter's `Disconnected` event: a dead session stays the keep-alive's business.
|
||||
- **Behaviour against today's servers.** A pair that only ever reports Healthy levels (OtOpcUa 250/240 without Warm, or a server that always reports 255) never starts a streak: no probe, no switch — the default-on setting changes nothing there. It does act on a non-Warm OtOpcUa node that has dropped to 100/110 or 0/10 while its peer is ≥ 200, which is the desired move.
|
||||
|
||||
**Configuration** (per connection, on the **primary** endpoint's OPC UA JSON — the backup's copy is ignored; edited in the Central UI's *Backup endpoint* card next to *Failover Retry Count*; via the CLI it is part of `--primary-config`):
|
||||
|
||||
```json
|
||||
"serviceLevelFailover": { "enabled": true, "degradedWindowSeconds": 10, "minDwellSeconds": 60 }
|
||||
```
|
||||
|
||||
Absent or `null` means the defaults (enabled, 10 s, 60 s), so stored connections need no migration; the object travels inside the protocol configuration, which is encrypted at rest with the rest of it. Validation: window 1–3600 s, dwell 0–86400 s.
|
||||
|
||||
**Alarms across a promotion.** On Standby → Active the promoted OtOpcUa node runs a ConditionRefresh on every live event item on its own (design §4.5). The existing items therefore receive an **unsolicited** RefreshStart, the retained conditions, RefreshEnd. `RealOpcUaClient` frames it like the refresh it triggers on subscribe — `SnapshotStart` / `Snapshot`… / `SnapshotComplete` under a fresh epoch, live derivation resuming after RefreshEnd — and the actor's epoch-scoped snapshot cache replaces the source's state with it, dropping conditions the promoted node no longer retains. Covered by `RealOpcUaPromotionRefreshTests` and `DataConnectionPromotionRefreshTests`.
|
||||
|
||||
### In-place reconfiguration (register row 97)
|
||||
|
||||
A `ReconfigureConnectionCommand` replaces a **live** connection's configuration without stopping the connection actor. It is the same adapter swap the failover path above performs — the two share one `SwapAdapter` helper — so the diagram applies verbatim from *dispose adapter* onwards:
|
||||
@@ -186,6 +214,9 @@ All settings are parsed from the data connection's configuration JSON dictionari
|
||||
| `MaxMonitoredItemsPerSubscription` | int | `5000` | Monitored-item budget per OPC UA subscription. Above it the adapter **shards** onto an additional subscription on the same session (37,500 tags → 8 shards). Per endpoint, because item-count ceilings are a property of the target server. |
|
||||
| `SecurityMode` | string | `None` | Preferred endpoint security: `None`, `Sign`, or `SignAndEncrypt` |
|
||||
| `AutoAcceptUntrustedCerts` | bool | `true` | Accept untrusted server certificates |
|
||||
| `ServiceLevelFailover.Enabled` | bool | `true` | ServiceLevel-aware failover (#138). Read from the **primary** endpoint only; needs a backup. JSON: `serviceLevelFailover.enabled` |
|
||||
| `ServiceLevelFailover.DegradedWindowSeconds` | int | `10` | How long the connected endpoint's ServiceLevel must stay below 200 before the other endpoint is probed (1–3600) |
|
||||
| `ServiceLevelFailover.MinDwellSeconds` | int | `60` | Minimum time between ServiceLevel-driven switches (0–86400) |
|
||||
|
||||
### OPC UA application identity (deployment-wide, `ScadaBridge:OpcUa`)
|
||||
|
||||
@@ -242,6 +273,8 @@ These are configured via `DataConnectionOptions` in `appsettings.json`, not per-
|
||||
| `SubscribeBatchDelay` | 50ms | Delay BETWEEN reconnect re-subscribe chunks. Not applied on the instance-driven subscribe path — the Deployment Manager already staggers instance startup there. |
|
||||
| `QualityFlushInterval` | 1s | Coalescing window for pushing tag-quality counters to the health collector |
|
||||
| `MxSupervisoryAdviseParallelism` | 16 | Max in-flight supervisory advise commands on the MxGateway bulk-subscribe path when the endpoint has no write-user context |
|
||||
| `ServiceLevelPollInterval` | 2s | How often the connected OPC UA endpoint's `Server_ServiceLevel` is read when ServiceLevel-aware failover is in force (#138) |
|
||||
| `ServiceLevelProbeTimeout` | 15s | Budget for one ServiceLevel read, and for probing the other endpoint (connect + read); no answer in time = unreachable, no switch |
|
||||
| `MxAlarmStreamReopenDebounce` | 200ms | Window over which MxGateway alarm subscribes are coalesced into ONE reopen of the shared alarm stream (0–5 s; `0` = reopen per subscribe, the pre-2026-10-21 behaviour). Every new source still receives a fresh standing-state snapshot. |
|
||||
|
||||
## Subscription Management
|
||||
@@ -347,6 +380,7 @@ The registered `SourceReference` is a **NodeId** for OPC UA (the picker, CSV and
|
||||
- **State gating**: `SubscribeAlarmsRequest` is handled only in the **Connected** state; requests arriving while **Connecting**/**Reconnecting** are coalesced to the latest request per subscriber actor/source and processed on entering Connected. Unsubscribes and subscriber termination are handled in every state.
|
||||
- **Capability check**: if `_adapter is not IAlarmSubscribableConnection`, the actor replies `SubscribeAlarmsResponse(Success = false, ...)`.
|
||||
- **Reconnect handling**: on entering **Reconnecting**, the actor pushes a `NativeAlarmSourceUnavailable` to every alarm subscriber (consumers mark mirrored alarms uncertain rather than clearing them). On successful reconnection it re-subscribes the feed; the adapter re-emits a snapshot, reconciling state.
|
||||
- **Server-initiated refresh**: a RefreshStart/RefreshEnd pair the server sends on its own on an existing feed (OtOpcUa's promotion refresh under Warm redundancy, #138) takes the same snapshot path — forwarded in order under its own epoch, and it replaces the cached state for the source.
|
||||
- **Shared condition filter (last-subscriber-wins)**: because the feed is opened once per source, it carries a single condition filter. A second subscriber that registers a *different* filter for the same source overwrites it (last writer wins) and the actor logs a warning — co-subscribers to one source are expected to agree on the filter.
|
||||
- **Routing index**: subscribed sources are bucketed by the **first path segment** of their reference (everything before the first `.`), and a transition is matched only against the bucket for its own first segment, plus a residue list of sources that carry no separator at all (a prefix shorter than one segment, which can match other buckets). This is sound because a source reference containing a separator can only prefix a reference sharing its entire first segment. The per-source `StartsWith` test and the condition-type gate inside the bucket are unchanged, so routing decisions are identical to the previous linear scan over every subscribed source; snapshot boundaries bypass the condition filter, but OPC UA boundaries are sent only to the exact source binding. Connection-wide Mx boundaries broadcast to all subscribers while preserving their refresh epoch.
|
||||
- **Gateway union filter (MxGateway)**: `StreamAlarms` carries a single `alarm_filter_prefix`, so the adapter opens the stream on the **longest common prefix** of the currently subscribed source references and restarts it for each new source, including one covered by the existing prefix, so every newly bound source receives standing state. Restarts are **coalesced** (review 2026-10-21): subscribes arriving within `MxAlarmStreamReopenDebounce` (200 ms) of the first share ONE restart on the union prefix, so a staggered deploy batch of N alarm-bound instances costs one close/reopen and one connection-wide snapshot replay instead of N (each of which was broadcast to every already-subscribed source — O(N·M) boundary messages). A reconnect, teardown or last unsubscribe invalidates a pending restart. Each transport refresh has connection-wide start/end markers and a distinct epoch. An unsubscribe never restarts the stream — the prefix it leaves behind is at worst too broad, which may increase bandwidth. Correctness also depends on the epoch-scoped snapshot cache merging concurrent live updates and rejecting stale snapshot ends (DC-5); a broad filter alone provides no such guarantee. The prefix is a bandwidth optimisation only; the actor's per-source + condition-type gate remains authoritative, mirroring the OPC UA server-side WhereClause stance.
|
||||
|
||||
@@ -165,6 +165,40 @@
|
||||
min="1" max="20" @bind="_formFailoverRetryCount" />
|
||||
<div class="form-text">Retries before failing over to backup endpoint.</div>
|
||||
</div>
|
||||
@if (_protocol != "MxGateway")
|
||||
{
|
||||
<div class="mb-2" data-test="servicelevel-failover">
|
||||
<div class="form-check">
|
||||
<input type="checkbox" class="form-check-input" id="sl-failover-enabled"
|
||||
@bind="_formSlEnabled" />
|
||||
<label class="form-check-label small" for="sl-failover-enabled">
|
||||
ServiceLevel failover
|
||||
</label>
|
||||
</div>
|
||||
<div class="form-text">
|
||||
Fail over when the connected endpoint's ServiceLevel stays below 200 (Degraded)
|
||||
for the window below and the other endpoint reports a Healthy (200+), higher level.
|
||||
Needed for an OtOpcUa pair in Warm redundancy; a 250/240 pair never triggers it.
|
||||
</div>
|
||||
@if (_formSlEnabled)
|
||||
{
|
||||
<div class="row g-2 mt-1">
|
||||
<div class="col-auto">
|
||||
<label class="form-label small" for="sl-failover-window">Degraded window (s)</label>
|
||||
<input type="number" id="sl-failover-window" class="form-control form-control-sm"
|
||||
style="max-width: 120px;" min="1" max="3600" @bind="_formSlWindowSeconds" />
|
||||
@RenderPrimaryError("ServiceLevelFailover.DegradedWindowSeconds")
|
||||
</div>
|
||||
<div class="col-auto">
|
||||
<label class="form-label small" for="sl-failover-dwell">Minimum dwell (s)</label>
|
||||
<input type="number" id="sl-failover-dwell" class="form-control form-control-sm"
|
||||
style="max-width: 120px;" min="0" max="86400" @bind="_formSlDwellSeconds" />
|
||||
@RenderPrimaryError("ServiceLevelFailover.MinDwellSeconds")
|
||||
</div>
|
||||
</div>
|
||||
}
|
||||
</div>
|
||||
}
|
||||
<div class="mb-0">
|
||||
<button type="button" class="btn btn-outline-danger btn-sm"
|
||||
@onclick="RemoveBackup">Remove Backup</button>
|
||||
@@ -207,6 +241,10 @@
|
||||
private bool _backupIsLegacy;
|
||||
private bool _showBackup;
|
||||
private int _formFailoverRetryCount = 3;
|
||||
// ScadaBridge#138: connection-level ServiceLevel failover, stored on the PRIMARY OPC UA endpoint.
|
||||
private bool _formSlEnabled = true;
|
||||
private int _formSlWindowSeconds = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds;
|
||||
private int _formSlDwellSeconds = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds;
|
||||
private ValidationResult? _primaryErrors;
|
||||
private ValidationResult? _backupErrors;
|
||||
private string? _formError;
|
||||
@@ -276,6 +314,10 @@
|
||||
{
|
||||
(_primaryConfig, _primaryIsLegacy) =
|
||||
OpcUaEndpointConfigSerializer.Deserialize(conn.PrimaryConfiguration);
|
||||
var sl = _primaryConfig.ServiceLevelFailover ?? new OpcUaServiceLevelFailoverConfig();
|
||||
_formSlEnabled = sl.Enabled;
|
||||
_formSlWindowSeconds = sl.DegradedWindowSeconds;
|
||||
_formSlDwellSeconds = sl.MinDwellSeconds;
|
||||
if (!string.IsNullOrWhiteSpace(conn.BackupConfiguration))
|
||||
{
|
||||
(_backupConfig, _backupIsLegacy) =
|
||||
@@ -313,6 +355,15 @@
|
||||
}
|
||||
else
|
||||
{
|
||||
// Only meaningful with a backup; without one the setting is not stored at all.
|
||||
_primaryConfig.ServiceLevelFailover = _showBackup
|
||||
? new OpcUaServiceLevelFailoverConfig
|
||||
{
|
||||
Enabled = _formSlEnabled,
|
||||
DegradedWindowSeconds = _formSlWindowSeconds,
|
||||
MinDwellSeconds = _formSlDwellSeconds,
|
||||
}
|
||||
: null;
|
||||
_primaryErrors = OpcUaEndpointConfigValidator.Validate(_primaryConfig, "Primary.");
|
||||
_backupErrors = _showBackup
|
||||
? OpcUaEndpointConfigValidator.Validate(_backupConfig, "Backup.")
|
||||
@@ -350,6 +401,17 @@
|
||||
_backupMx = new MxGatewayEndpointConfig();
|
||||
_backupIsLegacy = false;
|
||||
_formFailoverRetryCount = 3;
|
||||
_formSlEnabled = true;
|
||||
_formSlWindowSeconds = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds;
|
||||
_formSlDwellSeconds = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds;
|
||||
}
|
||||
|
||||
private RenderFragment? RenderPrimaryError(string field)
|
||||
{
|
||||
var match = _primaryErrors?.Errors.FirstOrDefault(e => e.EntityName == "Primary." + field);
|
||||
return match is null
|
||||
? null
|
||||
: @<div class="text-danger small">@match.Message</div>;
|
||||
}
|
||||
|
||||
private void GoBack() => NavigationManager.NavigateTo("/design/connections");
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
|
||||
/// <summary>
|
||||
/// Optional capability for an <see cref="IDataConnection"/> whose server publishes a redundancy
|
||||
/// ServiceLevel (OPC UA <c>Server_ServiceLevel</c>, <c>i=2267</c>). The DataConnectionActor polls it
|
||||
/// on the connected endpoint, and probes the other endpoint with it, to drive ServiceLevel-aware
|
||||
/// failover (ScadaBridge#138). An adapter without the capability is never switched on ServiceLevel.
|
||||
/// </summary>
|
||||
public interface IServiceLevelReadable
|
||||
{
|
||||
/// <summary>
|
||||
/// Reads the server's current ServiceLevel. Never raises the connection's
|
||||
/// <see cref="IDataConnection.Disconnected"/> event: a lost session is the reconnect machine's
|
||||
/// business, not this read's.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">Bounds the read.</param>
|
||||
/// <returns>
|
||||
/// The ServiceLevel (0–255), or null when it is unknown: not connected, a Bad status, a value
|
||||
/// that is not a byte, or a failed read.
|
||||
/// </returns>
|
||||
Task<byte?> ReadServiceLevelAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
@@ -83,6 +83,15 @@ public readonly record struct OpcUaConfigParseResult
|
||||
/// </summary>
|
||||
public static class OpcUaEndpointConfigSerializer
|
||||
{
|
||||
/// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.Enabled"/>.</summary>
|
||||
public const string ServiceLevelFailoverEnabledKey = "ServiceLevelFailover.Enabled";
|
||||
|
||||
/// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.DegradedWindowSeconds"/>.</summary>
|
||||
public const string ServiceLevelFailoverWindowKey = "ServiceLevelFailover.DegradedWindowSeconds";
|
||||
|
||||
/// <summary>Flat-dict key for <see cref="OpcUaServiceLevelFailoverConfig.MinDwellSeconds"/>.</summary>
|
||||
public const string ServiceLevelFailoverDwellKey = "ServiceLevelFailover.MinDwellSeconds";
|
||||
|
||||
private static readonly JsonSerializerOptions JsonOpts = new()
|
||||
{
|
||||
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
|
||||
@@ -216,6 +225,12 @@ public static class OpcUaEndpointConfigSerializer
|
||||
dict["Deadband.Type"] = db.Type.ToString();
|
||||
dict["Deadband.Value"] = db.Value.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
}
|
||||
if (config.ServiceLevelFailover is { } sl)
|
||||
{
|
||||
dict[ServiceLevelFailoverEnabledKey] = sl.Enabled.ToString();
|
||||
dict[ServiceLevelFailoverWindowKey] = sl.DegradedWindowSeconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
dict[ServiceLevelFailoverDwellKey] = sl.MinDwellSeconds.ToString(System.Globalization.CultureInfo.InvariantCulture);
|
||||
}
|
||||
return dict;
|
||||
}
|
||||
|
||||
@@ -291,6 +306,18 @@ public static class OpcUaEndpointConfigSerializer
|
||||
c.Deadband = db;
|
||||
}
|
||||
|
||||
if (dict.ContainsKey(ServiceLevelFailoverEnabledKey)
|
||||
|| dict.ContainsKey(ServiceLevelFailoverWindowKey)
|
||||
|| dict.ContainsKey(ServiceLevelFailoverDwellKey))
|
||||
{
|
||||
var sl = new OpcUaServiceLevelFailoverConfig();
|
||||
if (dict.TryGetValue(ServiceLevelFailoverEnabledKey, out var slEn) && bool.TryParse(slEn, out var slEnabled))
|
||||
sl.Enabled = slEnabled;
|
||||
TryAssignInt(dict, ServiceLevelFailoverWindowKey, v => sl.DegradedWindowSeconds = v);
|
||||
TryAssignInt(dict, ServiceLevelFailoverDwellKey, v => sl.MinDwellSeconds = v);
|
||||
c.ServiceLevelFailover = sl;
|
||||
}
|
||||
|
||||
return c;
|
||||
}
|
||||
|
||||
|
||||
@@ -96,4 +96,12 @@ public sealed class OpcUaEndpointConfig
|
||||
/// Heartbeat configuration for connection monitoring.
|
||||
/// </summary>
|
||||
public OpcUaHeartbeatConfig? Heartbeat { get; set; }
|
||||
|
||||
// Redundancy (optional; null = defaults, see OpcUaServiceLevelFailoverConfig)
|
||||
/// <summary>
|
||||
/// ServiceLevel-aware failover settings (ScadaBridge#138). Connection-level: only the primary
|
||||
/// endpoint's value is read, and only when the connection has a backup. Null means the
|
||||
/// defaults (enabled, 10 s window, 60 s dwell).
|
||||
/// </summary>
|
||||
public OpcUaServiceLevelFailoverConfig? ServiceLevelFailover { get; set; }
|
||||
}
|
||||
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
namespace ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
|
||||
|
||||
/// <summary>
|
||||
/// ServiceLevel-aware failover for an OPC UA connection with a primary AND a backup endpoint
|
||||
/// (ScadaBridge#138; OtOpcUa Warm redundancy, lmxopcua#717). A connection-level setting: only the
|
||||
/// <b>primary</b> endpoint's copy is read, and it has no effect on a connection without a backup.
|
||||
///
|
||||
/// <para>
|
||||
/// The connected endpoint's <c>Server_ServiceLevel</c> (<c>i=2267</c>) is polled. When it stays in
|
||||
/// the Degraded band (below 200, OPC UA Part 4 §6.6.2.4.2) for <see cref="DegradedWindowSeconds"/>
|
||||
/// and the other endpoint is reachable and reports a Healthy-band level (200 or more) that is also
|
||||
/// higher, the connection fails over to it. A switch is not repeated within
|
||||
/// <see cref="MinDwellSeconds"/>. Both endpoints degraded, or the other one unreachable, means the
|
||||
/// connection stays where it is. On connect, a Degraded endpoint is left immediately (no window) for
|
||||
/// a Healthy, higher other endpoint, subject to the same dwell.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// A stored configuration without this object behaves as the defaults: enabled, 10 s window,
|
||||
/// 60 s dwell. With today's OtOpcUa pair (250 Primary / 240 Secondary, both Healthy band) it never
|
||||
/// triggers.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class OpcUaServiceLevelFailoverConfig
|
||||
{
|
||||
/// <summary>Default for <see cref="DegradedWindowSeconds"/>.</summary>
|
||||
public const int DefaultDegradedWindowSeconds = 10;
|
||||
|
||||
/// <summary>Default for <see cref="MinDwellSeconds"/>.</summary>
|
||||
public const int DefaultMinDwellSeconds = 60;
|
||||
|
||||
/// <summary>Whether ServiceLevel can trigger a failover. Default true.</summary>
|
||||
public bool Enabled { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// How long, in seconds, the connected endpoint must report a Degraded-band ServiceLevel
|
||||
/// without a break before the other endpoint is checked. Default 10.
|
||||
/// </summary>
|
||||
public int DegradedWindowSeconds { get; set; } = DefaultDegradedWindowSeconds;
|
||||
|
||||
/// <summary>
|
||||
/// Minimum time, in seconds, after a ServiceLevel-driven switch before another one may happen
|
||||
/// (anti-flap). Default 60.
|
||||
/// </summary>
|
||||
public int MinDwellSeconds { get; set; } = DefaultMinDwellSeconds;
|
||||
}
|
||||
@@ -74,6 +74,14 @@ public static class OpcUaEndpointConfigValidator
|
||||
if (config.Deadband is { } db && db.Value <= 0)
|
||||
errors.Add(Err("Deadband.Value", "Must be > 0."));
|
||||
|
||||
if (config.ServiceLevelFailover is { } sl)
|
||||
{
|
||||
if (sl.DegradedWindowSeconds < 1 || sl.DegradedWindowSeconds > 3600)
|
||||
errors.Add(Err("ServiceLevelFailover.DegradedWindowSeconds", "Must be between 1 and 3600."));
|
||||
if (sl.MinDwellSeconds < 0 || sl.MinDwellSeconds > 86400)
|
||||
errors.Add(Err("ServiceLevelFailover.MinDwellSeconds", "Must be between 0 and 86400."));
|
||||
}
|
||||
|
||||
return errors.Count == 0
|
||||
? ValidationResult.Success()
|
||||
: ValidationResult.FromErrors(errors.ToArray());
|
||||
|
||||
+320
@@ -0,0 +1,320 @@
|
||||
using System.Globalization;
|
||||
using Akka.Actor;
|
||||
using Akka.Event;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138 — ServiceLevel-aware failover between a primary and a backup endpoint.
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Why.</b> Until now the connection failed over only when the link itself misbehaved
|
||||
/// (<c>FailoverRetryCount</c> consecutive connect failures or unstable disconnects). Under OtOpcUa
|
||||
/// Warm redundancy (lmxopcua#717) the standby keeps a perfectly healthy session but serves
|
||||
/// <c>BadOutOfService</c> data at a standby ServiceLevel (90 healthy / 50 degraded / 1 booting), so a
|
||||
/// connection that lands there would sit on Bad quality indefinitely.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// <b>What.</b> While Connected, the connected endpoint's <c>Server_ServiceLevel</c> is polled every
|
||||
/// <see cref="DataConnectionOptions.ServiceLevelPollInterval"/>. A level below
|
||||
/// <see cref="HealthyServiceLevelMinimum"/> (200, the bottom of the Part 4 Healthy band) starts a
|
||||
/// degraded streak; any Healthy or unreadable sample ends it. Once the streak has lasted the
|
||||
/// connection's degraded window — or immediately on the first sample after a (re)connect, which is
|
||||
/// the "prefer the better endpoint on connect" rule — the other endpoint is probed with a
|
||||
/// short-lived session. The connection switches only when that probe reads a Healthy level that is
|
||||
/// also higher than the connected one, and never within the dwell time of the previous
|
||||
/// ServiceLevel switch. Otherwise it stays put and the window re-arms.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// <b>Unchanged behaviour.</b> A level of 200 or more never starts a streak, so today's OtOpcUa pair
|
||||
/// (250 Primary / 240 Secondary) never triggers anything; an adapter without
|
||||
/// <see cref="IServiceLevelReadable"/> (MxGateway) and a connection without a backup are never
|
||||
/// monitored at all. The switch itself reuses <see cref="SwitchToOtherEndpoint"/> and the reconnect
|
||||
/// cycle, exactly like the connection-driven failovers.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public partial class DataConnectionActor
|
||||
{
|
||||
/// <summary>The lowest ServiceLevel in the OPC UA Healthy band (Part 4 §6.6.2.4.2).</summary>
|
||||
internal const byte HealthyServiceLevelMinimum = 200;
|
||||
|
||||
private const string ServiceLevelPollTimerKey = "servicelevel-poll";
|
||||
|
||||
private ServiceLevelFailoverPolicy _serviceLevelPolicy;
|
||||
|
||||
/// <summary>When the current unbroken run of Degraded samples began; null when not degraded.</summary>
|
||||
private DateTimeOffset? _degradedSince;
|
||||
|
||||
/// <summary>When the last ServiceLevel-driven switch happened (dwell / anti-flap).</summary>
|
||||
private DateTimeOffset _lastServiceLevelSwitchAt = DateTimeOffset.MinValue;
|
||||
|
||||
private bool _serviceLevelReadInFlight;
|
||||
private bool _serviceLevelProbeInFlight;
|
||||
|
||||
/// <summary>True until the first sample of a session is evaluated (connect-time preference).</summary>
|
||||
private bool _serviceLevelConnectCheckPending;
|
||||
|
||||
/// <summary>Why <see cref="BecomeReconnecting"/> is re-establishing the link.</summary>
|
||||
internal enum ReconnectCause
|
||||
{
|
||||
/// <summary>The link was lost (disconnect, connection-level failure).</summary>
|
||||
ConnectionLost,
|
||||
|
||||
/// <summary>An in-place reconfiguration swapped the adapter.</summary>
|
||||
Reconfigured,
|
||||
|
||||
/// <summary>A sustained Degraded ServiceLevel swapped to the other endpoint (#138).</summary>
|
||||
ServiceLevelFailover,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when this connection's ServiceLevel is watched: the policy is enabled, there is a
|
||||
/// backup to go to, and the live adapter can read a ServiceLevel.
|
||||
/// </summary>
|
||||
private bool ServiceLevelFailoverInForce =>
|
||||
_serviceLevelPolicy.Enabled && _backupConfig != null && _adapter is IServiceLevelReadable;
|
||||
|
||||
/// <summary>Called on entering Connected. Arms the poll (first tick immediately).</summary>
|
||||
private void StartServiceLevelMonitoring()
|
||||
{
|
||||
ResetServiceLevelSession();
|
||||
if (!ServiceLevelFailoverInForce)
|
||||
return;
|
||||
|
||||
_serviceLevelConnectCheckPending = true;
|
||||
Timers.StartPeriodicTimer(ServiceLevelPollTimerKey, ServiceLevelPollTick.Instance,
|
||||
TimeSpan.Zero, _options.ServiceLevelPollInterval);
|
||||
}
|
||||
|
||||
/// <summary>Called on leaving Connected. Results still in flight are fenced by generation.</summary>
|
||||
private void StopServiceLevelMonitoring()
|
||||
{
|
||||
Timers.Cancel(ServiceLevelPollTimerKey);
|
||||
ResetServiceLevelSession();
|
||||
}
|
||||
|
||||
private void ResetServiceLevelSession()
|
||||
{
|
||||
_degradedSince = null;
|
||||
_serviceLevelReadInFlight = false;
|
||||
_serviceLevelProbeInFlight = false;
|
||||
_serviceLevelConnectCheckPending = false;
|
||||
}
|
||||
|
||||
private void HandleServiceLevelPollTick()
|
||||
{
|
||||
// One read at a time, and none while the other endpoint is being probed: the probe's
|
||||
// verdict is taken against the level it started from.
|
||||
if (!ServiceLevelFailoverInForce || _serviceLevelReadInFlight || _serviceLevelProbeInFlight)
|
||||
return;
|
||||
|
||||
_serviceLevelReadInFlight = true;
|
||||
var generation = _adapterGeneration;
|
||||
var reader = (IServiceLevelReadable)_adapter;
|
||||
var timeout = _options.ServiceLevelProbeTimeout;
|
||||
Task.Run(() => reader.ReadServiceLevelAsync().WaitAsync(timeout))
|
||||
.ContinueWith(t => new ServiceLevelSampled(generation,
|
||||
t.IsCompletedSuccessfully ? t.Result : null), TaskScheduler.Default)
|
||||
.PipeTo(_self);
|
||||
}
|
||||
|
||||
private void HandleServiceLevelSampled(ServiceLevelSampled sampled)
|
||||
{
|
||||
if (sampled.AdapterGeneration != _adapterGeneration)
|
||||
return;
|
||||
|
||||
_serviceLevelReadInFlight = false;
|
||||
var connectCheck = _serviceLevelConnectCheckPending;
|
||||
_serviceLevelConnectCheckPending = false;
|
||||
|
||||
if (sampled.Level is not { } level || level >= HealthyServiceLevelMinimum)
|
||||
{
|
||||
// Healthy, or unknown: either way the run of Degraded samples is broken. Unknown never
|
||||
// counts toward a switch — acting on missing information is how a pair flaps.
|
||||
if (_degradedSince != null)
|
||||
_log.Info("[{0}] {1} endpoint ServiceLevel is {2} — degraded watch cleared",
|
||||
_connectionName, _activeEndpoint, sampled.Level?.ToString(CultureInfo.InvariantCulture) ?? "unknown");
|
||||
_degradedSince = null;
|
||||
return;
|
||||
}
|
||||
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
if (_degradedSince is null)
|
||||
{
|
||||
_degradedSince = now;
|
||||
_log.Warning("[{0}] {1} endpoint reports ServiceLevel {2} (Degraded band, < {3}) — "
|
||||
+ "will check the other endpoint {4}",
|
||||
_connectionName, _activeEndpoint, level, HealthyServiceLevelMinimum,
|
||||
connectCheck ? "now (connect-time preference)"
|
||||
: $"if it persists for {_serviceLevelPolicy.DegradedWindow.TotalSeconds:F0}s");
|
||||
}
|
||||
|
||||
if (!connectCheck && now - _degradedSince.Value < _serviceLevelPolicy.DegradedWindow)
|
||||
return;
|
||||
|
||||
if (now - _lastServiceLevelSwitchAt < _serviceLevelPolicy.MinDwell)
|
||||
{
|
||||
_log.Debug("[{0}] ServiceLevel {1} still Degraded but the last ServiceLevel switch was {2:F0}s ago "
|
||||
+ "(dwell {3:F0}s) — staying", _connectionName, level,
|
||||
(now - _lastServiceLevelSwitchAt).TotalSeconds, _serviceLevelPolicy.MinDwell.TotalSeconds);
|
||||
return;
|
||||
}
|
||||
|
||||
StartServiceLevelProbe(level);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the OTHER endpoint's ServiceLevel through a short-lived adapter built from its
|
||||
/// configuration — connect, read, dispose — off the actor thread, bounded by
|
||||
/// <see cref="DataConnectionOptions.ServiceLevelProbeTimeout"/>. The live adapter is untouched.
|
||||
/// </summary>
|
||||
private void StartServiceLevelProbe(byte currentLevel)
|
||||
{
|
||||
_serviceLevelProbeInFlight = true;
|
||||
var generation = _adapterGeneration;
|
||||
var otherConfig = _activeEndpoint == ActiveEndpoint.Primary ? _backupConfig! : _primaryConfig;
|
||||
var factory = _factory;
|
||||
var protocol = _protocolType;
|
||||
var timeout = _options.ServiceLevelProbeTimeout;
|
||||
|
||||
Task.Run(() => ProbeServiceLevelAsync(factory, protocol, otherConfig).WaitAsync(timeout))
|
||||
.ContinueWith(t => t.IsCompletedSuccessfully
|
||||
? new ServiceLevelProbed(generation, currentLevel, t.Result.Level, t.Result.Error)
|
||||
: new ServiceLevelProbed(generation, currentLevel, null,
|
||||
t.Exception?.GetBaseException() is TimeoutException
|
||||
? $"no answer within {timeout.TotalSeconds:F0}s"
|
||||
: t.Exception?.GetBaseException().Message ?? "probe cancelled"),
|
||||
TaskScheduler.Default)
|
||||
.PipeTo(_self);
|
||||
}
|
||||
|
||||
private static async Task<(byte? Level, string? Error)> ProbeServiceLevelAsync(
|
||||
IDataConnectionFactory factory, string protocol, IDictionary<string, string> config)
|
||||
{
|
||||
IDataConnection? probe = null;
|
||||
try
|
||||
{
|
||||
probe = factory.Create(protocol, config);
|
||||
if (probe is not IServiceLevelReadable reader)
|
||||
return (null, "the adapter cannot read a ServiceLevel");
|
||||
|
||||
await probe.ConnectAsync(config).ConfigureAwait(false);
|
||||
var level = await reader.ReadServiceLevelAsync().ConfigureAwait(false);
|
||||
return (level, level is null ? "ServiceLevel unreadable" : null);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
return (null, ex.GetBaseException().Message);
|
||||
}
|
||||
finally
|
||||
{
|
||||
// Fire-and-forget: closing a session against a black-holed server can block for the
|
||||
// operation timeout, and the probe's verdict must not wait for it.
|
||||
if (probe != null)
|
||||
_ = probe.DisposeAsync().AsTask().ContinueWith(
|
||||
t => _ = t.Exception, TaskContinuationOptions.OnlyOnFaulted);
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleServiceLevelProbed(ServiceLevelProbed probed)
|
||||
{
|
||||
if (probed.AdapterGeneration != _adapterGeneration)
|
||||
return;
|
||||
|
||||
_serviceLevelProbeInFlight = false;
|
||||
if (!ServiceLevelFailoverInForce || _degradedSince is null)
|
||||
return;
|
||||
|
||||
var other = _activeEndpoint == ActiveEndpoint.Primary ? ActiveEndpoint.Backup : ActiveEndpoint.Primary;
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
|
||||
if (probed.OtherLevel is not { } otherLevel)
|
||||
{
|
||||
_log.Warning("[{0}] {1} endpoint ServiceLevel {2} is Degraded but the {3} endpoint is not "
|
||||
+ "available ({4}) — staying", _connectionName, _activeEndpoint, probed.CurrentLevel, other,
|
||||
probed.Error);
|
||||
_degradedSince = now; // re-arm: look again after another full window
|
||||
return;
|
||||
}
|
||||
|
||||
if (otherLevel < HealthyServiceLevelMinimum || otherLevel <= probed.CurrentLevel)
|
||||
{
|
||||
_log.Warning("[{0}] {1} endpoint ServiceLevel {2} is Degraded and the {3} endpoint reports {4} "
|
||||
+ "(not a Healthy, higher level) — staying", _connectionName, _activeEndpoint,
|
||||
probed.CurrentLevel, other, otherLevel);
|
||||
_degradedSince = now;
|
||||
return;
|
||||
}
|
||||
|
||||
var degradedFor = now - _degradedSince.Value;
|
||||
var previousEndpoint = SwitchToOtherEndpoint();
|
||||
_consecutiveFailures = 0;
|
||||
_consecutiveUnstableDisconnects = 0;
|
||||
_lastServiceLevelSwitchAt = now;
|
||||
|
||||
_log.Warning("[{0}] Failing over from {1} to {2} (ServiceLevel {3} Degraded for {4:F0}s; {2} reports {5})",
|
||||
_connectionName, previousEndpoint, _activeEndpoint, probed.CurrentLevel,
|
||||
degradedFor.TotalSeconds, otherLevel);
|
||||
|
||||
if (_siteEventLogger != null)
|
||||
{
|
||||
_ = _siteEventLogger.LogEventAsync(
|
||||
"connection", "Warning", null, _connectionName,
|
||||
$"Failover from {previousEndpoint} to {_activeEndpoint} (ServiceLevel)",
|
||||
$"{previousEndpoint} reported ServiceLevel {probed.CurrentLevel} for {degradedFor.TotalSeconds:F0}s; "
|
||||
+ $"{_activeEndpoint} reports {otherLevel}");
|
||||
}
|
||||
|
||||
BecomeReconnecting(ReconnectCause.ServiceLevelFailover);
|
||||
}
|
||||
|
||||
/// <summary>Poll tick for the connected endpoint's ServiceLevel.</summary>
|
||||
internal sealed record ServiceLevelPollTick
|
||||
{
|
||||
/// <summary>The single instance.</summary>
|
||||
public static readonly ServiceLevelPollTick Instance = new();
|
||||
}
|
||||
|
||||
/// <summary>One ServiceLevel sample of the connected endpoint (null = unknown).</summary>
|
||||
internal sealed record ServiceLevelSampled(int AdapterGeneration, byte? Level);
|
||||
|
||||
/// <summary>The other endpoint's ServiceLevel (null = unreachable / unreadable, see Error).</summary>
|
||||
internal sealed record ServiceLevelProbed(int AdapterGeneration, byte CurrentLevel, byte? OtherLevel, string? Error);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The connection-level ServiceLevel failover settings, read from the PRIMARY endpoint's flattened
|
||||
/// configuration (<see cref="OpcUaServiceLevelFailoverConfig"/>). Absent keys mean the defaults.
|
||||
/// </summary>
|
||||
internal readonly record struct ServiceLevelFailoverPolicy(bool Enabled, TimeSpan DegradedWindow, TimeSpan MinDwell)
|
||||
{
|
||||
/// <summary>Parses the policy; malformed or out-of-range values fall back to safe values.</summary>
|
||||
/// <param name="primaryConfig">The primary endpoint's flattened configuration.</param>
|
||||
/// <returns>The policy in force.</returns>
|
||||
public static ServiceLevelFailoverPolicy From(IDictionary<string, string>? primaryConfig)
|
||||
{
|
||||
var enabled = true;
|
||||
var window = OpcUaServiceLevelFailoverConfig.DefaultDegradedWindowSeconds;
|
||||
var dwell = OpcUaServiceLevelFailoverConfig.DefaultMinDwellSeconds;
|
||||
|
||||
if (primaryConfig != null)
|
||||
{
|
||||
if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverEnabledKey, out var e)
|
||||
&& bool.TryParse(e, out var parsedEnabled))
|
||||
enabled = parsedEnabled;
|
||||
if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverWindowKey, out var w)
|
||||
&& int.TryParse(w, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsedWindow))
|
||||
window = Math.Max(1, parsedWindow);
|
||||
if (primaryConfig.TryGetValue(OpcUaEndpointConfigSerializer.ServiceLevelFailoverDwellKey, out var d)
|
||||
&& int.TryParse(d, NumberStyles.Integer, CultureInfo.InvariantCulture, out var parsedDwell))
|
||||
dwell = Math.Max(0, parsedDwell);
|
||||
}
|
||||
|
||||
return new ServiceLevelFailoverPolicy(enabled, TimeSpan.FromSeconds(window), TimeSpan.FromSeconds(dwell));
|
||||
}
|
||||
}
|
||||
@@ -29,7 +29,7 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
/// Health reporting (connection status + tag resolution counts).
|
||||
/// Subscription lifecycle (register on create, cleanup on stop).
|
||||
/// </summary>
|
||||
public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
public partial class DataConnectionActor : UntypedActor, IWithTimers
|
||||
{
|
||||
public enum ActiveEndpoint { Primary, Backup }
|
||||
|
||||
@@ -328,6 +328,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
_siteEventLogger = siteEventLogger;
|
||||
_connectionDetails = _primaryConfig;
|
||||
_tagResolutionInterval = _options.TagResolutionRetryInterval;
|
||||
_serviceLevelPolicy = ServiceLevelFailoverPolicy.From(_primaryConfig);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -491,6 +492,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// No session yet — a probe would fail for every tag. ReSubscribeAll's chunk
|
||||
// completions re-arm the timer once the link is up.
|
||||
break;
|
||||
case ServiceLevelPollTick or ServiceLevelSampled or ServiceLevelProbed:
|
||||
// ScadaBridge#138: belongs to a session that is over; BecomeConnected restarts it.
|
||||
break;
|
||||
case QualityFlushTick:
|
||||
FlushQualityCountersIfDirty();
|
||||
break;
|
||||
@@ -538,6 +542,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// Register row 54 — the only transition that raises the connected gauge.
|
||||
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connected);
|
||||
Become(Connected);
|
||||
// ScadaBridge#138: watch this endpoint's ServiceLevel (no-op unless in force).
|
||||
StartServiceLevelMonitoring();
|
||||
foreach (var pending in _pendingTagSubscriptions.Values)
|
||||
Self.Tell(pending.Request, pending.Owner);
|
||||
_pendingTagSubscriptions.Clear();
|
||||
@@ -592,6 +598,15 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
case QualityFlushTick:
|
||||
FlushQualityCountersIfDirty();
|
||||
break;
|
||||
case ServiceLevelPollTick:
|
||||
HandleServiceLevelPollTick();
|
||||
break;
|
||||
case ServiceLevelSampled sampled:
|
||||
HandleServiceLevelSampled(sampled);
|
||||
break;
|
||||
case ServiceLevelProbed probed:
|
||||
HandleServiceLevelProbed(probed);
|
||||
break;
|
||||
case AdapterDisconnected disconnected:
|
||||
// Fence the disconnect on its adapter generation. The outgoing adapter's
|
||||
// event can fire concurrently with SwapAdapter's unhook, so a Disconnected from
|
||||
@@ -629,28 +644,36 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
|
||||
// ── Reconnecting State ──
|
||||
|
||||
/// <param name="afterReconfigure">
|
||||
/// True when this transition was caused by an in-place reconfiguration
|
||||
/// (<see cref="HandleReconfigure"/>) rather than by losing the link. It suppresses the
|
||||
/// unstable-disconnect accounting and the failover check — the adapter has just been
|
||||
/// swapped ON PURPOSE against a brand-new configuration, so counting it as an unstable
|
||||
/// cycle would let a reconfigure trip a failover onto the new backup before the new primary
|
||||
/// has been tried even once — and it reconnects immediately instead of waiting out
|
||||
/// <param name="cause">
|
||||
/// Why the link is being re-established. <see cref="ReconnectCause.Reconfigured"/> (an
|
||||
/// in-place reconfiguration, <see cref="HandleReconfigure"/>) and
|
||||
/// <see cref="ReconnectCause.ServiceLevelFailover"/> (ScadaBridge#138) both mean the adapter
|
||||
/// has just been swapped ON PURPOSE rather than the link lost. They suppress the
|
||||
/// unstable-disconnect accounting and the failover check — counting a deliberate swap as an
|
||||
/// unstable cycle would let it trip a second, connection-driven failover before the new
|
||||
/// endpoint has been tried even once — and reconnect immediately instead of waiting out
|
||||
/// <see cref="DataConnectionOptions.ReconnectInterval"/>, since there is nothing to wait for.
|
||||
/// </param>
|
||||
private void BecomeReconnecting(bool afterReconfigure = false)
|
||||
private void BecomeReconnecting(ReconnectCause cause = ReconnectCause.ConnectionLost)
|
||||
{
|
||||
var deliberateSwap = cause != ReconnectCause.ConnectionLost;
|
||||
_log.Warning("[{0}] Entering Reconnecting state{1}", _connectionName,
|
||||
afterReconfigure ? " (reconfigured in place)" : string.Empty);
|
||||
// The session is over: abandon its re-seed (02/L4).
|
||||
cause switch
|
||||
{
|
||||
ReconnectCause.Reconfigured => " (reconfigured in place)",
|
||||
ReconnectCause.ServiceLevelFailover => " (ServiceLevel failover)",
|
||||
_ => string.Empty,
|
||||
});
|
||||
// The session is over: abandon its re-seed (02/L4) and its ServiceLevel watch (#138).
|
||||
CancelSessionWork();
|
||||
StopServiceLevelMonitoring();
|
||||
_healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected);
|
||||
_healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected");
|
||||
// Register row 54 — drops this connection out of the connected gauge while the
|
||||
// total holds, which is exactly the connected < total alert condition.
|
||||
ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected);
|
||||
|
||||
if (!afterReconfigure)
|
||||
if (!deliberateSwap)
|
||||
{
|
||||
// Track unstable connections toward failover.
|
||||
// If we were connected for less than the stability threshold, this counts
|
||||
@@ -671,15 +694,10 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// Failover if we keep connecting and going stale repeatedly
|
||||
if (_backupConfig != null && _consecutiveUnstableDisconnects >= _failoverRetryCount)
|
||||
{
|
||||
var previousEndpoint = _activeEndpoint;
|
||||
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
|
||||
? ActiveEndpoint.Backup
|
||||
: ActiveEndpoint.Primary;
|
||||
var previousEndpoint = SwitchToOtherEndpoint();
|
||||
_consecutiveUnstableDisconnects = 0;
|
||||
_consecutiveFailures = 0;
|
||||
|
||||
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
|
||||
|
||||
_log.Warning("[{0}] Failing over from {1} to {2} (unstable connection pattern)",
|
||||
_connectionName, previousEndpoint, _activeEndpoint);
|
||||
|
||||
@@ -697,10 +715,13 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
if (_siteEventLogger != null)
|
||||
{
|
||||
_ = _siteEventLogger.LogEventAsync(
|
||||
"connection", afterReconfigure ? "Info" : "Warning", null, _connectionName,
|
||||
afterReconfigure
|
||||
? "Connection reconfigured — reconnecting on the new configuration"
|
||||
: "Connection lost — entering reconnect cycle",
|
||||
"connection", cause == ReconnectCause.ConnectionLost ? "Warning" : "Info", null, _connectionName,
|
||||
cause switch
|
||||
{
|
||||
ReconnectCause.Reconfigured => "Connection reconfigured — reconnecting on the new configuration",
|
||||
ReconnectCause.ServiceLevelFailover => $"ServiceLevel failover — reconnecting on the {_activeEndpoint} endpoint",
|
||||
_ => "Connection lost — entering reconnect cycle",
|
||||
},
|
||||
null);
|
||||
}
|
||||
|
||||
@@ -716,7 +737,25 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// Schedule reconnect attempt. StartSingleTimer REPLACES any reconnect already armed,
|
||||
// so a reconfigure landing mid-backoff cannot leave two AttemptConnect messages racing.
|
||||
Timers.StartSingleTimer("reconnect", new AttemptConnect(),
|
||||
afterReconfigure ? TimeSpan.Zero : _options.ReconnectInterval);
|
||||
deliberateSwap ? TimeSpan.Zero : _options.ReconnectInterval);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Flips <see cref="_activeEndpoint"/> to the other configured endpoint and swaps the adapter
|
||||
/// onto it — the one endpoint switch shared by every failover trigger: consecutive connect
|
||||
/// failures (<see cref="CountFailureAndMaybeFailover"/>), the unstable-connection pattern
|
||||
/// (<see cref="BecomeReconnecting"/>) and a sustained Degraded ServiceLevel (ScadaBridge#138).
|
||||
/// Callers reset whichever counters their trigger owns. Requires a backup configuration.
|
||||
/// </summary>
|
||||
/// <returns>The endpoint that was active before the switch.</returns>
|
||||
private ActiveEndpoint SwitchToOtherEndpoint()
|
||||
{
|
||||
var previousEndpoint = _activeEndpoint;
|
||||
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
|
||||
? ActiveEndpoint.Backup
|
||||
: ActiveEndpoint.Primary;
|
||||
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig!);
|
||||
return previousEndpoint;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -845,6 +884,8 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
_primaryConfig = command.PrimaryConnectionDetails;
|
||||
_backupConfig = command.BackupConnectionDetails;
|
||||
_failoverRetryCount = command.FailoverRetryCount;
|
||||
_serviceLevelPolicy = ServiceLevelFailoverPolicy.From(_primaryConfig);
|
||||
_lastServiceLevelSwitchAt = DateTimeOffset.MinValue;
|
||||
|
||||
_activeEndpoint = ActiveEndpoint.Primary;
|
||||
_consecutiveFailures = 0;
|
||||
@@ -852,7 +893,7 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
|
||||
SwapAdapter(_primaryConfig);
|
||||
|
||||
BecomeReconnecting(afterReconfigure: true);
|
||||
BecomeReconnecting(ReconnectCause.Reconfigured);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -930,6 +971,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// Ignore — the adapter has no live session; ReSubscribeAll's chunk
|
||||
// completions re-arm the retry timer after reconnect.
|
||||
break;
|
||||
case ServiceLevelPollTick or ServiceLevelSampled or ServiceLevelProbed:
|
||||
// ScadaBridge#138: belongs to a session that is over; BecomeConnected restarts it.
|
||||
break;
|
||||
case QualityFlushTick:
|
||||
FlushQualityCountersIfDirty();
|
||||
break;
|
||||
@@ -1219,14 +1263,9 @@ public class DataConnectionActor : UntypedActor, IWithTimers
|
||||
// Failover: switch endpoint after exhausting retry count (only if backup is configured)
|
||||
if (_backupConfig != null && _consecutiveFailures >= _failoverRetryCount)
|
||||
{
|
||||
var previousEndpoint = _activeEndpoint;
|
||||
_activeEndpoint = _activeEndpoint == ActiveEndpoint.Primary
|
||||
? ActiveEndpoint.Backup
|
||||
: ActiveEndpoint.Primary;
|
||||
var previousEndpoint = SwitchToOtherEndpoint();
|
||||
_consecutiveFailures = 0;
|
||||
|
||||
SwapAdapter(_activeEndpoint == ActiveEndpoint.Primary ? _primaryConfig : _backupConfig);
|
||||
|
||||
_log.Warning("[{0}] Failing over from {1} to {2}",
|
||||
_connectionName, previousEndpoint, _activeEndpoint);
|
||||
|
||||
|
||||
@@ -21,8 +21,11 @@ namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
|
||||
/// </summary>
|
||||
public class OpcUaDataConnection
|
||||
: IDataConnection, IBrowsableDataConnection, IAlarmSubscribableConnection, IAddressSpaceSearchable,
|
||||
IBatchSubscribableConnection
|
||||
IBatchSubscribableConnection, IServiceLevelReadable
|
||||
{
|
||||
/// <summary>The standard <c>Server_ServiceLevel</c> node (ns=0), OPC UA Part 5.</summary>
|
||||
internal const string ServiceLevelNodeId = "i=2267";
|
||||
|
||||
private readonly IOpcUaClientFactory _clientFactory;
|
||||
private readonly ILogger<OpcUaDataConnection> _logger;
|
||||
private IOpcUaClient? _client;
|
||||
@@ -291,6 +294,39 @@ public class OpcUaDataConnection
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
/// <remarks>
|
||||
/// ScadaBridge#138. Unlike <see cref="ReadAsync"/>, a failure here never raises
|
||||
/// <see cref="Disconnected"/>: the poll runs every few seconds alongside the keep-alive, and a
|
||||
/// dead session is already detected (and reconnected) there. A failure only means "unknown".
|
||||
/// </remarks>
|
||||
public async Task<byte?> ReadServiceLevelAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
var client = _client;
|
||||
if (client == null || !client.IsConnected)
|
||||
return null;
|
||||
|
||||
try
|
||||
{
|
||||
var (value, _, statusCode) = await client.ReadValueAsync(ServiceLevelNodeId, cancellationToken);
|
||||
if (MapStatusCode(statusCode) == QualityCode.Bad)
|
||||
return null;
|
||||
return value switch
|
||||
{
|
||||
byte b => b,
|
||||
sbyte or short or ushort or int or uint or long or ulong
|
||||
when Convert.ToInt64(value, System.Globalization.CultureInfo.InvariantCulture) is >= 0 and <= 255
|
||||
=> (byte)Convert.ToInt64(value, System.Globalization.CultureInfo.InvariantCulture),
|
||||
_ => null,
|
||||
};
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
_logger.LogDebug(ex, "OPC UA ServiceLevel read failed for {Endpoint}", _endpointUrl);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
|
||||
{
|
||||
|
||||
@@ -119,4 +119,19 @@ public class DataConnectionOptions
|
||||
/// <c>(SeedReadMaxAttempts - 1) × SeedReadRetryDelay</c> regardless of tag count.
|
||||
/// </summary>
|
||||
public TimeSpan SeedReadRetryDelay { get; set; } = TimeSpan.FromMilliseconds(250);
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138: how often a connected endpoint's ServiceLevel (<c>i=2267</c>) is read when
|
||||
/// ServiceLevel-aware failover is in force (OPC UA connection with a backup and the
|
||||
/// connection's <c>ServiceLevelFailover.Enabled</c>). One single-node Read per poll. Must be
|
||||
/// shorter than the smallest configured degraded window to make that window meaningful.
|
||||
/// </summary>
|
||||
public TimeSpan ServiceLevelPollInterval { get; set; } = TimeSpan.FromSeconds(2);
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138: budget for reading a ServiceLevel — the poll on the connected endpoint,
|
||||
/// and the probe of the other endpoint (connect a short-lived session, read, close). A probe
|
||||
/// that does not finish in time counts as "other endpoint unreachable", so no switch happens.
|
||||
/// </summary>
|
||||
public TimeSpan ServiceLevelProbeTimeout { get; set; } = TimeSpan.FromSeconds(15);
|
||||
}
|
||||
|
||||
@@ -64,5 +64,11 @@ public sealed class DataConnectionOptionsValidator : OptionsValidatorBase<DataCo
|
||||
builder.RequireThat(options.MxAlarmStreamReopenDebounce >= TimeSpan.Zero
|
||||
&& options.MxAlarmStreamReopenDebounce <= TimeSpan.FromSeconds(5),
|
||||
$"ScadaBridge:DataConnection:MxAlarmStreamReopenDebounce must be between 0 and 5 seconds (was {options.MxAlarmStreamReopenDebounce}).");
|
||||
|
||||
builder.RequireThat(options.ServiceLevelPollInterval > TimeSpan.Zero,
|
||||
$"ScadaBridge:DataConnection:ServiceLevelPollInterval must be a positive duration (was {options.ServiceLevelPollInterval}).");
|
||||
|
||||
builder.RequireThat(options.ServiceLevelProbeTimeout > TimeSpan.Zero,
|
||||
$"ScadaBridge:DataConnection:ServiceLevelProbeTimeout must be a positive duration (was {options.ServiceLevelProbeTimeout}).");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -167,4 +167,74 @@ public class DataConnectionFormTests : BunitContext
|
||||
Assert.Equal(60000,
|
||||
doc.RootElement.GetProperty("sessionTimeoutMs").GetInt32());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138: with a backup, the ServiceLevel failover fields shown next to Failover
|
||||
/// Retry Count are stored on the PRIMARY endpoint's JSON; the backup's copy stays null.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Save_WithBackup_StoresServiceLevelFailoverOnThePrimary()
|
||||
{
|
||||
DataConnection? captured = null;
|
||||
await _siteRepo.AddDataConnectionAsync(Arg.Do<DataConnection>(d => captured = d));
|
||||
|
||||
var cut = RenderForCreateSite(1);
|
||||
cut.FindAll("input[type='text']")
|
||||
.First(i => !i.HasAttribute("readonly")
|
||||
&& (i.GetAttribute("placeholder") is null
|
||||
|| !i.GetAttribute("placeholder")!.StartsWith("opc.tcp")))
|
||||
.Change("OtOpcUa-Pair");
|
||||
cut.FindAll("input[type='text']")
|
||||
.First(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true)
|
||||
.Change("opc.tcp://otopcua-a:4840");
|
||||
await cut.FindAll("button").First(b => b.TextContent.Trim() == "Add Backup Endpoint").ClickAsync(new());
|
||||
cut.FindAll("input[type='text']")
|
||||
.Where(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true)
|
||||
.ElementAt(1)
|
||||
.Change("opc.tcp://otopcua-b:4840");
|
||||
|
||||
// Defaults are on screen, then edited.
|
||||
Assert.Equal("10", cut.Find("#sl-failover-window").GetAttribute("value"));
|
||||
Assert.Equal("60", cut.Find("#sl-failover-dwell").GetAttribute("value"));
|
||||
cut.Find("#sl-failover-window").Change("15");
|
||||
cut.Find("#sl-failover-dwell").Change("90");
|
||||
|
||||
await cut.FindAll("button").First(b => b.TextContent.Trim() == "Save").ClickAsync(new());
|
||||
|
||||
Assert.NotNull(captured);
|
||||
using var primary = JsonDocument.Parse(captured!.PrimaryConfiguration!);
|
||||
var sl = primary.RootElement.GetProperty("serviceLevelFailover");
|
||||
Assert.True(sl.GetProperty("enabled").GetBoolean());
|
||||
Assert.Equal(15, sl.GetProperty("degradedWindowSeconds").GetInt32());
|
||||
Assert.Equal(90, sl.GetProperty("minDwellSeconds").GetInt32());
|
||||
|
||||
using var backup = JsonDocument.Parse(captured.BackupConfiguration!);
|
||||
Assert.Equal(JsonValueKind.Null, backup.RootElement.GetProperty("serviceLevelFailover").ValueKind);
|
||||
}
|
||||
|
||||
/// <summary>ScadaBridge#138: without a backup the setting is neither shown nor stored.</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Save_WithoutBackup_StoresNoServiceLevelFailover()
|
||||
{
|
||||
DataConnection? captured = null;
|
||||
await _siteRepo.AddDataConnectionAsync(Arg.Do<DataConnection>(d => captured = d));
|
||||
|
||||
var cut = RenderForCreateSite(1);
|
||||
Assert.Empty(cut.FindAll("#sl-failover-enabled"));
|
||||
cut.FindAll("input[type='text']")
|
||||
.First(i => !i.HasAttribute("readonly")
|
||||
&& (i.GetAttribute("placeholder") is null
|
||||
|| !i.GetAttribute("placeholder")!.StartsWith("opc.tcp")))
|
||||
.Change("PLC-1");
|
||||
cut.FindAll("input[type='text']")
|
||||
.First(i => i.GetAttribute("placeholder")?.StartsWith("opc.tcp") == true)
|
||||
.Change("opc.tcp://plant-a:4840");
|
||||
await cut.FindAll("button").First(b => b.TextContent.Trim() == "Save").ClickAsync(new());
|
||||
|
||||
Assert.NotNull(captured);
|
||||
using var primary = JsonDocument.Parse(captured!.PrimaryConfiguration!);
|
||||
Assert.Equal(JsonValueKind.Null, primary.RootElement.GetProperty("serviceLevelFailover").ValueKind);
|
||||
}
|
||||
}
|
||||
|
||||
+54
@@ -504,4 +504,58 @@ public class OpcUaEndpointConfigSerializerTests
|
||||
Assert.NotNull(config.UserIdentity);
|
||||
Assert.Equal(OpcUaUserTokenType.Anonymous, config.UserIdentity!.TokenType);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138: the ServiceLevel failover object round-trips through the stored JSON and
|
||||
/// through the flat dict the site hands the adapter, under its documented keys.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ServiceLevelFailover_RoundTrips_JsonAndFlatDict()
|
||||
{
|
||||
var original = new OpcUaEndpointConfig
|
||||
{
|
||||
EndpointUrl = "opc.tcp://otopcua-a:4840",
|
||||
ServiceLevelFailover = new OpcUaServiceLevelFailoverConfig
|
||||
{
|
||||
Enabled = false, DegradedWindowSeconds = 25, MinDwellSeconds = 120,
|
||||
},
|
||||
};
|
||||
|
||||
var json = OpcUaEndpointConfigSerializer.Serialize(original);
|
||||
Assert.Contains("\"serviceLevelFailover\":{\"enabled\":false,\"degradedWindowSeconds\":25,\"minDwellSeconds\":120}", json);
|
||||
var (round, isLegacy) = OpcUaEndpointConfigSerializer.Deserialize(json);
|
||||
Assert.False(isLegacy);
|
||||
Assert.NotNull(round.ServiceLevelFailover);
|
||||
Assert.False(round.ServiceLevelFailover!.Enabled);
|
||||
Assert.Equal(25, round.ServiceLevelFailover.DegradedWindowSeconds);
|
||||
Assert.Equal(120, round.ServiceLevelFailover.MinDwellSeconds);
|
||||
|
||||
var flat = OpcUaEndpointConfigSerializer.ToFlatDict(round);
|
||||
Assert.Equal("False", flat["ServiceLevelFailover.Enabled"]);
|
||||
Assert.Equal("25", flat["ServiceLevelFailover.DegradedWindowSeconds"]);
|
||||
Assert.Equal("120", flat["ServiceLevelFailover.MinDwellSeconds"]);
|
||||
var back = OpcUaEndpointConfigSerializer.FromFlatDict(flat).ServiceLevelFailover;
|
||||
Assert.NotNull(back);
|
||||
Assert.False(back!.Enabled);
|
||||
Assert.Equal(25, back.DegradedWindowSeconds);
|
||||
Assert.Equal(120, back.MinDwellSeconds);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// ScadaBridge#138, upgrade safety: a row stored before the setting existed has no
|
||||
/// <c>serviceLevelFailover</c>; it deserializes as Typed (not Malformed) with the object null,
|
||||
/// and its flat dict gains no new keys — so existing connections are not seen as changed.
|
||||
/// </summary>
|
||||
[Fact]
|
||||
public void ServiceLevelFailover_AbsentFromAnExistingRow_IsNull_AndAddsNoFlatKeys()
|
||||
{
|
||||
const string stored = "{\"endpointUrl\":\"opc.tcp://plant-a:4840\",\"securityMode\":\"none\",\"heartbeat\":null}";
|
||||
var result = OpcUaEndpointConfigSerializer.Deserialize(stored);
|
||||
Assert.Equal(OpcUaConfigParseStatus.Typed, result.Status);
|
||||
Assert.Null(result.Config.ServiceLevelFailover);
|
||||
|
||||
var flat = OpcUaEndpointConfigSerializer.ToFlatDict(result.Config);
|
||||
Assert.DoesNotContain(flat.Keys, k => k.StartsWith("ServiceLevelFailover.", StringComparison.Ordinal));
|
||||
Assert.Null(OpcUaEndpointConfigSerializer.FromFlatDict(flat).ServiceLevelFailover);
|
||||
}
|
||||
}
|
||||
|
||||
+26
@@ -220,4 +220,30 @@ public class OpcUaEndpointConfigValidatorTests
|
||||
var r = OpcUaEndpointConfigValidator.Validate(c, fieldPrefix: "Primary.");
|
||||
Assert.Contains(r.Errors, e => e.EntityName == "Primary.UserIdentity.Username");
|
||||
}
|
||||
|
||||
/// <summary>ScadaBridge#138: the ServiceLevel failover window and dwell are range-checked, prefixed.</summary>
|
||||
/// <param name="window">Degraded window, seconds.</param>
|
||||
/// <param name="dwell">Minimum dwell, seconds.</param>
|
||||
/// <param name="badField">The field expected to fail, or null when valid.</param>
|
||||
[Theory]
|
||||
[InlineData(10, 60, null)]
|
||||
[InlineData(1, 0, null)]
|
||||
[InlineData(3600, 86400, null)]
|
||||
[InlineData(0, 60, "Primary.ServiceLevelFailover.DegradedWindowSeconds")]
|
||||
[InlineData(3601, 60, "Primary.ServiceLevelFailover.DegradedWindowSeconds")]
|
||||
[InlineData(10, -1, "Primary.ServiceLevelFailover.MinDwellSeconds")]
|
||||
[InlineData(10, 86401, "Primary.ServiceLevelFailover.MinDwellSeconds")]
|
||||
public void Validate_ServiceLevelFailoverRanges(int window, int dwell, string? badField)
|
||||
{
|
||||
var c = Valid();
|
||||
c.ServiceLevelFailover = new OpcUaServiceLevelFailoverConfig
|
||||
{
|
||||
DegradedWindowSeconds = window, MinDwellSeconds = dwell,
|
||||
};
|
||||
var r = OpcUaEndpointConfigValidator.Validate(c, fieldPrefix: "Primary.");
|
||||
if (badField is null)
|
||||
Assert.True(r.IsValid);
|
||||
else
|
||||
Assert.Contains(r.Errors, e => e.EntityName == badField);
|
||||
}
|
||||
}
|
||||
|
||||
+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