diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs index 76d1265d..f7eaac28 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; using ZB.MOM.WW.OtOpcUa.Core.Resilience; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; namespace ZB.MOM.WW.OtOpcUa.Host.Drivers; @@ -23,22 +24,26 @@ public sealed class DriverResilienceStatusPublisherService : BackgroundService private readonly DriverResilienceStatusTracker _tracker; private readonly Func _actorSystemAccessor; private readonly ILogger _logger; + private readonly ITelemetryLocalHub _hub; private readonly TimeSpan _interval; /// Initializes a new instance of . /// The process-singleton resilience-status tracker to read each tick. /// Lazy accessor for the Akka whose DPS mediator publishes the snapshots. /// Logger for publish diagnostics. + /// The node-local live-telemetry hub each published snapshot is also emitted into (Phase 5). /// Publish cadence; defaults to 5 s when null. public DriverResilienceStatusPublisherService( DriverResilienceStatusTracker tracker, Func actorSystemAccessor, ILogger logger, + ITelemetryLocalHub hub, TimeSpan? interval = null) { _tracker = tracker; _actorSystemAccessor = actorSystemAccessor; _logger = logger; + _hub = hub; _interval = interval ?? DefaultInterval; } @@ -77,7 +82,12 @@ public sealed class DriverResilienceStatusPublisherService : BackgroundService var mediator = DistributedPubSub.Get(_actorSystemAccessor()).Mediator; foreach (var message in messages) + { mediator.Tell(new Publish(DriverResilienceStatusChanged.TopicName, message)); + // 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 — a strictly additive tap. + _hub.Emit(new TelemetryItem.Resilience(message)); + } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 3204646c..9d020640 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -34,6 +34,7 @@ using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Historian; using ZB.MOM.WW.OtOpcUa.Runtime.Scripting; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security; using ZB.MOM.WW.OtOpcUa.Runtime; using ZB.MOM.WW.Auth.Abstractions.Roles; @@ -310,7 +311,9 @@ if (hasDriver) ? parsedLevel : LogEventLevel.Information; builder.Services.AddSingleton(sp => - new DpsScriptLogPublisher(() => sp.GetRequiredService())); + new DpsScriptLogPublisher( + () => sp.GetRequiredService(), + sp.GetRequiredService())); builder.Services.AddSingleton(sp => new ScriptRootLogger( ScriptRootLoggerFactory.Build( sp.GetRequiredService(), scriptLogFilePath, scriptLogTopicMinLevel, Serilog.Log.Logger))); 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 4f8e7a59..6f93d39f 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/AkkaDriverHealthPublisher.cs @@ -2,6 +2,7 @@ using Akka.Actor; using Akka.Cluster.Tools.PublishSubscribe; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; namespace ZB.MOM.WW.OtOpcUa.Runtime.Drivers; @@ -17,10 +18,16 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher public const string TopicName = DriverHealthChanged.TopicName; private readonly ActorSystem _system; + private readonly ITelemetryLocalHub _hub; /// Initializes a new instance of . /// The Akka actor system used to resolve the DPS mediator. - public AkkaDriverHealthPublisher(ActorSystem system) => _system = system; + /// The node-local live-telemetry hub each snapshot is also emitted into (Phase 5). + public AkkaDriverHealthPublisher(ActorSystem system, ITelemetryLocalHub hub) + { + _system = system; + _hub = hub; + } /// public void Publish(string clusterId, string driverInstanceId, DriverHealth health, int errorCount5Min) @@ -34,5 +41,8 @@ public sealed class AkkaDriverHealthPublisher : IDriverHealthPublisher errorCount5Min, DateTime.UtcNow); 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. + _hub.Emit(new TelemetryItem.Health(msg)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index bc48abb2..c1025d56 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -26,6 +26,7 @@ using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.DeploymentCache; using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; using ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; using CommonsNodeId = ZB.MOM.WW.OtOpcUa.Commons.Types.NodeId; @@ -118,6 +119,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// private readonly IAlarmStateStore? _alarmStateStore; + /// Optional node-local live-telemetry hub (Phase 5): each Primary-gated native-alarm + /// alerts transition is also emitted here, and it is threaded into the spawned VirtualTag + + /// ScriptedAlarm hosts. Null (admin-only nodes / tests) skips the emit and passes null down. + private readonly ITelemetryLocalHub? _telemetryHub; + private readonly CommonsNodeId _localNode; private readonly IActorRef? _ackRouter; private readonly IDriverFactory _driverFactory; @@ -397,6 +403,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// driver-role node this is the replicated LocalDb store, so condition state persists with no /// ConfigDb; null skips spawning the alarm host (the ConfigDb-backed EF fallback was retired in /// Phase 4 Task 9). Defaults to null. + /// Per-cluster mesh Phase 5: optional node-local live-telemetry hub each + /// Primary-gated native-alarm transition is also emitted into, and which is threaded into the spawned + /// VirtualTag + ScriptedAlarm hosts. Defaults to null (admin-only nodes / tests skip the emit). /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory? dbFactory, @@ -420,7 +429,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers string? replicationPeerHost = null, bool fetchAndCacheMode = false, IDeploymentArtifactFetcher? artifactFetcher = null, - IAlarmStateStore? alarmStateStore = null) => + IAlarmStateStore? alarmStateStore = null, + ITelemetryLocalHub? telemetryHub = null) => // WARNING: this forwarding list is POSITIONAL, and Props.Create compiles it into an // expression tree. Six IActorRef? parameters and several interface-typed ones mean a // mis-ordered argument is usually type-compatible and therefore compiles clean, then binds @@ -431,7 +441,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers healthPublisher, virtualTagEvaluator, historyWriter, virtualTagHostOverride, loggerFactory, scriptRootLogger, scriptedAlarmHostOverride, invokerFactory, driverMemberCountProvider, deploymentArtifactCache, redundancyRoleView, replicationPeerHost, fetchAndCacheMode, artifactFetcher, - alarmStateStore)); + alarmStateStore, telemetryHub)); /// Initializes a new DriverHostActor with the specified dependencies. /// Database context factory for configuration database access. @@ -469,6 +479,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers /// Per-cluster mesh Phase 4: scripted-alarm condition-state store. On a /// driver-role node this is the replicated LocalDb store; null skips spawning the alarm host (the /// ConfigDb-backed EF fallback was retired in Phase 4 Task 9). + /// Per-cluster mesh Phase 5: optional node-local live-telemetry hub each + /// Primary-gated native-alarm transition is also emitted into, and which is threaded into the spawned + /// VirtualTag + ScriptedAlarm hosts. Null (admin-only nodes / tests) skips the emit and passes null down. public DriverHostActor( IDbContextFactory? dbFactory, CommonsNodeId localNode, @@ -491,8 +504,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers string? replicationPeerHost = null, bool fetchAndCacheMode = false, IDeploymentArtifactFetcher? artifactFetcher = null, - IAlarmStateStore? alarmStateStore = null) + IAlarmStateStore? alarmStateStore = null, + ITelemetryLocalHub? telemetryHub = null) { + _telemetryHub = telemetryHub; _deploymentArtifactCache = deploymentArtifactCache; _redundancyRoleView = redundancyRoleView; _replicationPeerHost = replicationPeerHost; @@ -577,7 +592,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers } _virtualTagHost = Context.ActorOf( - VirtualTagHostActor.Props(_opcUaPublishActor, _dependencyMux, _virtualTagEvaluator, _historyWriter), + VirtualTagHostActor.Props(_opcUaPublishActor, _dependencyMux, _virtualTagEvaluator, _historyWriter, _telemetryHub), "virtual-tag-host"); } @@ -630,7 +645,8 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var engine = new ScriptedAlarmEngine( upstream, store, new ScriptLoggerFactory(_scriptRootLogger.Logger), _scriptRootLogger.Logger); _scriptedAlarmHost = Context.ActorOf( - ScriptedAlarmHostActor.Props(_opcUaPublishActor, _dependencyMux, upstream, engine, _localNode), + ScriptedAlarmHostActor.Props( + _opcUaPublishActor, _dependencyMux, upstream, engine, _localNode, telemetryHub: _telemetryHub), "scripted-alarm-host"); } @@ -1402,7 +1418,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers var meta = _alarmMetaByNodeId.TryGetValue(nodeId, out var m) ? m : (EquipmentId: nodeId, Name: nodeId, AlarmType: "AlarmCondition", HistorizeToAveva: (bool?)null, ReferencingEquipmentPaths: (IReadOnlyList)Array.Empty()); - _mediator.Tell(new Publish(ScriptedAlarmHostActor.AlertsTopic, new AlarmTransitionEvent( + var alertEvent = new AlarmTransitionEvent( AlarmId: nodeId, EquipmentPath: meta.EquipmentId, AlarmName: meta.Name, @@ -1424,7 +1440,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers HistorizeToAveva: meta.HistorizeToAveva, // WP4: the Area/Line/Equipment UNS paths that reference this raw condition — the /alerts row // renders them as the equipment list (empty for a raw condition no equipment references yet). - ReferencingEquipmentPaths: meta.ReferencingEquipmentPaths))); + ReferencingEquipmentPaths: meta.ReferencingEquipmentPaths); + _mediator.Tell(new Publish(ScriptedAlarmHostActor.AlertsTopic, alertEvent)); + // Phase 5: fan the same transition into the node-local live-telemetry hub (no-op until a gRPC + // client subscribes). The DPS publish above is unchanged — a strictly additive tap, riding the + // same Primary gate (only reached when serviceAlertsAsPrimary). + _telemetryHub?.Emit(new TelemetryItem.Alarm(alertEvent)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs index 7b538c66..8f7c2074 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ScriptedAlarms/ScriptedAlarmHostActor.cs @@ -11,6 +11,7 @@ using ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; namespace ZB.MOM.WW.OtOpcUa.Runtime.ScriptedAlarms; @@ -116,6 +117,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// alerts-emit gate reads while the role is unknown. Null ⇒ read the live cluster state. private readonly Func? _driverMemberCountProvider; + /// Optional node-local live-telemetry hub (Phase 5): each cluster-wide alerts transition + /// is also emitted here for the gRPC streaming service. Null (tests / no hub wired) skips the emit. + private readonly ITelemetryLocalHub? _telemetryHub; + /// Monotonic load generation, bumped on every . The continuation that /// pipes back an captures the generation it was started under; a stale /// completion (an earlier generation arriving after a newer apply) is discarded in @@ -138,6 +143,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// The local cluster node id, used to read this node's /// from the redundancy-state topic so only the Primary publishes the cluster-wide alerts /// transition. Null (the default) leaves the role unknown ⇒ default-emit (single-node deploys + tests). + /// Test seam (archreview 03/S4): overrides the count of Up + /// driver-role cluster members the alerts-emit gate reads while the role is unknown. + /// Optional node-local live-telemetry hub (Phase 5) each cluster-wide + /// alerts transition is also emitted into; null (tests) skips the emit. /// The used to instantiate the actor. public static Props Props( IActorRef publishActor, @@ -145,8 +154,9 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor DependencyMuxTagUpstreamSource upstream, ScriptedAlarmEngine engine, NodeId? localNode = null, - Func? driverMemberCountProvider = null) => - Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider)); + Func? driverMemberCountProvider = null, + ITelemetryLocalHub? telemetryHub = null) => + Akka.Actor.Props.Create(() => new ScriptedAlarmHostActor(publishActor, mux, upstream, engine, localNode, driverMemberCountProvider, telemetryHub)); /// Initializes a new instance of the class. /// The OPC UA publish actor emissions are bridged to. @@ -159,13 +169,16 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor /// Test seam (archreview 03/S4): overrides the count of Up /// driver-role cluster members the alerts-emit gate reads while the role is unknown. Null reads /// the live cluster state (0 on a non-cluster ActorRefProvider ⇒ single-node default-emit). + /// Optional node-local live-telemetry hub (Phase 5) each cluster-wide + /// alerts transition is also emitted into; null (tests) skips the emit. public ScriptedAlarmHostActor( IActorRef publishActor, IActorRef? mux, DependencyMuxTagUpstreamSource upstream, ScriptedAlarmEngine engine, NodeId? localNode = null, - Func? driverMemberCountProvider = null) + Func? driverMemberCountProvider = null, + ITelemetryLocalHub? telemetryHub = null) { ArgumentNullException.ThrowIfNull(publishActor); ArgumentNullException.ThrowIfNull(upstream); @@ -176,6 +189,7 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor _engine = engine; _localNode = localNode; _driverMemberCountProvider = driverMemberCountProvider; + _telemetryHub = telemetryHub; // OnEvent fires on the engine's worker thread. NEVER touch Context / actor state here — // marshal onto the actor thread via the thread-safe Self.Tell. Keep the handler in a field @@ -377,6 +391,10 @@ public sealed class ScriptedAlarmHostActor : ReceiveActor } _mediator.Tell(new Publish(AlertsTopic, evt)); + // Phase 5: fan the same transition into the node-local live-telemetry hub (no-op until a gRPC + // client subscribes). The DPS publish above is unchanged — a strictly additive tap, and it rides + // the same Primary gate as the publish (only the Primary reaches this line). + _telemetryHub?.Emit(new TelemetryItem.Alarm(evt)); } /// Count of Up cluster members carrying the driver role, for the alerts-emit gate. Uses diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Scripting/DpsScriptLogPublisher.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Scripting/DpsScriptLogPublisher.cs index fa77729c..e5bba132 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Scripting/DpsScriptLogPublisher.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Scripting/DpsScriptLogPublisher.cs @@ -2,6 +2,7 @@ using Akka.Actor; using Akka.Cluster.Tools.PublishSubscribe; using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; using ZB.MOM.WW.OtOpcUa.Core.Scripting; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; using ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; namespace ZB.MOM.WW.OtOpcUa.Runtime.Scripting; @@ -27,15 +28,20 @@ namespace ZB.MOM.WW.OtOpcUa.Runtime.Scripting; public sealed class DpsScriptLogPublisher : IScriptLogPublisher { private readonly Func _system; + private readonly ITelemetryLocalHub _hub; /// Initializes a new instance of the class. /// /// Lazy accessor for the running . Invoked on each /// so registration does not depend on Akka having started yet. /// - /// Thrown when is null. - public DpsScriptLogPublisher(Func system) => + /// The node-local live-telemetry hub each script-log entry is also emitted into (Phase 5). + /// Thrown when or is null. + public DpsScriptLogPublisher(Func system, ITelemetryLocalHub hub) + { _system = system ?? throw new ArgumentNullException(nameof(system)); + _hub = hub ?? throw new ArgumentNullException(nameof(hub)); + } /// public void Publish(ScriptLogEntry entry) @@ -44,6 +50,9 @@ public sealed class DpsScriptLogPublisher : IScriptLogPublisher { var mediator = DistributedPubSub.Get(_system()).Mediator; mediator.Tell(new Publish(VirtualTagActor.ScriptLogsTopic, entry)); + // Phase 5: fan the same entry 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. + _hub.Emit(new TelemetryItem.Script(entry)); } catch (Exception ex) { diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs index 34fa0946..062ff3ad 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs @@ -495,7 +495,11 @@ public static class ServiceCollectionExtensions replicationPeerHost: replicationPeerHost, fetchAndCacheMode: fetchAndCacheMode, artifactFetcher: artifactFetcher, - alarmStateStore: alarmStateStore), + alarmStateStore: alarmStateStore, + // Phase 5: the node-local live-telemetry hub (registered in AddOtOpcUaRuntime). The host + // emits its Primary-gated native-alarm transitions into it and threads it into the + // VirtualTag + ScriptedAlarm hosts it spawns. + telemetryHub: resolver.GetService()), DriverHostActorName); registry.Register(driverHost); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs index d733a264..fe13c413 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagActor.cs @@ -6,6 +6,7 @@ using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; using ZB.MOM.WW.OtOpcUa.Commons.Observability; using ZB.MOM.WW.OtOpcUa.Commons.OpcUa; using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; @@ -61,6 +62,7 @@ public sealed class VirtualTagActor : ReceiveActor private readonly Func? _publisherFactory; private readonly IReadOnlyList _dependencyRefs; private readonly IActorRef? _mux; + private readonly ITelemetryLocalHub? _telemetryHub; private readonly ILoggingAdapter _log = Context.GetLogger(); private readonly Dictionary _dependencies = new(StringComparer.Ordinal); @@ -78,6 +80,8 @@ public sealed class VirtualTagActor : ReceiveActor /// Optional factory for creating DPS publishers. /// Optional list of dependency tag references; defaults to empty. /// Optional reference to a dependency multiplexer actor. + /// Optional node-local live-telemetry hub each script-log entry is also + /// emitted into (Phase 5); null (tests) simply skips the emit. /// A configured for creating the actor. public static Props Props( string virtualTagId, @@ -86,14 +90,16 @@ public sealed class VirtualTagActor : ReceiveActor string? scriptId = null, Func? publisherFactory = null, IReadOnlyList? dependencyRefs = null, - IActorRef? mux = null) => + IActorRef? mux = null, + ITelemetryLocalHub? telemetryHub = null) => Akka.Actor.Props.Create(() => new VirtualTagActor( virtualTagId, expression, evaluator ?? NullVirtualTagEvaluator.Instance, scriptId ?? virtualTagId, publisherFactory, dependencyRefs ?? Array.Empty(), - mux)); + mux, + telemetryHub)); /// Initializes a virtual tag actor with the given configuration and dependencies. /// Unique identifier for the virtual tag. @@ -103,6 +109,8 @@ public sealed class VirtualTagActor : ReceiveActor /// Optional factory for creating DPS publishers. /// List of dependency tag references that this tag depends on. /// Optional reference to a dependency multiplexer actor. + /// Optional node-local live-telemetry hub each script-log entry is also + /// emitted into (Phase 5); null (tests) simply skips the emit. public VirtualTagActor( string virtualTagId, string expression, @@ -110,7 +118,8 @@ public sealed class VirtualTagActor : ReceiveActor string scriptId, Func? publisherFactory, IReadOnlyList dependencyRefs, - IActorRef? mux) + IActorRef? mux, + ITelemetryLocalHub? telemetryHub = null) { _virtualTagId = virtualTagId; _scriptId = scriptId; @@ -119,6 +128,7 @@ public sealed class VirtualTagActor : ReceiveActor _publisherFactory = publisherFactory; _dependencyRefs = dependencyRefs; _mux = mux; + _telemetryHub = telemetryHub; Receive(OnDependencyChanged); Receive(_ => OnReassertValue()); @@ -259,6 +269,9 @@ public sealed class VirtualTagActor : ReceiveActor } DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(ScriptLogsTopic, entry)); + // Phase 5: fan the same entry into the node-local live-telemetry hub (no-op until a gRPC client + // subscribes). The DPS publish above is unchanged — a strictly additive tap. + _telemetryHub?.Emit(new TelemetryItem.Script(entry)); } } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs index 50592064..7005dfdc 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/VirtualTags/VirtualTagHostActor.cs @@ -6,6 +6,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.VirtualTags; using ZB.MOM.WW.OtOpcUa.OpcUaServer; using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Telemetry; namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags; @@ -39,6 +40,7 @@ public sealed class VirtualTagHostActor : ReceiveActor // Sink for historized VirtualTag results (plans with Historize=true). NullHistoryWriter when no // durable historian is wired, so OnResult always has a non-null target. private readonly IHistoryWriter _history; + private readonly ITelemetryLocalHub? _telemetryHub; private readonly ILoggingAdapter _log = Context.GetLogger(); // vtagId -> spawned child VirtualTagActor. @@ -59,18 +61,22 @@ public sealed class VirtualTagHostActor : ReceiveActor /// Sink for results whose plan has Historize=true. Null ⇒ /// (no durable historian wired), so existing call sites /// compile unchanged and never historize. + /// Optional node-local live-telemetry hub passed to each spawned child so + /// its script-log entries also emit to the Phase-5 hub; null (tests) skips the emit. /// The used to spawn a . public static Props Props(IActorRef publishActor, IActorRef? mux, IVirtualTagEvaluator evaluator, - IHistoryWriter? historyWriter = null) => - Akka.Actor.Props.Create(() => new VirtualTagHostActor(publishActor, mux, evaluator, historyWriter)); + IHistoryWriter? historyWriter = null, ITelemetryLocalHub? telemetryHub = null) => + Akka.Actor.Props.Create(() => new VirtualTagHostActor(publishActor, mux, evaluator, historyWriter, telemetryHub)); /// Initializes a new instance of the class. /// The OPC UA publish actor results are bridged to. /// Optional dependency multiplexer passed to each spawned child. /// The evaluator each child uses to compute its expression. /// Sink for historized results; null ⇒ . + /// Optional node-local live-telemetry hub passed to each spawned child so + /// its script-log entries also emit to the Phase-5 hub; null (tests) skips the emit. public VirtualTagHostActor(IActorRef publishActor, IActorRef? mux, IVirtualTagEvaluator evaluator, - IHistoryWriter? historyWriter = null) + IHistoryWriter? historyWriter = null, ITelemetryLocalHub? telemetryHub = null) { ArgumentNullException.ThrowIfNull(publishActor); ArgumentNullException.ThrowIfNull(evaluator); @@ -78,6 +84,7 @@ public sealed class VirtualTagHostActor : ReceiveActor _mux = mux; _evaluator = evaluator; _history = historyWriter ?? NullHistoryWriter.Instance; + _telemetryHub = telemetryHub; Receive(OnApply); Receive(OnResult); @@ -166,7 +173,8 @@ public sealed class VirtualTagHostActor : ReceiveActor scriptId: p.VirtualTagId, publisherFactory: null, dependencyRefs: p.DependencyRefs, - mux: _mux)); + mux: _mux, + telemetryHub: _telemetryHub)); Context.Watch(child); _children[p.VirtualTagId] = child; newlySpawned.Add(p.VirtualTagId); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs index 246e14eb..180eea82 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs @@ -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); } + + /// Per-cluster mesh Phase 5 — each snapshot the publish tick sends to the + /// driver-resilience-status DPS topic is also emitted into the node-local + /// as a . Runs the background + /// service against a single-node cluster (so DistributedPubSub resolves) for a few ticks and asserts the + /// hub captured the tracked pair. + [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.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(); + item.E.DriverInstanceId.ShouldBe("drv-1"); + item.E.HostName.ShouldBe("host-a"); + item.E.CurrentInFlight.ShouldBe(1); + } + finally + { + await sys.Terminate(); + } + } + + /// Fake hub recording every emit; is unused here. + private sealed class CapturingHub : ITelemetryLocalHub + { + public List Items { get; } = new(); + + public void Emit(TelemetryItem item) + { + lock (Items) Items.Add(item); + } + + public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException(); + } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs index 6354eb50..0157cb9b 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Scripting/DpsScriptLogPublisherTests.cs @@ -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(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(() => new DpsScriptLogPublisher(null!)); + Should.Throw(() => new DpsScriptLogPublisher(null!, new NoopHub())); + } + + /// No-op telemetry hub for the publisher tests (the hub tap is asserted in + /// PublishSeamEmitsToHubTests; here it must simply not interfere). + private sealed class NoopHub : ITelemetryLocalHub + { + public void Emit(TelemetryItem item) { } + + public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs new file mode 100644 index 00000000..6b1d821c --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Telemetry/PublishSeamEmitsToHubTests.cs @@ -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; + +/// +/// Per-cluster mesh Phase 5 — the four telemetry publish seams each additionally fan their DPS payload +/// into the node-local . Each test drives a producer with a capturing +/// fake hub and asserts exactly one 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. +/// +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 -------------------------- + + /// The driver-health publisher fans the SAME it publishes to + /// the driver-health DPS topic into the hub as a . + [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(); + 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 ------------------------------- + + /// The DPS script-log publisher fans the SAME instance it publishes + /// to the script-logs topic into the hub as a . + [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(); + item.E.ShouldBeSameAs(entry); // the exact instance published to DPS + } + + /// A null hub is rejected at construction (fail-fast, DI provides a real hub on driver nodes). + [Fact] + public void Script_log_publisher_rejects_null_hub() + { + Should.Throw(() => new DpsScriptLogPublisher(() => Sys, null!)); + } + + // --- alerts seam (DriverHostActor native alarm) → TelemetryItem.Alarm ------------------------------ + + /// A Primary-gated native-alarm transition fans the SAME it + /// publishes to the alerts topic into the hub as a . Proves the + /// hub was threaded through 's Props into the emit seam. + [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(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(); + item.E.AlarmId.ShouldBe(AlarmRawPath); + item.E.AlarmName.ShouldBe("temp_hi"); + item.E.TransitionKind.ShouldBe("Activated"); + item.E.Message.ShouldBe("temperature high"); + } + + /// 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. + private (IActorRef Actor, Akka.TestKit.TestProbe Publish) SpawnHostAndApply( + IDbContextFactory 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 { "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(TimeSpan.FromSeconds(5)).Outcome.ShouldBe(ApplyAckOutcome.Applied); + publish.ExpectMsg(TimeSpan.FromSeconds(5)); + + return (actor, publish); + } + + /// Seeds a Sealed v3 deployment with a single Boolean raw tag carrying an alarm object so + /// the composer projects it as a Part 9 condition at its RawPath (mirrors DriverHostActorNativeAlarmTests). + private static DeploymentId SeedV3AlarmDeployment(IDbContextFactory 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(), + 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; + } + + /// Minimal alarm-subscription handle for building . + private sealed class StubAlarmHandle : IAlarmSubscriptionHandle + { + public string DiagnosticId => "stub-alarm-sub"; + } + + /// Fake hub that records every . is + /// unused by the producer seams under test. + private sealed class CapturingHub : ITelemetryLocalHub + { + public List Items { get; } = new(); + + public void Emit(TelemetryItem item) => Items.Add(item); + + public ITelemetrySubscription Subscribe(int boundedCapacity) => throw new NotSupportedException(); + } +}