feat(mesh-phase5): tap the 4 telemetry publish seams into the local hub (DPS intact)

Each of the four telemetry publish seams now also emits its DPS payload into the node-local ITelemetryLocalHub (Phase 5), leaving the existing DistributedPubSub publish untouched: - driver-health AkkaDriverHealthPublisher -> TelemetryItem.Health - resilience DriverResilienceStatusPublisherService -> TelemetryItem.Resilience - alerts ScriptedAlarmHostActor + DriverHostActor (native) -> TelemetryItem.Alarm - script-logs VirtualTagActor + DpsScriptLogPublisher -> TelemetryItem.Script DI services take the hub as a required ctor param; actors take an optional nullable hub threaded through their Props and the real spawn sites (DriverHostActor <- ServiceCollectionExtensions; VirtualTagActor/ScriptedAlarmHostActor <- DriverHostActor), so existing test Props constructions keep compiling and simply do not emit. The hub is a no-op until a gRPC client subscribes, so the emit is safe on every node. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
2026-07-23 15:27:29 -04:00
parent 9882b1d250
commit 8e5090edf3
12 changed files with 429 additions and 25 deletions
@@ -1,7 +1,12 @@
+using Akka.Actor;
+using Akka.Cluster;
+using Akka.Configuration;
+using Microsoft.Extensions.Logging.Abstractions;
 using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Core.Resilience;
 using ZB.MOM.WW.OtOpcUa.Host.Drivers;
+using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;

 namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests;

@@ -53,4 +58,81 @@ public sealed class DriverResilienceStatusPublisherServiceTests
        healthy.CurrentInFlight.ShouldBe(1);
        healthy.PublishedUtc.ShouldBe(Published);
    }
+
+    /// <summary>Per-cluster mesh Phase 5 — each snapshot the publish tick sends to the
+    /// <c>driver-resilience-status</c> DPS topic is also emitted into the node-local
+    /// <see cref="ITelemetryLocalHub"/> as a <see cref="TelemetryItem.Resilience"/>. Runs the background
+    /// service against a single-node cluster (so DistributedPubSub resolves) for a few ticks and asserts the
+    /// hub captured the tracked pair.</summary>
+    [Fact]
+    public async Task ExecuteAsync_emits_Resilience_item_per_published_snapshot()
+    {
+        // Plain self-joined single-node cluster (mirrors SecretsReplicationRegistrationTests — the
+        // xunit.v3 project cannot use Akka.TestKit.Xunit2). public-hostname is load-bearing for the bind.
+        var hocon = ConfigurationFactory.ParseString("""
+            akka {
+              loglevel = WARNING
+              actor.provider = cluster
+              remote.dot-netty.tcp {
+                hostname = "127.0.0.1"
+                public-hostname = "127.0.0.1"
+                port = 0
+              }
+              cluster { roles = ["driver"] }
+            }
+            """);
+        var sys = ActorSystem.Create("resil-hub-test", hocon);
+        try
+        {
+            var cluster = Akka.Cluster.Cluster.Get(sys);
+            cluster.Join(cluster.SelfAddress);
+            var upDeadline = DateTime.UtcNow.AddSeconds(10);
+            while (!cluster.State.Members.Any(m => m.Status == MemberStatus.Up) && DateTime.UtcNow < upDeadline)
+                await Task.Delay(50, TestContext.Current.CancellationToken);
+            cluster.State.Members.ShouldContain(m => m.Status == MemberStatus.Up);
+
+            var tracker = new DriverResilienceStatusTracker();
+            tracker.RecordCallStart("drv-1", "host-a");
+
+            var hub = new CapturingHub();
+            var service = new DriverResilienceStatusPublisherService(
+                tracker, () => sys, NullLogger<DriverResilienceStatusPublisherService>.Instance, hub,
+                interval: TimeSpan.FromMilliseconds(50));
+
+            await service.StartAsync(TestContext.Current.CancellationToken);
+            try
+            {
+                var deadline = DateTime.UtcNow.AddSeconds(5);
+                while (hub.Items.Count == 0 && DateTime.UtcNow < deadline)
+                    await Task.Delay(50, TestContext.Current.CancellationToken);
+            }
+            finally
+            {
+                await service.StopAsync(TestContext.Current.CancellationToken);
+            }
+
+            hub.Items.ShouldNotBeEmpty();
+            var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Resilience>();
+            item.E.DriverInstanceId.ShouldBe("drv-1");
+            item.E.HostName.ShouldBe("host-a");
+            item.E.CurrentInFlight.ShouldBe(1);
+        }
+        finally
+        {
+            await sys.Terminate();
+        }
+    }
+
+    /// <summary>Fake hub recording every emit; <see cref="Subscribe"/> is unused here.</summary>
+    private sealed class CapturingHub : ITelemetryLocalHub
+    {
+        public List<TelemetryItem> Items { get; } = new();
+
+        public void Emit(TelemetryItem item)
+        {
+            lock (Items) Items.Add(item);
+        }
+
+        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
+    }
 }
