From 90ce099928fc6f50fa2658bfbb270ebb5fa25620 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:42:56 -0400 Subject: [PATCH 1/5] feat(transport): counter for import-session cap rejections (register row 66) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Row 53 shipped the scadabridge.central.transport.import_sessions.open gauge and a Warning log on a cap hit, but a log line is not something an operator can alert on. Adds the companion counter scadabridge.central.transport.import_sessions.rejected on the existing ZB.MOM.WW.ScadaBridge meter, so no ObservedMeters allowlist change is needed (an unlisted meter exports nothing, silently). The emit point is deliberately narrow. BundleSessionStore.Open tests the cap TWICE — once on entry and again after EvictExpired() — and only the second test is a rejection: a store full of merely expired sessions reclaims a slot and admits the caller. Counting the first would make the series fire routinely on any node whose sweep had not yet run, turning a page-worthy signal into background noise. Re-Opening an already-open session id skips the cap block entirely and is likewise not counted. The two instruments are diagnostic as a pair: rejections rising while the gauge oscillates below the cap is genuine concurrency (raise the cap); rejections rising while the gauge sits pinned at the cap is a session leak (register row 43's signature), where raising the cap only defers the problem. Tests assert a delta over a scoped action rather than an absolute value, so they do not inherit the register row 32 static-state flake that makes QueueDepthGaugeTests fail only under the full concurrent suite. Negative controls cover under-cap, expired-reclaim, and duplicate-session-id. Tests: 11 passed in BundleSessionObservabilityTests (6 existing + 5 new). --- docs/components/Transport.md | 3 + docs/requirements/Component-Transport.md | 2 +- .../Observability/ScadaBridgeTelemetry.cs | 25 ++++ .../Import/BundleSessionStore.cs | 10 ++ .../Import/BundleSessionObservabilityTests.cs | 136 ++++++++++++++++++ 5 files changed, 175 insertions(+), 1 deletion(-) diff --git a/docs/components/Transport.md b/docs/components/Transport.md index 9c39eb26..6104ad73 100644 --- a/docs/components/Transport.md +++ b/docs/components/Transport.md @@ -52,6 +52,9 @@ Without this, an abandoned wizard or a plain CLI preview held one of the eight ` **Session-cap observability (register row 53).** That leak was only ever inferable from rotating test failures, because nothing counted open sessions and nothing logged a rejection. Two signals now close the gap: - **Gauge `scadabridge.central.transport.import_sessions.open`** — open (unexpired) sessions on this node, on the `ZB.MOM.WW.ScadaBridge` meter, which is already in `SiteServiceRegistration.ObservedMeters` (the `ZbTelemetryOptions.Meters` allowlist; an unlisted meter exports nothing, silently). The provider is registered by `BundleSessionEvictionService.StartAsync` and identity-checked-cleared on `StopAsync`, mirroring the StoreAndForward queue-depth gauge. `IBundleSessionStore.OpenSessionCount` deliberately excludes expired-but-unswept entries — counting the raw dictionary would report the node as pinned at the cap for up to a sweep interval after a burst had actually drained, which is the exact false positive this instrument exists to rule out. Transport is registered only in the host's Central branch, so the gauge reports 0 on a site node. +- **Counter `scadabridge.central.transport.import_sessions.rejected`** (register row 66) — bundle-import sessions turned away because the cap was exhausted, on the same meter (so likewise no allowlist change). Row 53 shipped the gauge and a Warning log, but a log line cannot be alerted on; this is the series to alert on. It is incremented **only at the post-eviction recheck** in `BundleSessionStore.Open`: the cap is tested twice, and a store full of merely *expired* sessions reclaims and admits the caller, so counting the first test would fire the signal routinely on any node whose sweep has not yet run. Re-`Open`ing an already-open session id skips the cap block entirely and is likewise never counted. + + Read the two instruments **together** — the pair is what discriminates the failure modes. Rejections climbing while the gauge oscillates *below* the cap is genuine concurrency: raise `MaxConcurrentImportSessions`. Rejections climbing while the gauge sits *pinned at* the cap is a session leak (the register row 43 signature); raising the cap only defers it, so find the leak instead. - **Warning log on rejection** — `BundleSessionStore.Open` logs the open count, the cap, and the TTL when it rejects. It fires only on a real cap hit: a store full of *expired* sessions is reclaimed first and logs nothing. A count that stays pinned at the cap across repeated warnings is a leak; an occasional warning is genuine concurrency, and the fix there is to raise `ScadaBridge:Transport:MaxConcurrentImportSessions` (each slot pins up to `MaxBundleSizeMb` of decrypted plaintext, so the cap is a memory budget). ### Conflict resolution and `BundleImportId` correlation diff --git a/docs/requirements/Component-Transport.md b/docs/requirements/Component-Transport.md index de37bd1b..f8bb5f8b 100644 --- a/docs/requirements/Component-Transport.md +++ b/docs/requirements/Component-Transport.md @@ -134,7 +134,7 @@ flowchart TD class T muted ``` -The component is central-hosted. It is registered in `ZB.MOM.WW.ScadaBridge.Host` for central roles only, never for site roles — even when a bundle carries site-scoped config (T18), Transport only writes the central configuration database, never the site clusters. All persistence flows through existing audited repository interfaces in `ZB.MOM.WW.ScadaBridge.ConfigurationDatabase` — the component does not call `DbContext.SaveChangesAsync` directly. `BundleSessionStore` is in-process on the active central node (matching Blazor Server circuit affinity): 30-minute TTL, eviction on expiry, 3-strike passphrase lockout per session, and a `MaxConcurrentImportSessions` cap (default 8; arch-review 05 Task 24 — startup-validated as `> 0` by `TransportOptionsValidator`, since a zero/negative cap would reject every import session forever) bounding the decrypted-content footprint since each open session pins a fully-decrypted bundle in memory. **The cap is observable (register row 53):** open (unexpired) sessions are published as the `scadabridge.central.transport.import_sessions.open` gauge on the `ZB.MOM.WW.ScadaBridge` meter — already in the `ObservedMeters` allowlist, so no telemetry config change is needed — and a rejection at the cap logs at Warning with the current count, the cap and the TTL. Reclaiming a store full of *expired* sessions is not a cap hit and logs nothing. +The component is central-hosted. It is registered in `ZB.MOM.WW.ScadaBridge.Host` for central roles only, never for site roles — even when a bundle carries site-scoped config (T18), Transport only writes the central configuration database, never the site clusters. All persistence flows through existing audited repository interfaces in `ZB.MOM.WW.ScadaBridge.ConfigurationDatabase` — the component does not call `DbContext.SaveChangesAsync` directly. `BundleSessionStore` is in-process on the active central node (matching Blazor Server circuit affinity): 30-minute TTL, eviction on expiry, 3-strike passphrase lockout per session, and a `MaxConcurrentImportSessions` cap (default 8; arch-review 05 Task 24 — startup-validated as `> 0` by `TransportOptionsValidator`, since a zero/negative cap would reject every import session forever) bounding the decrypted-content footprint since each open session pins a fully-decrypted bundle in memory. **The cap is observable (register row 53):** open (unexpired) sessions are published as the `scadabridge.central.transport.import_sessions.open` gauge on the `ZB.MOM.WW.ScadaBridge` meter — already in the `ObservedMeters` allowlist, so no telemetry config change is needed — and a rejection at the cap logs at Warning with the current count, the cap and the TTL. Reclaiming a store full of *expired* sessions is not a cap hit and logs nothing. **A log line is not alertable, so register row 66 adds the counter `scadabridge.central.transport.import_sessions.rejected`** on the same meter, incremented at the same single genuine-rejection point (the post-eviction recheck — never on the reclaim-then-succeed path, and never when re-opening an already-open session id). Alert on the counter; interpret it against the gauge — rejections rising while the gauge stays below the cap means genuine concurrency (raise the cap), whereas rejections rising while the gauge is pinned at the cap means a leaked session (raising the cap only defers it). **Session release is explicit, not TTL-only (register row 43).** `ApplyAsync` releases the session it applied (zero-then-remove, T-007), and `IBundleImporter.Abandon(sessionId)` does the same for a session that will never be applied. Every abandonment path calls it: the Central UI wizard on new upload / Back-to-Upload / `ResetSessionState` / lockout reset / `Dispose` (navigation away or circuit teardown), the CLI `transport preview` handler (a preview never applies), and the CLI `transport import` handler for the aborts that precede apply (blocker rows, unresolved mappings). `Abandon` is idempotent — double release, an unknown id, and an already-expired id are all silent no-ops — and since session ids are random GUIDs handed only to their opener, abandoning the id you hold can never evict another operator's in-progress import. Relying on the TTL alone was an availability *and* secret-lifetime defect: eight abandoned wizards exhausted `MaxConcurrentImportSessions` and locked out every import on the active node for 30 minutes, each still holding its decrypted bundle content. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index b2431fe4..34d3bd1d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -93,6 +93,23 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("scadabridge.site.stream.events_dropped", unit: "1", description: "Events evicted from a site gRPC stream's bounded send channel, tagged by stream kind."); + /// + /// Register row 66: incremented each time BundleSessionStore.Open refuses a + /// bundle-import session because TransportOptions.MaxConcurrentImportSessions + /// (default 8) is genuinely exhausted — i.e. the store was still at the cap AFTER + /// expired sessions were reclaimed. Reclaim-then-succeed and re-opening an already + /// open session id are deliberately NOT counted: neither turns an operator away. + /// + /// Row 53 shipped the companion open-sessions gauge and a Warning log, but a log line + /// is not alertable. The pair is what distinguishes the two failure modes: rejections + /// climbing while the gauge oscillates below the cap is genuine concurrency (raise the + /// cap), whereas rejections climbing while the gauge sits pinned at the cap is a + /// session leak like register row 43's (find the leak — raising the cap only defers it). + /// + private static readonly Counter _transportImportSessionsRejected = + Meter.CreateCounter("scadabridge.central.transport.import_sessions.rejected", unit: "1", + description: "Bundle-import sessions rejected because the concurrent-import-session cap was exhausted."); + // ---------------- Histograms ---------------- /// @@ -186,6 +203,14 @@ public static class ScadaBridgeTelemetry /// Records one resync whose ack window expired. public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1); + /// + /// Records one bundle-import session rejected on a genuine concurrent-session cap hit + /// (register row 66). Call ONLY from the post-eviction recheck in + /// BundleSessionStore.Open — counting the pre-eviction check would inflate the + /// series with reclaim-then-succeed cases that never turned an operator away. + /// + public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1); + /// Records that a site connection opened (increments the up-count gauge). public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); diff --git a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs index ccac6231..fd713a69 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Transport/Import/BundleSessionStore.cs @@ -3,6 +3,7 @@ using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Transport; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Transport; namespace ZB.MOM.WW.ScadaBridge.Transport.Import; @@ -115,6 +116,15 @@ public sealed class BundleSessionStore : IBundleSessionStore EvictExpired(); if (_sessions.Count >= cap) { + // Register row 66: the alertable half of row 53's observability. The + // Warning below carries the diagnosis but a log line cannot be alerted + // on; this counter can. Deliberately incremented HERE, after + // EvictExpired() has already failed to free a slot — the outer check + // at the top of this block is not a rejection (a store full of stale + // entries reclaims and succeeds), so counting there would inflate the + // series with imports that were never actually turned away. + ScadaBridgeTelemetry.RecordImportSessionRejected(); + // Register row 53: this rejection used to be silent apart from the // exception the caller saw. A cap hit is either genuine concurrency // (many operators importing at once) or a session leak like register diff --git a/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs index ca9f0939..ea6d17e2 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Transport.Tests/Import/BundleSessionObservabilityTests.cs @@ -19,6 +19,7 @@ namespace ZB.MOM.WW.ScadaBridge.Transport.Tests.Import; public sealed class BundleSessionObservabilityTests { private const string OpenSessionsGauge = "scadabridge.central.transport.import_sessions.open"; + private const string RejectedSessionsCounter = "scadabridge.central.transport.import_sessions.rejected"; // ------------------------------------------------------------------ // Warning log on cap rejection @@ -83,6 +84,108 @@ public sealed class BundleSessionObservabilityTests Assert.Empty(logger.Entries); } + // ------------------------------------------------------------------ + // Rejection counter (register row 66) + // ------------------------------------------------------------------ + + [Fact] + public void Open_AtCap_IncrementsTheRejectionCounter() + { + // Row 53 shipped the gauge + Warning; a log line is not alertable. This counter is + // the series an operator can page on, so it must move on a genuine cap hit. + var clock = new TestClock(DateTimeOffset.UtcNow); + var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 }); + var sut = new BundleSessionStore(opts, clock, NullLogger.Instance); + var future = clock.GetUtcNow().AddMinutes(30); + + for (var i = 0; i < 3; i++) + { + sut.Open(SessionExpiringAt(future)); + } + + var delta = ReadRejectionCounter(() => + Assert.Throws(() => sut.Open(SessionExpiringAt(future)))); + + Assert.Equal(1, delta); + } + + [Fact] + public void Open_RepeatedCapHits_AccumulateOnTheRejectionCounter() + { + // A counter that saturated at 1 would hide exactly the case it exists for: a + // sustained rejection rate is what distinguishes real contention from a one-off. + var clock = new TestClock(DateTimeOffset.UtcNow); + var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 1 }); + var sut = new BundleSessionStore(opts, clock, NullLogger.Instance); + var future = clock.GetUtcNow().AddMinutes(30); + sut.Open(SessionExpiringAt(future)); + + var delta = ReadRejectionCounter(() => + { + for (var i = 0; i < 3; i++) + { + Assert.Throws(() => sut.Open(SessionExpiringAt(future))); + } + }); + + Assert.Equal(3, delta); + } + + [Fact] + public void Open_UnderCap_DoesNotIncrementTheRejectionCounter() + { + // Negative control: the counter is a rejection signal, not an open-session signal. + var clock = new TestClock(DateTimeOffset.UtcNow); + var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 3 }); + var sut = new BundleSessionStore(opts, clock, NullLogger.Instance); + + var delta = ReadRejectionCounter(() => + sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)))); + + Assert.Equal(0, delta); + } + + [Fact] + public void Open_AtCapOfExpiredSessions_ReclaimsWithoutIncrementingTheRejectionCounter() + { + // The load-bearing negative control. The cap check runs TWICE — once before + // EvictExpired and once after — and only the post-eviction check is a rejection. + // Counting the first would make the series fire routinely on any node whose sweep + // has not yet run, turning a page-worthy signal into background noise. + var clock = new TestClock(DateTimeOffset.UtcNow); + var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 }); + var sut = new BundleSessionStore(opts, clock, NullLogger.Instance); + + sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5))); + sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(5))); + clock.Advance(TimeSpan.FromMinutes(10)); + + var delta = ReadRejectionCounter(() => + sut.Open(SessionExpiringAt(clock.GetUtcNow().AddMinutes(30)))); + + Assert.Equal(0, delta); + } + + [Fact] + public void Open_ReopeningAnExistingSessionIdAtCap_DoesNotIncrementTheRejectionCounter() + { + // Re-Opening an id already in the store skips the cap block entirely and overwrites + // in place. It consumes no new slot and turns nobody away, so it must not count. + var clock = new TestClock(DateTimeOffset.UtcNow); + var opts = Options.Create(new TransportOptions { MaxConcurrentImportSessions = 2 }); + var sut = new BundleSessionStore(opts, clock, NullLogger.Instance); + var future = clock.GetUtcNow().AddMinutes(30); + + var first = SessionExpiringAt(future); + sut.Open(first); + sut.Open(SessionExpiringAt(future)); + + // Store is now exactly at the cap of 2. + var delta = ReadRejectionCounter(() => sut.Open(first)); + + Assert.Equal(0, delta); + } + // ------------------------------------------------------------------ // OpenSessionCount / gauge // ------------------------------------------------------------------ @@ -197,6 +300,39 @@ public sealed class BundleSessionObservabilityTests return observed; } + /// + /// Runs with a listener attached to the import-session rejection + /// counter and returns the number of rejections recorded DURING that action. + /// + /// + /// Deliberately a delta over a scoped action rather than an absolute read. The counter + /// is a process-global static, so an absolute assertion would inherit the register + /// row 32 flake that makes QueueDepthGaugeTests fail only under the full + /// concurrent suite. A counter measured this way is immune: the listener exists only + /// for the duration of , and every increment is driven + /// synchronously by this test's own Open call on its own store instance, so a + /// concurrently-running test class cannot contribute a measurement to this window. + /// + /// The action to measure. + /// Rejections counted while ran. + private static long ReadRejectionCounter(Action act) + { + long delta = 0; + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName + && instrument.Name == RejectedSessionsCounter) + { + l.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, measurement, _, _) => delta += measurement); + listener.Start(); + act(); + return delta; + } + private static BundleSession SessionExpiringAt(DateTimeOffset expiresAt) => new() { SessionId = Guid.NewGuid(), From b5f120c370789e17d479c26b1fc7146e7a29cf8f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:53:12 -0400 Subject: [PATCH 2/5] feat(observability): honest stream-subscriber name + real data-connection gauges (row 54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scadabridge_site_connection_up never counted data connections. It counts open central->site gRPC subscriber streams (Debug View + the site-wide alarm feed), so it correctly reads 0 on a healthy site nobody is watching — and the name invited reading it as device health. That misreading has already cost a live debugging session: register row 48's 'connection_up is 0 on all six site nodes' was healthy steady state, not the fault it was diagnosed as. Two halves, per the row: RENAME. scadabridge.site.connection.up -> scadabridge.site.stream.subscribers.active (helpers SiteConnectionOpened/Closed -> SiteStreamSubscriberOpened/Closed), and the instrument gains the unit:"1" the other gauges already carried. Every reference updated: the gRPC server, both test suites, Component-Communication.md, components/Commons.md, docker/README.md and docker/seed-live-probe.sh. A test now pins the new name AND asserts the old one is gone, so it cannot creep back. NEW. scadabridge.site.data_connections.connected / .total — the first exported data-connection signal (the state existed on the health report and in the CLI, but never as a metric). connected < total is the alert condition; neither number alone separates 'a device is down' from 'nothing is configured', which is why the denominator ships with the numerator. Sourcing is deliberate on two counts. It is fed by the DCL at its four state transitions rather than derived from the health report, so the metric survives a dead report pipeline — the exact failure the 'metrics stale' flag exists to catch, and a derived gauge would go blind at the same moment. And it uses a static registry reached through ScadaBridgeTelemetry (the SiteConnectionOpened precedent) rather than a DI-injected holder: the DCL actors already take ISiteHealthCollector through their constructors, and threading a second dependency through both would ripple into every test that builds them. It also keeps SiteHealthCollector untouched, which matters right now — another agent is concurrently editing that file for the tag-resolution path. Removal is wired too: an undeployed connection leaves the registry, so it stops counting toward the total instead of latching connected < total on forever. Both gauges read 0 on the standby (connections open on the active node only) — healthy, documented, and called out in the docs so it does not become the next site_connection_up. No ObservedMeters allowlist change: all three instruments ride the existing ZB.MOM.WW.ScadaBridge meter. (An unlisted meter exports nothing, silently.) Tests: Commons 711 passed (9 new, delta-based against a baseline with unique per-test connection names, so they do not inherit the register row 32 QueueDepthGaugeTests static-state flake); Communication 717; DCL 319. --- docker/seed-live-probe.sh | 15 +- docs/components/Commons.md | 2 +- docs/requirements/Component-Communication.md | 2 +- .../Component-HealthMonitoring.md | 29 +++ .../Observability/ScadaBridgeTelemetry.cs | 140 ++++++++++++- .../Grpc/SiteStreamGrpcServer.cs | 8 +- .../Actors/DataConnectionActor.cs | 10 + .../Actors/DataConnectionManagerActor.cs | 5 + .../Observability/DataConnectionGaugeTests.cs | 190 ++++++++++++++++++ .../Grpc/SiteStreamGrpcServerTests.cs | 14 +- .../SiteAlarmLiveCacheTelemetryTests.cs | 2 +- .../ScadaBridgeTelemetryTests.cs | 4 +- 12 files changed, 389 insertions(+), 32 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs diff --git a/docker/seed-live-probe.sh b/docker/seed-live-probe.sh index 54e738a9..c5841d76 100755 --- a/docker/seed-live-probe.sh +++ b/docker/seed-live-probe.sh @@ -112,9 +112,14 @@ echo "" echo "While a stream is open, the site node serving it reports both signals" echo "(metrics are on container port 8084, and the containers have no curl/wget):" echo " docker exec scadabridge-site-a-a bash -c 'exec 3<>/dev/tcp/localhost/8084 && printf \"GET /metrics HTTP/1.0\\r\\nHost: localhost\\r\\n\\r\\n\" >&3 && cat <&3' \\" -echo " | grep -E 'site_connection_up|stream_batch_size'" +echo " | grep -E 'stream_subscribers_active|stream_batch_size|data_connections'" echo "" -echo "NOTE: scadabridge_site_connection_up counts OPEN CENTRAL->SITE gRPC SUBSCRIBER" -echo "STREAMS, not OPC UA data connections. It reads 0 on a perfectly healthy site" -echo "with no subscriber attached — use the health report's dataConnectionEndpoints" -echo "for data-connection health." +echo "NOTE: scadabridge_site_stream_subscribers_active (renamed from the misleading" +echo "scadabridge_site_connection_up, register row 54) counts OPEN CENTRAL->SITE gRPC" +echo "SUBSCRIBER STREAMS, not OPC UA data connections. It reads 0 on a perfectly healthy" +echo "site with no subscriber attached — that is correct, not a fault." +echo "" +echo "For real data-connection health use the gauges added by the same row:" +echo " scadabridge_site_data_connections_connected / _total" +echo "connected < total is the alert condition. Both read 0 on the STANDBY node, since" +echo "connections are opened by the deployment path on the active node only." diff --git a/docs/components/Commons.md b/docs/components/Commons.md index 872f1bd6..394b4551 100644 --- a/docs/components/Commons.md +++ b/docs/components/Commons.md @@ -295,7 +295,7 @@ public record AlarmStateChanged( ### Observability -`Observability/ScadaBridgeTelemetry` defines the singleton `Meter` named `ZB.MOM.WW.ScadaBridge` and the application-wide instrument definitions. Components call the static emit helpers (`RecordDeploymentApplied`, `SiteConnectionOpened`, etc.) rather than creating their own meters. Instruments are no-ops until an OTel listener attaches, so uninstrumented hosts pay no overhead. +`Observability/ScadaBridgeTelemetry` defines the singleton `Meter` named `ZB.MOM.WW.ScadaBridge` and the application-wide instrument definitions. Components call the static emit helpers (`RecordDeploymentApplied`, `SiteStreamSubscriberOpened`, etc.) rather than creating their own meters. Instruments are no-ops until an OTel listener attaches, so uninstrumented hosts pay no overhead. ## Usage diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 45512df9..479d0264 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -108,7 +108,7 @@ The streaming protocol is defined in `sitestream.proto` (`src/ZB.MOM.WW.ScadaBri - **Service**: `SiteStreamService` — hosted on each site node by `SiteStreamGrpcServer` — exposes six RPCs. Two are real-time **server-streaming** subscriptions (`SubscribeInstance`, `SubscribeSite`); the other four are **unary request/response** calls added by the Audit Log (#23) and Site Call Audit (#22) components. These are distinct from the command/control gRPC services (`CentralControlService` on central, `SiteCommandService` on the site) — `SiteStreamService` is the streaming + audit-pull surface, not the command/control channel: - `SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent)` — the per-instance real-time debug stream (§6). - - `SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent)` — the **site-wide, alarm-only** aggregated stream (§6.1) added for the operator Alarm Summary live cache. Additive (new RPC + new `SiteStreamRequest { correlation_id }` message; no field renumbering). The server handler mirrors `SubscribeInstance` — but with **its own, larger send channel** (`GrpcSiteAlarmStreamChannelCapacity`, default 20 000, `DropOldest`; WP2.3): sharing the Debug View's 1000-slot channel meant an alarm burst during a WAN stall silently evicted operator-visible transitions to make room for diagnostics traffic. Drops are tagged `stream=site-alarms` on `scadabridge.site.stream.events_dropped`. Otherwise identical — same `GrpcMaxConcurrentStreams`/`GrpcMaxStreamLifetime` limits, `SiteConnectionOpened/Closed` telemetry, and `StreamRelayActor` mapping — but subscribes via `SiteStreamManager.SubscribeSiteAlarms` (all instances, `AlarmStateChanged` only; attribute events dropped, no per-instance filter). + - `SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent)` — the **site-wide, alarm-only** aggregated stream (§6.1) added for the operator Alarm Summary live cache. Additive (new RPC + new `SiteStreamRequest { correlation_id }` message; no field renumbering). The server handler mirrors `SubscribeInstance` — but with **its own, larger send channel** (`GrpcSiteAlarmStreamChannelCapacity`, default 20 000, `DropOldest`; WP2.3): sharing the Debug View's 1000-slot channel meant an alarm burst during a WAN stall silently evicted operator-visible transitions to make room for diagnostics traffic. Drops are tagged `stream=site-alarms` on `scadabridge.site.stream.events_dropped`. Otherwise identical — same `GrpcMaxConcurrentStreams`/`GrpcMaxStreamLifetime` limits, `SiteStreamSubscriberOpened/Closed` telemetry, and `StreamRelayActor` mapping — but subscribes via `SiteStreamManager.SubscribeSiteAlarms` (all instances, `AlarmStateChanged` only; attribute events dropped, no per-instance filter). - `IngestAuditEvents(AuditEventBatch) returns (IngestAck)` — a legacy central-side ingest surface on the *site*-hosted service; it is **dead in the shipped topology** because no site ever dials a site for ingest. The production audit-telemetry *push* path is site→central gRPC to `CentralControlService`, which routes the batch to the central `AuditLogIngestActor` proxy and returns the accepted `EventId`s. - `IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck)` — ingest receiving surface for the combined cached-call telemetry packet (audit row + `SiteCalls` operational upsert written in one transaction). - `PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse)` — central→site **reconciliation pull** for the Audit Log self-heal feed; the site serves `Pending`/`Forwarded` rows from its `ISiteAuditQueue`. **At-least-once (WP2.3):** rows are NOT retired when served. The site retires (`Reconciled`) everything at or before the cursor of the **next** pull, because that cursor is central's only proof of receipt — a fault between the response leaving the site and central committing it re-serves the rows instead of losing them (central dedups on `EventId`). The request carries an additive composite-keyset `after_id` (field 3) mirroring `PullSiteCallsRequest`; with it the read is a strict `(OccurredAtUtc, EventId)` keyset and the retirement is exact, without it the legacy inclusive `>=` read applies and only rows strictly older than the cursor instant are provably received. diff --git a/docs/requirements/Component-HealthMonitoring.md b/docs/requirements/Component-HealthMonitoring.md index ecc8f198..931007f3 100644 --- a/docs/requirements/Component-HealthMonitoring.md +++ b/docs/requirements/Component-HealthMonitoring.md @@ -101,6 +101,35 @@ These tiles are **point-in-time** like the Notification Outbox and Site Call Aud - Notification Outbox and Site Call Audit KPIs are not stored by Health Monitoring; they are computed point-in-time from the central `Notifications` and `SiteCalls` tables respectively each time the dashboard refreshes — consistent with the current-status-only philosophy. - Site connectivity history (online/offline transitions) may optionally be logged via the Audit Log or a separate mechanism if needed in the future. +## Prometheus Metrics (site node) + +Distinct from the health *report* above: these are scraped from the site node's `/metrics` +endpoint (`MetricsPort`, default 8084) rather than sent to central. + +| Series | Meaning | +| --- | --- | +| `scadabridge_site_data_connections_connected` | Data connections on this node in the `Connected` state | +| `scadabridge_site_data_connections_total` | Data connections known to this node, in any state | +| `scadabridge_site_stream_subscribers_active` | Open central→site gRPC subscriber streams | + +- The two `data_connections` gauges (register row 54) are the **only exported data-connection + signal**; before them the health report's `DataConnectionStatuses` was visible in the UI and + the CLI but never as a metric. `connected < total` is the alert condition — neither number + alone distinguishes "a device is down" from "nothing is configured". They are fed directly by + the Data Connection Layer at each connection state transition, deliberately **not** derived + from the health-report path, so the metric survives a dead report pipeline (the exact failure + the *metrics stale* flag exists to catch). +- Both read **0 on the standby node** — data connections are opened by the deployment path on + the active node only. A standby reporting 0 is healthy; alert on the active node, or on + `max()` across the pair. +- `scadabridge_site_stream_subscribers_active` was renamed from `scadabridge_site_connection_up` + by the same row. It has only ever counted open subscriber streams (Debug View + the site-wide + alarm feed), so it correctly reads 0 on a healthy site nobody is watching — the old name + invited reading it as data-connection health, and did cause exactly that misdiagnosis once. +- All three ride the `ZB.MOM.WW.ScadaBridge` meter, already in the `SiteServiceRegistration.ObservedMeters` + allowlist. `ZbTelemetryOptions.Meters` is an **allowlist**: an unlisted meter exports nothing, + with no error and no missing-metric signal. + ## No Alerting - Health monitoring is **display-only** for now — no automated notifications or alerts are triggered by health status changes. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index 34d3bd1d..cce04d8a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -1,4 +1,5 @@ using System.Diagnostics.Metrics; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.Commons.Observability; @@ -126,8 +127,27 @@ public static class ScadaBridgeTelemetry // ---------------- Observable gauges ---------------- - /// Current count of open site connections, mutated via . - private static long _siteConnectionsUp; + /// + /// Current count of open central→site gRPC subscriber streams, mutated via + /// . See 's gauge for + /// why this is NOT a data-connection count. + /// + private static long _siteStreamSubscriberCount; + + /// + /// Register row 54: last-known state of every data connection on this site node, keyed by + /// connection name. Written by the DCL at its four state-transition points and read by the + /// data-connection gauges below. + /// + /// + /// A static registry, matching 's precedent, rather + /// than a DI-injected holder: the DCL actors already receive ISiteHealthCollector + /// through their constructors, and threading a second dependency through both actor + /// constructors would ripple into every test that builds them. A site node hosts exactly + /// one DCL, so a process-global registry and a per-node one are the same thing here. + /// + private static readonly System.Collections.Concurrent.ConcurrentDictionary + _dataConnectionStates = new(StringComparer.Ordinal); /// Current count of running per-site live-alarm aggregators, mutated via . private static long _liveAlarmAggregatorsActive; @@ -144,11 +164,52 @@ public static class ScadaBridgeTelemetry private static Func? _importSessionCountProvider; #pragma warning disable IDE0052 // Held to keep the observable gauges alive for the meter's lifetime. - /// Gauge reporting the number of currently open site connections. - private static readonly ObservableGauge _siteConnectionUp = - Meter.CreateObservableGauge("scadabridge.site.connection.up", - () => Interlocked.Read(ref _siteConnectionsUp), - description: "Number of currently open site connections."); + /// + /// Gauge reporting the number of open central→site gRPC subscriber streams. + /// + /// + /// Register row 54 RENAMED this from scadabridge.site.connection.up + /// (scadabridge_site_connection_up in Prometheus). The old name read as a + /// data-connection health signal, but it has only ever counted open subscriber streams + /// on SiteStreamGrpcServer — Debug View and the site-wide alarm feed. It therefore + /// sits at 0 on a perfectly healthy site that nobody is watching, and an alert written + /// against the old name fired on an idle operator console rather than a broken device. + /// For real data-connection health use the scadabridge.site.data_connections.* + /// gauges below. Counts both stream kinds together (no stream tag). + /// + private static readonly ObservableGauge _siteStreamSubscribersActive = + Meter.CreateObservableGauge("scadabridge.site.stream.subscribers.active", + () => Interlocked.Read(ref _siteStreamSubscriberCount), + unit: "1", + description: "Open central→site gRPC subscriber streams (Debug View + site-wide alarm feed). NOT data connections."); + + /// + /// Register row 54: gauge reporting how many of this site node's data connections are + /// currently in — the real data-connection + /// signal the misnamed stream-subscriber gauge was being mistaken for. + /// + /// + /// Read together with : connected < total is the + /// alertable condition, and neither number alone distinguishes "one device down" from + /// "nothing configured". Expect **0 on the standby node** — connections are opened by the + /// deployment path on the active node only, so a standby reporting 0 is healthy, not a + /// fault. Alert on the active node, or on max() across the pair. + /// + private static readonly ObservableGauge _dataConnectionsConnected = + Meter.CreateObservableGauge("scadabridge.site.data_connections.connected", + CountDataConnections, + unit: "1", + description: "Data connections on this site node currently in the Connected state."); + + /// + /// Register row 54: gauge reporting how many data connections this site node knows about, + /// in any state. The denominator for . + /// + private static readonly ObservableGauge _dataConnectionsTotal = + Meter.CreateObservableGauge("scadabridge.site.data_connections.total", + () => (long)_dataConnectionStates.Count, + unit: "1", + description: "Data connections known to this site node, in any state."); /// Gauge reporting the current StoreAndForward queue depth via the registered provider. private static readonly ObservableGauge _storeAndForwardQueueDepth = @@ -211,11 +272,68 @@ public static class ScadaBridgeTelemetry /// public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1); - /// Records that a site connection opened (increments the up-count gauge). - public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); + /// + /// Records that a central→site gRPC subscriber stream opened (increments + /// scadabridge.site.stream.subscribers.active). Renamed from + /// SiteConnectionOpened by register row 54 — it never counted data connections. + /// + public static void SiteStreamSubscriberOpened() => Interlocked.Increment(ref _siteStreamSubscriberCount); - /// Records that a site connection closed (decrements the up-count gauge). - public static void SiteConnectionClosed() => Interlocked.Decrement(ref _siteConnectionsUp); + /// + /// Records that a central→site gRPC subscriber stream closed (decrements + /// scadabridge.site.stream.subscribers.active). Renamed from + /// SiteConnectionClosed by register row 54. + /// + public static void SiteStreamSubscriberClosed() => Interlocked.Decrement(ref _siteStreamSubscriberCount); + + /// + /// Records the current state of one data connection, feeding the + /// scadabridge.site.data_connections.* gauges (register row 54). Called by the DCL + /// at each connection state transition; safe to call repeatedly with the same value. + /// + /// The data connection's name (unique per site node). + /// The connection's new state. + public static void RecordDataConnectionState(string connectionName, ConnectionHealth state) + { + if (string.IsNullOrEmpty(connectionName)) + { + return; + } + + _dataConnectionStates[connectionName] = state; + } + + /// + /// Drops a data connection from the gauges (register row 54). Called when the DCL removes + /// a connection, so an undeployed connection stops counting toward the total rather than + /// lingering as a permanently-not-connected row that would alert forever. + /// + /// The data connection's name. + public static void RemoveDataConnection(string connectionName) + { + if (string.IsNullOrEmpty(connectionName)) + { + return; + } + + _dataConnectionStates.TryRemove(connectionName, out _); + } + + /// Counts data connections currently in . + /// The connected count. + private static long CountDataConnections() + { + long connected = 0; + foreach (var state in _dataConnectionStates.Values) + { + if (state == ConnectionHealth.Connected) + { + connected++; + } + } + + return connected; + } /// Records that a per-site live-alarm aggregator started (increments the active-aggregator gauge). public static void LiveAlarmAggregatorStarted() => Interlocked.Increment(ref _liveAlarmAggregatorsActive); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index 2acb2a91..9a257585 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -467,12 +467,12 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase // Telemetry follow-on: the connection is now fully established (Subscribe // succeeded, so no leak via the catch above). Count it up here and balance - // it in the finally below so the scadabridge.site.connection.up gauge is - // decremented on EVERY exit path — normal completion, client-cancel / + // it in the finally below so the scadabridge.site.stream.subscribers.active + // gauge is decremented on EVERY exit path — normal completion, client-cancel / // duplicate-replacement (OperationCanceledException), server shutdown // (CancelAllStreams -> Cts.Cancel), and any other exception — guaranteeing // exactly one Closed per Opened and a gauge that never drifts up. - ScadaBridgeTelemetry.SiteConnectionOpened(); + ScadaBridgeTelemetry.SiteStreamSubscriberOpened(); try { // R2 — event batching. The pump replaces the old per-event @@ -500,7 +500,7 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase } finally { - ScadaBridgeTelemetry.SiteConnectionClosed(); + ScadaBridgeTelemetry.SiteStreamSubscriberClosed(); _streamSubscriber.RemoveSubscriber(relayActor); _actorSystem!.Stop(relayActor); channel.Writer.TryComplete(); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 9cd2c598..659b7f8a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -3,6 +3,7 @@ using Akka.Event; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; @@ -312,6 +313,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _log.Info("[{0}] Entering Connecting state", _connectionName); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Connecting); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Connecting"); + // Register row 54 — mirror the state into the data-connection gauges. Kept alongside + // the health-collector write (rather than derived from it) so the metric never + // depends on the health-report path being alive. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connecting); Become(Connecting); Self.Tell(new AttemptConnect()); } @@ -394,6 +399,8 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount); var endpointLabel = _backupConfig == null ? "Connected" : $"Connected to {_activeEndpoint.ToString().ToLower()}"; _healthCollector.UpdateConnectionEndpoint(_connectionName, endpointLabel); + // Register row 54 — the only transition that raises the connected gauge. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connected); Become(Connected); Stash.UnstashAll(); } @@ -475,6 +482,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _log.Warning("[{0}] Entering Reconnecting state", _connectionName); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected"); + // Register row 54 — drops this connection out of the connected gauge while the + // total holds, which is exactly the connected < total alert condition. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected); // Track unstable connections toward failover. // If we were connected for less than the stability threshold, this counts diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs index 14de697d..b1467468 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Serialization; using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; @@ -307,6 +308,10 @@ public class DataConnectionManagerActor : ReceiveActor Context.Stop(actor); _connectionActors.Remove(command.ConnectionName); _healthCollector.RemoveConnection(command.ConnectionName); + // Register row 54 — drop it from the gauges too, so an undeployed connection + // stops counting toward the total instead of lingering as a permanently + // not-connected row that would hold the connected < total alert on forever. + ScadaBridgeTelemetry.RemoveDataConnection(command.ConnectionName); _log.Info("Removed DataConnectionActor for {0}", command.ConnectionName); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs new file mode 100644 index 00000000..98f1de8c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs @@ -0,0 +1,190 @@ +using System.Diagnostics.Metrics; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Observability; + +/// +/// Register row 54: the real per-site data-connection gauges. Before these existed the only +/// connection-ish series was scadabridge_site_connection_up, which counted open +/// central→site gRPC subscriber streams and sat at 0 on a healthy unwatched site — so there +/// was no metric that answered "are this site's devices actually connected?" at all. +/// +/// +/// Every assertion here is a DELTA against a baseline taken inside the test, and every +/// connection name is unique to the test. That is deliberate: the registry behind these +/// gauges is a process-global static, so absolute assertions would inherit the register +/// row 32 flake that makes QueueDepthGaugeTests pass alone and fail under the full +/// concurrent suite. Each test also removes its own connections in a finally so it +/// leaves the shared registry exactly as it found it. +/// +public sealed class DataConnectionGaugeTests +{ + private const string ConnectedGauge = "scadabridge.site.data_connections.connected"; + private const string TotalGauge = "scadabridge.site.data_connections.total"; + + [Fact] + public void ConnectedState_RaisesConnectedAndTotal() + { + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + + Assert.Equal(connectedBefore + 1, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Theory] + [InlineData(ConnectionHealth.Disconnected)] + [InlineData(ConnectionHealth.Connecting)] + [InlineData(ConnectionHealth.Error)] + public void NonConnectedState_RaisesTotalOnly(ConnectionHealth state) + { + // The load-bearing shape of the pair: a device that is configured but not talking + // must show up in the denominator and NOT in the numerator, because connected < total + // is the whole alert condition. A gauge that only counted connected rows would read + // identically for "one device down" and "device deleted". + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, state); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Fact] + public void TransitionFromConnectedToDisconnected_DropsConnectedButKeepsTotal() + { + // The DCL calls Record* once per state transition on the SAME name, so a transition + // must overwrite rather than accumulate — otherwise a flapping connection would + // inflate the total without bound. + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Disconnected); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Fact] + public void RemoveDataConnection_DropsItFromBothGauges() + { + // An undeployed connection must leave the denominator. If it lingered as a + // permanently not-connected row, connected < total would latch on forever and the + // alert would be useless after the first undeploy. + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RemoveDataConnection(name); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore, ReadGauge(TotalGauge)); + } + + [Fact] + public void RemoveDataConnection_IsIdempotent() + { + var name = UniqueName(); + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RemoveDataConnection(name); + + var ex = Record.Exception(() => ScadaBridgeTelemetry.RemoveDataConnection(name)); + + Assert.Null(ex); + } + + [Fact] + public void NullOrEmptyConnectionName_IsIgnored() + { + // Guards the gauges against a malformed caller creating a phantom row that could + // never be removed by name. + var totalBefore = ReadGauge(TotalGauge); + + ScadaBridgeTelemetry.RecordDataConnectionState(string.Empty, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RecordDataConnectionState(null!, ConnectionHealth.Connected); + + Assert.Equal(totalBefore, ReadGauge(TotalGauge)); + } + + [Fact] + public void StreamSubscriberGauge_IsPublishedUnderTheRenamedName() + { + // Register row 54's rename half. Pins the honest name so a future edit cannot + // silently restore scadabridge.site.connection.up, whose meaning (open gRPC + // subscriber streams) never matched what operators read it as. + var names = PublishedInstrumentNames(); + + Assert.Contains("scadabridge.site.stream.subscribers.active", names); + Assert.DoesNotContain("scadabridge.site.connection.up", names); + } + + private static string UniqueName() => $"conn-{Guid.NewGuid():N}"; + + /// Forces one observation of the named gauge and returns its value. + /// Instrument name to observe. + /// The observed value, or 0 if the instrument never reported. + private static long ReadGauge(string instrumentName) + { + long observed = 0; + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName + && instrument.Name == instrumentName) + { + l.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, measurement, _, _) => observed = measurement); + listener.Start(); + listener.RecordObservableInstruments(); + return observed; + } + + /// Collects every instrument name published on the ScadaBridge meter. + /// The published instrument names. + private static IReadOnlyCollection PublishedInstrumentNames() + { + var names = new List(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, _) => + { + if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName) + { + names.Add(instrument.Name); + } + }; + listener.Start(); + return names; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs index 43a3c331..84c870e1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs @@ -443,12 +443,12 @@ public class SiteStreamGrpcServerTests : TestKit } [Fact] - public async Task SiteConnectionUpGauge_GoesToOneOnConnect_AndBackToZeroOnCancel() + public async Task SiteStreamSubscribersActiveGauge_GoesToOneOnConnect_AndBackToZeroOnCancel() { - // Telemetry follow-on: the scadabridge.site.connection.up gauge must read + // Telemetry follow-on: the scadabridge.site.stream.subscribers.active gauge must read // exactly 1 while a site stream is established and return to 0 once the - // stream terminates on the cancel path — proving SiteConnectionOpened() is - // matched by exactly one SiteConnectionClosed() in the handler's finally. + // stream terminates on the cancel path — proving SiteStreamSubscriberOpened() is + // matched by exactly one SiteStreamSubscriberClosed() in the handler's finally. var server = CreateServer(); server.SetReady(Sys); @@ -459,7 +459,7 @@ public class SiteStreamGrpcServerTests : TestKit listener.InstrumentPublished = (instrument, l) => { if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName && - instrument.Name == "scadabridge.site.connection.up") + instrument.Name == "scadabridge.site.stream.subscribers.active") { l.EnableMeasurementEvents(instrument); } @@ -485,7 +485,7 @@ public class SiteStreamGrpcServerTests : TestKit // (possibly parallel) tests left behind — read relative so the assertion // is robust to test interleaving on the process-wide static counter. // - // SiteConnectionOpened() runs AFTER the _activeStreams insertion that + // SiteStreamSubscriberOpened() runs AFTER the _activeStreams insertion that // WaitForConditionAsync(ActiveStreamCount == 1) keys off (see // SiteStreamGrpcServer.SubscribeInstance), so a one-shot ReadGauge() here // can race the increment under full-suite CPU oversubscription. Poll the @@ -499,7 +499,7 @@ public class SiteStreamGrpcServerTests : TestKit // After the cancel path runs the finally, the gauge is balanced back to // the baseline — no leaked "up" count. Poll here too: the gauge decrement // and the stream-count drop are independent observations, so under load - // give the finally's SiteConnectionClosed() time to land. + // give the finally's SiteStreamSubscriberClosed() time to land. AwaitAssert(() => Assert.Equal(baseline, ReadGauge()), TimeSpan.FromSeconds(5)); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs index 87549372..67729234 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// Telemetry guards for the aggregated live alarm cache (plan #10, Task 6): the /// active-aggregator observable gauge reflects Started/Stopped, and the reconnect /// counter increments on each recorded reconnect. Mirrors the MeterListener read -/// pattern used by SiteStreamGrpcServerTests.SiteConnectionUpGauge_*; reads are +/// pattern used by SiteStreamGrpcServerTests.SiteStreamSubscribersActiveGauge_*; reads are /// relative to a baseline so the process-wide static instruments are robust to any /// parallel test interleaving. /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs index 6c89f559..c17b290b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs @@ -22,8 +22,8 @@ public class ScadaBridgeTelemetryTests { ScadaBridgeTelemetry.RecordDeploymentApplied(); ScadaBridgeTelemetry.RecordInboundApiRequest("X"); - ScadaBridgeTelemetry.SiteConnectionOpened(); - ScadaBridgeTelemetry.SiteConnectionClosed(); + ScadaBridgeTelemetry.SiteStreamSubscriberOpened(); + ScadaBridgeTelemetry.SiteStreamSubscriberClosed(); ScadaBridgeTelemetry.LiveAlarmAggregatorStarted(); ScadaBridgeTelemetry.LiveAlarmAggregatorStopped(); ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect(); From 103f2f1931fc3868ecbe1ba91003c1d17cfb9708 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:53:26 -0400 Subject: [PATCH 3/5] feat(host+rig): give central a dedicated Prometheus scrape port (register row 56) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The row read 'central nodes expose no Prometheus endpoint'. That is not quite what was wrong, and the difference matters. Central already maps /metrics (Program.cs central branch) and telemetry is registered for both roles through the shared BindSharedOptions, so the endpoint rides endpoint routing and has always been served on central's web port (:5000 on the rig, anonymous — there is no FallbackPolicy). What central lacked was the site branch's dedicated MetricsPort, which is the port the established live-scrape recipe probes: docker exec bash -c 'exec 3<>/dev/tcp/localhost/8084 ...' So a probe of central on 8084 failed and central-only series (KPI/outbox counters, the bundle-import session gauge + its new rejection counter, alarm-cache gauges) read as unscrapeable while they were in fact being exported on a different port. Central now binds a dedicated HTTP/1.1 listener on ScadaBridge:Node:MetricsPort, so one probe recipe covers all eight rig nodes. HTTP/1.1-capable because a standard Prometheus scraper cannot negotiate h2c, which is why this cannot share the gRPC port. Read as a PRESENCE check, not GetValue's defaulted read. NodeOptions defaults MetricsPort to 8084, so a defaulted read would make every existing central node start binding a new port on upgrade — and a port already in use is a hard boot failure, not a degraded one. Central gains the listener only where an operator asked for it; site nodes are unchanged and still bind theirs unconditionally. The bind is also skipped when the port would collide with the gRPC port or an ASPNETCORE_URLS port, since Kestrel throws on a duplicate ListenAnyIP and /metrics is reachable on those ports anyway. Config only — no host port mapping added, because the probe runs inside the container. Applied to both docker/ centrals and both docker-env2/ centrals. FILE EDITS ONLY: no docker command was run, and deploy.sh was not invoked. --- .../central-node-a/appsettings.Central.json | 4 +- .../central-node-b/appsettings.Central.json | 4 +- docker/README.md | 39 +++++++++++++--- .../central-node-a/appsettings.Central.json | 4 +- .../central-node-b/appsettings.Central.json | 4 +- src/ZB.MOM.WW.ScadaBridge.Host/Program.cs | 44 +++++++++++++++++-- 6 files changed, 86 insertions(+), 13 deletions(-) diff --git a/docker-env2/central-node-a/appsettings.Central.json b/docker-env2/central-node-a/appsettings.Central.json index 02398abf..f4f6ba5e 100644 --- a/docker-env2/central-node-a/appsettings.Central.json +++ b/docker-env2/central-node-a/appsettings.Central.json @@ -4,7 +4,9 @@ "Role": "Central", "NodeName": "central-a", "NodeHostname": "scadabridge-env2-central-a", - "RemotingPort": 8081 + "RemotingPort": 8081, + "_metricsPort": "Register row 56 — mirrors docker/central-node-a. Dedicated HTTP/1.1 Prometheus scrape listener; central binds it only when this key is present. In-container port only, no host mapping.", + "MetricsPort": 8084 }, "Cluster": { "SeedNodes": [ diff --git a/docker-env2/central-node-b/appsettings.Central.json b/docker-env2/central-node-b/appsettings.Central.json index 41b93443..8850c3ca 100644 --- a/docker-env2/central-node-b/appsettings.Central.json +++ b/docker-env2/central-node-b/appsettings.Central.json @@ -4,7 +4,9 @@ "Role": "Central", "NodeName": "central-b", "NodeHostname": "scadabridge-env2-central-b", - "RemotingPort": 8081 + "RemotingPort": 8081, + "_metricsPort": "Register row 56 — mirrors docker/central-node-a. Dedicated HTTP/1.1 Prometheus scrape listener; central binds it only when this key is present. In-container port only, no host mapping.", + "MetricsPort": 8084 }, "Cluster": { "SeedNodes": [ diff --git a/docker/README.md b/docker/README.md index c59d0d8f..d2831de4 100644 --- a/docker/README.md +++ b/docker/README.md @@ -181,12 +181,28 @@ Two traps it exists to avoid: Binding to them gives a `Connected` data connection with `Good` quality and *zero* event traffic, which looks healthy and probes nothing. Use the built-in simulated nodes (`SpikeData`, `RandomSignedInt32`, `StepUp`, `FastUInt1..5`, `SlowUInt1..5`). -- **`scadabridge_site_connection_up` does not mean what its name suggests.** It counts - open central→site gRPC *subscriber streams*, not OPC UA data connections, so it sits - at 0 on a perfectly healthy site with nobody watching. For data-connection health read - `dataConnectionEndpoints` / `dataConnectionTagQuality` off `scadabridge health site`. +- **`scadabridge_site_stream_subscribers_active` is not data-connection health.** It counts + open central→site gRPC *subscriber streams*, so it sits at 0 on a perfectly healthy site + with nobody watching. (Register row 54 renamed it from `scadabridge_site_connection_up`, + whose name invited exactly that misreading — it is what made row 48's "0 on all six site + nodes" look like a fault when it was healthy steady state.) -Verify (metrics are on container port 8084; the containers ship no curl/wget): +For real data-connection health, the same row added two honest gauges: + +| Series | Meaning | +| --- | --- | +| `scadabridge_site_data_connections_connected` | Connections currently in the `Connected` state | +| `scadabridge_site_data_connections_total` | Connections known to this node, in any state | + +`connected < total` is the alert condition — neither number alone separates "a device is +down" from "nothing is configured". Both read **0 on the standby node**, which is healthy: +connections are opened by the deployment path on the active node only, so alert on the +active node or on `max()` across the pair. The health report's `dataConnectionEndpoints` / +`dataConnectionTagQuality` (via `scadabridge health site`) remain the per-connection detail. + +Verify (metrics are on container port 8084 — **all eight nodes** as of register row 56, which +gave the two central nodes the same dedicated scrape listener the six site nodes already had; +the containers ship no curl/wget): ```bash scadabridge debug snapshot --id @@ -195,9 +211,20 @@ scadabridge debug stream --id # open a stream, then in another s docker exec scadabridge-site-a-a bash -c \ 'exec 3<>/dev/tcp/localhost/8084 && printf "GET /metrics HTTP/1.0\r\nHost: localhost\r\n\r\n" >&3 && cat <&3' \ - | grep -E 'site_connection_up|stream_batch_size' + | grep -E 'stream_subscribers_active|stream_batch_size|data_connections' + +# Central-only series (KPI/outbox counters, bundle-import session gauge + rejection +# counter, alarm-cache gauges) — same recipe, central container: +docker exec scadabridge-central-a bash -c \ + 'exec 3<>/dev/tcp/localhost/8084 && printf "GET /metrics HTTP/1.0\r\nHost: localhost\r\n\r\n" >&3 && cat <&3' \ + | grep -E 'import_sessions|alarm_cache' ``` +> Central binds 8084 only because `docker/central-node-*/appsettings.Central.json` now sets +> `ScadaBridge:Node:MetricsPort`. Unlike a site node, central reads that key **only when +> present** — a central deployment without it is unchanged and still serves `/metrics` on its +> web port, since the endpoint rides endpoint routing and is served on every bound listener. + ### After Code Changes Rebuild and redeploy. The Docker build cache skips NuGet restore when only source files change: diff --git a/docker/central-node-a/appsettings.Central.json b/docker/central-node-a/appsettings.Central.json index dea71fc3..0184c93e 100644 --- a/docker/central-node-a/appsettings.Central.json +++ b/docker/central-node-a/appsettings.Central.json @@ -4,7 +4,9 @@ "Role": "Central", "NodeName": "central-a", "NodeHostname": "scadabridge-central-a", - "RemotingPort": 8081 + "RemotingPort": 8081, + "_metricsPort": "Register row 56. Central maps /metrics on every listener it binds, so it was always reachable on the web port (:5000) — but not on 8084, which is the port the live-scrape recipe probes (docker exec ... bash -c ' { // The HTTP/1.1 (+ HTTP/2) surface from the configured URLs, re-declared so it @@ -217,6 +238,21 @@ try { listenOptions.Protocols = Microsoft.AspNetCore.Server.Kestrel.Core.HttpProtocols.Http2; }); + + // The Prometheus scrape listener (register row 56), when configured. Skipped + // when it would double-bind a port already declared above — Kestrel throws on a + // duplicate ListenAnyIP, and /metrics is reachable on that port either way since + // it rides endpoint routing. HTTP/1.1-capable on purpose: a standard Prometheus + // scraper cannot negotiate h2c, which is why this cannot share the gRPC port. + if (centralMetricsPort is { } cmp + && cmp != centralGrpcPort + && !httpPorts.Contains(cmp)) + { + options.ListenAnyIP(cmp, listenOptions => + { + listenOptions.Protocols = Microsoft.AspNetCore.Server.Kestrel.Core.HttpProtocols.Http1AndHttp2; + }); + } }); // Shared components @@ -698,9 +734,11 @@ try builder.Configuration.AddConfiguration(configuration); // Captive-dependency validation (deferred register row 50) — the SITE half of the - // pair. This branch is the one that actually gains coverage: the docker rig sets - // ASPNETCORE_ENVIRONMENT=Development on the central containers but NOT on the site - // containers, so a site node has always built its provider unvalidated. + // pair. This branch is the one that actually gains coverage: the docker rig long + // set ASPNETCORE_ENVIRONMENT=Development on the central containers but NOT on the + // site containers, so a site node built its provider unvalidated. Register row 63 + // closed that — all six rig site containers (and docker-env2's two) now set it, so + // this validation actually runs there. builder.Host.UseDefaultServiceProvider(ServiceProviderValidation.Configure); // Serilog From 0d2d6de6c67003dd5f1c6a1fcd4ec14293f40668 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:53:33 -0400 Subject: [PATCH 4/5] chore(rig): set ASPNETCORE_ENVIRONMENT on the site containers (register row 63) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The central containers have always set it; the six site containers never did, so the Development-only DI scope validation from register row 50 (ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on exactly the half of the topology that has six of the eight nodes. Program.cs even carried a comment saying the site branch 'is the one that actually gains coverage' while the rig gave it none; that comment is corrected here too. Now that the site graph validates clean this buys live coverage: a captive dependency or an unconstructible site-only descriptor fails the container at boot instead of surfacing later as a mid-run resolve error. Applied to all six docker/ site containers and both docker-env2/ site containers, keeping the two rigs consistent. Dev/rig only — never production. FILE EDITS ONLY: no docker command was run, and deploy.sh was not invoked. Both compose files re-parsed with yaml.safe_load; all 8 + 4 services confirmed to carry the expected env. --- docker-env2/docker-compose.yml | 8 ++++++ docker/docker-compose.yml | 48 ++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/docker-env2/docker-compose.yml b/docker-env2/docker-compose.yml index b7d3d80f..fc7f85d4 100644 --- a/docker-env2/docker-compose.yml +++ b/docker-env2/docker-compose.yml @@ -70,6 +70,10 @@ services: container_name: scadabridge-env2-site-x-a environment: SCADABRIDGE_CONFIG: Site + # Register row 63 — mirrors docker/docker-compose.yml. Enables the Development-only + # DI scope validation (register row 50) on the site nodes, which previously ran + # unvalidated because only the central containers set this. Dev/rig only. + ASPNETCORE_ENVIRONMENT: Development ports: - "9121:8082" # Akka remoting - "9123:8083" # gRPC streaming @@ -90,6 +94,10 @@ services: container_name: scadabridge-env2-site-x-b environment: SCADABRIDGE_CONFIG: Site + # Register row 63 — mirrors docker/docker-compose.yml. Enables the Development-only + # DI scope validation (register row 50) on the site nodes, which previously ran + # unvalidated because only the central containers set this. Dev/rig only. + ASPNETCORE_ENVIRONMENT: Development ports: - "9122:8082" # Akka remoting - "9124:8083" # gRPC streaming diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 3c05b583..ee1d358d 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -213,6 +213,14 @@ services: # file). site-b and site-c deliberately carry neither. <<: [*secrets-store-env, *secrets-hub-env, *secrets-hub-site-env] SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9021:8082" # Akka remoting (host access for debugging) - "9023:8083" # gRPC streaming @@ -241,6 +249,14 @@ services: # file). site-b and site-c deliberately carry neither. <<: [*secrets-store-env, *secrets-hub-env, *secrets-hub-site-env] SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9022:8082" # Akka remoting - "9024:8083" # gRPC streaming @@ -267,6 +283,14 @@ services: environment: <<: *secrets-store-env SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9031:8082" # Akka remoting - "9033:8083" # gRPC streaming @@ -293,6 +317,14 @@ services: environment: <<: *secrets-store-env SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9032:8082" # Akka remoting - "9034:8083" # gRPC streaming @@ -319,6 +351,14 @@ services: environment: <<: *secrets-store-env SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9041:8082" # Akka remoting - "9043:8083" # gRPC streaming @@ -345,6 +385,14 @@ services: environment: <<: *secrets-store-env SCADABRIDGE_CONFIG: Site + # Register row 63. The central containers have always set this; the site containers + # never did, so the Development-only DI scope validation (register row 50, + # ServiceProviderValidation.Configure — ValidateScopes/ValidateOnBuild) was dead on + # exactly the half of the topology that has six nodes. The site graph now validates + # clean, so turning it on here buys live coverage: a captive dependency or an + # unconstructible site-only descriptor fails the container at boot instead of + # surfacing as a mid-run resolve error. Dev/rig only — never set in production. + ASPNETCORE_ENVIRONMENT: Development ports: - "9042:8082" # Akka remoting - "9044:8083" # gRPC streaming From 8da60f57acdd9f7fdb36dc8bbc251e6a0f644df4 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 13:02:35 -0400 Subject: [PATCH 5/5] feat(health): KPI history + sustained-backlog badge for LocalDb replication (row 33) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LocalDbOplogBacklog was reported and Prometheus-exported with nothing consuming it: not sampled into KPI history, so a purge-burst spike could not be reviewed after the fact, and no threshold treatment anywhere. Both halves shipped. KPI SAMPLING. SiteHealthKpiSampleSource now emits localDbOplogBacklog and localDbReplicationConnected (0/1), and the backlog is charted on the Health dashboard's per-site trend panel. Charting is the point: a point-in-time tile can never answer 'did that spike drain?', which is the only thing separating an expected purge burst from a stalled peer. Both are CONDITIONALLY emitted rather than coerced with ?? 0 like the auditBacklogPending metric beside them. Both health-report fields are nullable and null means 'no reading' — replication not configured (rig site-b/site-c) or a failed poll — never zero. A fabricated 0 would render a pair that cannot read its own oplog, or one deliberately unreplicated, as perfectly healthy once a minute forever, burying the real series. So a site emits 12 metrics when unreplicated and 14 when replicated, and a GAP in these two series truthfully means 'not replicated / not reporting'. Note false and null differ for the connected flag: a configured link that is DOWN samples as 0; no link samples as nothing. Both fold as Gauge (summing an oplog depth, or a 0/1 flag, per hour is meaningless). THRESHOLD SEMANTICS — sustained, not instantaneous. This is the design, not a refinement. The site_events retention purge is CDC-captured like any other write, so the backlog jumps by the deleted batch and drains as the peer acks it; a bare 'backlog > threshold' badge would fire on every replicated site every day and be ignored within a week. Healthy is defined by the backlog RETURNING to ~0, which only a dwell can express. CentralHealthAggregator holds LocalDbBacklogAboveSince — the instant the backlog first rose above LocalDbBacklogThreshold (default 1000) — and the badge appears only after LocalDbBacklogSustainedFor (default 15 min). A report at or below the threshold clears the anchor; one above leaves the ORIGINAL instant intact, so the window is sustained rather than sliding. The null case is deliberately a third branch: a null backlog leaves the anchor UNCHANGED rather than clearing it. Null is evidence neither of a drain nor of a backlog, so a node going blind mid-climb must not silently reset its own dwell, and an unreplicated site (null forever) never accumulates an anchor and can never be badged. Validation rejects a non-positive threshold (0 would badge every replicated site permanently, since an idle backlog already sits at it) and a dwell shorter than one ReportInterval (satisfiable by a single sample, which collapses the check back into the bare threshold it replaces). DOC CORRECTION worth flagging: this row's premise was 'mirror the SiteAuditBacklog configurable threshold badge'. That badge does not exist — Component-HealthMonitoring.md described it in two places but no option, no comparison and no markup were ever implemented. The two stale claims are corrected here rather than left to send the next reader looking for it; the LocalDb badge is built fresh on the Metrics-stale badge's pattern, which is a real configurable-threshold badge. Tests: HealthMonitoring 115 (9 new aggregator tests incl. purge-burst-drains, unreplicated-never-anchors, null-mid-climb, exact-threshold, stale-report; 5 new validator tests; 4 new sampler tests incl. the null-semantics and false-vs-null cases); CentralUI 1042 (5 new badge/chart tests incl. the purge-burst negative control); Commons Kpi 93; KpiHistory 54. --- docs/deployment/topology-guide.md | 26 +++ .../Component-HealthMonitoring.md | 46 ++++- docs/requirements/Component-KpiHistory.md | 11 ++ .../Components/Pages/Monitoring/Health.razor | 45 +++++ .../Types/Kpi/KpiMetrics.cs | 17 ++ .../CentralHealthAggregator.cs | 46 ++++- .../HealthMonitoringOptions.cs | 28 +++ .../HealthMonitoringOptionsValidator.cs | 24 +++ .../Kpi/SiteHealthKpiSampleSource.cs | 28 +++ .../SiteHealthState.cs | 22 +++ .../Pages/HealthPageTests.cs | 97 ++++++++++ .../Kpi/KpiMetricAggregationTests.cs | 8 + .../HealthMonitoringOptionsValidatorTests.cs | 57 ++++++ .../Kpi/SiteHealthKpiSampleSourceTests.cs | 86 ++++++++- .../LocalDbBacklogSustainedTests.cs | 176 ++++++++++++++++++ 15 files changed, 711 insertions(+), 6 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/LocalDbBacklogSustainedTests.cs diff --git a/docs/deployment/topology-guide.md b/docs/deployment/topology-guide.md index d06e5211..cc279f92 100644 --- a/docs/deployment/topology-guide.md +++ b/docs/deployment/topology-guide.md @@ -255,6 +255,32 @@ LocalDbReplicationConnected staying true with the backlog returning to ~0 is the An unreplicated node logs only the first line. If a backlog spike has no such line near it in the active node's log, the purge is *not* the explanation and the spike is worth investigating. +**Two dashboard surfaces make this reviewable rather than a log-archaeology exercise** (register +row 33): + +- **The "Replication backlog" badge** on the site tile of the Health dashboard. It is deliberately + a *sustained* signal, not a bare threshold: the badge appears only once the backlog has stayed + above `ScadaBridge:HealthMonitoring:LocalDbBacklogThreshold` (default **1000**) continuously for + `ScadaBridge:HealthMonitoring:LocalDbBacklogSustainedFor` (default **15 minutes**). The daily + purge burst described above drains in seconds to a couple of minutes, so it never reaches the + dwell and never badges the site — which is the entire reason the dwell exists. A bare threshold + would fire once a day on every replicated site and be ignored within a week. The dwell is + validated to be at least one `ReportInterval`, since a shorter window could be satisfied by a + single sample and would collapse back into a bare threshold. +- **The "LocalDb Replication Backlog" trend chart** in the dashboard's per-site Site Health Trends + panel, backed by the KPI history store (sampled every minute, retained 90 days raw / 365 days + hourly). This is what answers the question a point-in-time reading cannot: *did that spike + drain?* A past purge burst appears as a spike that returns to ~0; a real fault appears as a + level that never comes back down. + +**Both surfaces are silent on an unreplicated site, by design.** `LocalDbOplogBacklog` and +`LocalDbReplicationConnected` are nullable, and null means "no reading", never zero. The KPI +sampler therefore writes **no sample at all** rather than a fabricated 0, so site-b/site-c show an +empty chart rather than a flat zero line, and they never accumulate the dwell anchor the badge +keys off. A gap in those series means "not replicated / not reporting" — which is the truth. Note +that `false` and null differ for the connected flag: a configured link that is DOWN samples as 0, +while no link at all samples as nothing. + ### Site Pair Upgrades — stop and start BOTH nodes together **A rolling upgrade of a site pair, one node at a time, is no longer supported.** It worked while diff --git a/docs/requirements/Component-HealthMonitoring.md b/docs/requirements/Component-HealthMonitoring.md index 931007f3..e33f8c40 100644 --- a/docs/requirements/Component-HealthMonitoring.md +++ b/docs/requirements/Component-HealthMonitoring.md @@ -34,7 +34,7 @@ Site clusters (metric collection and reporting). Central cluster (aggregation an | Notification Outbox queue depth | Notification Outbox (central) | Count of `Pending` + `Retrying` notifications — central-computed, not site-reported | | Notification Outbox stuck count | Notification Outbox (central) | Count of `Pending` / `Retrying` notifications older than the configurable stuck-age threshold — central-computed, not site-reported | | Notification Outbox parked count | Notification Outbox (central) | Count of `Parked` notifications — central-computed, not site-reported | -| `SiteAuditBacklog` | Audit Log (site) | Count of `Pending` rows in the site-local `AuditLog` plus oldest-pending-age plus on-disk bytes. A configurable threshold drives a Health dashboard warning on the affected site tile. | +| `SiteAuditBacklog` | Audit Log (site) | Count of `Pending` rows in the site-local `AuditLog` plus oldest-pending-age plus on-disk bytes. Surfaced on the Audit KPI tiles (which border-highlight any non-zero backlog) and sampled into KPI history as `auditBacklogPending`. **No configurable per-site threshold badge exists for this metric** — earlier revisions of this document described one, but none was ever implemented and no such option exists. The only configurable sustained-threshold site badge is the LocalDb replication backlog (register row 33, below). | | `SiteAuditWriteFailures` | Audit Log (site) | Count of failed hot-path audit appends at the site since the last health report. | | `AuditRedactionFailure` | Audit Log (central) | Count of payload redactor errors (over-redacted payloads, safety-net hit) since the last interval. | | `LocalDbReplicationConnected` | Consolidated site LocalDb | Whether a peer sync session is currently running on this node. **Nullable — null means "no reading", not "disconnected".** Reported only by a node with replication configured. | @@ -90,7 +90,7 @@ The Audit Log spans both sites (hot-path append + telemetry forward) and central - **Audit volume** — events/min landing in the central `AuditLog` table, shown global plus per-site sparkline; sourced from the Audit Log component on the active central node. - **Audit error rate** — percent of central `AuditLog` rows with `Status` other than `Success` / `Delivered` / `Enqueued` over a rolling 5-minute window. This is the operational error rate of audited operations (HTTP 5xx, transient failures, parked deliveries, etc.) — NOT the audit writer's own health. Audit-writer issues surface separately via `AuditRedactionFailure`. -- **Audit backlog** — global aggregate of `SiteAuditBacklog` across reporting sites (count of `Pending` site-local audit rows, oldest pending age, on-disk bytes); click drills into a per-site breakdown. The per-site tile surfaces a warning badge when its `SiteAuditBacklog` crosses the configurable threshold. +- **Audit backlog** — global aggregate of `SiteAuditBacklog` across reporting sites (count of `Pending` site-local audit rows, oldest pending age, on-disk bytes); click drills into a per-site breakdown. The tile border-highlights a non-zero aggregate backlog; there is **no** configurable per-site `SiteAuditBacklog` threshold badge (see the note in Monitored Metrics — earlier revisions described one that was never built). These tiles are **point-in-time** like the Notification Outbox and Site Call Audit KPI tiles — no time-series store; consistent with Health Monitoring's "current status only" philosophy. The site-scoped `SiteAuditBacklog` / `SiteAuditWriteFailures` metrics arrive in the existing site health report; the central-scoped `AuditRedactionFailure` metric is central-computed alongside the existing central KPIs. @@ -101,6 +101,46 @@ These tiles are **point-in-time** like the Notification Outbox and Site Call Aud - Notification Outbox and Site Call Audit KPIs are not stored by Health Monitoring; they are computed point-in-time from the central `Notifications` and `SiteCalls` tables respectively each time the dashboard refreshes — consistent with the current-status-only philosophy. - Site connectivity history (online/offline transitions) may optionally be logged via the Audit Log or a separate mechanism if needed in the future. +## Sustained LocalDb Replication Backlog Badge + +The per-site tile shows a **Replication backlog** warning badge when `LocalDbOplogBacklog` has +stayed above `LocalDbBacklogThreshold` (default **1000**) continuously for +`LocalDbBacklogSustainedFor` (default **15 minutes**). Both are configurable under +`ScadaBridge:HealthMonitoring` and validated at startup. + +- **Sustained, not instantaneous — and that is the design, not a refinement.** The daily + `site_events` retention purge and the storage-cap trim are CDC-captured like any other write, so + the backlog jumps by the deleted batch size and drains as the peer acks it. A bare + `backlog > threshold` badge would therefore fire on every replicated site every day. Healthy is + defined by the backlog *returning to ~0*, which only a dwell can express. `LocalDbBacklogSustainedFor` + is validated to be at least one `ReportInterval`: a shorter dwell could be satisfied by a single + sample and would collapse back into the bare threshold. +- **Implementation.** `CentralHealthAggregator` holds `SiteHealthState.LocalDbBacklogAboveSince`, + the instant the backlog first rose above the threshold. Reports at or below the threshold clear + it; reports above it leave the original instant intact (a sustained window, not a sliding one). +- **Null handling is load-bearing.** A report with a `null` backlog leaves the anchor *unchanged* + rather than clearing it — null means "no reading" (unreplicated site, or a failed poll), which is + evidence neither of a drain nor of a backlog. An unreplicated site (rig site-b/site-c) reports + null forever, so it never accumulates an anchor and can never be badged. +- Offline sites are not badged; the Offline badge is the signal that matters there. + +## KPI History Sampling of Replication State + +`SiteHealthKpiSampleSource` samples `localDbOplogBacklog` and `localDbReplicationConnected` (0/1) +into the KPI history store, and the backlog is charted in the Health dashboard's per-site Site +Health Trends panel as **LocalDb Replication Backlog**. This is the retrospective half of the +signal: a point-in-time tile can never answer "did that spike drain?", which is the only thing +distinguishing an expected purge burst from a stalled peer. + +Both metrics are **conditionally emitted** — a site reporting `null` writes no sample rather than a +coerced `0`, unlike the `auditBacklogPending` metric alongside them. A fabricated 0 would render a +pair that cannot read its own oplog, or one deliberately not replicated, as perfectly healthy once +a minute forever. Consequently a site's sample count varies: **12 metrics for an unreplicated site, +14 for a replicated one**, and a gap in these two series truthfully means "not replicated / not +reporting". Note `false` and `null` are different states for the connected flag — a configured link +that is down samples as `0`; no link at all samples as nothing. Both fold as **Gauge** in the hourly +rollup catalog (summing an oplog depth, or a 0/1 flag, per hour would be meaningless). + ## Prometheus Metrics (site node) Distinct from the health *report* above: these are scraped from the site node's `/metrics` @@ -151,4 +191,4 @@ endpoint (`MetricsPort`, default 8084) rather than sent to central. - **Central UI**: Health Monitoring Dashboard displays aggregated metrics. - **Communication Layer**: Health reports flow as periodic messages. -- **KPI History (#26)**: emits `IKpiSampleSource` (`SiteHealthKpiSampleSource`, per-Site) consumed by the KpiHistory recorder (#26). It reads the in-memory `ICentralHealthAggregator.GetAllSiteStates()` (no DB), turning the per-site snapshot — previously sequence-numbered every 30s but discarded — into 12 sampled `KpiSample` metrics per site, persisted every minute: `connectionsUp`/`connectionsDown`, `scriptErrors`, `alarmEvalErrors`, `sfBufferDepth`, `deadLetters`, `parkedMessages`, `deployedInstances`/`enabledInstances`/`disabledInstances`, `auditBacklogPending`, `eventLogWriteFailures`. Of these, **only four are charted** in the dashboard's per-site `KpiTrendChart` panel — `connectionsDown`, `scriptErrors`, `sfBufferDepth`, and `deadLetters` (the metrics carried in the shared `KpiMetrics.SiteHealth` Commons catalog and fetched by `Health.razor`'s trend selector). The remaining eight are **sampled and persisted but not (yet) charted** — they are retained in the `KpiSample` store and available for future trend surfaces or ad-hoc query. The synthetic central self-report (`CentralHealthReportLoop.CentralSiteId = "$central"`, fed by the leader-only report loop via `CollectReport($central)`) appears in `GetAllSiteStates()` and is **intentionally sampled into the per-Site KPI store as a real `KpiScopes.Site` series** keyed `$central`; the Central UI deliberately **pins `$central` first** in the per-site trend selector (rendered as "Central Cluster"), so the central cluster's own trends sit alongside the real sites'. (`$central`'s zero-valued connection/instance/S&F metrics reflect the central self-report carrying no site-runtime data, not a defect.) See [Component-KpiHistory.md](Component-KpiHistory.md). +- **KPI History (#26)**: emits `IKpiSampleSource` (`SiteHealthKpiSampleSource`, per-Site) consumed by the KpiHistory recorder (#26). It reads the in-memory `ICentralHealthAggregator.GetAllSiteStates()` (no DB), turning the per-site snapshot — previously sequence-numbered every 30s but discarded — into 12–14 sampled `KpiSample` metrics per site, persisted every minute: `connectionsUp`/`connectionsDown`, `scriptErrors`, `alarmEvalErrors`, `sfBufferDepth`, `deadLetters`, `parkedMessages`, `deployedInstances`/`enabledInstances`/`disabledInstances`, `auditBacklogPending`, `eventLogWriteFailures`, plus the two conditionally-emitted LocalDb replication metrics `localDbOplogBacklog` and `localDbReplicationConnected` (register row 33 — emitted only when the site actually reports a reading, hence 12 for an unreplicated site and 14 for a replicated one; see *KPI History Sampling of Replication State* above). Of these, **only five are charted** in the dashboard's per-site `KpiTrendChart` panel — `connectionsDown`, `scriptErrors`, `sfBufferDepth`, `deadLetters`, and `localDbOplogBacklog` (the metrics carried in the shared `KpiMetrics.SiteHealth` Commons catalog and fetched by `Health.razor`'s trend selector). The rest are **sampled and persisted but not (yet) charted** — they are retained in the `KpiSample` store and available for future trend surfaces or ad-hoc query. The synthetic central self-report (`CentralHealthReportLoop.CentralSiteId = "$central"`, fed by the leader-only report loop via `CollectReport($central)`) appears in `GetAllSiteStates()` and is **intentionally sampled into the per-Site KPI store as a real `KpiScopes.Site` series** keyed `$central`; the Central UI deliberately **pins `$central` first** in the per-site trend selector (rendered as "Central Cluster"), so the central cluster's own trends sit alongside the real sites'. (`$central`'s zero-valued connection/instance/S&F metrics reflect the central self-report carrying no site-runtime data, not a defect.) See [Component-KpiHistory.md](Component-KpiHistory.md). diff --git a/docs/requirements/Component-KpiHistory.md b/docs/requirements/Component-KpiHistory.md index 11f3b15b..7069b2fc 100644 --- a/docs/requirements/Component-KpiHistory.md +++ b/docs/requirements/Component-KpiHistory.md @@ -143,6 +143,17 @@ Reads `ICentralHealthAggregator.GetAllSiteStates()` (in-memory, no DB). Scope: * - `disabledInstances` - `auditBacklogPending` - `eventLogWriteFailures` +- `localDbOplogBacklog` — **conditionally emitted** (register row 33) +- `localDbReplicationConnected` — **conditionally emitted** (register row 33), 0/1 + +The last two are written **only when the site actually reports a reading**. Both underlying health-report +fields are nullable and null means "no reading" (replication not configured, or a failed poll), never +zero — so unlike `auditBacklogPending` above, they are not coerced with `?? 0`. A site therefore emits +**12 metrics when unreplicated and 14 when replicated**, and a gap in these two series truthfully means +"not replicated / not reporting" rather than a flat zero line that would render an unreplicated pair, or +one that cannot read its own oplog, as perfectly healthy. Both fold as **Gauge** in +`KpiMetricAggregationCatalog` (they are instantaneous readings; summing an oplog depth or a 0/1 flag per +hour would be meaningless). `localDbOplogBacklog` is charted on the Health dashboard's per-site trend panel. ## Query + UI diff --git a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/Health.razor b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/Health.razor index bbcd16a6..c2b0f5be 100644 --- a/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/Health.razor +++ b/src/ZB.MOM.WW.ScadaBridge.CentralUI/Components/Pages/Monitoring/Health.razor @@ -170,6 +170,17 @@ IsAvailable="@_sfBufferDepthAvailable" ErrorMessage="@_sfBufferDepthError" /> + @* Register row 33. The point of charting the backlog is retrospective: + the daily purge burst is expected, so what an operator needs is to see + whether a past spike DRAINED — which a point-in-time tile can never + answer. An unreplicated site emits no samples, so this chart is + legitimately empty there rather than a flat, misleading zero line. *@ +
+ +
} @@ -248,6 +259,14 @@ Metrics stale } + @if (IsLocalDbBacklogSustained(state)) + { + + Replication backlog + + } @if (!state.IsOnline && state.LastStatusChangeAt is { } changedAt) { offline since @changedAt.ToString("u") @@ -526,6 +545,27 @@ private string StaleTimeoutDisplay => FormatDuration(HealthOptions.Value.MetricsStaleTimeout); + // Human-friendly rendering of the configured sustained-backlog dwell for the + // "Replication backlog" badge tooltip (e.g. "15 minutes"). + private string BacklogSustainedDisplay => + FormatDuration(HealthOptions.Value.LocalDbBacklogSustainedFor); + + // Register row 33: the sustained LocalDb replication backlog badge. + // + // Reads the anchor the aggregator maintains (LocalDbBacklogAboveSince) rather than the + // instantaneous backlog on the report, because a bare "backlog > threshold" badge would + // fire on EVERY replicated site EVERY day: the site_events retention purge is CDC-captured + // like any other write, so the backlog jumps by the deleted batch size and then drains. + // Healthy is "the backlog comes back down"; only a dwell can express that. + // + // A site that reports no backlog at all (unreplicated — rig site-b/site-c — or a failed + // poll) never gets an anchor, so it can never be badged. That is the whole null-semantics + // requirement: null means "no reading", not zero. + private bool IsLocalDbBacklogSustained(SiteHealthState state) => + state.IsOnline + && state.LocalDbBacklogAboveSince is { } since + && DateTimeOffset.UtcNow - since >= HealthOptions.Value.LocalDbBacklogSustainedFor; + // Online central nodes, from the same ClusterNodes list the Nodes column renders. // Fewer than two means no standby, which disables the manual-failover control. This is // a display-side guard only — AkkaManualFailoverService re-checks against live cluster @@ -616,6 +656,9 @@ private IReadOnlyList? _sfBufferDepthSeries; private bool _sfBufferDepthAvailable = true; private string? _sfBufferDepthError; + private IReadOnlyList? _localDbBacklogSeries; + private bool _localDbBacklogAvailable = true; + private string? _localDbBacklogError; private static bool SiteHasActiveErrors(SiteHealthState state) { @@ -739,6 +782,8 @@ await LoadTrendSeriesAsync(KpiMetrics.SiteHealth.ScriptErrors, siteId, fromUtc, toUtc); (_sfBufferDepthSeries, _sfBufferDepthAvailable, _sfBufferDepthError) = await LoadTrendSeriesAsync(KpiMetrics.SiteHealth.SfBufferDepth, siteId, fromUtc, toUtc); + (_localDbBacklogSeries, _localDbBacklogAvailable, _localDbBacklogError) = + await LoadTrendSeriesAsync(KpiMetrics.SiteHealth.LocalDbOplogBacklog, siteId, fromUtc, toUtc); } finally { diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs index ffedf771..2d11a0df 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs @@ -120,5 +120,22 @@ public static class KpiMetrics /// Summed store-and-forward buffer depth across all buffers. public const string SfBufferDepth = "sfBufferDepth"; + + /// + /// Unacked LocalDb replication oplog depth (register row 33). Emitted ONLY by sites + /// that actually report a backlog — an unreplicated site reports null, and null means + /// "no reading", never zero, so no sample is written at all rather than a fabricated 0. + /// A gap in this series therefore means "not replicated / not reporting", which is + /// exactly what it should mean. + /// + public const string LocalDbOplogBacklog = "localDbOplogBacklog"; + + /// + /// LocalDb replication link state as 1/0 (register row 33). Emitted ONLY when the site + /// reports a value; null (replication not configured) writes no sample. Note that + /// false and null are different states here — false is a configured link that is + /// down, null is no link configured — which is why this cannot be coerced to 0. + /// + public const string LocalDbReplicationConnected = "localDbReplicationConnected"; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/CentralHealthAggregator.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/CentralHealthAggregator.cs index 811a6e5f..450054a2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/CentralHealthAggregator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/CentralHealthAggregator.cs @@ -50,7 +50,8 @@ public class CentralHealthAggregator : BackgroundService, ICentralHealthAggregat FirstSeenAt = now, LastSequenceNumber = report.SequenceNumber, IsOnline = true, - IsMetricsStale = false + IsMetricsStale = false, + LocalDbBacklogAboveSince = NextBacklogAboveSince(null, report, now) // LastStatusChangeAt stays null: first observation is not a flip. }; @@ -84,7 +85,9 @@ public class CentralHealthAggregator : BackgroundService, ICentralHealthAggregat IsMetricsStale = false, // Stamp a flip only when this report brings the site back online; // a report from an already-online site leaves the timestamp be. - LastStatusChangeAt = existing.IsOnline ? existing.LastStatusChangeAt : now + LastStatusChangeAt = existing.IsOnline ? existing.LastStatusChangeAt : now, + LocalDbBacklogAboveSince = + NextBacklogAboveSince(existing.LocalDbBacklogAboveSince, report, now) }; if (_siteStates.TryUpdate(report.SiteId, updated, existing)) @@ -320,4 +323,43 @@ public class CentralHealthAggregator : BackgroundService, ICentralHealthAggregat } } } + + /// + /// Advances the sustained-backlog anchor for one incoming report (register row 33). + /// + /// + /// Three cases, and the null one is the subtle one: + /// + /// + /// Backlog null — no reading. An unreplicated site (rig site-b/site-c) reports this + /// forever, and a replicated site reports it when a poll fails. Carry the previous anchor + /// through UNCHANGED: null is not zero, so it is neither evidence of a drain (which would + /// justify clearing) nor of a backlog (which would justify starting one). An unreplicated + /// site therefore never accumulates an anchor and can never be badged. + /// + /// + /// Backlog above the threshold — start the clock if it is not already running. Keeping the + /// ORIGINAL start instant is what makes the window sustained rather than sliding. + /// + /// + /// Backlog at or below the threshold — clear. This is the daily purge burst's exit path: + /// the spike drains within a few reports and the anchor is discarded long before the dwell + /// elapses, so the expected burst never badges the site. + /// + /// + /// + /// The anchor currently held for this site, if any. + /// The incoming health report. + /// The current time. + /// The anchor to store. + private DateTimeOffset? NextBacklogAboveSince( + DateTimeOffset? current, SiteHealthReport report, DateTimeOffset now) + { + if (report.LocalDbOplogBacklog is not { } backlog) + { + return current; + } + + return backlog > _options.LocalDbBacklogThreshold ? current ?? now : null; + } } diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptions.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptions.cs index 645da52d..ee4fba66 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptions.cs @@ -32,4 +32,32 @@ public class HealthMonitoringOptions /// (~4 missed 30s reports). /// public TimeSpan MetricsStaleTimeout { get; set; } = TimeSpan.FromMinutes(2); + + /// + /// Unacked LocalDb replication oplog depth above which a site is considered backlogged + /// (register row 33). Crossing this alone shows nothing — the backlog must also stay + /// above it for before the Health dashboard + /// badges the site. Must be positive. Default: 1000 entries. + /// + public long LocalDbBacklogThreshold { get; set; } = 1000; + + /// + /// How long the oplog backlog must stay continuously above + /// before the Health dashboard badges the site + /// (register row 33). + /// + /// + /// This dwell time is the whole point of the signal, not a refinement of it. A bare + /// threshold would fire every single day on a healthy replicated site: the `site_events` + /// retention purge and the storage-cap trim are CDC-captured like any other write, so the + /// backlog jumps by the deleted batch size and then drains as the peer acks it. That spike + /// is expected and documented (see docs/deployment/topology-guide.md → *Reading the + /// replication backlog*). Healthy means the backlog RETURNS TO ~0; unhealthy means it + /// stays up. Only a sustained window can tell those apart, so the default is set well + /// clear of a purge drain. Must be positive and no shorter than + /// — a window shorter than one report interval could be + /// satisfied by a single sample, which is exactly the bare threshold this exists to avoid. + /// Default: 15 minutes. + /// + public TimeSpan LocalDbBacklogSustainedFor { get; set; } = TimeSpan.FromMinutes(15); } diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptionsValidator.cs index 64df0047..f0e69950 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthMonitoringOptionsValidator.cs @@ -64,5 +64,29 @@ public sealed class HealthMonitoringOptionsValidator : OptionsValidatorBase= ReportInterval ({options.ReportInterval}): a stale window shorter than one " + "report interval would flag every healthy site, since reports cannot arrive faster than " + "they are emitted."); + + builder.RequireThat(options.LocalDbBacklogThreshold > 0, + $"ScadaBridge:HealthMonitoring:LocalDbBacklogThreshold must be positive " + + $"(was {options.LocalDbBacklogThreshold}); a zero/negative threshold would badge every " + + "replicated site permanently, since an idle backlog of 0 already sits at or above it."); + + builder.RequireThat(options.LocalDbBacklogSustainedFor > TimeSpan.Zero, + $"ScadaBridge:HealthMonitoring:LocalDbBacklogSustainedFor must be a positive duration " + + $"(was {options.LocalDbBacklogSustainedFor}); it is the dwell time a backlog must stay " + + "above the threshold before the site is badged."); + + // A dwell shorter than one report interval can be satisfied by a single sample, + // degrading the sustained check back into the bare threshold it exists to replace — + // which would fire on every daily purge burst. Same De Morgan guard shape as the + // pairs above: silent unless BOTH fields are positive and the dwell is too short, + // so the dedicated positive-value checks own that failure. + builder.RequireThat( + !(options.ReportInterval > TimeSpan.Zero + && options.LocalDbBacklogSustainedFor > TimeSpan.Zero + && options.LocalDbBacklogSustainedFor < options.ReportInterval), + $"ScadaBridge:HealthMonitoring:LocalDbBacklogSustainedFor ({options.LocalDbBacklogSustainedFor}) " + + $"must be >= ReportInterval ({options.ReportInterval}): a dwell shorter than one report " + + "interval can be satisfied by a single sample, which turns the sustained check back into " + + "a bare threshold and re-introduces the daily purge-burst false positive it exists to avoid."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/SiteHealthKpiSampleSource.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/SiteHealthKpiSampleSource.cs index 501cb967..c5b0cef6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/SiteHealthKpiSampleSource.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/Kpi/SiteHealthKpiSampleSource.cs @@ -49,6 +49,8 @@ public sealed class SiteHealthKpiSampleSource : IKpiSampleSource private const string MetricDisabledInstances = "disabledInstances"; private const string MetricAuditBacklogPending = "auditBacklogPending"; private const string MetricEventLogWriteFailures = KpiMetrics.SiteHealth.EventLogWriteFailures; + private const string MetricLocalDbOplogBacklog = KpiMetrics.SiteHealth.LocalDbOplogBacklog; + private const string MetricLocalDbReplicationConnected = KpiMetrics.SiteHealth.LocalDbReplicationConnected; private readonly ICentralHealthAggregator _aggregator; @@ -136,6 +138,32 @@ public sealed class SiteHealthKpiSampleSource : IKpiSampleSource report.SiteAuditBacklog?.PendingCount ?? 0)); samples.Add(Sample( capturedAtUtc, MetricEventLogWriteFailures, siteId, report.SiteEventLogWriteFailures)); + + // Register row 33: LocalDb replication. Both fields are NULLABLE and null is + // load-bearing — it means "no reading" (replication not configured, as on rig + // site-b/site-c, or a failed poll), NOT zero. So unlike the audit-backlog line + // above, these are CONDITIONALLY emitted rather than coerced with `?? 0`: + // + // - A fabricated 0 backlog would render a pair that cannot read its own oplog, + // or one that simply isn't replicated, as perfectly healthy — and would do it + // once a minute, forever, burying the real series for the sites that are. + // - For the connected flag, `false` and null are genuinely different states: a + // configured link that is DOWN versus no link at all. Collapsing null to 0 + // would alarm on every unreplicated site permanently. + // + // The cost is that a site's sample count varies (12 unreplicated, 14 replicated). + // That is the intended shape: a GAP in these two series means "not replicated / not + // reporting", which is a truthful reading, whereas a flat 0 line is a lie. + if (report.LocalDbOplogBacklog is { } backlog) + { + samples.Add(Sample(capturedAtUtc, MetricLocalDbOplogBacklog, siteId, backlog)); + } + + if (report.LocalDbReplicationConnected is { } connected) + { + samples.Add(Sample( + capturedAtUtc, MetricLocalDbReplicationConnected, siteId, connected ? 1 : 0)); + } } private static KpiSample Sample( diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthState.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthState.cs index 7767aef2..8c02ee6a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthState.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/SiteHealthState.cs @@ -67,6 +67,28 @@ public sealed record SiteHealthState /// public bool IsMetricsStale { get; init; } + /// + /// Gets the instant this site's LocalDb replication oplog backlog first rose above + /// and stayed there, or + /// null if the backlog is not currently above it (register row 33). + /// + /// + /// The anchor for the *sustained* backlog badge: the badge shows only once + /// has elapsed since this + /// instant. Holding a start time rather than a boolean is what makes the signal survive the + /// expected daily purge burst — that spike raises the backlog for a drain's worth of + /// reports and then clears this back to null, never reaching the dwell. + /// + /// Null is set two different ways and they mean different things. A report whose backlog is + /// at or below the threshold CLEARS it (the backlog demonstrably drained). A report whose + /// backlog is null — an unreplicated site, or a failed poll — LEAVES IT UNCHANGED: + /// null means "no reading", never zero, so a node that stops reporting mid-climb must not + /// silently reset the clock, and an unreplicated site (which reports null forever) never + /// starts one. + /// + /// + public DateTimeOffset? LocalDbBacklogAboveSince { get; init; } + /// /// Gets the instant of the last online↔offline flip, or null if the /// site has never changed status since it was first observed. Answers "when diff --git a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Pages/HealthPageTests.cs b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Pages/HealthPageTests.cs index c7905911..4ee067cf 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Pages/HealthPageTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Pages/HealthPageTests.cs @@ -346,6 +346,103 @@ public class HealthPageTests : BunitContext }); } + // ------------------------------------------------------------------ + // Sustained LocalDb replication backlog badge (register row 33) + // ------------------------------------------------------------------ + + [Fact] + public void Renders_ReplicationBacklogBadge_WhenBacklogSustainedPastTheDwell() + { + SeedStates(new SiteHealthState + { + SiteId = "site-a", + IsOnline = true, + LastHeartbeatAt = DateTimeOffset.UtcNow, + // Default dwell is 15 minutes; 30 minutes above threshold is well past it. + LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-30), + }); + + var cut = RenderHealthPage(); + + cut.WaitForAssertion(() => + { + Assert.Contains("data-test=\"localdb-backlog-badge\"", cut.Markup); + Assert.Contains("Replication backlog", cut.Markup); + }); + } + + [Fact] + public void DoesNotRender_ReplicationBacklogBadge_DuringAPurgeBurst() + { + // THE false-positive this badge is shaped to avoid. The daily site_events purge is + // CDC-captured, so the backlog spikes and then drains. Two minutes above the + // threshold is a purge burst, not a fault, and must stay silent — otherwise the + // badge fires on every replicated site every day and gets ignored. + SeedStates(new SiteHealthState + { + SiteId = "site-a", + IsOnline = true, + LastHeartbeatAt = DateTimeOffset.UtcNow, + LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-2), + }); + + var cut = RenderHealthPage(); + + cut.WaitForAssertion(() => Assert.Contains("site-a", cut.Markup)); + Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup); + } + + [Fact] + public void DoesNotRender_ReplicationBacklogBadge_ForAnUnreplicatedSite() + { + // Rig site-b/site-c report a null backlog forever, so the aggregator never gives them + // an anchor. A deliberately-unreplicated site must never be badged. + SeedStates(new SiteHealthState + { + SiteId = "site-b", + IsOnline = true, + LastHeartbeatAt = DateTimeOffset.UtcNow, + LocalDbBacklogAboveSince = null, + }); + + var cut = RenderHealthPage(); + + cut.WaitForAssertion(() => Assert.Contains("site-b", cut.Markup)); + Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup); + } + + [Fact] + public void DoesNotRender_ReplicationBacklogBadge_ForAnOfflineSite() + { + // An offline site's last-known anchor is stale by definition; the Offline badge is the + // signal that matters, and stacking a backlog warning on top is noise. + SeedStates(new SiteHealthState + { + SiteId = "site-a", + IsOnline = false, + LastHeartbeatAt = DateTimeOffset.UtcNow.AddMinutes(-30), + LocalDbBacklogAboveSince = DateTimeOffset.UtcNow.AddMinutes(-30), + }); + + var cut = RenderHealthPage(); + + cut.WaitForAssertion(() => Assert.Contains("site-a", cut.Markup)); + Assert.DoesNotContain("data-test=\"localdb-backlog-badge\"", cut.Markup); + } + + [Fact] + public void Renders_LocalDbBacklogTrendChart() + { + // The retrospective half of row 33: a point-in-time tile can never answer "did that + // spike drain?", which is the only question that distinguishes healthy from faulty. + SeedSites("site-a"); + + var cut = RenderHealthPage(); + + cut.WaitForAssertion(() => + Assert.Contains("LocalDb Replication Backlog", cut.Markup)); + } + [Fact] public void Renders_OfflineSince_ForOfflineSiteWithStatusChange() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiMetricAggregationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiMetricAggregationTests.cs index a3cb9b08..d6df9cbf 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiMetricAggregationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Kpi/KpiMetricAggregationTests.cs @@ -47,6 +47,12 @@ public class KpiMetricAggregationTests (KpiSources.SiteHealth, "disabledInstances"), (KpiSources.SiteHealth, "auditBacklogPending"), (KpiSources.SiteHealth, "eventLogWriteFailures"), + // Register row 33. Both are instantaneous readings, so both fold as Gauge (the + // default) — summing an oplog depth per hour would be meaningless, and summing a + // 0/1 link flag doubly so. Both are CONDITIONALLY emitted: a site that reports no + // reading writes no sample rather than a fabricated 0. + (KpiSources.SiteHealth, "localDbOplogBacklog"), + (KpiSources.SiteHealth, "localDbReplicationConnected"), }; // The (source, metric) pairs that must fold as Rate (sum-per-hour). @@ -76,6 +82,8 @@ public class KpiMetricAggregationTests (KpiSources.SiteHealth, "connectionsDown"), (KpiSources.SiteHealth, "sfBufferDepth"), (KpiSources.SiteHealth, "deployedInstances"), + (KpiSources.SiteHealth, "localDbOplogBacklog"), + (KpiSources.SiteHealth, "localDbReplicationConnected"), }; public static IEnumerable AllEmittedCases() => diff --git a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthMonitoringOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthMonitoringOptionsValidatorTests.cs index c4e7fbb5..91406d83 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthMonitoringOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthMonitoringOptionsValidatorTests.cs @@ -93,4 +93,61 @@ public class HealthMonitoringOptionsValidatorTests Assert.True(result.Failed); Assert.Contains("MetricsStaleTimeout", result.FailureMessage); } + + // ------------------------------------------------------------------ + // LocalDb sustained-backlog badge (register row 33) + // ------------------------------------------------------------------ + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void NonPositiveLocalDbBacklogThreshold_IsRejected(long threshold) + { + // An idle backlog of 0 already sits at or above a zero/negative threshold, so this + // would badge every replicated site permanently. + var result = Validate(new HealthMonitoringOptions { LocalDbBacklogThreshold = threshold }); + + Assert.True(result.Failed); + Assert.Contains("LocalDbBacklogThreshold", result.FailureMessage); + } + + [Fact] + public void ZeroLocalDbBacklogSustainedFor_IsRejected() + { + var result = Validate(new HealthMonitoringOptions + { + LocalDbBacklogSustainedFor = TimeSpan.Zero, + }); + + Assert.True(result.Failed); + Assert.Contains("LocalDbBacklogSustainedFor", result.FailureMessage); + } + + [Fact] + public void LocalDbBacklogSustainedFor_ShorterThanReportInterval_IsRejected() + { + // A dwell shorter than one report interval can be satisfied by a single sample, which + // collapses the sustained check back into the bare threshold it replaced — and that + // fires on every daily purge burst. + var result = Validate(new HealthMonitoringOptions + { + ReportInterval = TimeSpan.FromSeconds(30), + LocalDbBacklogSustainedFor = TimeSpan.FromSeconds(10), + }); + + Assert.True(result.Failed); + Assert.Contains("LocalDbBacklogSustainedFor", result.FailureMessage); + } + + [Fact] + public void DefaultSustainedWindow_ComfortablyOutlastsAPurgeDrain() + { + // Guards the intent of the default rather than its digits: the dwell must be long + // enough that an expected purge spike drains inside it, or the badge becomes daily + // noise and gets ignored. + var options = new HealthMonitoringOptions(); + + Assert.True(options.LocalDbBacklogSustainedFor >= TimeSpan.FromMinutes(5)); + Assert.True(options.LocalDbBacklogSustainedFor >= options.ReportInterval); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/SiteHealthKpiSampleSourceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/SiteHealthKpiSampleSourceTests.cs index dc8f4176..8bba11bb 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/SiteHealthKpiSampleSourceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/Kpi/SiteHealthKpiSampleSourceTests.cs @@ -99,11 +99,95 @@ public class SiteHealthKpiSampleSourceTests Assert.Equal(9, byMetric["auditBacklogPending"]); Assert.Equal(11, byMetric["eventLogWriteFailures"]); - // All 12 metrics emitted, exactly once each, for the one populated site. + // 12 metrics for an UNREPLICATED site — this fixture leaves both LocalDb replication + // fields null, and register row 33 emits those two conditionally rather than coercing + // null to 0. A replicated site emits 14; see the LocalDb tests below. Assert.Equal(12, byMetric.Count); Assert.Equal(12, samples.Count); } + // ------------------------------------------------------------------ + // LocalDb replication metrics (register row 33) + // ------------------------------------------------------------------ + + [Fact] + public async Task CollectAsync_NullLocalDbFields_EmitsNeitherLocalDbMetric() + { + // THE null-semantics case, and the reason these two are not `?? 0` like the audit + // backlog above. Null means "no reading" — an unreplicated site (rig site-b/site-c) or + // a failed poll. A fabricated 0 backlog would render a pair that cannot read its own + // oplog, or one deliberately not replicated, as perfectly healthy once a minute forever. + var report = MinimalReport("site-a") with + { + LocalDbOplogBacklog = null, + LocalDbReplicationConnected = null, + }; + var aggregator = new StubAggregator + { + States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } }, + }; + + var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt); + + Assert.DoesNotContain(samples, s => s.Metric == "localDbOplogBacklog"); + Assert.DoesNotContain(samples, s => s.Metric == "localDbReplicationConnected"); + Assert.Equal(12, samples.Count); + } + + [Fact] + public async Task CollectAsync_ReplicatedSite_EmitsBothLocalDbMetrics() + { + var report = MinimalReport("site-a") with + { + LocalDbOplogBacklog = 4242, + LocalDbReplicationConnected = true, + }; + var aggregator = new StubAggregator + { + States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } }, + }; + + var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt); + + Assert.Equal(4242, samples.Single(s => s.Metric == "localDbOplogBacklog").Value); + Assert.Equal(1, samples.Single(s => s.Metric == "localDbReplicationConnected").Value); + Assert.Equal(14, samples.Count); + } + + [Fact] + public async Task CollectAsync_ZeroBacklog_IsStillEmitted() + { + // A real, measured 0 is genuinely different from "no reading" and must be recorded — + // otherwise a healthy replicated site would be indistinguishable from an unreplicated + // one, and the drain half of a purge burst would never appear in the series. + var report = MinimalReport("site-a") with { LocalDbOplogBacklog = 0 }; + var aggregator = new StubAggregator + { + States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } }, + }; + + var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt); + + Assert.Equal(0, samples.Single(s => s.Metric == "localDbOplogBacklog").Value); + } + + [Fact] + public async Task CollectAsync_DisconnectedReplication_EmitsZeroNotAbsent() + { + // false and null are different states: a configured link that is DOWN versus no link + // at all. Only the former is a fault, and it must show up as a 0 sample rather than a + // gap, or a broken link would look exactly like an unreplicated site. + var report = MinimalReport("site-a") with { LocalDbReplicationConnected = false }; + var aggregator = new StubAggregator + { + States = { ["site-a"] = new SiteHealthState { SiteId = "site-a", LatestReport = report } }, + }; + + var samples = await new SiteHealthKpiSampleSource(aggregator).CollectAsync(CapturedAt); + + Assert.Equal(0, samples.Single(s => s.Metric == "localDbReplicationConnected").Value); + } + [Fact] public async Task CollectAsync_NullAuditBacklog_EmitsZeroAuditBacklogPending() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/LocalDbBacklogSustainedTests.cs b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/LocalDbBacklogSustainedTests.cs new file mode 100644 index 00000000..014eb3c0 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/LocalDbBacklogSustainedTests.cs @@ -0,0 +1,176 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests; + +/// +/// Register row 33: the sustained-backlog anchor behind the Health dashboard's +/// "Replication backlog" badge. +/// +/// +/// The design constraint these tests exist to pin: a bare "backlog > threshold" badge is +/// unusable, because the daily `site_events` retention purge is CDC-captured like any other +/// write and jumps the backlog by the deleted batch size before it drains. That spike is +/// expected and documented. So the aggregator holds the instant the backlog FIRST rose above +/// the threshold, and the badge only appears once the configured dwell has elapsed — healthy +/// means the backlog comes back down, which no point-in-time check can express. +/// +public sealed class LocalDbBacklogSustainedTests +{ + private const long Threshold = 1000; + + [Fact] + public void BacklogAboveThreshold_StartsTheClock() + { + var (aggregator, time) = NewAggregator(); + var start = time.GetUtcNow(); + + aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); + + Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void BacklogStayingAbove_KeepsTheOriginalStartInstant() + { + // The window must be sustained, not sliding: if each above-threshold report reset the + // anchor, the dwell would never elapse and the badge could never appear at all. + var (aggregator, time) = NewAggregator(); + var start = time.GetUtcNow(); + + aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); + time.Advance(TimeSpan.FromMinutes(10)); + aggregator.ProcessReport(Report("site-a", 2, backlog: Threshold + 5000)); + + Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void BacklogAtOrBelowThreshold_ClearsTheClock() + { + // The purge-burst exit path. A spike that drains must discard the anchor well before + // the dwell elapses, so the expected daily burst never badges the site. + var (aggregator, time) = NewAggregator(); + + aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 50_000)); + time.Advance(TimeSpan.FromMinutes(2)); + aggregator.ProcessReport(Report("site-a", 2, backlog: 0)); + + Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void BacklogExactlyAtThreshold_DoesNotStartTheClock() + { + // The comparison is strictly greater-than, so the threshold is the highest ACCEPTABLE + // depth rather than the lowest alarming one. + var (aggregator, _) = NewAggregator(); + + aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold)); + + Assert.Null(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void UnreplicatedSite_NeverAccumulatesAnAnchor() + { + // Rig site-b/site-c: a null backlog forever. Null means "no reading", never zero, so + // it must never start a clock — otherwise every unreplicated site would eventually be + // badged for a backlog it does not have. + var (aggregator, time) = NewAggregator(); + + aggregator.ProcessReport(Report("site-b", 1, backlog: null)); + time.Advance(TimeSpan.FromHours(6)); + aggregator.ProcessReport(Report("site-b", 2, backlog: null)); + + Assert.Null(aggregator.GetSiteState("site-b")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void NullBacklogMidClimb_LeavesTheAnchorUntouched() + { + // A failed poll on a replicated node reports null. That is not evidence the backlog + // drained, so it must not clear the clock — a node that goes blind mid-climb would + // otherwise silently reset its own dwell and never be badged. + var (aggregator, time) = NewAggregator(); + var start = time.GetUtcNow(); + + aggregator.ProcessReport(Report("site-a", 1, backlog: Threshold + 1)); + time.Advance(TimeSpan.FromMinutes(5)); + aggregator.ProcessReport(Report("site-a", 2, backlog: null)); + + Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void FirstReportForANewSite_CanStartTheClock() + { + // The registration path is a separate branch from the update path; a site whose very + // first report is already backlogged must not be missed. + var (aggregator, time) = NewAggregator(); + var start = time.GetUtcNow(); + + aggregator.ProcessReport(Report("brand-new", 1, backlog: Threshold + 1)); + + Assert.Equal(start, aggregator.GetSiteState("brand-new")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void ThresholdIsConfigurable() + { + var (aggregator, _) = NewAggregator(threshold: 10); + + aggregator.ProcessReport(Report("site-a", 1, backlog: 11)); + + Assert.NotNull(aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + [Fact] + public void StaleReport_DoesNotDisturbTheAnchor() + { + // Out-of-order reports are rejected wholesale by the sequence-number guard; the anchor + // must ride along with that rejection rather than being updated from stale data. + var (aggregator, time) = NewAggregator(); + var start = time.GetUtcNow(); + + aggregator.ProcessReport(Report("site-a", 5, backlog: Threshold + 1)); + time.Advance(TimeSpan.FromMinutes(5)); + aggregator.ProcessReport(Report("site-a", 2, backlog: 0)); + + Assert.Equal(start, aggregator.GetSiteState("site-a")!.LocalDbBacklogAboveSince); + } + + private static (CentralHealthAggregator Aggregator, TestTimeProvider Time) NewAggregator( + long threshold = Threshold) + { + var time = new TestTimeProvider(DateTimeOffset.UtcNow); + var opts = new HealthMonitoringOptions + { + OfflineTimeout = TimeSpan.FromSeconds(60), + LocalDbBacklogThreshold = threshold, + }; + var aggregator = new CentralHealthAggregator( + Options.Create(opts), NullLogger.Instance, time); + return (aggregator, time); + } + + private static SiteHealthReport Report(string siteId, long seq, long? backlog) => + new( + SiteId: siteId, + SequenceNumber: seq, + ReportTimestamp: DateTimeOffset.UtcNow, + DataConnectionStatuses: new Dictionary(), + TagResolutionCounts: new Dictionary(), + ScriptErrorCount: 0, + AlarmEvaluationErrorCount: 0, + StoreAndForwardBufferDepths: new Dictionary(), + DeadLetterCount: 0, + DeployedInstanceCount: 0, + EnabledInstanceCount: 0, + DisabledInstanceCount: 0) + { + LocalDbOplogBacklog = backlog, + }; +}