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

feat(historian-sidecar): live aahClientManaged alarm-event write path (C.1)

Browse Source 
SdkAlarmHistorianWriteBackend.WriteBatchAsync replaces the RetryPlease
placeholder with the real entry point — HistorianAccess.AddStreamedValue
(HistorianEvent, out HistorianAccessError) in aahClientManaged, pinned by
decompiling the installed SDK.

The write path opens its own ReadOnly=false connection: the query-side
HistorianDataSource opens ReadOnly sessions and AddStreamedValue fails on
those with WriteToReadOnlyFile. IHistorianConnectionFactory gains a readOnly
parameter (default true, query path unchanged); BuildConnectionArgs is
extracted as a pure helper. HistorianClusterEndpointPicker is shared for
node failover; connection-class errors abort the batch as RetryPlease and
reset the connection, malformed-input codes map to PermanentFail.

Tests: connection-unavailable batch deferral, ClassifyOutcome error-code
table, BuildConnectionArgs read-vs-write shaping (80 pass, 2 rig-skipped).
Live_* round-trip tests stay Skip-gated for the D.1 rollout smoke.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
2026-05-18 16:08:32 -04:00
parent 419eda256b
commit cd2306db66
6 changed files with 544 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware/Backend/IHistorianConnectionFactory.cs
View File

@@ -10,32 +10,24 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Backend
/// </summary>
internal interface IHistorianConnectionFactory
{
HistorianAccess CreateAndConnect(HistorianConfiguration config, HistorianConnectionType type);
/// <summary>
/// Opens a Historian SDK connection. <paramref name="readOnly"/> defaults to
/// <c>true</c> for the query path; the alarm-event write backend passes
/// <c>false</c> because <c>HistorianAccess.AddStreamedValue</c> fails with
/// <c>WriteToReadOnlyFile</c> on a read-only session.
/// </summary>
HistorianAccess CreateAndConnect(
HistorianConfiguration config, HistorianConnectionType type, bool readOnly = true);
}
/// <summary>Production implementation — opens real Historian SDK connections.</summary>
internal sealed class SdkHistorianConnectionFactory : IHistorianConnectionFactory
{
public HistorianAccess CreateAndConnect(HistorianConfiguration config, HistorianConnectionType type)
public HistorianAccess CreateAndConnect(
HistorianConfiguration config, HistorianConnectionType type, bool readOnly = true)
{
var conn = new HistorianAccess();
var args = new HistorianConnectionArgs
{
ServerName = config.ServerName,
TcpPort = (ushort)config.Port,
IntegratedSecurity = config.IntegratedSecurity,
UseArchestrAUser = config.IntegratedSecurity,
ConnectionType = type,
ReadOnly = true,
PacketTimeout = (uint)(config.CommandTimeoutSeconds * 1000)
};
if (!config.IntegratedSecurity)
{
args.UserName = config.UserName ?? string.Empty;
args.Password = config.Password ?? string.Empty;
}
var args = BuildConnectionArgs(config, type, readOnly);
if (!conn.OpenConnection(args, out var error))
{
@@ -69,5 +61,32 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Backend
throw new TimeoutException(
$"Historian SDK connection to {config.ServerName}:{config.Port} timed out after {config.CommandTimeoutSeconds}s");
}
/// <summary>
/// Builds the <see cref="HistorianConnectionArgs"/> for a connection. Pure (no SDK
/// side effects) so the read-only-vs-write argument shaping is unit-testable.
/// </summary>
internal static HistorianConnectionArgs BuildConnectionArgs(
HistorianConfiguration config, HistorianConnectionType type, bool readOnly)
{
var args = new HistorianConnectionArgs
{
ServerName = config.ServerName,
TcpPort = (ushort)config.Port,
IntegratedSecurity = config.IntegratedSecurity,
UseArchestrAUser = config.IntegratedSecurity,
ConnectionType = type,
ReadOnly = readOnly,
PacketTimeout = (uint)(config.CommandTimeoutSeconds * 1000)
};
if (!config.IntegratedSecurity)
{
args.UserName = config.UserName ?? string.Empty;
args.Password = config.Password ?? string.Empty;
}
return args;
}
}
}

351
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware/Backend/SdkAlarmHistorianWriteBackend.cs
View File

