diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs index c22b012be..2a1eb77d3 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs @@ -65,6 +65,13 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; /// capability. Additive, never a state change. Same by-reference value-equality caveat as /// . /// +/// +/// The ClusterNode.NodeId of the node whose driver actor produced this snapshot (lmxopcua#847), or +/// null from a node that predates the field / a harness with no role info. Both nodes of a redundant pair +/// run the same driver instance, so DriverInstanceId alone is NOT a unique key: keyed by it, a driver +/// Faulted on central-1 and Healthy on central-2 flipped between the two on every publish. Consumers that keep +/// a last-value cache key by (DriverInstanceId, ReportingNodeId). Additive — an old central ignores it. +/// public sealed record DriverHealthChanged( string ClusterId, string DriverInstanceId, @@ -77,7 +84,8 @@ public sealed record DriverHealthChanged( string? RediscoveryReason = null, IReadOnlyList? HostStatuses = null, SkippedTagReport? SkippedTags = null, - IReadOnlyDictionary? Diagnostics = null) + IReadOnlyDictionary? Diagnostics = null, + string? ReportingNodeId = null) { /// /// DPS topic name. Both the runtime AkkaDriverHealthPublisher and the AdminUI diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto index e2fdd7258..99a07325f 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto @@ -88,6 +88,10 @@ message DriverHealth { // flag carries that bit - the same shape as has_host_statuses. bool has_diagnostics = 13; map diagnostics = 14; + // The ClusterNode.NodeId of the reporting node (lmxopcua#847). Both nodes of a redundant pair run the same + // driver instance, so central keys its last-value cache by (driver_instance_id, reporting_node_id). Additive: + // absent from an old node (central then falls back to the driver id alone). + optional string reporting_node_id = 15; } // Mirrors ZB.MOM.WW.OtOpcUa.Core.Abstractions.SkippedTagReport. diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor index 119529cf2..4fdf06d08 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor @@ -1,8 +1,8 @@ @page "/hosts" @* Two views, top to bottom. (1) Akka cluster topology: each member's NodeId (host:port), roles, leader status — the cluster-membership view. (2) Driver instances: live driver-health grouped by - cluster. The health feed (DriverHealthChanged) carries no per-Akka-member identity, so the rows - are cluster-scoped (keyed per driver instance across the cluster, not per member). The section + cluster. Each health snapshot carries its reporting node (DriverHealthChanged.ReportingNodeId, #847), so the + rows are one per (driver instance, node) — a driver Faulted on one node and Healthy on its peer shows both. The section reads the in-process driver-health snapshot store directly + reloads its config from the ConfigDB. *@ @attribute [Authorize(Policy = AdminUiPolicies.AuthenticatedRead)] @rendermode RenderMode.InteractiveServer @@ -152,6 +152,7 @@ else Driver + Node Type Status Hosts @@ -166,14 +167,14 @@ else @if (g.Drivers.Count == 0) { - No drivers. + No drivers. } else { @foreach (var d in g.Drivers) { - + @(d.Name ?? d.DriverInstanceId) @if (d.RediscoveryNeededUtc is not null) @@ -203,6 +204,16 @@ else } + + @if (d.NodeId is null) + { + — + } + else + { + @d.NodeId + } + @(d.DriverType ?? "—") @d.State diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs index e659ae7d5..555ef0d73 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs @@ -58,6 +58,9 @@ public sealed record HostsDriverInstanceInfo( /// when no instance matched. Fed to the row's expandable DriverStatusPanel. An /// unmatched row defaults to enabled on purpose: the row only exists because a live snapshot arrived, /// so telling the operator it is "not deployed" would contradict the evidence in front of them. +/// The cluster node that reported this snapshot (lmxopcua#847); null from a node that predates +/// the field. Both nodes of a redundant pair run the same driver instance, so the page shows one row per +/// (driver, node) — a driver Faulted on one node and Healthy on the other used to flip between the two. public sealed record HostsDriverRow( string DriverInstanceId, string? Name, string? DriverType, string State, DateTime? LastSuccessfulReadUtc, string? LastError, int ErrorCount5Min, DateTime PublishedUtc, @@ -65,8 +68,12 @@ public sealed record HostsDriverRow( IReadOnlyList? HostStatuses = null, SkippedTagReport? SkippedTags = null, bool Enabled = true, - IReadOnlyDictionary? Diagnostics = null) + IReadOnlyDictionary? Diagnostics = null, + string? NodeId = null) { + /// The row's render key: unique per (driver instance, reporting node). + public string RowKey => $"{DriverInstanceId}@{NodeId}"; + /// Diagnostics as ordered key/value pairs for rendering; empty when there are none, in which /// case the page renders nothing. public IReadOnlyList> DiagnosticItems { get; } = @@ -117,7 +124,8 @@ public sealed record HostsDriverRow( /// /// Cluster identifier. /// Configured nodes in this cluster, ordered by node id. -/// Driver rows in this cluster, ordered by name then instance id. +/// Driver rows in this cluster — one per (driver instance, reporting node) — ordered by +/// name, then instance id, then node id. public sealed record HostsClusterGroup(string ClusterId, IReadOnlyList Nodes, IReadOnlyList Drivers); /// @@ -188,10 +196,12 @@ public static class HostsDriverView s.HostStatuses, s.SkippedTags, instance?.Enabled ?? true, - s.Diagnostics); + s.Diagnostics, + s.ReportingNodeId); }) .OrderBy(d => d.Name ?? d.DriverInstanceId, StringComparer.OrdinalIgnoreCase) .ThenBy(d => d.DriverInstanceId, StringComparer.OrdinalIgnoreCase) + .ThenBy(d => d.NodeId ?? "", StringComparer.OrdinalIgnoreCase) .ToList(); groups.Add(new HostsClusterGroup(clusterId, clusterNodes, clusterDrivers)); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs index 120c91e49..51918c5c5 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs @@ -15,14 +15,16 @@ public interface IDriverStatusSnapshotStore /// The driver health snapshot to store. void Upsert(DriverHealthChanged snapshot); - /// Attempts to retrieve the last-known health snapshot for a driver instance. + /// Attempts to retrieve the most recently published health snapshot for a driver instance, across + /// every node reporting it (both nodes of a redundant pair run the same instance — lmxopcua#847; use + /// for the per-node view). /// The driver instance's stable logical ID. /// The stored snapshot, if found. /// if a snapshot was found; otherwise . bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot); - /// Returns a point-in-time snapshot of every driver instance's last-known health. - /// A read-only collection of the last-known health snapshot for each driver instance. + /// Returns a point-in-time snapshot of every (driver instance, reporting node)'s last-known health. + /// One last-known snapshot per driver instance PER reporting node (lmxopcua#847). IReadOnlyCollection GetAll(); /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs index 395fa6708..e6bba87d7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs @@ -6,11 +6,17 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs; /// /// Thread-safe in-memory implementation of . -/// Keyed by ; last write wins. +/// Keyed by (, ); +/// last write wins per key. /// +/// +/// Keyed per reporting node since lmxopcua#847: both nodes of a redundant pair run the same driver instance, +/// so a driver-id-only key made a driver Faulted on central-1 and Healthy on central-2 flip between the two on +/// every publish. A snapshot from a node that predates ReportingNodeId keys under the empty node. +/// public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotStore { - private readonly ConcurrentDictionary _byInstance = new(); + private readonly ConcurrentDictionary<(string DriverInstanceId, string NodeId), DriverHealthChanged> _byInstance = new(); private readonly ILogger? _logger; private readonly FailureLogThrottle _failureLog = new(); @@ -22,10 +28,12 @@ public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotSto /// Raised whenever a driver status snapshot is stored or updated. public event Action? SnapshotChanged; + private static (string, string) KeyOf(DriverHealthChanged s) => (s.DriverInstanceId, s.ReportingNodeId ?? ""); + /// public void Upsert(DriverHealthChanged snapshot) { - _byInstance[snapshot.DriverInstanceId] = snapshot; + _byInstance[KeyOf(snapshot)] = snapshot; // Capture-then-invoke so a concurrent unsubscribe can't null the delegate mid-raise; one // subscriber at a time so a throwing one cannot starve the rest (or escape onto the bridge actor). SafeMulticast.Invoke(SnapshotChanged, snapshot, @@ -36,11 +44,11 @@ public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotSto public int EvictUnconfiguredStale(IReadOnlySet configuredInstanceIds, DateTime olderThanUtc) { var removed = 0; - foreach (var (id, snapshot) in _byInstance) + foreach (var (key, snapshot) in _byInstance) { - if (configuredInstanceIds.Contains(id) || snapshot.PublishedUtc >= olderThanUtc) continue; + if (configuredInstanceIds.Contains(key.DriverInstanceId) || snapshot.PublishedUtc >= olderThanUtc) continue; // Remove only the exact snapshot inspected: a fresh publish that raced in is kept. - if (_byInstance.TryRemove(new KeyValuePair(id, snapshot))) removed++; + if (_byInstance.TryRemove(new KeyValuePair<(string, string), DriverHealthChanged>(key, snapshot))) removed++; } return removed; @@ -48,7 +56,17 @@ public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotSto /// public bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot) - => _byInstance.TryGetValue(driverInstanceId, out snapshot!); + { + DriverHealthChanged? latest = null; + foreach (var (key, s) in _byInstance) + { + if (!string.Equals(key.DriverInstanceId, driverInstanceId, StringComparison.Ordinal)) continue; + if (latest is null || s.PublishedUtc > latest.PublishedUtc) latest = s; + } + + snapshot = latest!; + return latest is not null; + } /// public IReadOnlyCollection GetAll() => _byInstance.Values.ToArray(); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs index c975196c4..09afcf4b7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs @@ -131,7 +131,8 @@ public static class TelemetryProtoMapCentral RediscoveryReason: msg.HasRediscoveryReason ? msg.RediscoveryReason : null, HostStatuses: ToHostStatuses(msg), SkippedTags: ToSkippedTags(msg), - Diagnostics: msg.HasDiagnostics ? new Dictionary(msg.Diagnostics) : null); + Diagnostics: msg.HasDiagnostics ? new Dictionary(msg.Diagnostics) : null, + ReportingNodeId: msg.HasReportingNodeId ? msg.ReportingNodeId : null); } /// diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs index 772ea7ea0..a5d82f35e 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs @@ -132,6 +132,8 @@ public static class TelemetryProtoMapNode msg.RediscoveryNeededUtc = ToUtcTimestamp(e.RediscoveryNeededUtc.Value); if (e.RediscoveryReason is not null) msg.RediscoveryReason = e.RediscoveryReason; + if (e.ReportingNodeId is not null) + msg.ReportingNodeId = e.ReportingNodeId; // Presence flag first — an empty repeated field cannot say whether the driver has a probe at all. if (e.HostStatuses is not null) diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs index bef91b671..188729f1f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs @@ -1,5 +1,6 @@ using Akka.Actor; using Akka.Cluster.Tools.PublishSubscribe; +using ZB.MOM.WW.OtOpcUa.Commons.Interfaces; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; @@ -19,14 +20,19 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher private readonly ActorSystem _system; private readonly ITelemetryLocalHub _hub; + private readonly string? _localNodeId; /// Initializes a new instance of . /// The Akka actor system used to resolve the DPS mediator. /// The node-local live-telemetry hub each snapshot is also emitted into (Phase 5). - public AkkaDriverHealthPublisher(ActorSystem system, ITelemetryLocalHub hub) + /// This node's role info; its LocalNode stamps every snapshot's + /// (lmxopcua#847). Optional for harnesses. + public AkkaDriverHealthPublisher(ActorSystem system, ITelemetryLocalHub hub, IClusterRoleInfo? roleInfo = null) { _system = system; _hub = hub; + var local = roleInfo?.LocalNode.Value; + _localNodeId = string.IsNullOrWhiteSpace(local) ? null : local; } /// @@ -53,7 +59,8 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher rediscoveryReason, hostStatuses, skippedTags, - diagnostics); + diagnostics, + _localNodeId); DistributedPubSub.Get(_system).Mediator.Tell(new Publish(TopicName, msg)); // Phase 5: fan the same snapshot into the node-local live-telemetry hub (no-op until a gRPC // client subscribes). The DPS publish above is unchanged — the hub is a strictly additive tap. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/DriverStatusSnapshotStoreTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/DriverStatusSnapshotStoreTests.cs index 410c5e53a..3c7e740fa 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/DriverStatusSnapshotStoreTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/DriverStatusSnapshotStoreTests.cs @@ -56,4 +56,39 @@ public sealed class DriverStatusSnapshotStoreTests count.ShouldBe(1); } + + [Fact] + public void Snapshots_from_two_nodes_for_one_driver_are_kept_side_by_side_847() + { + var store = new InMemoryDriverStatusSnapshotStore(); + var t0 = new DateTime(2026, 11, 11, 0, 0, 0, DateTimeKind.Utc); + store.Upsert(new DriverHealthChanged("MAIN", "drv", "Faulted", null, "x", 1, t0, ReportingNodeId: "central-1")); + store.Upsert(new DriverHealthChanged("MAIN", "drv", "Healthy", null, null, 0, t0.AddSeconds(1), ReportingNodeId: "central-2")); + // A re-publish from central-1 replaces only central-1's entry. + store.Upsert(new DriverHealthChanged("MAIN", "drv", "Faulted", null, "y", 2, t0.AddSeconds(2), ReportingNodeId: "central-1")); + + var all = store.GetAll(); + all.Count.ShouldBe(2); + all.Single(s => s.ReportingNodeId == "central-1").LastError.ShouldBe("y"); + all.Single(s => s.ReportingNodeId == "central-2").State.ShouldBe("Healthy"); + + // TryGet (the per-driver panel / hub) answers the most recently published across nodes. + store.TryGet("drv", out var latest).ShouldBeTrue(); + latest.ReportingNodeId.ShouldBe("central-1"); + latest.LastError.ShouldBe("y"); + store.TryGet("other", out _).ShouldBeFalse(); + } + + [Fact] + public void Eviction_removes_every_node_entry_of_an_unconfigured_stale_driver_847() + { + var store = new InMemoryDriverStatusSnapshotStore(); + var old = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + store.Upsert(new DriverHealthChanged("MAIN", "gone", "Healthy", null, null, 0, old, ReportingNodeId: "a")); + store.Upsert(new DriverHealthChanged("MAIN", "gone", "Healthy", null, null, 0, old, ReportingNodeId: "b")); + store.Upsert(new DriverHealthChanged("MAIN", "kept", "Healthy", null, null, 0, old, ReportingNodeId: "a")); + + store.EvictUnconfiguredStale(new HashSet { "kept" }, old.AddMinutes(1)).ShouldBe(2); + store.GetAll().ShouldHaveSingleItem().DriverInstanceId.ShouldBe("kept"); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hosts/HostsDriverViewTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hosts/HostsDriverViewTests.cs index 9a37c04ba..c069336ed 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hosts/HostsDriverViewTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hosts/HostsDriverViewTests.cs @@ -356,4 +356,35 @@ public sealed class HostsDriverViewTests row.HasSkippedTags.ShouldBeFalse(); row.SkippedTagsTitle().ShouldBeEmpty(); } + + [Fact] + public void A_driver_reported_by_two_nodes_yields_one_row_per_node_847() + { + // lmxopcua#847: both nodes of a redundant pair run the same driver; keyed by driver id alone the row + // flipped between central-1's Faulted and central-2's Healthy on every publish. + var snaps = new[] + { + new DriverHealthChanged("c", "drv", "Healthy", null, null, 0, When, ReportingNodeId: "central-2"), + new DriverHealthChanged("c", "drv", "Faulted", null, "boom", 1, When, ReportingNodeId: "central-1"), + }; + + var rows = HostsDriverView.Build(snaps, null, [Instance("drv", "c", "Line 1", "Modbus")]) + .ShouldHaveSingleItem().Drivers; + + rows.Count.ShouldBe(2); + rows.Select(r => r.NodeId).ShouldBe(["central-1", "central-2"]); + rows.Select(r => r.State).ShouldBe(["Faulted", "Healthy"]); + rows.Select(r => r.RowKey).Distinct().Count().ShouldBe(2); + rows.ShouldAllBe(r => r.Name == "Line 1"); + } + + [Fact] + public void Hosts_page_renders_a_node_column_and_keys_rows_by_driver_and_node_847() + { + var src = AdminUiSource.Read("Components", "Pages", "Hosts.razor"); + src.ShouldContain("@key=\"@($\"row-{d.RowKey}\")\""); + src.ShouldNotContain("@key=\"@($\"row-{d.DriverInstanceId}\")\""); + src.ShouldContain(">Node"); + src.ShouldContain("@d.NodeId"); + } }