From 9882b1d250d57785f66dbd94d86de4553142d97e Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 23 Jul 2026 15:08:56 -0400 Subject: [PATCH] feat(mesh-phase5): node-local telemetry hub (snapshot-replay + drop-oldest fan-out) Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW --- .../ServiceCollectionExtensions.cs | 5 + .../Telemetry/ITelemetryLocalHub.cs | 80 ++++++++ .../Telemetry/TelemetryLocalHub.cs | 121 ++++++++++++ .../Telemetry/TelemetryLocalHubTests.cs | 173 ++++++++++++++++++ 4 files changed, 379 insertions(+) create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs create mode 100644 src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 477a6344..34fa0946 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -28,6 +28,7 @@ using ZB.MOM.WW.OtOpcUa.Runtime.Health; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; using ZB.MOM.WW.LocalDb; @@ -70,6 +71,10 @@ public static class ServiceCollectionExtensions services.TryAddSingleton(NullOpcUaAddressSpaceSink.Instance); services.TryAddSingleton(NullServiceLevelPublisher.Instance); services.TryAddSingleton(); + // Per-cluster mesh Phase 5: the node-local live-telemetry fan-out hub. Feeds this node's OWN + // telemetry (never DPS) to the Phase-5 gRPC streaming service central dials in on. AddOtOpcUaRuntime + // runs only inside Program.cs's hasDriver block, so this lands on driver-role nodes only. + services.TryAddSingleton(); return services; } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs new file mode 100644 index 00000000..127cb755 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/ITelemetryLocalHub.cs @@ -0,0 +1,80 @@ +using System.Threading.Channels; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +/// +/// A single live-telemetry item flowing through the node-local hub. A closed union over the four +/// wire-facing telemetry records — carried as the domain records themselves, NOT as proto (the +/// gRPC projection happens at the streaming-service seam, one layer out). +/// +/// +/// Two of the four are snapshot-style (a last-value-per-key view: keyed +/// by DriverInstanceId, keyed by (DriverInstanceId, HostName)) +/// and are cached so a newly-attaching subscriber is primed immediately. The other two +/// (, ) are append-style logs and are forwarded live-only. +/// +public abstract record TelemetryItem +{ + private TelemetryItem() { } + + /// An append-style alarm transition. Live-forwarded only; never cached/replayed. + public sealed record Alarm(AlarmTransitionEvent E) : TelemetryItem; + + /// An append-style script-log line. Live-forwarded only; never cached/replayed. + public sealed record Script(ScriptLogEntry E) : TelemetryItem; + + /// A snapshot-style driver-health change, cached last-value per DriverInstanceId. + public sealed record Health(DriverHealthChanged E) : TelemetryItem; + + /// A snapshot-style resilience-status change, cached last-value per (DriverInstanceId, HostName). + public sealed record Resilience(DriverResilienceStatusChanged E) : TelemetryItem; +} + +/// +/// One attached reader on the . Disposing detaches it from the hub +/// and completes its channel; the reader first yields the cached snapshots, then live deltas. +/// +public interface ITelemetrySubscription : IDisposable +{ + /// The bounded reader this subscription drains. Completed on . + ChannelReader Reader { get; } +} + +/// +/// Process-wide, node-local fan-out hub sitting between this node's own telemetry producers and the +/// (Phase-5) gRPC streaming service central dials into. +/// +/// +/// +/// CRITICAL INVARIANT — this hub carries ONLY this node's own telemetry. It is fed +/// directly, in-process, by the node's producers and MUST NEVER subscribe to cluster +/// DistributedPubSub. On the single mesh, DPS delivers every node's events to every node; if the +/// hub read DPS it would stream peers' events and central — which dials each node individually — +/// would double-count. +/// +/// +/// Fan-out is lossy under backpressure by design: each subscriber owns a bounded channel with +/// , so a slow consumer sheds its own oldest items +/// and can never block the producer / node-actor threads. +/// +/// +public interface ITelemetryLocalHub +{ + /// + /// Fans to every currently-attached subscriber (non-blocking, drop-oldest + /// on a full channel) and, for the two snapshot-style items, updates the last-value cache that + /// primes future subscribers. + /// + void Emit(TelemetryItem item); + + /// + /// Attaches a new subscriber. Its reader first yields the cached snapshots (all cached + /// , then all cached ), + /// then live deltas — with no delta lost across the attach boundary. Dispose to detach. + /// + /// Per-subscriber channel capacity (must be positive). + ITelemetrySubscription Subscribe(int boundedCapacity); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs new file mode 100644 index 00000000..6aaf9ba9 --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Telemetry/TelemetryLocalHub.cs @@ -0,0 +1,121 @@ +using System.Collections.Concurrent; +using System.Threading.Channels; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +/// +/// Default : a process-wide singleton on driver-role nodes. +/// +/// +/// +/// No DPS. This hub is fed only by this node's own in-process producers (a later Phase-5 +/// task taps them). It never subscribes to cluster DistributedPubSub — see the invariant on +/// . +/// +/// +/// Snapshot-then-attach ordering (no lost delta). updates the snapshot +/// cache and fans out under a single gate; writes the cached snapshots +/// into the new channel and only THEN adds it to the subscriber set — all under that same gate. +/// Because both operations are serialized by the gate, an racing a +/// is resolved one of two ways, both correct: it runs entirely before the +/// subscribe (its value is in the snapshot, and it is not delivered live because the channel is +/// not yet attached), or entirely after (it is delivered live, strictly after the snapshot). No +/// item is dropped or duplicated across the boundary. Fan-out is a non-blocking +/// into bounded/DropOldest channels, so holding the gate +/// never blocks on a slow consumer. +/// +/// +public sealed class TelemetryLocalHub : ITelemetryLocalHub +{ + private readonly object _gate = new(); + private readonly ConcurrentDictionary> _subscribers = new(); + + // Snapshot last-value caches. Only ever mutated under _gate (kept ConcurrentDictionary so a + // Subscribe reading them under the gate is trivially safe even against any future lock-free read). + private readonly ConcurrentDictionary _healthCache = new(); + private readonly ConcurrentDictionary<(string InstanceId, string HostName), TelemetryItem.Resilience> _resilienceCache = new(); + + /// + public void Emit(TelemetryItem item) + { + ArgumentNullException.ThrowIfNull(item); + + lock (_gate) + { + switch (item) + { + case TelemetryItem.Health h: + _healthCache[h.E.DriverInstanceId] = h; + break; + case TelemetryItem.Resilience r: + _resilienceCache[(r.E.DriverInstanceId, r.E.HostName)] = r; + break; + // Alarm / Script are append-style — never cached. + } + + foreach (var channel in _subscribers.Values) + channel.Writer.TryWrite(item); // bounded + DropOldest ⇒ non-blocking, sheds oldest when full + } + } + + /// + public ITelemetrySubscription Subscribe(int boundedCapacity) + { + if (boundedCapacity < 1) + throw new ArgumentOutOfRangeException(nameof(boundedCapacity), boundedCapacity, "Capacity must be positive."); + + var channel = Channel.CreateBounded(new BoundedChannelOptions(boundedCapacity) + { + FullMode = BoundedChannelFullMode.DropOldest, + SingleReader = true, + SingleWriter = false, + }); + + var id = Guid.NewGuid(); + lock (_gate) + { + // Snapshot FIRST (Health then Resilience), attach SECOND — both under the gate so a + // concurrent Emit cannot slip a delta between the snapshot read and the attach. + foreach (var health in _healthCache.Values) + channel.Writer.TryWrite(health); + foreach (var resilience in _resilienceCache.Values) + channel.Writer.TryWrite(resilience); + + _subscribers[id] = channel; + } + + return new Subscription(this, id, channel); + } + + private void Detach(Guid id) + { + lock (_gate) + { + if (_subscribers.TryRemove(id, out var channel)) + channel.Writer.TryComplete(); + } + } + + private sealed class Subscription : ITelemetrySubscription + { + private readonly TelemetryLocalHub _hub; + private readonly Guid _id; + private int _disposed; + + public Subscription(TelemetryLocalHub hub, Guid id, Channel channel) + { + _hub = hub; + _id = id; + Reader = channel.Reader; + } + + public ChannelReader Reader { get; } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; // idempotent + _hub.Detach(_id); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs new file mode 100644 index 00000000..81613977 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/TelemetryLocalHubTests.cs @@ -0,0 +1,173 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry; + +/// +/// Verifies the node-local telemetry hub (per-cluster mesh Phase 5): snapshot-replay of the two +/// snapshot-style channels (Health / Resilience), live-only forwarding of the append-style logs +/// (Alarm / Script), independent bounded per-subscriber fan-out, DropOldest backpressure, and +/// detach-on-dispose. The hub carries ONLY this node's own telemetry — it never touches DPS. +/// +public sealed class TelemetryLocalHubTests +{ + private static DriverHealthChanged Health(string instanceId, string state = "Healthy") => + new("cluster-a", instanceId, state, null, null, 0, DateTime.UtcNow); + + private static DriverResilienceStatusChanged Resilience(string instanceId, string host) => + new(instanceId, host, false, 0, 0, null, DateTime.UtcNow, DateTime.UtcNow); + + private static AlarmTransitionEvent Alarm(string id) => + new(id, "Area/Line/Equip", "AlarmName", "Activated", 500, "msg", "system", DateTime.UtcNow); + + private static ScriptLogEntry Script(string id) => + new(id, "Information", "log", DateTime.UtcNow, null, null, null); + + private static List Drain(ITelemetrySubscription sub) + { + var items = new List(); + while (sub.Reader.TryRead(out var item)) + items.Add(item); + return items; + } + + [Fact] + public void New_subscriber_is_primed_with_cached_Health_snapshot() + { + var hub = new TelemetryLocalHub(); + hub.Emit(new TelemetryItem.Health(Health("drv-1", "Faulted"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + var items = Drain(sub); + items.ShouldHaveSingleItem(); + var health = items[0].ShouldBeOfType(); + health.E.DriverInstanceId.ShouldBe("drv-1"); + health.E.State.ShouldBe("Faulted"); + } + + [Fact] + public void New_subscriber_is_primed_with_cached_Resilience_snapshot() + { + var hub = new TelemetryLocalHub(); + hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + var items = Drain(sub); + items.ShouldHaveSingleItem(); + var res = items[0].ShouldBeOfType(); + res.E.DriverInstanceId.ShouldBe("drv-1"); + res.E.HostName.ShouldBe("host-a"); + } + + [Fact] + public void Alarm_and_Script_are_not_replayed_to_a_new_subscriber() + { + var hub = new TelemetryLocalHub(); + hub.Emit(new TelemetryItem.Alarm(Alarm("a-1"))); + hub.Emit(new TelemetryItem.Script(Script("s-1"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + Drain(sub).ShouldBeEmpty(); + } + + [Fact] + public void Two_concurrent_subscribers_each_receive_a_post_subscribe_emit() + { + var hub = new TelemetryLocalHub(); + using var s1 = hub.Subscribe(boundedCapacity: 16); + using var s2 = hub.Subscribe(boundedCapacity: 16); + + hub.Emit(new TelemetryItem.Alarm(Alarm("a-1"))); + + Drain(s1).ShouldHaveSingleItem().ShouldBeOfType().E.AlarmId.ShouldBe("a-1"); + Drain(s2).ShouldHaveSingleItem().ShouldBeOfType().E.AlarmId.ShouldBe("a-1"); + } + + [Fact] + public void Health_snapshot_keeps_only_the_latest_per_instance() + { + var hub = new TelemetryLocalHub(); + hub.Emit(new TelemetryItem.Health(Health("drv-1", "Reconnecting"))); + hub.Emit(new TelemetryItem.Health(Health("drv-1", "Healthy"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + var items = Drain(sub); + items.ShouldHaveSingleItem(); + items[0].ShouldBeOfType().E.State.ShouldBe("Healthy"); + } + + [Fact] + public void Health_snapshot_keeps_distinct_instances() + { + var hub = new TelemetryLocalHub(); + hub.Emit(new TelemetryItem.Health(Health("drv-1"))); + hub.Emit(new TelemetryItem.Health(Health("drv-2"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + var ids = Drain(sub) + .OfType() + .Select(h => h.E.DriverInstanceId) + .OrderBy(x => x) + .ToArray(); + ids.ShouldBe(new[] { "drv-1", "drv-2" }); + } + + [Fact] + public void Resilience_snapshot_keeps_only_latest_per_instance_and_host() + { + var hub = new TelemetryLocalHub(); + // Same instance, two distinct hosts → two cache entries. + hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a"))); + hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-b"))); + // Overwrite (drv-1, host-a). + hub.Emit(new TelemetryItem.Resilience(Resilience("drv-1", "host-a"))); + + using var sub = hub.Subscribe(boundedCapacity: 16); + + var keys = Drain(sub) + .OfType() + .Select(r => (r.E.DriverInstanceId, r.E.HostName)) + .OrderBy(x => x.HostName) + .ToArray(); + keys.ShouldBe(new[] { ("drv-1", "host-a"), ("drv-1", "host-b") }); + } + + [Fact] + public void Bounded_channel_drops_oldest_not_newest_when_full() + { + var hub = new TelemetryLocalHub(); + using var sub = hub.Subscribe(boundedCapacity: 2); + + // Append-style (uncached) items so the snapshot doesn't consume capacity. + hub.Emit(new TelemetryItem.Alarm(Alarm("a-1"))); + hub.Emit(new TelemetryItem.Alarm(Alarm("a-2"))); + hub.Emit(new TelemetryItem.Alarm(Alarm("a-3"))); + + var ids = Drain(sub).OfType().Select(a => a.E.AlarmId).ToArray(); + // a-1 (oldest) evicted; the two newest survive in order. + ids.ShouldBe(new[] { "a-2", "a-3" }); + } + + [Fact] + public void Dispose_detaches_the_subscriber_and_completes_its_channel() + { + var hub = new TelemetryLocalHub(); + var sub = hub.Subscribe(boundedCapacity: 16); + + sub.Dispose(); + + // A further Emit must neither throw nor deliver to the detached subscriber. + Should.NotThrow(() => hub.Emit(new TelemetryItem.Alarm(Alarm("a-1")))); + sub.Reader.Completion.IsCompleted.ShouldBeTrue(); + sub.Reader.TryRead(out _).ShouldBeFalse(); + } +}