fix(historian): trim only HistoryServerCapabilities/AggregateFunctions; null set = all (#682 review)
ServerCapabilities/AggregateFunctions (i=2997) is Part 5's list of every aggregate the server supports, and live AggregateFilter subscriptions still compute all 37 SDK aggregates, so it keeps the SDK's full set; only i=11201 follows the backend. A null SupportedAggregates is treated as the interface default (all) by both the advertisement and HistoryRead-Processed (it used to NRE inside the batch). Docs: i=2997/i=11201 split and the boot window. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Joseph Doherty
@@ -6,24 +6,23 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #682 — make the two advertised aggregate-function folders, <c>ServerCapabilities/AggregateFunctions</c>
|
||||
/// (i=2997) and <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201), list exactly the aggregates the wired
|
||||
/// historian backend serves (<see cref="IHistorianDataSource.SupportedAggregates"/>), instead of the SDK's full
|
||||
/// registered set.
|
||||
/// Gitea #682 — make <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201) list exactly the aggregates the
|
||||
/// wired historian backend serves for HistoryRead (<see cref="IHistorianDataSource.SupportedAggregates"/>), instead of
|
||||
/// the SDK's full registered set.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// The SDK's <c>AggregateManager.RegisterFactory</c> both registers a calculator factory AND adds an
|
||||
/// <c>Organizes</c> reference from each folder to the aggregate's object node. Only the REFERENCES are trimmed
|
||||
/// here: the factories stay, because a live-data MonitoredItem with an <c>AggregateFilter</c> is validated
|
||||
/// against (and computed by) the AggregateManager, not the historian — removing a factory would break live
|
||||
/// aggregate subscriptions for every aggregate the historian happens not to serve.
|
||||
/// ⚠️ <c>ServerCapabilities/AggregateFunctions</c> (i=2997) is deliberately NOT trimmed. Part 5 defines it as every
|
||||
/// aggregate the SERVER supports, and the server still supports all the SDK's registered aggregates for live
|
||||
/// data: a MonitoredItem <c>AggregateFilter</c> is validated and computed by the SDK's <c>AggregateManager</c>,
|
||||
/// not the historian. The SDK itself models the split — <c>DiagnosticsNodeManager.AddAggregateFunction</c> always
|
||||
/// adds to i=2997 and adds to i=11201 only when the aggregate is historical. i=11201 is the history surface, so it
|
||||
/// is the one that follows the backend.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The two folders are separate nodes that reference the same aggregate object nodes (both directions). The
|
||||
/// aggregate nodes themselves are kept (they are standard type-system nodes a client may still Read); only the
|
||||
/// folder ⇄ aggregate <c>Organizes</c> pair is removed or restored. Idempotent and reversible, so re-wiring a
|
||||
/// different source re-syncs both folders.
|
||||
/// Only the folder ⇄ aggregate <c>Organizes</c> reference pair is removed or restored (both directions); the
|
||||
/// calculator factories and the aggregate-function object nodes stay. Idempotent and reversible, so re-wiring a
|
||||
/// different source re-syncs the folder.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal static class AdvertisedAggregateFunctions
|
||||
@@ -35,20 +34,26 @@ internal static class AdvertisedAggregateFunctions
|
||||
.Select(f => (NodeId)f.GetValue(null)!)
|
||||
.ToArray();
|
||||
|
||||
/// <summary>The two advertised folders (both are trimmed — they are separate nodes).</summary>
|
||||
internal static readonly NodeId[] Folders =
|
||||
[
|
||||
ObjectIds.Server_ServerCapabilities_AggregateFunctions,
|
||||
ObjectIds.HistoryServerCapabilities_AggregateFunctions,
|
||||
];
|
||||
/// <summary>The trimmed folder — the HISTORY aggregate list only (see remarks for why not i=2997).</summary>
|
||||
internal static readonly NodeId Folder = ObjectIds.HistoryServerCapabilities_AggregateFunctions;
|
||||
|
||||
/// <summary>
|
||||
/// Sync both folders to <paramref name="served"/>: an aggregate-function node is organized under a folder iff the
|
||||
/// server maps it (<see cref="AggregateFunctionMap"/>) AND <paramref name="served"/> contains it.
|
||||
/// The set a source serves, treating a <c>null</c> <see cref="IHistorianDataSource.SupportedAggregates"/> (an
|
||||
/// out-of-tree source that returns null despite the non-nullable contract) as <see cref="HistoryAggregateTypes.All"/>,
|
||||
/// the interface default — the ONE null rule both the advertisement and HistoryRead-Processed use.
|
||||
/// </summary>
|
||||
/// <param name="source">The historian source.</param>
|
||||
/// <returns>Its served set, never null.</returns>
|
||||
public static IReadOnlySet<HistoryAggregateType> ServedBy(IHistorianDataSource source) =>
|
||||
source.SupportedAggregates ?? HistoryAggregateTypes.All;
|
||||
|
||||
/// <summary>
|
||||
/// Sync i=11201 to <paramref name="served"/>: an aggregate-function node is organized under it iff the server maps
|
||||
/// it (<see cref="AggregateFunctionMap"/>) AND <paramref name="served"/> contains it.
|
||||
/// </summary>
|
||||
/// <param name="server">The running server.</param>
|
||||
/// <param name="served">The backend's served aggregates.</param>
|
||||
/// <returns>The advertised NodeIds, or <c>null</c> when the diagnostics node manager is not available yet.</returns>
|
||||
/// <returns>The advertised NodeIds, or <c>null</c> when the diagnostics node manager or folder is not available.</returns>
|
||||
public static IReadOnlyList<NodeId>? Apply(IServerInternal server, IReadOnlySet<HistoryAggregateType> served)
|
||||
{
|
||||
if (server?.DiagnosticsNodeManager is not CustomNodeManager2 diagnostics) return null;
|
||||
@@ -56,10 +61,8 @@ internal static class AdvertisedAggregateFunctions
|
||||
var advertised = new List<NodeId>();
|
||||
lock (diagnostics.Lock)
|
||||
{
|
||||
var folders = Folders
|
||||
.Select(id => diagnostics.FindPredefinedNode(id, typeof(BaseObjectState)))
|
||||
.Where(f => f is not null)
|
||||
.ToList();
|
||||
var folder = diagnostics.FindPredefinedNode(Folder, typeof(BaseObjectState));
|
||||
if (folder is null) return null;
|
||||
|
||||
foreach (var aggregateId in StandardAggregateIds)
|
||||
{
|
||||
@@ -69,19 +72,16 @@ internal static class AdvertisedAggregateFunctions
|
||||
var advertise = AggregateFunctionMap.Map(aggregateId) is { } aggregate && served.Contains(aggregate);
|
||||
if (advertise) advertised.Add(aggregateId);
|
||||
|
||||
foreach (var folder in folders)
|
||||
var linked = folder.ReferenceExists(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
if (advertise && !linked)
|
||||
{
|
||||
var linked = folder!.ReferenceExists(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
if (advertise && !linked)
|
||||
{
|
||||
folder.AddReference(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
aggregateNode.AddReference(ReferenceTypeIds.Organizes, true, folder.NodeId);
|
||||
}
|
||||
else if (!advertise && linked)
|
||||
{
|
||||
folder.RemoveReference(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
aggregateNode.RemoveReference(ReferenceTypeIds.Organizes, true, folder.NodeId);
|
||||
}
|
||||
folder.AddReference(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
aggregateNode.AddReference(ReferenceTypeIds.Organizes, true, folder.NodeId);
|
||||
}
|
||||
else if (!advertise && linked)
|
||||
{
|
||||
folder.RemoveReference(ReferenceTypeIds.Organizes, false, aggregateId);
|
||||
aggregateNode.RemoveReference(ReferenceTypeIds.Organizes, true, folder.NodeId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -345,7 +345,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
/// node-manager <c>Lock</c>, so the override may block-bridge to this (async) source.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Gitea #682: assigning a source also re-syncs the advertised <c>AggregateFunctions</c> folders to its
|
||||
/// Gitea #682: assigning a source also re-syncs <c>HistoryServerCapabilities/AggregateFunctions</c> to its
|
||||
/// <see cref="IHistorianDataSource.SupportedAggregates"/> (see <see cref="SyncAdvertisedAggregates"/>).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
@@ -360,9 +360,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #682 — trim <c>ServerCapabilities/AggregateFunctions</c> (i=2997) and
|
||||
/// <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201) to the current source's
|
||||
/// <see cref="IHistorianDataSource.SupportedAggregates"/>. Called on every <see cref="HistorianDataSource"/>
|
||||
/// Gitea #682 — trim <c>HistoryServerCapabilities/AggregateFunctions</c> (i=11201) to the current source's
|
||||
/// <see cref="IHistorianDataSource.SupportedAggregates"/> (null ⇒ all, <see cref="AdvertisedAggregateFunctions.ServedBy"/>).
|
||||
/// <c>ServerCapabilities/AggregateFunctions</c> (i=2997) keeps the SDK's full set — the server still computes every
|
||||
/// one of them for live <c>AggregateFilter</c> subscriptions. Called on every <see cref="HistorianDataSource"/>
|
||||
/// assignment and by <see cref="OtOpcUaSdkServer"/> right after the SDK registers its aggregate factories, so the
|
||||
/// Null default is applied from boot. Only the folder references change — the AggregateManager's calculator
|
||||
/// factories (live <c>AggregateFilter</c> subscriptions) are untouched. Never throws: a failure leaves the previous
|
||||
@@ -373,11 +374,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
try
|
||||
{
|
||||
var source = _historianDataSource;
|
||||
var advertised = AdvertisedAggregateFunctions.Apply(Server, source.SupportedAggregates);
|
||||
var advertised = AdvertisedAggregateFunctions.Apply(Server, AdvertisedAggregateFunctions.ServedBy(source));
|
||||
if (advertised is not null)
|
||||
{
|
||||
_logger.LogInformation(
|
||||
"OtOpcUaNodeManager: advertising {Count} aggregate function(s) for historian {Source}: {Aggregates}",
|
||||
"OtOpcUaNodeManager: advertising {Count} history aggregate function(s) for historian {Source}: {Aggregates}",
|
||||
advertised.Count, source.GetType().Name,
|
||||
string.Join(", ", advertised.Select(id => AggregateFunctionMap.Map(id)?.ToString() ?? id.ToString())));
|
||||
}
|
||||
@@ -5091,7 +5092,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// #677: a client AggregateConfiguration other than the server defaults is refused per node, never silently ignored.
|
||||
var configurationServable = ServerAggregateConfiguration.IsServable(details.AggregateConfiguration);
|
||||
nodesToProcess = AuthorizeHistoryRead(context, nodesToProcess, results, errors);
|
||||
var source = HistorianDataSource;
|
||||
var served = AdvertisedAggregateFunctions.ServedBy(HistorianDataSource);
|
||||
WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () =>
|
||||
{
|
||||
using var deadline = CreateDeadline();
|
||||
@@ -5102,11 +5103,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
// AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by
|
||||
// the base dispatcher). handle.Index is the node's position in that collection.
|
||||
var aggregateNodeId = details.AggregateType[handle.Index];
|
||||
// #682: one authority — the source's SupportedAggregates (also what the AggregateFunctions folders
|
||||
// advertise). An aggregate the server cannot name, or the backend does not serve, is refused up front
|
||||
// without a backend call.
|
||||
// #682: one authority — the source's SupportedAggregates (also what HistoryServerCapabilities/
|
||||
// AggregateFunctions advertises; null ⇒ all, the same rule). An aggregate the server cannot name, or the
|
||||
// backend does not serve, is refused up front without a backend call.
|
||||
var aggregate = AggregateFunctionMap.Map(aggregateNodeId);
|
||||
if (aggregate is null || !source.SupportedAggregates.Contains(aggregate.Value))
|
||||
if (aggregate is null || !served.Contains(aggregate.Value))
|
||||
{
|
||||
errors[handle.Index] = StatusCodes.BadAggregateNotSupported;
|
||||
continue;
|
||||
|
||||
@@ -202,9 +202,10 @@ public sealed class OtOpcUaSdkServer : StandardServer
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #682 — keep the SDK's full set of calculator factories (live-data <c>AggregateFilter</c> subscriptions
|
||||
/// need them), then trim the advertised <c>AggregateFunctions</c> folders to the node manager's current historian
|
||||
/// source. At this point that is the Null default (no aggregates); the Host's
|
||||
/// <see cref="SetHistorianDataSource"/> re-syncs to the configured backend's set right after start.
|
||||
/// need them, and <c>ServerCapabilities/AggregateFunctions</c> keeps listing all of them), then trim
|
||||
/// <c>HistoryServerCapabilities/AggregateFunctions</c> to the node manager's current historian source. At this
|
||||
/// point that is the Null default (no history aggregates); the Host's <see cref="SetHistorianDataSource"/>
|
||||
/// re-syncs to the configured backend's set right after start.
|
||||
/// </summary>
|
||||
/// <inheritdoc />
|
||||
protected override AggregateManager CreateAggregateManager(
|
||||
|
||||
Reference in New Issue
Block a user