@@ -4,6 +4,7 @@ using Shouldly;
 using Xunit;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
 using ZB.MOM.WW.OtOpcUa.Runtime.Scripting;
+using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
 using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
 using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;

@@ -40,7 +41,7 @@ public sealed class DpsScriptLogPublisherTests : RuntimeActorTestBase
        probe.Send(mediator, new Subscribe(VirtualTagActor.ScriptLogsTopic, probe.Ref));
        probe.ExpectMsg<SubscribeAck>(TimeSpan.FromSeconds(10));

-        var publisher = new DpsScriptLogPublisher(() => Sys);
+        var publisher = new DpsScriptLogPublisher(() => Sys, new NoopHub());
        var entry = SampleEntry();

        // The SubscribeAck confirms the subscribe was registered, but DPS topic membership is
@@ -61,7 +62,7 @@ public sealed class DpsScriptLogPublisherTests : RuntimeActorTestBase
    public void Publish_does_not_throw_when_the_system_accessor_throws()
    {
        var publisher = new DpsScriptLogPublisher(
-            () => throw new InvalidOperationException("system not ready"));
+            () => throw new InvalidOperationException("system not ready"), new NoopHub());

        Should.NotThrow(() => publisher.Publish(SampleEntry()));
    }
@@ -70,6 +71,15 @@ public sealed class DpsScriptLogPublisherTests : RuntimeActorTestBase
    [Fact]
    public void Null_system_accessor_throws_ArgumentNullException()
    {
-        Should.Throw<ArgumentNullException>(() => new DpsScriptLogPublisher(null!));
+        Should.Throw<ArgumentNullException>(() => new DpsScriptLogPublisher(null!, new NoopHub()));
+    }
+
+    /// <summary>No-op telemetry hub for the publisher tests (the hub tap is asserted in
+    /// <c>PublishSeamEmitsToHubTests</c>; here it must simply not interfere).</summary>
+    private sealed class NoopHub : ITelemetryLocalHub
+    {
+        public void Emit(TelemetryItem item) { }
+
+        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
    }
 }
