feat(historian): OPC UA HistoryUpdate in the node manager + HistoryUpdate authorization (T6)
- OtOpcUaNodeManager overrides HistoryUpdateData / HistoryDeleteRawModified / HistoryDeleteAtTime, block-bridging to the optional IHistorianDataUpdater under the shared HistoryRead batch limiter + deadline. Insert/Replace/Update map to HistoryUpdateMode; Remove, IsDeleteModified, events and structure data are BadHistoryOperationUnsupported. Per-item codes stitched back in request order (MinValue SourceTimestamp => BadInvalidTimestamp, untypeable value => BadTypeMismatch); success clears the base's pre-seeded Unsupported error. - HistoryWrite AccessLevel bit OR-ed in for historized nodes when an updater is wired (raw + UNS); it is dispatch-load-bearing in the SDK base. OtOpcUaSdkServer.SetHistorianDataUpdater; the hosted service assigns it from an optional HistorianUpdaterHolder before SetSink, clears it on stop. - AuthorizeHistoryUpdate: ACL on => scoped OpcUaOperation.HistoryUpdate decision (replaces the role check); ACL off => #563 key re-check + new OpcUaDataPlaneRoles.HistoryUpdate role, conferred by ApiKeyInterimRoleMapper only from a Cluster-scoped HistoryUpdate grant (WriteOperate does not). - AdminUI ACL editor offers NodePermissions.HistoryUpdate. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -163,7 +163,7 @@ else
|
||||
NodePermissions.Browse, NodePermissions.Read, NodePermissions.Subscribe, NodePermissions.HistoryRead,
|
||||
NodePermissions.WriteOperate, NodePermissions.WriteTune, NodePermissions.WriteConfigure,
|
||||
NodePermissions.AlarmRead, NodePermissions.AlarmAcknowledge, NodePermissions.AlarmConfirm, NodePermissions.AlarmShelve,
|
||||
NodePermissions.MethodCall,
|
||||
NodePermissions.MethodCall, NodePermissions.HistoryUpdate,
|
||||
];
|
||||
|
||||
[Parameter] public string ClusterId { get; set; } = "";
|
||||
|
||||
@@ -54,7 +54,7 @@ public static class ApiKeyInterimRoleMapper
|
||||
effective |= grant.PermissionFlags;
|
||||
}
|
||||
|
||||
var roles = new List<string>(2);
|
||||
var roles = new List<string>(3);
|
||||
if (effective.HasFlag(NodePermissions.WriteOperate)) roles.Add(OpcUaDataPlaneRoles.WriteOperate);
|
||||
|
||||
// AlarmAcknowledge is the flag an operator authors; AlarmAck is the role the Part 9 method gates
|
||||
@@ -62,6 +62,10 @@ public static class ApiKeyInterimRoleMapper
|
||||
// separately — mapping them would imply a granularity the gate does not have.
|
||||
if (effective.HasFlag(NodePermissions.AlarmAcknowledge)) roles.Add(OpcUaDataPlaneRoles.AlarmAck);
|
||||
|
||||
// HistoryUpdate has its own role: WriteOperate (live writes) never implies authority to rewrite
|
||||
// recorded history, so the role is conferred by the HistoryUpdate flag alone.
|
||||
if (effective.HasFlag(NodePermissions.HistoryUpdate)) roles.Add(OpcUaDataPlaneRoles.HistoryUpdate);
|
||||
|
||||
return roles;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -45,6 +45,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
private readonly ILoggerFactory _loggerFactory;
|
||||
private readonly ILogger<OtOpcUaServerHostedService> _logger;
|
||||
private readonly IDriverPrimaryRoleSource? _primaryRoleSource;
|
||||
private readonly HistorianUpdaterHolder _historianUpdater;
|
||||
|
||||
private OpcUaApplicationHost? _appHost;
|
||||
private OtOpcUaSdkServer? _server;
|
||||
@@ -94,6 +95,9 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
/// Wired onto the node manager so a native-condition Acknowledge on a non-Primary node is refused rather than
|
||||
/// applied and then reverted. Optional: a node with no driver role registers none, and null means "allow" —
|
||||
/// the single-node posture <c>UnmanagedDriverPrimaryRoleSource</c> also answers with.</param>
|
||||
/// <param name="historianUpdater">The selected historian backend's optional HistoryUpdate capability.
|
||||
/// Optional: when unregistered it resolves to <see cref="HistorianUpdaterHolder.None"/> and HistoryUpdate is
|
||||
/// reported unsupported on every node.</param>
|
||||
public OtOpcUaServerHostedService(
|
||||
IOptions<OpcUaApplicationHostOptions> options,
|
||||
DeferredAddressSpaceSink deferredSink,
|
||||
@@ -108,7 +112,8 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
IOptions<AclOptions> aclOptions,
|
||||
TimeProvider time,
|
||||
ILoggerFactory loggerFactory,
|
||||
IDriverPrimaryRoleSource? primaryRoleSource = null)
|
||||
IDriverPrimaryRoleSource? primaryRoleSource = null,
|
||||
HistorianUpdaterHolder? historianUpdater = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(aclOptions);
|
||||
ArgumentNullException.ThrowIfNull(time);
|
||||
@@ -131,6 +136,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_actorRegistry = actorRegistry;
|
||||
_loggerFactory = loggerFactory;
|
||||
_primaryRoleSource = primaryRoleSource;
|
||||
_historianUpdater = historianUpdater ?? HistorianUpdaterHolder.None;
|
||||
_logger = loggerFactory.CreateLogger<OtOpcUaServerHostedService>();
|
||||
}
|
||||
|
||||
@@ -168,6 +174,13 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
return;
|
||||
}
|
||||
|
||||
// The HistoryUpdate backend MUST be wired before SetSink: the node manager decides each historized
|
||||
// node's HistoryWrite AccessLevel bit at materialise time from whether an updater is present, and the
|
||||
// SDK base admits a node to HistoryUpdate only when that bit is set. Installing the real sink below is
|
||||
// what lets OpcUaPublishActor's deploy applies start materialising nodes, so assigning the updater
|
||||
// first guarantees every node sees it (nodes are never retro-updated).
|
||||
_server.SetHistorianDataUpdater(_historianUpdater.Updater);
|
||||
|
||||
_deferredSink.SetSink(new SdkAddressSpaceSink(_server.NodeManager));
|
||||
|
||||
// Wire the reverse-path inbound-alarm-command router: a client Acknowledge/Confirm/Shelve that
|
||||
@@ -371,6 +384,7 @@ public sealed class OtOpcUaServerHostedService : IHostedService, IAsyncDisposabl
|
||||
_server?.SetNativeAlarmAckPrimaryGate(null);
|
||||
// Restore the Null historian so a late HistoryRead doesn't hit a disposed read client.
|
||||
_server?.SetHistorianDataSource(null);
|
||||
_server?.SetHistorianDataUpdater(null);
|
||||
// Restore the empty authority so a late session activation cannot authorize against a stopping node.
|
||||
_server?.SetDataPlaneAuthority(null);
|
||||
// Detach the ACL gate too. Leaving it attached alongside an emptied authority would deny every late
|
||||
|
||||
@@ -309,6 +309,29 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
set => _historianDataSource = value ?? NullHistorianDataSource.Instance;
|
||||
}
|
||||
|
||||
private volatile IHistorianDataUpdater? _historianDataUpdater;
|
||||
|
||||
/// <summary>
|
||||
/// Optional server-side backend for the OPC UA HistoryUpdate service (insert / replace / update /
|
||||
/// delete of historized samples). <c>null</c> (the default) means the selected historian backend
|
||||
/// cannot modify history: every HistoryUpdate is reported <c>BadHistoryOperationUnsupported</c>.
|
||||
/// <para>
|
||||
/// ⚠️ <b>Assign it before any node materialises.</b> Whether an updater is present decides the
|
||||
/// <c>HistoryWrite</c> AccessLevel bit a historized node is created with
|
||||
/// (<see cref="ComposeAccessLevel"/>), and the SDK base admits a node to HistoryUpdate only when
|
||||
/// that bit is set — so the bit is dispatch-load-bearing, not cosmetic. Setting the updater after
|
||||
/// nodes exist does NOT retro-update their AccessLevel. The Host assigns it in
|
||||
/// <c>OtOpcUaServerHostedService.StartAsync</c> before installing the address-space sink, which
|
||||
/// precedes every deploy apply. Volatile for the same start-thread / request-thread handoff as
|
||||
/// <see cref="HistorianDataSource"/>.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public IHistorianDataUpdater? HistorianDataUpdater
|
||||
{
|
||||
get => _historianDataUpdater;
|
||||
set => _historianDataUpdater = value;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// #520 stage 4 — the node's current data-plane authorization inputs: which cluster it serves,
|
||||
/// the generation those were resolved at, and the ACL grants themselves.
|
||||
@@ -2861,7 +2884,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
var historized = historianTagname is not null;
|
||||
// The SDK exposes the flags separately (no CurrentReadWrite composite): ReadWrite is
|
||||
// CurrentRead | CurrentWrite, with the HistoryRead bit OR-ed in when historized.
|
||||
var access = ComposeAccessLevel(writable, historized);
|
||||
var access = ComposeAccessLevel(writable, historized, _historianDataUpdater is not null);
|
||||
var variable = new BaseDataVariableState(parent)
|
||||
{
|
||||
NodeId = new NodeId(variableNodeId, ns),
|
||||
@@ -2902,11 +2925,19 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
/// Historizing flags — the single source of truth shared by <see cref="EnsureVariable"/> (node
|
||||
/// creation) and <see cref="UpdateTagAttributes"/> (F10b surgical in-place update). The SDK exposes
|
||||
/// the flags separately (no CurrentReadWrite composite): ReadWrite is CurrentRead | CurrentWrite,
|
||||
/// with the HistoryRead bit OR-ed in when historized. OR-ing byte constants promotes to int, so cast back.</summary>
|
||||
private static byte ComposeAccessLevel(bool writable, bool historized)
|
||||
/// with the HistoryRead bit OR-ed in when historized, and the HistoryWrite bit OR-ed in when historized AND
|
||||
/// <paramref name="historyWritable"/> (a <see cref="HistorianDataUpdater"/> is wired). The HistoryWrite bit is
|
||||
/// what admits a node to the SDK base's HistoryUpdate dispatch at all. OR-ing byte constants promotes to int,
|
||||
/// so cast back.</summary>
|
||||
/// <param name="writable">The node accepts client value writes.</param>
|
||||
/// <param name="historized">The node is historized.</param>
|
||||
/// <param name="historyWritable">A HistoryUpdate backend is wired.</param>
|
||||
/// <returns>The access-level byte.</returns>
|
||||
internal static byte ComposeAccessLevel(bool writable, bool historized, bool historyWritable)
|
||||
{
|
||||
byte access = writable ? (byte)(AccessLevels.CurrentRead | AccessLevels.CurrentWrite) : AccessLevels.CurrentRead;
|
||||
if (historized) access = (byte)(access | AccessLevels.HistoryRead);
|
||||
if (historized && historyWritable) access = (byte)(access | AccessLevels.HistoryWrite);
|
||||
return access;
|
||||
}
|
||||
|
||||
@@ -2949,7 +2980,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
{
|
||||
if (!_variables.TryGetValue(key, out var v)) return false;
|
||||
var historized = historianTagname is not null;
|
||||
var access = ComposeAccessLevel(writable, historized);
|
||||
var access = ComposeAccessLevel(writable, historized, _historianDataUpdater is not null);
|
||||
v.AccessLevel = access;
|
||||
v.UserAccessLevel = access;
|
||||
v.Historizing = historized;
|
||||
@@ -5148,6 +5179,301 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------------------------
|
||||
// HistoryUpdate (pluggable historian, spec §5). The SDK's public HistoryUpdate admits a variable only when
|
||||
// its AccessLevel carries HistoryWrite (see ComposeAccessLevel), pre-seeds every admitted slot
|
||||
// BadHistoryOperationUnsupported, RELEASES the node-manager Lock and dispatches by details type to the
|
||||
// protected arms below — so, as for HistoryRead, block-bridging the async updater here is safe. Event and
|
||||
// structure-data details are not overridden: the base leaves them Unsupported.
|
||||
// ---------------------------------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// Drops the handles in a HistoryUpdate batch the calling session may not update, marking each denied
|
||||
/// slot <c>BadUserAccessDenied</c>. Unlike <see cref="AuthorizeHistoryRead"/> this gate exists in BOTH
|
||||
/// modes, because rewriting recorded history was never ungated.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// ACL flag <b>on</b>: the scoped decision for <see cref="OpcUaOperation.HistoryUpdate"/> REPLACES the
|
||||
/// role check (never chained — see <see cref="OnEquipmentTagWrite"/> for why chaining would leave
|
||||
/// every narrow grant inert).
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// ACL flag <b>off</b>: the session's key is re-checked first (Gitea #563 — an expired or revoked key
|
||||
/// loses its activation-time roles), then the dedicated
|
||||
/// <see cref="OpcUaDataPlaneRoles.HistoryUpdate"/> role is required. <c>WriteOperate</c> is
|
||||
/// deliberately NOT reused: writing a live value and rewriting history are different authorities.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
/// <param name="context">The SDK context carrying the session identity.</param>
|
||||
/// <param name="nodesToProcess">The handles the base dispatcher admitted.</param>
|
||||
/// <param name="results">The service-level results list.</param>
|
||||
/// <param name="errors">The service-level errors list.</param>
|
||||
/// <returns>The handles that may proceed.</returns>
|
||||
private List<NodeHandle> AuthorizeHistoryUpdate(
|
||||
ServerSystemContext context,
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors)
|
||||
{
|
||||
var identity = (context as ISessionOperationContext)?.UserIdentity as RoleCarryingUserIdentity
|
||||
?? context.OperationContext?.UserIdentity as RoleCarryingUserIdentity;
|
||||
|
||||
var allowed = new List<NodeHandle>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
var veto = AclEnforced
|
||||
? _authorizer!.Authorize(identity, OpcUaOperation.HistoryUpdate, handle.NodeId, RealmOf(handle.NodeId))
|
||||
: _authorizer?.AuthorizeKeyUsable(identity, OpcUaOperation.HistoryUpdate, handle.NodeId)
|
||||
?? EvaluateHistoryUpdateRoleGate(identity);
|
||||
if (veto is null)
|
||||
{
|
||||
allowed.Add(handle);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Both slots explicitly: "you may not update this" is a different fact from the pre-seeded
|
||||
// "this node has no updatable history".
|
||||
errors[handle.Index] = veto;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = veto.StatusCode };
|
||||
}
|
||||
|
||||
return allowed;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The flag-off role gate for HistoryUpdate: a null identity or one without the
|
||||
/// <see cref="OpcUaDataPlaneRoles.HistoryUpdate"/> role ⇒ <c>BadUserAccessDenied</c>; otherwise
|
||||
/// <c>null</c> (proceed). Case-insensitive, matching the other role gates.
|
||||
/// </summary>
|
||||
/// <param name="identity">The session's role-carrying identity, or null.</param>
|
||||
/// <returns><c>null</c> to proceed; otherwise the veto.</returns>
|
||||
internal static ServiceResult? EvaluateHistoryUpdateRoleGate(RoleCarryingUserIdentity? identity) =>
|
||||
identity is null || !identity.Roles.Contains(OpcUaDataPlaneRoles.HistoryUpdate, StringComparer.OrdinalIgnoreCase)
|
||||
? new ServiceResult(StatusCodes.BadUserAccessDenied)
|
||||
: null;
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void HistoryUpdateData(
|
||||
ServerSystemContext context,
|
||||
IList<UpdateDataDetails> nodesToUpdate,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IDictionary<NodeId, NodeState> cache)
|
||||
{
|
||||
ServeHistoryUpdate(context, nodesToProcess, results, errors, handle =>
|
||||
{
|
||||
var details = nodesToUpdate[handle.Index];
|
||||
HistoryUpdateMode mode;
|
||||
switch (details.PerformInsertReplace)
|
||||
{
|
||||
case PerformUpdateType.Insert: mode = HistoryUpdateMode.Insert; break;
|
||||
case PerformUpdateType.Replace: mode = HistoryUpdateMode.Replace; break;
|
||||
case PerformUpdateType.Update: mode = HistoryUpdateMode.Update; break;
|
||||
default: return null; // Remove (and anything unknown) ⇒ unsupported
|
||||
}
|
||||
|
||||
var items = (IList<DataValue>?)details.UpdateValues ?? Array.Empty<DataValue>();
|
||||
return (updater, tagname, ct) => UpdateDataAsync(updater, tagname, mode, items, ct);
|
||||
});
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void HistoryDeleteRawModified(
|
||||
ServerSystemContext context,
|
||||
IList<DeleteRawModifiedDetails> nodesToUpdate,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IDictionary<NodeId, NodeState> cache)
|
||||
{
|
||||
ServeHistoryUpdate(context, nodesToProcess, results, errors, handle =>
|
||||
{
|
||||
var details = nodesToUpdate[handle.Index];
|
||||
// We never keep modified-value history, so there is none to delete.
|
||||
if (details.IsDeleteModified) return null;
|
||||
var start = details.StartTime;
|
||||
var end = details.EndTime;
|
||||
return async (updater, tagname, ct) =>
|
||||
{
|
||||
var code = await updater.DeleteRawAsync(tagname, start, end, ct).ConfigureAwait(false);
|
||||
return new HistoryUpdateResult { StatusCode = code };
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override void HistoryDeleteAtTime(
|
||||
ServerSystemContext context,
|
||||
IList<DeleteAtTimeDetails> nodesToUpdate,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IDictionary<NodeId, NodeState> cache)
|
||||
{
|
||||
ServeHistoryUpdate(context, nodesToProcess, results, errors, handle =>
|
||||
{
|
||||
var times = nodesToUpdate[handle.Index].ReqTimes?.ToList() ?? new List<DateTime>();
|
||||
return async (updater, tagname, ct) =>
|
||||
{
|
||||
var codes = times.Count == 0
|
||||
? Array.Empty<uint>()
|
||||
: await updater.DeleteAtTimeAsync(tagname, times, ct).ConfigureAwait(false);
|
||||
if (codes.Count != times.Count)
|
||||
throw new InvalidOperationException(
|
||||
$"historian updater returned {codes.Count} result(s) for {times.Count} time(s)");
|
||||
return new HistoryUpdateResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
OperationResults = new StatusCodeCollection(codes.Select(c => new StatusCode(c))),
|
||||
};
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Applies one node's UpdateDataDetails: items the server can refuse on its own
|
||||
/// (<c>SourceTimestamp == DateTime.MinValue</c> ⇒ <c>BadInvalidTimestamp</c>; a value
|
||||
/// <see cref="HistorianValue.TryFromObject"/> cannot type ⇒ <c>BadTypeMismatch</c>) are answered locally,
|
||||
/// the rest go to the updater in ONE call, and the per-item codes are stitched back in request order.
|
||||
/// </summary>
|
||||
/// <param name="updater">The wired updater.</param>
|
||||
/// <param name="tagname">The node's historian tagname.</param>
|
||||
/// <param name="mode">The update mode.</param>
|
||||
/// <param name="items">The request's values, in order.</param>
|
||||
/// <param name="ct">The request deadline.</param>
|
||||
/// <returns>The node's result, carrying one operation result per item.</returns>
|
||||
private static async Task<HistoryUpdateResult> UpdateDataAsync(
|
||||
IHistorianDataUpdater updater, string tagname, HistoryUpdateMode mode, IList<DataValue> items,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var codes = new uint[items.Count];
|
||||
var toSend = new List<HistorizationValue>(items.Count);
|
||||
var sentIndex = new List<int>(items.Count);
|
||||
for (var i = 0; i < items.Count; i++)
|
||||
{
|
||||
var item = items[i];
|
||||
if (item is null)
|
||||
{
|
||||
codes[i] = StatusCodes.BadInvalidArgument;
|
||||
}
|
||||
else if (item.SourceTimestamp == DateTime.MinValue)
|
||||
{
|
||||
codes[i] = StatusCodes.BadInvalidTimestamp;
|
||||
}
|
||||
else if (!HistorianValue.TryFromObject(item.Value, out var value))
|
||||
{
|
||||
codes[i] = StatusCodes.BadTypeMismatch;
|
||||
}
|
||||
else
|
||||
{
|
||||
toSend.Add(new HistorizationValue(item.SourceTimestamp, value, item.StatusCode.Code));
|
||||
sentIndex.Add(i);
|
||||
}
|
||||
}
|
||||
|
||||
if (toSend.Count > 0)
|
||||
{
|
||||
var backend = await updater.UpdateDataAsync(tagname, mode, toSend, ct).ConfigureAwait(false);
|
||||
if (backend.Count != toSend.Count)
|
||||
throw new InvalidOperationException(
|
||||
$"historian updater returned {backend.Count} result(s) for {toSend.Count} value(s)");
|
||||
for (var j = 0; j < backend.Count; j++) codes[sentIndex[j]] = backend[j];
|
||||
}
|
||||
|
||||
// Per-node StatusCode stays Good even when item results are mixed — Part 11 reports item outcomes in
|
||||
// OperationResults; the node code is for whole-node failures only.
|
||||
return new HistoryUpdateResult
|
||||
{
|
||||
StatusCode = StatusCodes.Good,
|
||||
OperationResults = new StatusCodeCollection(codes.Select(c => new StatusCode(c))),
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Shared driver for the three HistoryUpdate arms: authorize, then under the shared HistoryRead batch
|
||||
/// limiter + per-request deadline resolve each node's tagname and run its operation with bounded
|
||||
/// parallelism. A node that is not historized, has no updater, or whose details are out of scope
|
||||
/// (<paramref name="plan"/> returns null) is <c>BadHistoryOperationUnsupported</c>.
|
||||
/// </summary>
|
||||
/// <param name="context">The SDK context.</param>
|
||||
/// <param name="nodesToProcess">The handles the base dispatcher admitted.</param>
|
||||
/// <param name="results">The service-level results list.</param>
|
||||
/// <param name="errors">The service-level errors list.</param>
|
||||
/// <param name="plan">Builds a node's operation from its details, or returns null when unsupported.</param>
|
||||
private void ServeHistoryUpdate(
|
||||
ServerSystemContext context,
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
Func<NodeHandle, Func<IHistorianDataUpdater, string, CancellationToken, Task<HistoryUpdateResult>>?> plan)
|
||||
{
|
||||
nodesToProcess = AuthorizeHistoryUpdate(context, nodesToProcess, results, errors);
|
||||
if (nodesToProcess.Count == 0) return;
|
||||
|
||||
WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () =>
|
||||
{
|
||||
using var deadline = CreateDeadline();
|
||||
var ct = deadline?.Token ?? CancellationToken.None;
|
||||
var updater = _historianDataUpdater;
|
||||
var work = new List<Func<Task>>(nodesToProcess.Count);
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
var operation = updater is null ? null : plan(handle);
|
||||
if (operation is null || !TryGetHistorizedTagname(handle.NodeId, out var tagname) || tagname is null)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
|
||||
continue;
|
||||
}
|
||||
|
||||
var h = handle;
|
||||
var tag = tagname;
|
||||
work.Add(() => ServeUpdateNodeAsync(h, results, errors, token => operation(updater!, tag, token), ct));
|
||||
}
|
||||
|
||||
RunBounded(work, HistoryReadBatchParallelism);
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs one node's HistoryUpdate operation and fills its slots. Never throws (the
|
||||
/// <see cref="RunBounded"/> contract): a deadline cancellation is <c>BadTimeout</c>, any backend failure
|
||||
/// is logged and reported <c>BadInternalError</c> for THIS node only.
|
||||
/// </summary>
|
||||
/// <param name="handle">The node handle.</param>
|
||||
/// <param name="results">The service-level results list.</param>
|
||||
/// <param name="errors">The service-level errors list.</param>
|
||||
/// <param name="operation">The node's operation.</param>
|
||||
/// <param name="ct">The request deadline.</param>
|
||||
private async Task ServeUpdateNodeAsync(
|
||||
NodeHandle handle,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
Func<CancellationToken, Task<HistoryUpdateResult>> operation,
|
||||
CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
results[handle.Index] = await operation(ct).ConfigureAwait(false);
|
||||
// The base pre-seeded BadHistoryOperationUnsupported, and the master node manager copies a non-Good
|
||||
// error over the result's StatusCode — so a success MUST clear it.
|
||||
errors[handle.Index] = ServiceResult.Good;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadTimeout;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadTimeout };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "OtOpcUaNodeManager: HistoryUpdate failed for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadInternalError;
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadInternalError };
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Clamp a <see cref="uint"/> request cap to a non-negative <see cref="int"/> for the
|
||||
/// event-read surface (whose <c>maxEvents</c> is signed): values above <see cref="int.MaxValue"/>
|
||||
/// saturate to <see cref="int.MaxValue"/>.</summary>
|
||||
@@ -5298,7 +5624,37 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IList<SdkHistoryReadResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
Action serve)
|
||||
Action serve) =>
|
||||
WithHistoryBatchLimiter(nodesToProcess, errors, serve, handle =>
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations });
|
||||
|
||||
/// <summary>
|
||||
/// HistoryUpdate counterpart of <see cref="WithHistoryReadBatchLimiter(List{NodeHandle}, IList{SdkHistoryReadResult}, IList{ServiceResult}, Action)"/>:
|
||||
/// the SAME process-wide limiter and admission wait, so update batches and read batches share one
|
||||
/// concurrency budget against the historian backend.
|
||||
/// </summary>
|
||||
/// <param name="nodesToProcess">The batch's handles — every slot is failed on limiter saturation.</param>
|
||||
/// <param name="results">The service-level HistoryUpdate results list.</param>
|
||||
/// <param name="errors">The service-level errors list.</param>
|
||||
/// <param name="serve">The arm body to run once a slot is acquired.</param>
|
||||
private void WithHistoryReadBatchLimiter(
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IList<HistoryUpdateResult> results,
|
||||
IList<ServiceResult> errors,
|
||||
Action serve) =>
|
||||
WithHistoryBatchLimiter(nodesToProcess, errors, serve, handle =>
|
||||
results[handle.Index] = new HistoryUpdateResult { StatusCode = StatusCodes.BadTooManyOperations });
|
||||
|
||||
/// <summary>Shared body of the HistoryRead / HistoryUpdate batch limiter.</summary>
|
||||
/// <param name="nodesToProcess">The batch's handles.</param>
|
||||
/// <param name="errors">The service-level errors list.</param>
|
||||
/// <param name="serve">The arm body.</param>
|
||||
/// <param name="rejectResult">Fills the per-node result slot when the batch is rejected.</param>
|
||||
private void WithHistoryBatchLimiter(
|
||||
List<NodeHandle> nodesToProcess,
|
||||
IList<ServiceResult> errors,
|
||||
Action serve,
|
||||
Action<NodeHandle> rejectResult)
|
||||
{
|
||||
var gate = GetHistoryReadBatchLimiter();
|
||||
|
||||
@@ -5312,7 +5668,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
foreach (var handle in nodesToProcess)
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadTooManyOperations;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations };
|
||||
rejectResult(handle);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -128,6 +128,25 @@ public sealed class OtOpcUaSdkServer : StandardServer
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wire the optional HistoryUpdate backend (the <see cref="IHistorianDataUpdater"/> the node manager's
|
||||
/// HistoryUpdate overrides block-bridge to) onto the created <see cref="OtOpcUaNodeManager"/>. Mirrors
|
||||
/// <see cref="SetHistorianDataSource"/>; <c>null</c> means "this backend cannot modify history".
|
||||
/// <para>
|
||||
/// ⚠️ Call this BEFORE any node materialises: whether an updater is wired decides the
|
||||
/// <c>HistoryWrite</c> AccessLevel bit at materialise time, and the SDK base admits a node to
|
||||
/// HistoryUpdate only when that bit is set. Nodes materialised earlier are not retro-updated.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
/// <param name="updater">The update backend, or <c>null</c> for none.</param>
|
||||
/// <returns><c>true</c> when set on a live node manager; <c>false</c> when no node manager exists yet.</returns>
|
||||
public bool SetHistorianDataUpdater(IHistorianDataUpdater? updater)
|
||||
{
|
||||
if (_otOpcUaNodeManager is null) return false;
|
||||
_otOpcUaNodeManager.HistorianDataUpdater = updater;
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Wire the node's data-plane authorization inputs (#520 stage 4) onto the created
|
||||
/// <see cref="OtOpcUaNodeManager"/> — the cluster it serves, the current auth generation, and the
|
||||
|
||||
@@ -28,4 +28,10 @@ public static class OpcUaDataPlaneRoles
|
||||
/// <c>OnWriteValue</c> handler to route the value to the backing driver; absent it the
|
||||
/// write is denied with <c>BadUserAccessDenied</c> before any driver call.</summary>
|
||||
public const string WriteOperate = "WriteOperate";
|
||||
|
||||
/// <summary>The role that grants OPC UA HistoryUpdate authority (insert / replace / update / delete of
|
||||
/// historized samples) while <c>Security:Acl:Enabled</c> is off. Conferred only by a Cluster-scoped grant
|
||||
/// carrying <c>NodePermissions.HistoryUpdate</c>; <see cref="WriteOperate"/> does NOT confer it — writing
|
||||
/// a live value and rewriting recorded history are different authorities.</summary>
|
||||
public const string HistoryUpdate = "HistoryUpdate";
|
||||
}
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Runtime.Historian;
|
||||
|
||||
/// <summary>
|
||||
/// DI carrier for the optional <see cref="IHistorianDataUpdater"/> the selected historian backend supplies.
|
||||
/// A holder (rather than a nullable registration) lets "this backend cannot modify history" be registered
|
||||
/// explicitly as <see cref="None"/>, so consumers resolve one type and branch on <see cref="Updater"/>.
|
||||
/// </summary>
|
||||
/// <param name="Updater">The backend's updater, or <c>null</c> when HistoryUpdate is unsupported.</param>
|
||||
public sealed record HistorianUpdaterHolder(IHistorianDataUpdater? Updater)
|
||||
{
|
||||
/// <summary>No updater: every HistoryUpdate is reported unsupported.</summary>
|
||||
public static readonly HistorianUpdaterHolder None = new((IHistorianDataUpdater?)null);
|
||||
}
|
||||
Reference in New Issue
Block a user