dohertj2/lmxopcua
1
0
Fork 0
Code Issues 32 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(opcuaclient): review fixes - drop driver-side OperationLimits chunking (SDK batches), diagnostics as OTel counter, safe discovery selection, options validation at load (#273 #275 #276 #277 #278)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 14:55:53 -04:00
parent 935ba8d1d4
commit 1219b7b13e
13 changed files with 406 additions and 652 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Contracts/OpcUaClientDriverOptions.cs
+10 -15
View File
@@ -203,33 +203,20 @@ public sealed record OpcUaClientDriverOptions
/// <summary>
/// Publish cycles with no data before the server sends a keep-alive (default 10). Must be &gt;= 1.
/// </summary>
[Range(1, int.MaxValue)]
public uint SubscriptionKeepAliveCount { get; init; } = 10;
/// <summary>
/// Publish cycles with no Publish request before the server drops the subscription (default 1000).
/// Must be at least 3x <see cref="SubscriptionKeepAliveCount"/> (OPC UA Part 4 §5.13.2).
/// </summary>
[Range(3, int.MaxValue)]
public uint SubscriptionLifetimeCount { get; init; } = 1000;
/// <summary>Max notifications per Publish response; <c>0</c> (default) = server decides.</summary>
public uint SubscriptionMaxNotificationsPerPublish { get; init; }
/// <summary>Relative subscription priority 0-255 (default 0).</summary>
[Range(0, 255)]
public byte SubscriptionPriority { get; init; }
// ---- #275 ----
/// <summary>
/// When <c>true</c> (default) the driver reads the remote server's
/// <c>ServerCapabilities/OperationLimits</c> after connect and splits Read / Write / CreateMonitoredItems
/// requests that exceed them (a declared limit of 0 = unlimited). Set <c>false</c> to send every request
/// whole and let the server answer <c>BadTooManyOperations</c> itself.
/// </summary>
public bool HonorServerOperationLimits { get; init; } = true;
// ---- #277 ----
/// <summary>Certificate-revocation / issuer-store settings. Defaults preserve the SDK behaviour.</summary>
@@ -241,11 +228,19 @@ public sealed record OpcUaClientDriverOptions
/// Optional discovery endpoint (a Local Discovery Server or the server itself), e.g.
/// <c>opc.tcp://lds.internal:4840</c>. When set, <c>InitializeAsync</c> calls <c>FindServers</c> and
/// <c>GetEndpoints</c> on it and tries the discovered URLs that match <see cref="SecurityPolicy"/> +
/// <see cref="SecurityMode"/> <i>before</i> the statically configured <see cref="EndpointUrls"/> /
/// <see cref="EndpointUrl"/>. A discovery failure is non-fatal: the static list is used. Discovery
/// <see cref="SecurityMode"/> <i>after</i> the statically configured <see cref="EndpointUrls"/> /
/// <see cref="EndpointUrl"/> (pin the server with <see cref="DiscoveryServerUri"/>). A discovery failure is non-fatal: the static list is used. Discovery
/// itself always runs over an unsecured channel, as the spec requires. <c>null</c> (default) = off.
/// </summary>
public string? DiscoveryUrl { get; init; }
/// <summary>
/// ApplicationUri of the one server discovery may resolve to. An LDS usually registers several servers;
/// with this set only that server's endpoints are used. When unset and discovery finds more than one
/// server, connect fails with an error listing their ApplicationUris. Discovered endpoints are always
/// tried AFTER the configured <see cref="EndpointUrls"/>.
/// </summary>
public string? DiscoveryServerUri { get; init; }
}
/// <summary>
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDiagnostics.cs
+32 -86
View File
@@ -1,102 +1,48 @@
using ZB.MOM.WW.OtOpcUa.Commons.Observability;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// <summary>
/// Thread-safe counters for the OPC UA Client driver (#276), surfaced through
/// <c>DriverHealth.Diagnostics</c>. Writers are the SDK's notification / keep-alive threads and
/// the request paths, so every field is touched only through <see cref="Interlocked"/>.
/// Publishes an OPC UA Client driver instance's session/publish events (#276) on the shared
/// <c>otopcua.driver.opcuaclient.events</c> OTel counter, tagged with <c>driver_instance</c> and
/// <c>event</c>. The counter is thread-safe, so the SDK's notification / keep-alive threads record
/// directly with no locking here.
/// </summary>
internal sealed class OpcUaClientDiagnostics
{
private long _publishResponses;
private long _notifications;
private long _dataChanges;
private long _badStatusValues;
private long _publishErrors;
private long _keepAliveMisses;
private long _reconnectAttempts;
private long _reconnectsSucceeded;
private long _reconnectsExhausted;
private long _lastReconnectUtcTicks;
private long _readCalls;
private long _readFailures;
private long _writeCalls;
private long _writeFailures;
private long _chunkedRequests;
/// <summary>A NotificationMessage arrived (one per Publish response that carried data).</summary>
public void RecordPublishResponse() => Interlocked.Increment(ref _publishResponses);
/// <summary>A Publish failed (missing publish requests, subscription errors).</summary>
public void RecordPublishError() => Interlocked.Increment(ref _publishErrors);
public const string PublishResponse = "publish_response";
/// <summary>Items carried by notification messages.</summary>
public const string Notification = "notification";
/// <summary>One monitored-item data change was delivered.</summary>
/// <param name="bad">Whether the value's status code was Bad.</param>
public void RecordDataChange(bool bad)
{
Interlocked.Increment(ref _dataChanges);
if (bad) Interlocked.Increment(ref _badStatusValues);
}
/// <summary>Count raw notification items carried by a message.</summary>
/// <param name="count">Number of items in the message.</param>
public void RecordNotifications(int count) => Interlocked.Add(ref _notifications, count);
public const string DataChange = "data_change";
/// <summary>A delivered data change carried a Bad status.</summary>
public const string BadStatusValue = "bad_status_value";
/// <summary>A Publish failed (missing publish requests, subscription errors).</summary>
public const string PublishError = "publish_error";
/// <summary>A keep-alive came back Bad.</summary>
public void RecordKeepAliveMiss() => Interlocked.Increment(ref _keepAliveMisses);
public const string KeepAliveMiss = "keepalive_miss";
/// <summary>The reconnect handler was armed.</summary>
public void RecordReconnectAttempt() => Interlocked.Increment(ref _reconnectAttempts);
public const string ReconnectAttempt = "reconnect_attempt";
/// <summary>A reconnect landed a new session.</summary>
public void RecordReconnectSucceeded()
{
Interlocked.Increment(ref _reconnectsSucceeded);
Interlocked.Exchange(ref _lastReconnectUtcTicks, DateTime.UtcNow.Ticks);
}
public const string ReconnectSuccess = "reconnect_success";
/// <summary>A reconnect handler exhausted its retry window.</summary>
public void RecordReconnectExhausted() => Interlocked.Increment(ref _reconnectsExhausted);
public const string ReconnectExhausted = "reconnect_exhausted";
/// <summary>A Read request (all chunks) finished.</summary>
/// <param name="failed">Whether any chunk failed on the wire.</param>
public void RecordRead(bool failed)
private readonly string _driverInstanceId;
/// <summary>Creates the recorder for one driver instance.</summary>
/// <param name="driverInstanceId">Value of the <c>driver_instance</c> tag.</param>
public OpcUaClientDiagnostics(string driverInstanceId) => _driverInstanceId = driverInstanceId;
/// <summary>Add <paramref name="count"/> to the counter for <paramref name="evt"/>.</summary>
/// <param name="evt">One of the event constants.</param>
/// <param name="count">Amount to add (default 1).</param>
public void Record(string evt, long count = 1)
{
Interlocked.Increment(ref _readCalls);
if (failed) Interlocked.Increment(ref _readFailures);
if (count <= 0) return;
OtOpcUaTelemetry.OpcUaClientEvents.Add(count,
new KeyValuePair<string, object?>(OtOpcUaTelemetry.PollOverrunDriverInstanceTag, _driverInstanceId),
new KeyValuePair<string, object?>("event", evt));
}
/// <summary>A Write request (all chunks) finished.</summary>
/// <param name="failed">Whether any chunk failed on the wire.</param>
public void RecordWrite(bool failed)
{
Interlocked.Increment(ref _writeCalls);
if (failed) Interlocked.Increment(ref _writeFailures);
}
/// <summary>A logical request was split into more than one wire call to honour a server OperationLimit.</summary>
public void RecordChunkedRequest() => Interlocked.Increment(ref _chunkedRequests);
/// <summary>
/// Point-in-time copy for <c>DriverHealth.Diagnostics</c>. Keys are stable and
/// <c>PascalCase</c>; <c>LastReconnectUtcTicks</c> is 0 until a reconnect succeeds.
/// </summary>
/// <returns>A fresh dictionary the caller may keep.</returns>
public IReadOnlyDictionary<string, double> Snapshot() => new Dictionary<string, double>
{
["PublishResponseCount"] = Interlocked.Read(ref _publishResponses),
["NotificationCount"] = Interlocked.Read(ref _notifications),
["DataChangeCount"] = Interlocked.Read(ref _dataChanges),
["BadStatusValueCount"] = Interlocked.Read(ref _badStatusValues),
["PublishErrorCount"] = Interlocked.Read(ref _publishErrors),
["KeepAliveMissCount"] = Interlocked.Read(ref _keepAliveMisses),
["ReconnectAttemptCount"] = Interlocked.Read(ref _reconnectAttempts),
["ReconnectSuccessCount"] = Interlocked.Read(ref _reconnectsSucceeded),
["ReconnectExhaustedCount"] = Interlocked.Read(ref _reconnectsExhausted),
["LastReconnectUtcTicks"] = Interlocked.Read(ref _lastReconnectUtcTicks),
["ReadCallCount"] = Interlocked.Read(ref _readCalls),
["ReadFailureCount"] = Interlocked.Read(ref _readFailures),
["WriteCallCount"] = Interlocked.Read(ref _writeCalls),
["WriteFailureCount"] = Interlocked.Read(ref _writeFailures),
["ChunkedRequestCount"] = Interlocked.Read(ref _chunkedRequests),
};
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriver.cs
+137 -163
View File
@@ -79,6 +79,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
_options = options;
_driverInstanceId = driverInstanceId;
_diag = new OpcUaClientDiagnostics(driverInstanceId);
_logger = logger ?? NullLogger<OpcUaClientDriver>.Instance;
_secretResolver = secretResolver ?? NullSecretResolver.Instance;
}
@@ -144,21 +145,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <param name="map">The namespace map to install, or <c>null</c> to clear it.</param>
internal void SetNamespaceMapForTest(NamespaceMap? map) => _namespaceMap = map;
/// <summary>Test-only seam installing the OperationLimits that connect would have read (#275).</summary>
/// <param name="limits">The limits to apply to subsequent Read / Write / subscribe calls.</param>
internal void SetOperationLimitsForTest(OpcUaClientOperationLimits limits) => _limits = limits;
/// <summary>Test-only view of the diagnostics counters (#276).</summary>
internal OpcUaClientDiagnostics DiagnosticsForTest => _diag;
private DriverHealth _health = new(DriverState.Unknown, null, null);
/// <summary>Counters surfaced through <see cref="GetHealth"/> (#276).</summary>
private readonly OpcUaClientDiagnostics _diag = new();
/// <summary>Server OperationLimits read at connect (#275); <c>None</c> when unread / disabled.
/// Replaced wholesale (never mutated), so readers just snapshot the reference.</summary>
private OpcUaClientOperationLimits _limits = OpcUaClientOperationLimits.None;
/// <summary>Per-instance event counters published as OTel metrics (#276).</summary>
private readonly OpcUaClientDiagnostics _diag;
private NotificationEventHandler? _notificationHandler;
private PublishErrorEventHandler? _publishErrorHandler;
@@ -238,9 +231,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
IReadOnlyList<string> discovered = [];
if (!string.IsNullOrWhiteSpace(_options.DiscoveryUrl))
{
IReadOnlyList<DiscoveredEndpoint> found = [];
try
{
discovered = await DiscoverEndpointsAsync(
found = await DiscoverEndpointsAsync(
appConfig, _options.DiscoveryUrl!, _options.SecurityPolicy, _options.SecurityMode,
cancellationToken).ConfigureAwait(false);
}
@@ -249,8 +243,12 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
_logger.LogWarning(ex,
"OpcUaClientDriver '{DriverInstanceId}': discovery against '{DiscoveryUrl}' failed; " +
"falling back to the configured endpoint list.", _driverInstanceId, _options.DiscoveryUrl);
"using the configured endpoint list only.", _driverInstanceId, _options.DiscoveryUrl);
}
// Selection is OUTSIDE the try on purpose: an ambiguous LDS must fail the connect with a
// clear message, not be swallowed into "discovery unavailable".
discovered = SelectDiscoveredUrls(found, _options.DiscoveryServerUri);
}
var candidates = ResolveEndpointCandidates(_options, discovered);
@@ -306,9 +304,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_keepAliveHandler = OnKeepAlive;
session.KeepAlive += _keepAliveHandler;
AttachDiagnostics(session);
_limits = _options.HonorServerOperationLimits
? await OpcUaClientOperationLimits.ReadAsync(session, cancellationToken).ConfigureAwait(false)
: OpcUaClientOperationLimits.None;
// Build the bidirectional namespace map from the freshly negotiated session's
// NamespaceUris (driver-specs.md §8 "Namespace Remapping"). Stored NodeIds carry
@@ -460,10 +455,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
ResolveEndpointCandidates(opts, []);
/// <summary>
/// Resolve the ordered failover list with optional discovery results (#278). Discovered URLs are
/// <b>prepended</b> so a discovery sweep gets first attempt; static entries follow, de-duplicated
/// case-insensitively so one server listed in both is not attempted twice in a row. With no
/// discovered URLs the result is exactly the pre-discovery behaviour.
/// Resolve the ordered failover list with optional discovery results (#278). The configured
/// endpoints always come FIRST — a discovered endpoint can only ever be a fallback, never outrank or
/// replace the server the operator named. Discovered URLs follow, de-duplicated case-insensitively.
/// With no discovered URLs the result is exactly the pre-discovery behaviour.
/// </summary>
/// <param name="opts">Driver options containing endpoint configuration.</param>
/// <param name="discovered">URLs resolved via <see cref="DiscoverEndpointsAsync"/> (may be empty).</param>
@@ -476,10 +471,10 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var seen = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
var merged = new List<string>(discovered.Count + staticList.Count);
foreach (var u in discovered)
if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u);
foreach (var u in staticList)
if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u);
foreach (var u in discovered)
if (!string.IsNullOrWhiteSpace(u) && seen.Add(u)) merged.Add(u);
return merged;
}
@@ -495,7 +490,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
/// <param name="mode">Required security mode.</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>Matching endpoint URLs in discovery order, de-duplicated.</returns>
internal static async Task<IReadOnlyList<string>> DiscoverEndpointsAsync(
internal static async Task<IReadOnlyList<DiscoveredEndpoint>> DiscoverEndpointsAsync(
ApplicationConfiguration appConfig, string discoveryUrl,
OpcUaSecurityPolicy policy, OpcUaSecurityMode mode, CancellationToken ct)
{
@@ -509,7 +504,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
servers = await lds.FindServersAsync(null, ct).ConfigureAwait(false);
}
var urls = new List<string>();
var found = new List<DiscoveredEndpoint>();
foreach (var server in servers)
{
if (server.DiscoveryUrls is null) continue;
@@ -526,10 +521,47 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
catch (OperationCanceledException) when (ct.IsCancellationRequested) { throw; }
catch { continue; }
urls.AddRange(FilterEndpointUrls(endpoints, wantedPolicyUri, wantedMode));
found.AddRange(FilterEndpointUrls(endpoints, wantedPolicyUri, wantedMode)
.Select(u => new DiscoveredEndpoint(server.ApplicationUri ?? string.Empty, u)));
}
}
return urls.Distinct(StringComparer.OrdinalIgnoreCase).ToList();
return found;
}
/// <summary>One endpoint URL found by discovery, with the ApplicationUri of the server advertising it.</summary>
/// <param name="ApplicationUri">The advertising server's ApplicationUri (empty when it published none).</param>
/// <param name="Url">The endpoint URL.</param>
internal readonly record struct DiscoveredEndpoint(string ApplicationUri, string Url);
/// <summary>
/// Decide which discovered endpoints may be used (#278). An LDS can register many servers (identical
/// line PLCs, a gateway plus a historian), and the first one that accepts a connection would be bound
/// silently — mis-attributing every value and write. So: with <paramref name="serverUri"/> set, only
/// that ApplicationUri's endpoints; unset and exactly one distinct server, that server; unset and
/// several, throw naming them so the operator pins one.
/// </summary>
/// <param name="found">Matching endpoints from every server the LDS listed.</param>
/// <param name="serverUri">Optional <c>DiscoveryServerUri</c> filter.</param>
/// <returns>The usable URLs, de-duplicated, in discovery order (possibly empty).</returns>
/// <exception cref="InvalidOperationException">Several servers discovered and no filter configured.</exception>
internal static IReadOnlyList<string> SelectDiscoveredUrls(
IReadOnlyList<DiscoveredEndpoint> found, string? serverUri)
{
IEnumerable<DiscoveredEndpoint> chosen = found;
if (!string.IsNullOrWhiteSpace(serverUri))
{
chosen = found.Where(e => string.Equals(e.ApplicationUri, serverUri, StringComparison.Ordinal));
}
else
{
var servers = found.Select(e => e.ApplicationUri).Distinct(StringComparer.Ordinal).ToList();
if (servers.Count > 1)
throw new InvalidOperationException(
"OpcUaClient discovery found endpoints for more than one server " +
$"({string.Join(", ", servers.Select(u => string.IsNullOrEmpty(u) ? "<no ApplicationUri>" : u))}). " +
"Set DiscoveryServerUri to the ApplicationUri of the server this driver should use.");
}
return chosen.Select(e => e.Url).Distinct(StringComparer.OrdinalIgnoreCase).ToList();
}
/// <summary>The endpoint URLs in <paramref name="endpoints"/> matching the wanted policy URI + mode.</summary>
@@ -816,7 +848,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (sessionToClose is not null) DetachDiagnostics(sessionToClose);
Session = null;
}
_limits = OpcUaClientOperationLimits.None;
try { if (sessionToClose is Session s) await s.CloseAsync(cancellationToken).ConfigureAwait(false); }
catch { /* best-effort */ }
@@ -867,7 +898,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
/// <inheritdoc />
public DriverHealth GetHealth() => _health with { Diagnostics = _diag.Snapshot() };
public DriverHealth GetHealth() => _health;
// ---- IReadable ----
@@ -917,59 +948,44 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (toSend.Count == 0) return results;
// #275: one wire Read per OperationLimits.MaxNodesPerRead-sized chunk. A chunk that fails
// only fails ITS tags — the rest of the batch keeps its values.
var chunks = OpcUaClientOperationLimits.Chunk(toSend.Count, _limits.MaxNodesPerRead).ToList();
if (chunks.Count > 1) _diag.RecordChunkedRequest();
Exception? firstFailure = null;
foreach (var (start, count) in chunks)
try
{
var slice = new ReadValueIdCollection(count);
for (var w = start; w < start + count; w++) slice.Add(toSend[w]);
try
{
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Both,
nodesToRead: slice,
ct: cancellationToken).ConfigureAwait(false);
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Both,
nodesToRead: toSend,
ct: cancellationToken).ConfigureAwait(false);
var values = resp.Results;
for (var w = 0; w < values.Count && w < count; w++)
{
var r = indexMap[start + w];
var dv = values[w];
// Preserve the upstream StatusCode verbatim — including Bad codes per
// §8's cascading-quality rule. Also preserve SourceTimestamp so downstream
// clients can detect stale upstream data.
results[r] = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp);
}
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
var values = resp.Results;
for (var w = 0; w < values.Count; w++)
{
throw;
}
catch (Exception ex)
{
// Transport / timeout / session-dropped — fan out the same fault across every
// tag in this chunk. Per-tag StatusCode stays BadCommunicationError (not
// BadInternalError) so operators distinguish "upstream unreachable" from
// "driver bug".
for (var w = start; w < start + count; w++)
results[indexMap[w]] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
firstFailure ??= ex;
var r = indexMap[w];
var dv = values[w];
// Preserve the upstream StatusCode verbatim — including Bad codes per
// §8's cascading-quality rule. Also preserve SourceTimestamp so downstream
// clients can detect stale upstream data.
results[r] = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp);
}
_health = new DriverHealth(DriverState.Healthy, now, null);
}
catch (Exception ex)
{
// Transport / timeout / session-dropped — fan out the same fault across every
// tag in this batch. Per-tag StatusCode stays BadCommunicationError (not
// BadInternalError) so operators distinguish "upstream unreachable" from
// "driver bug".
for (var w = 0; w < indexMap.Count; w++)
{
var r = indexMap[w];
results[r] = new DataValueSnapshot(null, StatusBadCommunicationError, null, now);
}
_health = new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, ex.Message);
}
_diag.RecordRead(failed: firstFailure is not null);
_health = firstFailure is null
? new DriverHealth(DriverState.Healthy, now, null)
: new DriverHealth(DriverState.Degraded, _health.LastSuccessfulRead, firstFailure.Message);
}
finally { _gate.Release(); }
return results;
@@ -1019,55 +1035,40 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
if (toSend.Count == 0) return results;
// #275: one wire Write per OperationLimits.MaxNodesPerWrite-sized chunk. Chunks are sent in
// order and each fails independently: a later chunk that times out leaves earlier chunks'
// real results intact (those writes DID happen).
var chunks = OpcUaClientOperationLimits.Chunk(toSend.Count, _limits.MaxNodesPerWrite).ToList();
if (chunks.Count > 1) _diag.RecordChunkedRequest();
var anyFailed = false;
foreach (var (start, count) in chunks)
try
{
var slice = new WriteValueCollection(count);
for (var w = start; w < start + count; w++) slice.Add(toSend[w]);
try
{
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: slice,
ct: cancellationToken).ConfigureAwait(false);
var resp = await session.WriteAsync(
requestHeader: null,
nodesToWrite: toSend,
ct: cancellationToken).ConfigureAwait(false);
var codes = resp.Results;
for (var w = 0; w < codes.Count && w < count; w++)
{
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[indexMap[start + w]] = new WriteResult(codes[w].Code);
}
}
catch (OperationCanceledException)
var codes = resp.Results;
for (var w = 0; w < codes.Count; w++)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue. Every not-yet-answered chunk (this one and
// the ones after it) gets the same code.
const uint StatusBadTimeout = 0x800A0000u;
anyFailed = true;
for (var w = start; w < toSend.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
break;
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
anyFailed = true;
for (var w = start; w < start + count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
var r = indexMap[w];
// Pass upstream WriteResult StatusCode through verbatim. Success codes
// include Good (0) and any warning-level Good* variants; anything with
// the severity bits set is a Bad.
results[r] = new WriteResult(codes[w].Code);
}
}
_diag.RecordWrite(anyFailed);
catch (OperationCanceledException)
{
// Timeout / cancellation after the wire request may have been dispatched.
// Writes are non-idempotent — BadTimeout ("outcome unknown, do not blindly
// retry") is more honest than BadCommunicationError ("definitely did not
// happen"). Downstream callers that need retry semantics check for BadTimeout
// and can decide whether to re-issue.
const uint StatusBadTimeout = 0x800A0000u;
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadTimeout);
}
catch (Exception)
{
// Pre-wire transport failure — the write definitely did not reach the server.
for (var w = 0; w < indexMap.Count; w++)
results[indexMap[w]] = new WriteResult(StatusBadCommunicationError);
}
}
finally { _gate.Release(); }
return results;
@@ -1334,27 +1335,16 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
nodesToRead.Add(new ReadValueId { NodeId = pv.NodeId, AttributeId = Attributes.Historizing });
}
// #275: split the attribute reads to the server's MaxNodesPerRead (each variable contributes 4).
var values = new DataValueCollection(nodesToRead.Count);
DataValueCollection values;
try
{
foreach (var (start, count) in OpcUaClientOperationLimits.Chunk(nodesToRead.Count, _limits.MaxNodesPerRead))
{
var slice = new ReadValueIdCollection(count);
for (var w = start; w < start + count; w++) slice.Add(nodesToRead[w]);
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Neither,
nodesToRead: slice,
ct: ct).ConfigureAwait(false);
values.AddRange(resp.Results);
}
if (values.Count < nodesToRead.Count) throw new InvalidOperationException("Short enrichment read.");
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
throw;
var resp = await session.ReadAsync(
requestHeader: null,
maxAge: 0,
timestampsToReturn: TimestampsToReturn.Neither,
nodesToRead: nodesToRead,
ct: ct).ConfigureAwait(false);
values = resp.Results;
}
catch
{
@@ -1507,7 +1497,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// captures `handle`, so we must hold the exact delegate instance returned by
// `+=` to be able to remove it.
var itemHandlers = new List<MonitoredItemNotificationHandle>();
var itemsAdded = 0;
foreach (var fullRef in fullReferences)
{
if (!TryResolveTagNode(session, fullRef, out var nodeId)) continue;
@@ -1532,15 +1521,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
item.Notification += notifHandler;
itemHandlers.Add(new MonitoredItemNotificationHandle(item, notifHandler));
subscription.AddItem(item);
itemsAdded++;
// #275: flush a CreateMonitoredItems call whenever the server's per-call cap is reached.
// CreateItemsAsync only sends the items not yet created, so calling it per batch is safe.
if (_limits.MaxMonitoredItemsPerCall > 0 && itemsAdded % _limits.MaxMonitoredItemsPerCall == 0)
{
_diag.RecordChunkedRequest();
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
}
}
await subscription.CreateItemsAsync(cancellationToken).ConfigureAwait(false);
@@ -1590,7 +1570,8 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
var dv = mn.Value;
if (dv is null) return;
var fullRef = (item.Handle as string) ?? item.DisplayName ?? string.Empty;
_diag.RecordDataChange(bad: StatusCode.IsBad(dv.StatusCode));
_diag.Record(OpcUaClientDiagnostics.DataChange);
if (StatusCode.IsBad(dv.StatusCode)) _diag.Record(OpcUaClientDiagnostics.BadStatusValue);
var snapshot = new DataValueSnapshot(
Value: dv.Value,
StatusCode: dv.StatusCode.Code,
@@ -2249,13 +2230,13 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
{
_notificationHandler ??= (_, e) =>
{
_diag.RecordPublishResponse();
_diag.Record(OpcUaClientDiagnostics.PublishResponse);
var msg = e.NotificationMessage;
if (msg?.NotificationData is { } data)
foreach (var nd in data)
_diag.RecordNotifications(CountNotifications(nd));
_diag.Record(OpcUaClientDiagnostics.Notification, CountNotifications(nd));
};
_publishErrorHandler ??= (_, _) => _diag.RecordPublishError();
_publishErrorHandler ??= (_, _) => _diag.Record(OpcUaClientDiagnostics.PublishError);
session.Notification -= _notificationHandler;
session.Notification += _notificationHandler;
session.PublishError -= _publishErrorHandler;
@@ -2280,12 +2261,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
_ => 1,
};
private async Task RefreshOperationLimitsAsync(ISession session)
{
try { _limits = await OpcUaClientOperationLimits.ReadAsync(session, CancellationToken.None).ConfigureAwait(false); }
catch { /* best-effort: keep the previous limits */ }
}
/// <summary>
/// Session keep-alive handler. On a healthy ping, bumps HostState back to Running
/// (typical bounce after a transient network blip). On a bad ping, starts the SDK's
@@ -2301,7 +2276,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
}
TransitionTo(HostState.Stopped);
_diag.RecordKeepAliveMiss();
_diag.Record(OpcUaClientDiagnostics.KeepAliveMiss);
// Kick off the SDK's reconnect loop exactly once per drop. Keep-alive callbacks fire
// from the SDK keep-alive timer thread and the SDK can fire this handler repeatedly
@@ -2321,7 +2296,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// BeginReconnect is started outside the lock — it does no _reconnectHandler mutation
// and we don't want to hold _probeLock across an SDK call.
_diag.RecordReconnectAttempt();
_diag.Record(OpcUaClientDiagnostics.ReconnectAttempt);
handler.BeginReconnect(
sender,
(int)_options.ReconnectPeriod.TotalMilliseconds,
@@ -2388,8 +2363,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
RecordSkippedTags(_namespaceMap);
TransitionTo(HostState.Running);
_health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null);
_diag.RecordReconnectSucceeded();
if (_options.HonorServerOperationLimits) _ = RefreshOperationLimitsAsync(newSession);
_diag.Record(OpcUaClientDiagnostics.ReconnectSuccess);
return;
}
@@ -2402,7 +2376,7 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
// the last-known session for an always-on gateway rather than abandoning recovery
// entirely.
TransitionTo(HostState.Faulted);
_diag.RecordReconnectExhausted();
_diag.Record(OpcUaClientDiagnostics.ReconnectExhausted);
_health = new DriverHealth(
DriverState.Faulted, _health.LastSuccessfulRead,
"OPC UA session reconnect exhausted its retry window without recovering. " +
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientDriverFactoryExtensions.cs
+11 -1
View File
@@ -86,8 +86,18 @@ public static class OpcUaClientDriverFactoryExtensions
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
ArgumentException.ThrowIfNullOrWhiteSpace(driverConfigJson);
return JsonSerializer.Deserialize<OpcUaClientDriverOptions>(driverConfigJson, JsonOptions)
var options = JsonSerializer.Deserialize<OpcUaClientDriverOptions>(driverConfigJson, JsonOptions)
?? throw new InvalidOperationException(
$"OpcUaClient driver config for '{driverInstanceId}' deserialised to null");
// #273/#278: nothing runs DataAnnotations on driver options, so range-check here — at options load —
// and fail with a clear message instead of a driver that faults and retries forever.
try { OpcUaClientDriver.ValidateTuning(options); }
catch (InvalidOperationException ex)
{
throw new InvalidOperationException(
$"OpcUaClient driver config for '{driverInstanceId}' is invalid: {ex.Message}", ex);
}
return options;
}
}
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient/OpcUaClientOperationLimits.cs
-94
View File
@@ -1,94 +0,0 @@
using Opc.Ua;
using Opc.Ua.Client;
namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
/// <summary>
/// The remote server's <c>Server/ServerCapabilities/OperationLimits</c> that bound how many
/// items one service call may carry (#275). Every field is <c>0</c> = "no limit declared" (the
/// OPC UA convention), so <see cref="None"/> is the safe value when the server publishes nothing
/// or the read fails.
/// </summary>
/// <param name="MaxNodesPerRead">Items per Read.</param>
/// <param name="MaxNodesPerWrite">Items per Write.</param>
/// <param name="MaxNodesPerBrowse">Nodes per Browse.</param>
/// <param name="MaxMonitoredItemsPerCall">Items per CreateMonitoredItems / Modify / SetMonitoringMode.</param>
/// <param name="MaxNodesPerHistoryReadData">Nodes per HistoryRead (raw / processed / at-time).</param>
/// <param name="MaxNodesPerHistoryReadEvents">Nodes per HistoryRead (events).</param>
internal sealed record OpcUaClientOperationLimits(
uint MaxNodesPerRead,
uint MaxNodesPerWrite,
uint MaxNodesPerBrowse,
uint MaxMonitoredItemsPerCall,
uint MaxNodesPerHistoryReadData,
uint MaxNodesPerHistoryReadEvents)
{
/// <summary>No limits declared — every call goes out in one request.</summary>
public static OpcUaClientOperationLimits None { get; } = new(0, 0, 0, 0, 0, 0);
/// <summary>
/// Split <paramref name="total"/> items into consecutive <c>(Start, Count)</c> chunks of at most
/// <paramref name="limit"/> each. <c>limit == 0</c> (no limit) or a total that already fits yields a
/// single chunk; <c>total == 0</c> yields none.
/// </summary>
/// <param name="total">Number of items to send.</param>
/// <param name="limit">The server's per-call cap; 0 = unlimited.</param>
/// <returns>The chunk ranges, in order, covering <c>[0, total)</c> exactly once.</returns>
public static IEnumerable<(int Start, int Count)> Chunk(int total, uint limit)
{
if (total <= 0) yield break;
if (limit == 0 || limit >= (uint)total)
{
yield return (0, total);
yield break;
}
var size = (int)Math.Min(limit, int.MaxValue);
for (var start = 0; start < total; start += size)
yield return (start, Math.Min(size, total - start));
}
/// <summary>The standard variables read to populate the limits, in <see cref="FromValues"/> order.</summary>
internal static readonly NodeId[] LimitNodeIds =
[
VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerRead,
VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerWrite,
VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerBrowse,
VariableIds.Server_ServerCapabilities_OperationLimits_MaxMonitoredItemsPerCall,
VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryReadData,
VariableIds.Server_ServerCapabilities_OperationLimits_MaxNodesPerHistoryReadEvents,
];
/// <summary>
/// Build limits from the values read for <see cref="LimitNodeIds"/>. A Bad or non-<c>uint</c> value
/// (server doesn't publish that limit) is treated as 0 = unlimited.
/// </summary>
/// <param name="values">One value per <see cref="LimitNodeIds"/> entry, in order.</param>
/// <returns>The parsed limits.</returns>
public static OpcUaClientOperationLimits FromValues(IReadOnlyList<DataValue> values)
{
uint At(int i) =>
i < values.Count && values[i] is { } dv && StatusCode.IsGood(dv.StatusCode) && dv.Value is uint u ? u : 0;
return new OpcUaClientOperationLimits(At(0), At(1), At(2), At(3), At(4), At(5));
}
/// <summary>
/// Read the server's declared limits. Best-effort: any failure (older server without the
/// capability object, transport error) yields <see cref="None"/> so connect never fails over this.
/// </summary>
/// <param name="session">The freshly connected session.</param>
/// <param name="ct">Cancellation token.</param>
/// <returns>The declared limits, or <see cref="None"/>.</returns>
public static async Task<OpcUaClientOperationLimits> ReadAsync(ISession session, CancellationToken ct)
{
try
{
var nodes = new ReadValueIdCollection(LimitNodeIds.Length);
foreach (var id in LimitNodeIds)
nodes.Add(new ReadValueId { NodeId = id, AttributeId = Attributes.Value });
var resp = await session.ReadAsync(null, 0, TimestampsToReturn.Neither, nodes, ct).ConfigureAwait(false);
return FromValues(resp.Results);
}
catch (OperationCanceledException) { throw; }
catch { return None; }
}
}