@@ -0,0 +1,216 @@
+using System.Text.Json;
+using Akka.Actor;
+using Microsoft.EntityFrameworkCore;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Engines;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Alerts;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging;
+using ZB.MOM.WW.OtOpcUa.Commons.Types;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+using ZB.MOM.WW.OtOpcUa.Runtime.Drivers;
+using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa;
+using ZB.MOM.WW.OtOpcUa.Runtime.Scripting;
+using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry;
+using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Telemetry;
+
+/// <summary>
+///     Per-cluster mesh Phase 5 — the four telemetry publish seams each additionally fan their DPS payload
+///     into the node-local <see cref="ITelemetryLocalHub"/>. Each test drives a producer with a capturing
+///     fake hub and asserts exactly one <see cref="TelemetryItem"/> of the correct subtype carrying the
+///     SAME payload that goes onto the DistributedPubSub topic. The DPS publish itself is unchanged and is
+///     covered by the existing per-producer tests — these only assert the additive tap.
+/// </summary>
+public sealed class PublishSeamEmitsToHubTests : RuntimeActorTestBase
+{
+    private static readonly NodeId TestNode = NodeId.Parse("telemetry-seam-test");
+    private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64));
+    private static readonly DateTime Ts = new(2026, 7, 23, 10, 0, 0, DateTimeKind.Utc);
+    private const string AlarmRawPath = "Plant/Modbus/dev1/temp_hi";
+
+    // --- driver-health seam (AkkaDriverHealthPublisher) → TelemetryItem.Health --------------------------
+
+    /// <summary>The driver-health publisher fans the SAME <see cref="DriverHealthChanged"/> it publishes to
+    /// the <c>driver-health</c> DPS topic into the hub as a <see cref="TelemetryItem.Health"/>.</summary>
+    [Fact]
+    public void Driver_health_publish_emits_one_Health_item_with_same_payload()
+    {
+        var hub = new CapturingHub();
+        var publisher = new AkkaDriverHealthPublisher(Sys, hub);
+
+        var health = new DriverHealth(DriverState.Degraded, Ts, "some error");
+        publisher.Publish("c1", "drv-1", health, errorCount5Min: 4);
+
+        hub.Items.Count.ShouldBe(1);
+        var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Health>();
+        item.E.ClusterId.ShouldBe("c1");
+        item.E.DriverInstanceId.ShouldBe("drv-1");
+        item.E.State.ShouldBe(DriverState.Degraded.ToString());
+        item.E.LastError.ShouldBe("some error");
+        item.E.ErrorCount5Min.ShouldBe(4);
+    }
+
+    // --- script-logs seam (DpsScriptLogPublisher) → TelemetryItem.Script -------------------------------
+
+    /// <summary>The DPS script-log publisher fans the SAME <see cref="ScriptLogEntry"/> instance it publishes
+    /// to the <c>script-logs</c> topic into the hub as a <see cref="TelemetryItem.Script"/>.</summary>
+    [Fact]
+    public void Script_log_publish_emits_one_Script_item_with_same_payload()
+    {
+        var hub = new CapturingHub();
+        var publisher = new DpsScriptLogPublisher(() => Sys, hub);
+
+        var entry = new ScriptLogEntry(
+            ScriptId: "S1",
+            Level: "Information",
+            Message: "hello",
+            TimestampUtc: Ts,
+            VirtualTagId: "V1",
+            AlarmId: null,
+            EquipmentId: "EQ1");
+
+        publisher.Publish(entry);
+
+        hub.Items.Count.ShouldBe(1);
+        var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Script>();
+        item.E.ShouldBeSameAs(entry); // the exact instance published to DPS
+    }
+
+    /// <summary>A null hub is rejected at construction (fail-fast, DI provides a real hub on driver nodes).</summary>
+    [Fact]
+    public void Script_log_publisher_rejects_null_hub()
+    {
+        Should.Throw<ArgumentNullException>(() => new DpsScriptLogPublisher(() => Sys, null!));
+    }
+
+    // --- alerts seam (DriverHostActor native alarm) → TelemetryItem.Alarm ------------------------------
+
+    /// <summary>A Primary-gated native-alarm transition fans the SAME <see cref="AlarmTransitionEvent"/> it
+    /// publishes to the <c>alerts</c> topic into the hub as a <see cref="TelemetryItem.Alarm"/>. Proves the
+    /// hub was threaded through <see cref="DriverHostActor"/>'s Props into the emit seam.</summary>
+    [Fact]
+    public void Native_alarm_publish_emits_one_Alarm_item_with_matching_payload()
+    {
+        var db = NewInMemoryDbFactory();
+        var deploymentId = SeedV3AlarmDeployment(db, RevA);
+
+        var hub = new CapturingHub();
+        var (actor, publish) = SpawnHostAndApply(db, deploymentId, hub);
+
+        actor.Tell(new DriverInstanceActor.AttributeAlarmPublished("drv-1", new AlarmEventArgs(
+            new StubAlarmHandle(),
+            SourceNodeId: "Temp",
+            ConditionId: AlarmRawPath,
+            AlarmType: "OffNormalAlarm",
+            Message: "temperature high",
+            Severity: AlarmSeverity.High,
+            SourceTimestampUtc: Ts,
+            Kind: AlarmTransitionKind.Raise)));
+
+        // The OPC UA condition update confirms the transition was processed (ordered before the alerts fan-out).
+        publish.ExpectMsg<OpcUaPublishActor.AlarmStateUpdate>(TimeSpan.FromSeconds(5));
+
+        // Exactly one Alarm item, carrying the same transition the alerts topic received.
+        AwaitAssert(() => hub.Items.Count.ShouldBe(1), TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(100));
+        var item = hub.Items[0].ShouldBeOfType<TelemetryItem.Alarm>();
+        item.E.AlarmId.ShouldBe(AlarmRawPath);
+        item.E.AlarmName.ShouldBe("temp_hi");
+        item.E.TransitionKind.ShouldBe("Activated");
+        item.E.Message.ShouldBe("temperature high");
+    }
+
+    /// <summary>Spawns the host wired with the capturing hub, dispatches the deployment, and waits for the
+    /// Applied ack + the RebuildAddressSpace so the alarm map is built before the test raises an alarm.</summary>
+    private (IActorRef Actor, Akka.TestKit.TestProbe Publish) SpawnHostAndApply(
+        IDbContextFactory<OtOpcUaConfigDbContext> db, DeploymentId deploymentId, ITelemetryLocalHub hub)
+    {
+        var coordinator = CreateTestProbe();
+        var publish = CreateTestProbe();
+        var mux = CreateTestProbe();
+        var vtHost = CreateTestProbe();
+
+        var actor = Sys.ActorOf(DriverHostActor.Props(
+            db, TestNode, coordinator.Ref,
+            localRoles: new HashSet<string> { "driver" },
+            dependencyMux: mux.Ref,
+            opcUaPublishActor: publish.Ref,
+            virtualTagEvaluator: NullVirtualTagEvaluator.Instance,
+            virtualTagHostOverride: vtHost.Ref,
+            telemetryHub: hub));
+
+        actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
+        coordinator.ExpectMsg<ApplyAck>(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied);
+        publish.ExpectMsg<OpcUaPublishActor.RebuildAddressSpace>(TimeSpan.FromSeconds(5));
+
+        return (actor, publish);
+    }
+
+    /// <summary>Seeds a Sealed v3 deployment with a single Boolean raw tag carrying an <c>alarm</c> object so
+    /// the composer projects it as a Part 9 condition at its RawPath (mirrors DriverHostActorNativeAlarmTests).</summary>
+    private static DeploymentId SeedV3AlarmDeployment(IDbContextFactory<OtOpcUaConfigDbContext> db, RevisionHash rev)
+    {
+        var artifact = JsonSerializer.SerializeToUtf8Bytes(new
+        {
+            RawFolders = new[] { new { RawFolderId = "rf-plant", ParentRawFolderId = (string?)null, Name = "Plant", ClusterId = "c1" } },
+            DriverInstances = new[]
+            {
+                new { DriverInstanceId = "drv-1", RawFolderId = "rf-plant", Name = "Modbus", DriverType = "Modbus", DriverConfig = "{}", ClusterId = "c1", Enabled = false },
+            },
+            Devices = new[]
+            {
+                new { DeviceId = "drv-1:dev1", DriverInstanceId = "drv-1", Name = "dev1", DeviceConfig = "{}" },
+            },
+            TagGroups = Array.Empty<object>(),
+            Tags = new[]
+            {
+                new
+                {
+                    TagId = "tag-0",
+                    DeviceId = "drv-1:dev1",
+                    TagGroupId = (string?)null,
+                    Name = "temp_hi",
+                    DataType = "Boolean",
+                    AccessLevel = 0,
+                    TagConfig = JsonSerializer.Serialize(new { alarm = new { alarmType = "OffNormalAlarm", severity = 700 } }),
+                },
+            },
+        });
+
+        var id = DeploymentId.NewId();
+        using var ctx = db.CreateDbContext();
+        ctx.Deployments.Add(new Deployment
+        {
+            DeploymentId = id.Value,
+            RevisionHash = rev.Value,
+            Status = DeploymentStatus.Sealed,
+            CreatedBy = "test",
+            SealedAtUtc = DateTime.UtcNow,
+            ArtifactBlob = artifact,
+        });
+        ctx.SaveChanges();
+        return id;
+    }
+
+    /// <summary>Minimal alarm-subscription handle for building <see cref="AlarmEventArgs"/>.</summary>
+    private sealed class StubAlarmHandle : IAlarmSubscriptionHandle
+    {
+        public string DiagnosticId => "stub-alarm-sub";
+    }
+
+    /// <summary>Fake hub that records every <see cref="ITelemetryLocalHub.Emit"/>. <see cref="Subscribe"/> is
+    /// unused by the producer seams under test.</summary>
+    private sealed class CapturingHub : ITelemetryLocalHub
+    {
+        public List<TelemetryItem> Items { get; } = new();
+
+        public void Emit(TelemetryItem item) => Items.Add(item);
+
+        public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException();
+    }
+}