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

fix(hosts): driver health is per (driver, node) — ReportingNodeId on DriverHealthChanged + telemetry proto (#847)

Browse Source
This commit is contained in:
Joseph Doherty
2026-11-11 03:41:59 -05:00
parent 3b1b32f0c1
commit 038aa007ac
11 changed files with 150 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Core/ZB.MOM.WW.OtOpcUa.Commons/Messages/Drivers/DriverHealthChanged.cs
+9 -1
View File
@@ -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
/// <paramref name="HostStatuses"/>.
/// </param>
/// <param name="ReportingNodeId">
/// The <c>ClusterNode.NodeId</c> 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 <c>DriverInstanceId</c> 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.
/// </param>
public sealed record DriverHealthChanged(
string ClusterId,
string DriverInstanceId,
@@ -77,7 +84,8 @@ public sealed record DriverHealthChanged(
string? RediscoveryReason = null,
IReadOnlyList<HostConnectivityStatus>? HostStatuses = null,
SkippedTagReport? SkippedTags = null,
IReadOnlyDictionary<string, string>? Diagnostics = null)
IReadOnlyDictionary<string, string>? Diagnostics = null,
string? ReportingNodeId = null)
{
/// <summary>
/// DPS topic name. Both the runtime <c>AkkaDriverHealthPublisher</c> and the AdminUI
src/Core/ZB.MOM.WW.OtOpcUa.Commons/Protos/telemetry.proto
+4
View File
@@ -88,6 +88,10 @@ message DriverHealth {
// flag carries that bit - the same shape as has_host_statuses.
bool has_diagnostics = 13;
map<string, string> 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.
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Pages/Hosts.razor
+15 -4
View File
@@ -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
<thead>
<tr>
<th>Driver</th>
<th title="The node reporting this row — each node of a redundant pair runs the driver, so a driver has one row per node">Node</th>
<th>Type</th>
<th>Status</th>
<th>Hosts</th>
@@ -166,14 +167,14 @@ else
@if (g.Drivers.Count == 0)
{
<tr>
<td colspan="9"><span class="text-muted">No drivers.</span></td>
<td colspan="10"><span class="text-muted">No drivers.</span></td>
</tr>
}
else
{
@foreach (var d in g.Drivers)
{
<tr @key="@($"row-{d.DriverInstanceId}")">
<tr @key="@($"row-{d.RowKey}")">
<td>
@(d.Name ?? d.DriverInstanceId)
@if (d.RediscoveryNeededUtc is not null)
@@ -203,6 +204,16 @@ else
</div>
}
</td>
<td>
@if (d.NodeId is null)
{
<span class="text-muted" title="Reported by a node that predates per-node health (#847)">—</span>
}
else
{
<span class="mono">@d.NodeId</span>
}
</td>
<td>@(d.DriverType ?? "—")</td>
<td><span class="chip @DriverChipClass(d.State)">@d.State</span></td>
<td>
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hosts/HostsDriverView.cs
+13 -3
View File
@@ -58,6 +58,9 @@ public sealed record HostsDriverInstanceInfo(
/// when no instance matched. Fed to the row's expandable <c>DriverStatusPanel</c>. 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.</param>
/// <param name="NodeId">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.</param>
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<HostConnectivityStatus>? HostStatuses = null,
SkippedTagReport? SkippedTags = null,
bool Enabled = true,
IReadOnlyDictionary<string, string>? Diagnostics = null)
IReadOnlyDictionary<string, string>? Diagnostics = null,
string? NodeId = null)
{
/// <summary>The row's render key: unique per (driver instance, reporting node).</summary>
public string RowKey => $"{DriverInstanceId}@{NodeId}";
/// <summary>Diagnostics as ordered key/value pairs for rendering; empty when there are none, in which
/// case the page renders nothing.</summary>
public IReadOnlyList<KeyValuePair<string, string>> DiagnosticItems { get; } =
@@ -117,7 +124,8 @@ public sealed record HostsDriverRow(
/// </summary>
/// <param name="ClusterId">Cluster identifier.</param>
/// <param name="Nodes">Configured nodes in this cluster, ordered by node id.</param>
/// <param name="Drivers">Driver rows in this cluster, ordered by name then instance id.</param>
/// <param name="Drivers">Driver rows in this cluster — one per (driver instance, reporting node) — ordered by
/// name, then instance id, then node id.</param>
public sealed record HostsClusterGroup(string ClusterId, IReadOnlyList<HostsNodeInfo> Nodes, IReadOnlyList<HostsDriverRow> Drivers);
/// <summary>
@@ -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));
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/IDriverStatusSnapshotStore.cs
+5 -3
View File
@@ -15,14 +15,16 @@ public interface IDriverStatusSnapshotStore
/// <param name="snapshot">The driver health snapshot to store.</param>
void Upsert(DriverHealthChanged snapshot);
/// <summary>Attempts to retrieve the last-known health snapshot for a driver instance.</summary>
/// <summary>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
/// <see cref="GetAll"/> for the per-node view).</summary>
/// <param name="driverInstanceId">The driver instance's stable logical ID.</param>
/// <param name="snapshot">The stored snapshot, if found.</param>
/// <returns><see langword="true"/> if a snapshot was found; otherwise <see langword="false"/>.</returns>
bool TryGet(string driverInstanceId, out DriverHealthChanged snapshot);
/// <summary>Returns a point-in-time snapshot of every driver instance's last-known health.</summary>
/// <returns>A read-only collection of the last-known health snapshot for each driver instance.</returns>
/// <summary>Returns a point-in-time snapshot of every (driver instance, reporting node)'s last-known health.</summary>
/// <returns>One last-known snapshot per driver instance PER reporting node (lmxopcua#847).</returns>
IReadOnlyCollection<DriverHealthChanged> GetAll();
/// <summary>
src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Hubs/InMemoryDriverStatusSnapshotStore.cs
+25 -7
View File
@@ -6,11 +6,17 @@ namespace ZB.MOM.WW.OtOpcUa.AdminUI.Hubs;
/// <summary>
/// Thread-safe in-memory implementation of <see cref="IDriverStatusSnapshotStore"/>.
/// Keyed by <see cref="DriverHealthChanged.DriverInstanceId"/>; last write wins.
/// Keyed by (<see cref="DriverHealthChanged.DriverInstanceId"/>, <see cref="DriverHealthChanged.ReportingNodeId"/>);
/// last write wins per key.
/// </summary>
/// <remarks>
/// 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 <c>ReportingNodeId</c> keys under the empty node.
/// </remarks>
public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotStore
{
private readonly ConcurrentDictionary<string, DriverHealthChanged> _byInstance = new();
private readonly ConcurrentDictionary<(string DriverInstanceId, string NodeId), DriverHealthChanged> _byInstance = new();
private readonly ILogger<InMemoryDriverStatusSnapshotStore>? _logger;
private readonly FailureLogThrottle _failureLog = new();
@@ -22,10 +28,12 @@ public sealed class InMemoryDriverStatusSnapshotStore : IDriverStatusSnapshotSto
/// <summary>Raised whenever a driver status snapshot is stored or updated.</summary>
public event Action<DriverHealthChanged>? SnapshotChanged;
private static (string, string) KeyOf(DriverHealthChanged s) => (s.DriverInstanceId, s.ReportingNodeId ?? "");
/// <inheritdoc />
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<string> 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<string, DriverHealthChanged>(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
/// <inheritdoc />
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;
}
/// <inheritdoc />
public IReadOnlyCollection<DriverHealthChanged> GetAll() => _byInstance.Values.ToArray();
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Telemetry/TelemetryProtoMapCentral.cs
+2 -1
View File
@@ -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<string, string>(msg.Diagnostics) : null);
Diagnostics: msg.HasDiagnostics ? new Dictionary<string, string>(msg.Diagnostics) : null,
ReportingNodeId: msg.HasReportingNodeId ? msg.ReportingNodeId : null);
}
/// <summary>
src/Server/ZB.MOM.WW.OtOpcUa.Host/Grpc/TelemetryProtoMapNode.cs
+2
View File
@@ -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)
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs
+9 -2
View File
@@ -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;
/// <summary>Initializes a new instance of <see cref="AkkaDriverHealthPublisher"/>.</summary>
/// <param name="system">The Akka actor system used to resolve the DPS mediator.</param>
/// <param name="hub">The node-local live-telemetry hub each snapshot is also emitted into (Phase 5).</param>
public AkkaDriverHealthPublisher(ActorSystem system, ITelemetryLocalHub hub)
/// <param name="roleInfo">This node's role info; its <c>LocalNode</c> stamps every snapshot's
/// <see cref="DriverHealthChanged.ReportingNodeId"/> (lmxopcua#847). Optional for harnesses.</param>
public AkkaDriverHealthPublisher(ActorSystem system, ITelemetryLocalHub hub, IClusterRoleInfo? roleInfo = null)
{
_system = system;
_hub = hub;
var local = roleInfo?.LocalNode.Value;
_localNodeId = string.IsNullOrWhiteSpace(local) ? null : local;
}
/// <inheritdoc />
@@ -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.
tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/DriverStatusSnapshotStoreTests.cs
+35
View File
@@ -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<string> { "kept" }, old.AddMinutes(1)).ShouldBe(2);
store.GetAll().ShouldHaveSingleItem().DriverInstanceId.ShouldBe("kept");
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.AdminUI.Tests/Hosts/HostsDriverViewTests.cs
+31
View File
@@ -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</th>");
src.ShouldContain("@d.NodeId");
}
}