diff --git a/archreview/plans/R2-10-resilience-observability-plan.md.tasks.json b/archreview/plans/R2-10-resilience-observability-plan.md.tasks.json index 39fc2774..882d5331 100644 --- a/archreview/plans/R2-10-resilience-observability-plan.md.tasks.json +++ b/archreview/plans/R2-10-resilience-observability-plan.md.tasks.json @@ -6,7 +6,7 @@ { "id": "T2", "subject": "Builder: OnClosed records breaker-close on the tracker (TDD in DriverResiliencePipelineBuilderTests)", "status": "completed", "blockedBy": ["T1"] }, { "id": "T3", "subject": "Invoker: successful Execute* resets ConsecutiveFailures (TDD in CapabilityInvokerTests)", "status": "completed", "blockedBy": ["T1"] }, { "id": "T4", "subject": "Commons: DriverResilienceStatusChanged record + driver-resilience-status TopicName", "status": "completed", "blockedBy": [] }, - { "id": "T5", "subject": "Host: DriverResilienceStatusPublisherService (BuildMessages mapping + 5s DPS timer shell; TDD mapping in Host.IntegrationTests)", "status": "pending", "blockedBy": ["T1", "T4"] }, + { "id": "T5", "subject": "Host: DriverResilienceStatusPublisherService (BuildMessages mapping + 5s DPS timer shell; TDD mapping in Host.IntegrationTests)", "status": "completed", "blockedBy": ["T1", "T4"] }, { "id": "T6", "subject": "Host DI: register the publisher in AddOtOpcUaDriverFactories + ResilienceStatusReaderRegistrationTests wiring guard (tracker HAS a production reader)", "status": "pending", "blockedBy": ["T5"] }, { "id": "T7", "subject": "AdminUI: IDriverResilienceStatusStore + InMemory impl + AddOtOpcUaDriverStatusServices registration (TDD mirroring DriverStatusSnapshotStoreTests)", "status": "pending", "blockedBy": ["T4"] }, { "id": "T8", "subject": "AdminUI: DriverResilienceStatusBridge DPS actor + spawn in WithOtOpcUaSignalRBridges", "status": "pending", "blockedBy": ["T7"] }, diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs new file mode 100644 index 00000000..76d1265d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverResilienceStatusPublisherService.cs @@ -0,0 +1,93 @@ +using Akka.Actor; +using Akka.Cluster.Tools.PublishSubscribe; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Drivers; +using ZB.MOM.WW.OtOpcUa.Core.Resilience; + +namespace ZB.MOM.WW.OtOpcUa.Host.Drivers; + +/// +/// The operator-facing reader for . Runs on driver-role +/// nodes (registered next to the tracker it reads) and every publishes the +/// full tracker snapshot — one per (instance, host) +/// — to the driver-resilience-status DistributedPubSub topic. The AdminUI +/// DriverResilienceStatusBridge subscribes and feeds the resilience section of the driver +/// status panel. Mirrors the driver-health flow; publishing the full set every tick self-heals +/// late-subscribing admin nodes and gives the panel staleness detection for free. +/// +public sealed class DriverResilienceStatusPublisherService : BackgroundService +{ + private static readonly TimeSpan DefaultInterval = TimeSpan.FromSeconds(5); + + private readonly DriverResilienceStatusTracker _tracker; + private readonly Func _actorSystemAccessor; + private readonly ILogger _logger; + 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. + /// Publish cadence; defaults to 5 s when null. + public DriverResilienceStatusPublisherService( + DriverResilienceStatusTracker tracker, + Func actorSystemAccessor, + ILogger logger, + TimeSpan? interval = null) + { + _tracker = tracker; + _actorSystemAccessor = actorSystemAccessor; + _logger = logger; + _interval = interval ?? DefaultInterval; + } + + /// + /// Map the tracker's current snapshot to one wire message per (instance, host) pair. + /// Pure + unit-testable; the timer/publish shell in calls this each tick. + /// + /// The tracker to read. + /// The publish timestamp stamped on every emitted message. + /// One per tracked pair (empty when the tracker is empty). + public static IReadOnlyList BuildMessages( + DriverResilienceStatusTracker tracker, DateTime publishedUtc) => + tracker.Snapshot() + .Select(entry => new DriverResilienceStatusChanged( + entry.DriverInstanceId, + entry.HostName, + entry.Snapshot.IsBreakerOpen, + entry.Snapshot.ConsecutiveFailures, + entry.Snapshot.CurrentInFlight, + entry.Snapshot.LastBreakerOpenUtc, + entry.Snapshot.LastSampledUtc, + publishedUtc)) + .ToList(); + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(_interval); + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) + { + try + { + var messages = BuildMessages(_tracker, DateTime.UtcNow); + if (messages.Count == 0) + continue; + + var mediator = DistributedPubSub.Get(_actorSystemAccessor()).Mediator; + foreach (var message in messages) + mediator.Tell(new Publish(DriverResilienceStatusChanged.TopicName, message)); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + // A transient DPS / actor-system hiccup must never kill the publisher — the next tick retries. + _logger.LogWarning(ex, "DriverResilienceStatusPublisherService: publish tick failed; retrying next interval"); + } + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs new file mode 100644 index 00000000..246e14eb --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.IntegrationTests/DriverResilienceStatusPublisherServiceTests.cs @@ -0,0 +1,56 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Resilience; +using ZB.MOM.WW.OtOpcUa.Host.Drivers; + +namespace ZB.MOM.WW.OtOpcUa.Host.IntegrationTests; + +/// +/// Pure-DI (no fixture) coverage of the publisher's snapshot→message mapping — the only unit-testable +/// seam of . The DPS Publish shell is +/// verified by the R2-10 live gate; this asserts the mapping the shell publishes. +/// +public sealed class DriverResilienceStatusPublisherServiceTests +{ + private static readonly DateTime Published = new(2026, 7, 13, 9, 0, 0, DateTimeKind.Utc); + private static readonly DateTime Sampled = new(2026, 7, 13, 8, 59, 0, DateTimeKind.Utc); + + [Fact] + public void BuildMessages_EmptyTracker_ReturnsEmpty() + { + var tracker = new DriverResilienceStatusTracker(); + + var messages = DriverResilienceStatusPublisherService.BuildMessages(tracker, Published); + + messages.ShouldBeEmpty(); + } + + [Fact] + public void BuildMessages_MapsTrackerSnapshot_OneMessagePerInstanceHost() + { + var tracker = new DriverResilienceStatusTracker(); + // (drv-1, host-a): an open breaker with a standing failure count. + tracker.RecordFailure("drv-1", "host-a", Sampled); + tracker.RecordFailure("drv-1", "host-a", Sampled); + tracker.RecordBreakerOpen("drv-1", "host-a", Sampled); + // (drv-1, host-b): a healthy host with in-flight calls. + tracker.RecordCallStart("drv-1", "host-b"); + + var messages = DriverResilienceStatusPublisherService.BuildMessages(tracker, Published); + + messages.Count.ShouldBe(2); + + var open = messages.Single(m => m.HostName == "host-a"); + open.DriverInstanceId.ShouldBe("drv-1"); + open.BreakerOpen.ShouldBeTrue(); + open.ConsecutiveFailures.ShouldBe(2); + open.LastBreakerOpenUtc.ShouldBe(Sampled); + open.PublishedUtc.ShouldBe(Published); + + var healthy = messages.Single(m => m.HostName == "host-b"); + healthy.BreakerOpen.ShouldBeFalse(); + healthy.ConsecutiveFailures.ShouldBe(0); + healthy.CurrentInFlight.ShouldBe(1); + healthy.PublishedUtc.ShouldBe(Published); + } +}