Joseph Doherty
250 lines
10 KiB
C#
250 lines
10 KiB
C#
using System.Collections.Concurrent;
|
|
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;
|
|
|
|
/// <summary>
|
|
/// 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.
|
|
/// </summary>
|
|
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<TelemetryItem> Drain(ITelemetrySubscription sub)
|
|
{
|
|
var items = new List<TelemetryItem>();
|
|
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<TelemetryItem.Health>();
|
|
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<TelemetryItem.Resilience>();
|
|
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 Multiple_subscribers_each_receive_their_own_copy_of_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<TelemetryItem.Alarm>().E.AlarmId.ShouldBe("a-1");
|
|
Drain(s2).ShouldHaveSingleItem().ShouldBeOfType<TelemetryItem.Alarm>().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<TelemetryItem.Health>().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<TelemetryItem.Health>()
|
|
.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<TelemetryItem.Resilience>()
|
|
.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<TelemetryItem.Alarm>().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 Subscribe_with_non_positive_capacity_throws()
|
|
{
|
|
var hub = new TelemetryLocalHub();
|
|
Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(0));
|
|
Should.Throw<ArgumentOutOfRangeException>(() => hub.Subscribe(-1));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Concurrent_emit_subscribe_dispose_is_safe_and_never_double_delivers_a_snapshot()
|
|
{
|
|
var hub = new TelemetryLocalHub();
|
|
var stop = new CancellationTokenSource(TimeSpan.FromSeconds(3));
|
|
var exceptions = new ConcurrentQueue<Exception>();
|
|
|
|
// Writers: hammer Emit with a mix of all four kinds across a small key space so the snapshot
|
|
// caches churn (same instance/host reused → last-value overwrites) while subscribers attach.
|
|
var writers = Enumerable.Range(0, 4).Select(w => Task.Run(() =>
|
|
{
|
|
var rnd = new Random(w * 7919 + 1);
|
|
try
|
|
{
|
|
var n = 0;
|
|
while (!stop.IsCancellationRequested)
|
|
{
|
|
var instance = "drv-" + rnd.Next(0, 4);
|
|
var host = "host-" + rnd.Next(0, 3);
|
|
TelemetryItem item = (n++ % 4) switch
|
|
{
|
|
0 => new TelemetryItem.Health(Health(instance, "s" + n)),
|
|
1 => new TelemetryItem.Resilience(Resilience(instance, host)),
|
|
2 => new TelemetryItem.Alarm(Alarm("a-" + n)),
|
|
_ => new TelemetryItem.Script(Script("s-" + n)),
|
|
};
|
|
hub.Emit(item);
|
|
}
|
|
}
|
|
catch (Exception ex) { exceptions.Enqueue(ex); }
|
|
})).ToArray();
|
|
|
|
// Subscriber/disposer churn: attach, drain, verify the exactly-once-across-attach invariant, dispose.
|
|
var churners = Enumerable.Range(0, 6).Select(_ => Task.Run(() =>
|
|
{
|
|
try
|
|
{
|
|
while (!stop.IsCancellationRequested)
|
|
{
|
|
using var sub = hub.Subscribe(boundedCapacity: 4096);
|
|
// The snapshot prelude is fully written before Subscribe returns, so any Health/
|
|
// Resilience item read here that was NOT part of the prelude must be a live delta —
|
|
// i.e. it arrived strictly after attach. The invariant we assert: for one snapshot
|
|
// KEY, the hub never delivers the SAME snapshot instance both in the prelude and
|
|
// again live. We approximate "same snapshot" by reference identity: a live re-emit
|
|
// is always a freshly-allocated record, so a reference-equal duplicate would be a
|
|
// genuine double-delivery of one cached object across the boundary.
|
|
var seen = new HashSet<TelemetryItem>(ReferenceEqualityComparer.Instance);
|
|
while (sub.Reader.TryRead(out var item))
|
|
{
|
|
// (c) every drained item is a valid, non-null TelemetryItem of a known kind.
|
|
item.ShouldNotBeNull();
|
|
item.ShouldBeAssignableTo<TelemetryItem>();
|
|
// (b) no cached snapshot object delivered twice to THIS subscriber.
|
|
seen.Add(item).ShouldBeTrue();
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex) { exceptions.Enqueue(ex); }
|
|
})).ToArray();
|
|
|
|
await Task.WhenAll(writers.Concat(churners));
|
|
|
|
// (a) no thread threw.
|
|
exceptions.ShouldBeEmpty();
|
|
}
|
|
|
|
[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();
|
|
}
|
|
}
|