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-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/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 ... /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/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/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-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..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,75 @@ 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` +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. @@ -122,4 +191,4 @@ These tiles are **point-in-time** like the Notification Outbox and Site Call Aud - **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/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.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/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index b2431fe4..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; @@ -93,6 +94,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 ---------------- /// @@ -109,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; @@ -127,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 = @@ -186,11 +264,76 @@ public static class ScadaBridgeTelemetry /// Records one resync whose ack window expired. public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1); - /// Records that a site connection opened (increments the up-count gauge). - public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); + /// + /// 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 closed (decrements the up-count gauge). - public static void SiteConnectionClosed() => Interlocked.Decrement(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 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.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.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/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/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs index 30c3091f..96b69c07 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Program.cs @@ -200,6 +200,27 @@ try ?? builder.WebHost.GetSetting("urls") ?? "http://+:5000"; var httpPorts = ParseHttpBindPorts(httpUrls); + + // Register row 56: central's dedicated Prometheus listener, OPT-IN. + // + // /metrics is already mapped on central (app.MapZbMetrics() below) and, because it + // rides endpoint routing, it is served on every listener this node binds — so it has + // always been reachable on the HTTP surface (:5000 on the rig). What central lacked + // was the site branch's dedicated MetricsPort, which is what the established + // 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 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.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.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.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, + }; +} 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(); 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(),