@@ -1,6 +1,8 @@
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using ArchestrA;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Ipc;
@@ -8,39 +10,85 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Backend
{
/// <summary>
/// Production <see cref="IAlarmHistorianWriteBackend"/> backed by AVEVA Historian's
/// <c>aahClientManaged</c> alarm-event write API. The exact SDK entry point is
/// pinned during the live-rig smoke in PR D.1 — until that gate, this backend
/// reports <see cref="AlarmHistorianWriteOutcome.RetryPlease"/> for every
/// event with a structured diagnostic so the lmxopcua-side
/// <c>SqliteStoreAndForwardSink</c> retains the queued events rather than dropping
/// or hard-failing them.
/// <c>aahClientManaged</c> SDK. Each <see cref="AlarmHistorianEventDto"/> is written via
/// <c>HistorianAccess.AddStreamedValue(HistorianEvent, out HistorianAccessError)</c> —
/// the alarm-event write entry point pinned during PR C.1.
/// </summary>
/// <remarks>
/// <para>
/// Cluster failover reuses <see cref="HistorianClusterEndpointPicker"/> via
/// the shared <see cref="HistorianDataSource"/> connection pool — there is
/// no second connection pool for writes. Wonderware Historian's alarm-event
/// write surface accepts the same <c>HistorianAccess</c> session a read
/// opens, so reusing the picker is parity-preserving with v1's
/// <c>GalaxyHistorianWriter</c>.
/// The write path needs its <b>own</b> connection. The query-side
/// <see cref="HistorianDataSource"/> opens <c>ReadOnly</c> sessions, and
/// <c>AddStreamedValue</c> on a read-only session fails with
/// <c>WriteToReadOnlyFile</c>. This backend therefore opens a dedicated
/// <c>ReadOnly = false</c> connection; it shares
/// <see cref="HistorianClusterEndpointPicker"/> for node selection and failover but
/// not the connection object itself.
/// </para>
/// <para>
/// Once D.1 confirms the SDK entry point, this class swaps the placeholder
/// body for the real call sequence. The mapping from raw HRESULT /
/// <c>HistorianError</c> codes onto <see cref="AlarmHistorianWriteOutcome"/>
/// is already shared via <see cref="AahClientManagedAlarmEventWriter.MapOutcome"/>
/// so the smoke-pinned change stays minimal.
/// Per-event <c>HistorianAccessError.ErrorValue</c> codes map onto
/// <see cref="AlarmHistorianWriteOutcome"/> via
/// <see cref="AahClientManagedAlarmEventWriter.MapOutcome"/>. A connection-class
/// error aborts the remainder of the batch as
/// <see cref="AlarmHistorianWriteOutcome.RetryPlease"/> and resets the connection so
/// the next drain tick reconnects — possibly to a different cluster node.
/// </para>
/// <para>
/// The exact <c>HistorianEvent</c> field set required by the Historian is confirmed
/// against a live install during the PR D.1 rollout smoke; <see cref="ToHistorianEvent"/>
/// maps the unambiguous fields and carries operator comment / condition id as event
/// properties.
/// </para>
/// </remarks>
public sealed class SdkAlarmHistorianWriteBackend : IAlarmHistorianWriteBackend
public sealed class SdkAlarmHistorianWriteBackend : IAlarmHistorianWriteBackend, IDisposable
{
private static readonly ILogger Log = Serilog.Log.ForContext<SdkAlarmHistorianWriteBackend>();
// ErrorValue codes that mean the connection/server is the problem (transient) rather
// than the event payload. These abort the rest of the batch and trigger a reconnect.
private static readonly HashSet<HistorianAccessError.ErrorValue> ConnectionErrors =
new HashSet<HistorianAccessError.ErrorValue>
{
HistorianAccessError.ErrorValue.FailedToConnect,
HistorianAccessError.ErrorValue.FailedToCreateSession,
HistorianAccessError.ErrorValue.NoReply,
HistorianAccessError.ErrorValue.NotReady,
HistorianAccessError.ErrorValue.NotInitialized,
HistorianAccessError.ErrorValue.Stopping,
HistorianAccessError.ErrorValue.Win32Exception,
HistorianAccessError.ErrorValue.InvalidResponse,
};
// ErrorValue codes that mean the event itself is malformed — permanent, never retried.
private static readonly HashSet<HistorianAccessError.ErrorValue> MalformedErrors =
new HashSet<HistorianAccessError.ErrorValue>
{
HistorianAccessError.ErrorValue.InvalidArgument,
HistorianAccessError.ErrorValue.ValidationFailed,
HistorianAccessError.ErrorValue.NullPointerArgument,
HistorianAccessError.ErrorValue.WriteToReadOnlyFile,
HistorianAccessError.ErrorValue.NotImplemented,
HistorianAccessError.ErrorValue.NotApplicable,
};
private readonly HistorianConfiguration _config;
private readonly IHistorianConnectionFactory _factory;
private readonly HistorianClusterEndpointPicker _picker;
private readonly object _connectionLock = new object();
private HistorianAccess? _connection;
private string? _activeNode;
private bool _disposed;
public SdkAlarmHistorianWriteBackend(HistorianConfiguration config)
: this(config, new SdkHistorianConnectionFactory(), null) { }
internal SdkAlarmHistorianWriteBackend(
HistorianConfiguration config,
IHistorianConnectionFactory factory,
HistorianClusterEndpointPicker? picker = null)
{
_config = config ?? throw new ArgumentNullException(nameof(config));
_factory = factory ?? throw new ArgumentNullException(nameof(factory));
_picker = picker ?? new HistorianClusterEndpointPicker(config);
}
public Task<AlarmHistorianWriteOutcome[]> WriteBatchAsync(
@@ -52,22 +100,265 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Wonderware.Backend
return Task.FromResult(new AlarmHistorianWriteOutcome[0]);
}
// Placeholder: pin the SDK entry point in PR D.1 against a live AVEVA
// Historian. Until then the call returns RetryPlease for every slot so
// the lmxopcua-side sink keeps the events queued rather than dropping
// them — same effect as the current NullAlarmHistorianSink fallback,
// but visible through the structured diagnostic + per-event outcome.
Log.Warning(
"Alarm historian SDK write path not yet pinned — returning RetryPlease for {Count} event(s) from server {Server}. PR D.1 swaps this for the live aahClientManaged call.",
events.Length,
_config.ServerName);
var outcomes = new AlarmHistorianWriteOutcome[events.Length];
for (var i = 0; i < outcomes.Length; i++)
HistorianAccess connection;
try
{
outcomes[i] = AlarmHistorianWriteOutcome.RetryPlease;
connection = EnsureConnected();
}
catch (ObjectDisposedException)
{
throw;
}
catch (Exception ex)
{
// No reachable node — defer the whole batch so the lmxopcua-side SQLite
// store-and-forward sink retains the rows for the next drain tick.
Log.Warning(ex,
"Alarm historian write connection unavailable — deferring {Count} event(s) as RetryPlease",
events.Length);
FillRemaining(outcomes, 0, AlarmHistorianWriteOutcome.RetryPlease);
return Task.FromResult(outcomes);
}
for (var i = 0; i < events.Length; i++)
{
cancellationToken.ThrowIfCancellationRequested();
try
{
var historianEvent = ToHistorianEvent(events[i]);
if (connection.AddStreamedValue(historianEvent, out var error))
{
outcomes[i] = AlarmHistorianWriteOutcome.Ack;
continue;
}
var code = error?.ErrorCode ?? HistorianAccessError.ErrorValue.Failure;
if (ConnectionErrors.Contains(code))
{
// Connection died mid-batch — drop it and defer this event + the rest.
Log.Warning(
"Alarm historian write hit connection-level error {Code} ({Desc}); resetting connection, deferring {Remaining} event(s)",
code, error?.ErrorDescription, events.Length - i);
HandleConnectionError(error?.ErrorDescription);
FillRemaining(outcomes, i, AlarmHistorianWriteOutcome.RetryPlease);
return Task.FromResult(outcomes);
}
outcomes[i] = ClassifyOutcome(code);
Log.Warning(
"Alarm historian write rejected event {EventId}: {Code} ({Desc}) -> {Outcome}",
events[i].EventId, code, error?.ErrorDescription, outcomes[i]);
}
catch (OperationCanceledException)
{
throw;
}
catch (Exception ex)
{
// Transport-level throw (SDK marshalling fault, broken connection) —
// reset and defer this event + the rest.
Log.Warning(ex,
"Alarm historian write threw for event {EventId}; resetting connection, deferring {Remaining} event(s)",
events[i].EventId, events.Length - i);
HandleConnectionError(ex.Message);
FillRemaining(outcomes, i, AlarmHistorianWriteOutcome.RetryPlease);
return Task.FromResult(outcomes);
}
}
return Task.FromResult(outcomes);
}
/// <summary>
/// Maps an <see cref="AlarmHistorianEventDto"/> onto the SDK's
/// <c>HistorianEvent</c>. Operator comment and originating condition id ride as
/// event properties — operator-comment fidelity is the field the value-driven
/// fallback path cannot carry.
/// </summary>
internal static HistorianEvent ToHistorianEvent(AlarmHistorianEventDto dto)
{
// The ArchestrA SDK marks these HistorianEvent members obsolete but still honours
// them on write; their successors aren't wired in the version we bind against.
// Using them is the documented v1 behaviour — mirrors HistorianDataSource.ToDto,
// suppressed locally so any other deprecated-surface use still surfaces as an error.
#pragma warning disable CS0618
var historianEvent = new HistorianEvent
{
IsAlarm = true,
Source = dto.SourceName ?? string.Empty,
EventType = string.IsNullOrEmpty(dto.AlarmType) ? "Alarm" : dto.AlarmType,
EventTime = new DateTime(dto.EventTimeUtcTicks, DateTimeKind.Utc),
ReceivedTime = DateTime.UtcNow,
Severity = dto.Severity,
DisplayText = dto.Message ?? string.Empty,
};
if (Guid.TryParse(dto.EventId, out var id))
{
historianEvent.Id = id;
}
#pragma warning restore CS0618
if (!string.IsNullOrEmpty(dto.AckComment))
{
historianEvent.AddProperty("Comment", dto.AckComment, out _);
}
if (!string.IsNullOrEmpty(dto.ConditionId))
{
historianEvent.AddProperty("ConditionId", dto.ConditionId, out _);
}
return historianEvent;
}
/// <summary>
/// Classifies a non-connection-class <c>HistorianAccessError.ErrorValue</c> into an
/// <see cref="AlarmHistorianWriteOutcome"/> by routing it through the shared
/// <see cref="AahClientManagedAlarmEventWriter.MapOutcome"/> mapping. Exposed for
/// unit tests — connection-class codes are handled separately by the batch loop.
/// </summary>
internal static AlarmHistorianWriteOutcome ClassifyOutcome(HistorianAccessError.ErrorValue code)
=> AahClientManagedAlarmEventWriter.MapOutcome(
(int)code,
isCommunicationError: ConnectionErrors.Contains(code),
isMalformedInput: MalformedErrors.Contains(code));
private static void FillRemaining(
AlarmHistorianWriteOutcome[] outcomes, int from, AlarmHistorianWriteOutcome value)
{
for (var i = from; i < outcomes.Length; i++)
{
outcomes[i] = value;
}
}
private HistorianAccess EnsureConnected()
{
if (_disposed)
{
throw new ObjectDisposedException(nameof(SdkAlarmHistorianWriteBackend));
}
var existing = Volatile.Read(ref _connection);
if (existing != null) return existing;
var (conn, node) = ConnectToAnyHealthyNode();
lock (_connectionLock)
{
if (_disposed)
{
SafeClose(conn);
throw new ObjectDisposedException(nameof(SdkAlarmHistorianWriteBackend));
}
if (_connection != null)
{
SafeClose(conn);
return _connection;
}
_connection = conn;
_activeNode = node;
Log.Information("Alarm historian write connection opened to {Server}:{Port}", node, _config.Port);
return conn;
}
}
private (HistorianAccess Connection, string Node) ConnectToAnyHealthyNode()
{
var candidates = _picker.GetHealthyNodes();
if (candidates.Count == 0)
{
throw new InvalidOperationException(
_picker.NodeCount == 0
? "No historian nodes configured"
: $"All {_picker.NodeCount} historian nodes are in cooldown — no healthy endpoints");
}
Exception? lastException = null;
foreach (var node in candidates)
{
try
{
var conn = _factory.CreateAndConnect(
CloneConfigWithServerName(node), HistorianConnectionType.Event, readOnly: false);
_picker.MarkHealthy(node);
return (conn, node);
}
catch (Exception ex)
{
_picker.MarkFailed(node, ex.Message);
lastException = ex;
Log.Warning(ex, "Alarm historian node {Node} failed during write-connect; trying next", node);
}
}
throw new InvalidOperationException(
$"All {candidates.Count} healthy historian candidate(s) failed during write-connect: " +
(lastException?.Message ?? "(no detail)"),
lastException);
}
private void HandleConnectionError(string? detail)
{
lock (_connectionLock)
{
if (_connection == null) return;
SafeClose(_connection);
_connection = null;
var failedNode = _activeNode;
_activeNode = null;
if (failedNode != null) _picker.MarkFailed(failedNode, detail ?? "mid-batch failure");
Log.Warning("Alarm historian write connection reset (node={Node})", failedNode ?? "(unknown)");
}
}
private static void SafeClose(HistorianAccess conn)
{
try
{
conn.CloseConnection(out _);
conn.Dispose();
}
catch (Exception ex)
{
Log.Debug(ex, "Error closing alarm historian write connection");
}
}
private HistorianConfiguration CloneConfigWithServerName(string serverName) => new HistorianConfiguration
{
Enabled = _config.Enabled,
ServerName = serverName,
ServerNames = _config.ServerNames,
FailureCooldownSeconds = _config.FailureCooldownSeconds,
IntegratedSecurity = _config.IntegratedSecurity,
UserName = _config.UserName,
Password = _config.Password,
Port = _config.Port,
CommandTimeoutSeconds = _config.CommandTimeoutSeconds,
MaxValuesPerRead = _config.MaxValuesPerRead,
RequestTimeoutSeconds = _config.RequestTimeoutSeconds,
};
public void Dispose()
{
if (_disposed) return;
_disposed = true;
lock (_connectionLock)
{
if (_connection != null)
{
SafeClose(_connection);
_connection = null;
}
}
}
}
}