diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Kpi/AuditLogKpiSampleSource.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Kpi/AuditLogKpiSampleSource.cs index 8b4e333d..198803b9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Kpi/AuditLogKpiSampleSource.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Kpi/AuditLogKpiSampleSource.cs @@ -3,6 +3,12 @@ using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi; +// Task 10 (arch-review 04): the append-only repository leaves the snapshot's +// BacklogTotal at zero; the real backlog is the cross-site health aggregate that +// only the central node knows. An optional IAuditBacklogProvider supplies it so +// the *history* sample stops recording a hardwired zero (the live tile already +// filled this in). Sites/tests leave it unregistered → null → snapshot fallback. + namespace ZB.MOM.WW.ScadaBridge.AuditLog.Kpi; /// @@ -44,18 +50,30 @@ public sealed class AuditLogKpiSampleSource : IKpiSampleSource private const string BacklogTotalMetric = KpiMetrics.AuditLog.BacklogTotal; private readonly IAuditLogRepository _repository; + private readonly IAuditBacklogProvider? _backlogProvider; /// /// Initializes a new instance of the class. /// /// /// Append-only Audit Log repository — its - /// supplies the volume, error, and backlog counts snapshotted here. + /// supplies the volume and error counts snapshotted here (its BacklogTotal + /// is always zero — see ). /// - public AuditLogKpiSampleSource(IAuditLogRepository repository) + /// + /// Optional cross-site backlog aggregate. When registered (central node), the + /// backlogTotal sample records its value; when absent (sites / unit tests, + /// where MS.DI resolves the default) the source falls back to the snapshot's own + /// BacklogTotal — i.e. zero, the pre-Task-10 behavior. Injecting the seam + /// keeps the append-only repo out of the in-memory health aggregator. + /// + public AuditLogKpiSampleSource( + IAuditLogRepository repository, + IAuditBacklogProvider? backlogProvider = null) { ArgumentNullException.ThrowIfNull(repository); _repository = repository; + _backlogProvider = backlogProvider; } /// @@ -78,11 +96,15 @@ public sealed class AuditLogKpiSampleSource : IKpiSampleSource return []; } + // Backlog: the provider's cross-site aggregate when registered (central), + // otherwise the snapshot's own value (zero) as the site/test fallback. + var backlogTotal = _backlogProvider?.GetPendingBacklogTotal() ?? snapshot.BacklogTotal; + return [ Sample(TotalEventsLastHourMetric, snapshot.TotalEventsLastHour, capturedAtUtc), Sample(ErrorEventsLastHourMetric, snapshot.ErrorEventsLastHour, capturedAtUtc), - Sample(BacklogTotalMetric, snapshot.BacklogTotal, capturedAtUtc), + Sample(BacklogTotalMetric, backlogTotal, capturedAtUtc), ]; } diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Kpi/IAuditBacklogProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Kpi/IAuditBacklogProvider.cs new file mode 100644 index 00000000..9945fa4a --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Kpi/IAuditBacklogProvider.cs @@ -0,0 +1,33 @@ +namespace ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi; + +/// +/// Supplies the system-wide pending Audit Log backlog for the +/// backlogTotal KPI history sample ("KPI History & Trends"). +/// +/// +/// +/// The append-only Audit Log repository leaves AuditLogKpiSnapshot.BacklogTotal +/// at zero — the backlog is not a central-table count but the sum of the latest +/// per-site SiteAuditBacklog.PendingCount health reports, which live only in +/// the in-memory central health aggregator. The live Health-dashboard "Audit" tile +/// already fills this in (AuditLogQueryService.GetKpiSnapshotAsync); this seam +/// lets the KPI history sample source read the same aggregate instead of +/// persisting a hardwired zero. +/// +/// +/// Best-effort: sites that have not yet reported (or whose reporter is disabled) +/// contribute zero — a missing snapshot is treated as "unknown, count as nothing", +/// mirroring the live tile. Central-only: registered on the central node where the +/// health aggregator lives; sites and unit tests leave it unregistered so the +/// sample source falls back to the snapshot's own BacklogTotal. +/// +/// +public interface IAuditBacklogProvider +{ + /// + /// Returns the current total pending (not-yet-forwarded) Audit Log backlog, + /// summed across every site's latest health report. Never negative; zero when + /// no site is reporting a backlog. + /// + long GetPendingBacklogTotal(); +} diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/CentralHealthAuditBacklogProvider.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/CentralHealthAuditBacklogProvider.cs new file mode 100644 index 00000000..56a18406 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/CentralHealthAuditBacklogProvider.cs @@ -0,0 +1,52 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi; + +namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Kpi; + +/// +/// Central — sums the latest per-site +/// SiteAuditBacklog.PendingCount across the in-memory +/// , the same aggregate the live +/// Health-dashboard "Audit" backlog tile shows +/// (AuditLogQueryService.GetKpiSnapshotAsync). +/// +/// +/// Registered only on the central node (where the aggregator lives), so the +/// +/// backing the backlogTotal trend records the real backlog instead of the +/// append-only repository's hardwired zero. Best-effort: sites with no report or +/// a null SiteAuditBacklog contribute nothing. +/// +public sealed class CentralHealthAuditBacklogProvider : IAuditBacklogProvider +{ + private readonly ICentralHealthAggregator _aggregator; + + /// + /// Initializes a new instance of the class. + /// + /// The central health aggregator whose per-site latest reports carry the backlog. + public CentralHealthAuditBacklogProvider(ICentralHealthAggregator aggregator) + { + ArgumentNullException.ThrowIfNull(aggregator); + _aggregator = aggregator; + } + + /// + public long GetPendingBacklogTotal() + { + // Ported verbatim from AuditLogQueryService.GetKpiSnapshotAsync: a null + // SiteAuditBacklog is "unknown" (contributes zero), and the `> 0` guard + // ignores any defensive negative/zero value so the tile and the history + // agree exactly. + long backlog = 0; + foreach (var state in _aggregator.GetAllSiteStates().Values) + { + var pending = state.LatestReport?.SiteAuditBacklog?.PendingCount; + if (pending is > 0) + { + backlog += pending.Value; + } + } + + return backlog; + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs index 5cba3b6d..5ab0cec1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/ServiceCollectionExtensions.cs @@ -63,6 +63,15 @@ public static class ServiceCollectionExtensions services.AddHostedService(sp => sp.GetRequiredService()); services.AddHostedService(); + // Task 10 (arch-review 04): the central backlog aggregate for the Audit Log + // KPI *history*. Only the central node runs the aggregator, so this seam is + // registered here; the AuditLogKpiSampleSource resolves it (optional ctor + // param) and records the real cross-site backlog instead of the append-only + // repository's hardwired zero. Sites leave it unregistered → snapshot fallback. + services.AddSingleton< + Commons.Interfaces.Kpi.IAuditBacklogProvider, + Kpi.CentralHealthAuditBacklogProvider>(); + // "KPI History & Trends": per-site Site Health KPI sample source. // Reads the in-memory central aggregator (a singleton) rather than a // repository; registered Scoped to match the recorder's per-tick scope diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Kpi/AuditLogKpiSampleSourceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Kpi/AuditLogKpiSampleSourceTests.cs index 68b12450..d6c732a5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Kpi/AuditLogKpiSampleSourceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Kpi/AuditLogKpiSampleSourceTests.cs @@ -1,5 +1,6 @@ using NSubstitute; using ZB.MOM.WW.ScadaBridge.AuditLog.Kpi; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Types; using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi; @@ -97,6 +98,57 @@ public class AuditLogKpiSampleSourceTests Assert.Empty(samples); } + [Fact] + public async Task CollectAsync_uses_backlog_provider_when_present() + { + // The repository's snapshot carries the hardwired zero the append-only + // repo always emits; the provider carries the real cross-site aggregate. + var snapshot = new AuditLogKpiSnapshot( + TotalEventsLastHour: 10, + ErrorEventsLastHour: 1, + BacklogTotal: 0, + AsOfUtc: CapturedAt); + + var repo = Substitute.For(); + repo.GetKpiSnapshotAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(snapshot); + + var backlog = Substitute.For(); + backlog.GetPendingBacklogTotal().Returns(42L); + + var sut = new AuditLogKpiSampleSource(repo, backlog); + + var samples = await sut.CollectAsync(CapturedAt); + + // Volume + error metrics still come from the snapshot; backlog is the + // provider's real aggregate, not the snapshot's zero. + AssertMetric(samples, "totalEventsLastHour", 10); + AssertMetric(samples, "errorEventsLastHour", 1); + AssertMetric(samples, "backlogTotal", 42); + } + + [Fact] + public async Task CollectAsync_falls_back_to_snapshot_backlog_when_provider_absent() + { + // No provider (sites / unit tests): the sample source degrades to the + // snapshot's own BacklogTotal — unchanged pre-Task-10 behavior. + var snapshot = new AuditLogKpiSnapshot( + TotalEventsLastHour: 10, + ErrorEventsLastHour: 1, + BacklogTotal: 7, + AsOfUtc: CapturedAt); + + var repo = Substitute.For(); + repo.GetKpiSnapshotAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(snapshot); + + var sut = new AuditLogKpiSampleSource(repo, backlogProvider: null); + + var samples = await sut.CollectAsync(CapturedAt); + + AssertMetric(samples, "backlogTotal", 7); + } + private static void AssertMetric( IReadOnlyList samples, string metric, diff --git a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/CentralHealthAuditBacklogProviderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/CentralHealthAuditBacklogProviderTests.cs new file mode 100644 index 00000000..0c6e4061 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/CentralHealthAuditBacklogProviderTests.cs @@ -0,0 +1,100 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring.Kpi; + +namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests.Kpi; + +/// +/// Task 10 (arch-review 04) coverage for . +/// The provider sums the latest per-site SiteAuditBacklog.PendingCount across the +/// in-memory — the same aggregate the live Audit +/// backlog tile shows — so the backlogTotal KPI history stops recording a hardwired +/// zero. Sites with no report or a null backlog snapshot contribute nothing. +/// +public class CentralHealthAuditBacklogProviderTests +{ + private static readonly DateTime At = new(2026, 6, 17, 12, 0, 0, DateTimeKind.Utc); + + [Fact] + public void GetPendingBacklogTotal_Sums_PendingCount_Across_Sites() + { + var aggregator = new StubAggregator + { + States = + { + ["site-a"] = StateWithBacklog("site-a", 9), + ["site-b"] = StateWithBacklog("site-b", 4), + ["site-c"] = new SiteHealthState { SiteId = "site-c", LatestReport = null }, // heartbeat-only + }, + }; + + var provider = new CentralHealthAuditBacklogProvider(aggregator); + + // 9 + 4; the null-report site contributes nothing. + Assert.Equal(13, provider.GetPendingBacklogTotal()); + } + + [Fact] + public void GetPendingBacklogTotal_Returns_Zero_When_No_Backlog_Reported() + { + var aggregator = new StubAggregator + { + States = + { + ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = MinimalReport("site-a") with { SiteAuditBacklog = null } }, + ["site-b"] = new SiteHealthState { SiteId = "site-b", LatestReport = null }, + }, + }; + + var provider = new CentralHealthAuditBacklogProvider(aggregator); + + Assert.Equal(0, provider.GetPendingBacklogTotal()); + } + + private static SiteHealthState StateWithBacklog(string siteId, int pending) => + new() + { + SiteId = siteId, + LatestReport = MinimalReport(siteId) with + { + SiteAuditBacklog = new SiteAuditBacklogSnapshot( + PendingCount: pending, OldestPendingUtc: null, OnDiskBytes: 0), + }, + }; + + private static SiteHealthReport MinimalReport(string siteId) => + new( + SiteId: siteId, + SequenceNumber: 1, + ReportTimestamp: At, + DataConnectionStatuses: new Dictionary(), + TagResolutionCounts: new Dictionary(), + ScriptErrorCount: 0, + AlarmEvaluationErrorCount: 0, + StoreAndForwardBufferDepths: new Dictionary(), + DeadLetterCount: 0, + DeployedInstanceCount: 0, + EnabledInstanceCount: 0, + DisabledInstanceCount: 0); + + /// + /// Hand-rolled stub — the HealthMonitoring.Tests + /// project has no mocking library. Only is exercised. + /// + private sealed class StubAggregator : ICentralHealthAggregator + { + public Dictionary States { get; } = new(); + + public IReadOnlyDictionary GetAllSiteStates() => States; + + public SiteHealthState? GetSiteState(string siteId) => + States.TryGetValue(siteId, out var state) ? state : null; + + public void ProcessReport(SiteHealthReport report) => throw new NotSupportedException(); + + public void MarkHeartbeat(string siteId, DateTimeOffset receivedAt) => throw new NotSupportedException(); + + public void PruneUnknownSites(IReadOnlyCollection knownSiteIds) => throw new NotSupportedException(); + } +}