From b5f120c370789e17d479c26b1fc7146e7a29cf8f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Sat, 15 Aug 2026 12:53:12 -0400 Subject: [PATCH] feat(observability): honest stream-subscriber name + real data-connection gauges (row 54) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scadabridge_site_connection_up never counted data connections. It counts open central->site gRPC subscriber streams (Debug View + the site-wide alarm feed), so it correctly reads 0 on a healthy site nobody is watching — and the name invited reading it as device health. That misreading has already cost a live debugging session: register row 48's 'connection_up is 0 on all six site nodes' was healthy steady state, not the fault it was diagnosed as. Two halves, per the row: RENAME. scadabridge.site.connection.up -> scadabridge.site.stream.subscribers.active (helpers SiteConnectionOpened/Closed -> SiteStreamSubscriberOpened/Closed), and the instrument gains the unit:"1" the other gauges already carried. Every reference updated: the gRPC server, both test suites, Component-Communication.md, components/Commons.md, docker/README.md and docker/seed-live-probe.sh. A test now pins the new name AND asserts the old one is gone, so it cannot creep back. NEW. scadabridge.site.data_connections.connected / .total — the first exported data-connection signal (the state existed on the health report and in the CLI, but never as a metric). connected < total is the alert condition; neither number alone separates 'a device is down' from 'nothing is configured', which is why the denominator ships with the numerator. Sourcing is deliberate on two counts. It is fed by the DCL at its four state transitions rather than derived from the health report, so the metric survives a dead report pipeline — the exact failure the 'metrics stale' flag exists to catch, and a derived gauge would go blind at the same moment. And it uses a static registry reached through ScadaBridgeTelemetry (the SiteConnectionOpened precedent) rather than a DI-injected holder: the DCL actors already take ISiteHealthCollector through their constructors, and threading a second dependency through both would ripple into every test that builds them. It also keeps SiteHealthCollector untouched, which matters right now — another agent is concurrently editing that file for the tag-resolution path. Removal is wired too: an undeployed connection leaves the registry, so it stops counting toward the total instead of latching connected < total on forever. Both gauges read 0 on the standby (connections open on the active node only) — healthy, documented, and called out in the docs so it does not become the next site_connection_up. No ObservedMeters allowlist change: all three instruments ride the existing ZB.MOM.WW.ScadaBridge meter. (An unlisted meter exports nothing, silently.) Tests: Commons 711 passed (9 new, delta-based against a baseline with unique per-test connection names, so they do not inherit the register row 32 QueueDepthGaugeTests static-state flake); Communication 717; DCL 319. --- docker/seed-live-probe.sh | 15 +- docs/components/Commons.md | 2 +- docs/requirements/Component-Communication.md | 2 +- .../Component-HealthMonitoring.md | 29 +++ .../Observability/ScadaBridgeTelemetry.cs | 140 ++++++++++++- .../Grpc/SiteStreamGrpcServer.cs | 8 +- .../Actors/DataConnectionActor.cs | 10 + .../Actors/DataConnectionManagerActor.cs | 5 + .../Observability/DataConnectionGaugeTests.cs | 190 ++++++++++++++++++ .../Grpc/SiteStreamGrpcServerTests.cs | 14 +- .../SiteAlarmLiveCacheTelemetryTests.cs | 2 +- .../ScadaBridgeTelemetryTests.cs | 4 +- 12 files changed, 389 insertions(+), 32 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs diff --git a/docker/seed-live-probe.sh b/docker/seed-live-probe.sh index 54e738a9..c5841d76 100755 --- a/docker/seed-live-probe.sh +++ b/docker/seed-live-probe.sh @@ -112,9 +112,14 @@ echo "" echo "While a stream is open, the site node serving it reports both signals" echo "(metrics are on container port 8084, and the containers have no curl/wget):" echo " docker exec scadabridge-site-a-a bash -c 'exec 3<>/dev/tcp/localhost/8084 && printf \"GET /metrics HTTP/1.0\\r\\nHost: localhost\\r\\n\\r\\n\" >&3 && cat <&3' \\" -echo " | grep -E 'site_connection_up|stream_batch_size'" +echo " | grep -E 'stream_subscribers_active|stream_batch_size|data_connections'" echo "" -echo "NOTE: scadabridge_site_connection_up counts OPEN CENTRAL->SITE gRPC SUBSCRIBER" -echo "STREAMS, not OPC UA data connections. It reads 0 on a perfectly healthy site" -echo "with no subscriber attached — use the health report's dataConnectionEndpoints" -echo "for data-connection health." +echo "NOTE: scadabridge_site_stream_subscribers_active (renamed from the misleading" +echo "scadabridge_site_connection_up, register row 54) counts OPEN CENTRAL->SITE gRPC" +echo "SUBSCRIBER STREAMS, not OPC UA data connections. It reads 0 on a perfectly healthy" +echo "site with no subscriber attached — that is correct, not a fault." +echo "" +echo "For real data-connection health use the gauges added by the same row:" +echo " scadabridge_site_data_connections_connected / _total" +echo "connected < total is the alert condition. Both read 0 on the STANDBY node, since" +echo "connections are opened by the deployment path on the active node only." diff --git a/docs/components/Commons.md b/docs/components/Commons.md index 872f1bd6..394b4551 100644 --- a/docs/components/Commons.md +++ b/docs/components/Commons.md @@ -295,7 +295,7 @@ public record AlarmStateChanged( ### Observability -`Observability/ScadaBridgeTelemetry` defines the singleton `Meter` named `ZB.MOM.WW.ScadaBridge` and the application-wide instrument definitions. Components call the static emit helpers (`RecordDeploymentApplied`, `SiteConnectionOpened`, etc.) rather than creating their own meters. Instruments are no-ops until an OTel listener attaches, so uninstrumented hosts pay no overhead. +`Observability/ScadaBridgeTelemetry` defines the singleton `Meter` named `ZB.MOM.WW.ScadaBridge` and the application-wide instrument definitions. Components call the static emit helpers (`RecordDeploymentApplied`, `SiteStreamSubscriberOpened`, etc.) rather than creating their own meters. Instruments are no-ops until an OTel listener attaches, so uninstrumented hosts pay no overhead. ## Usage diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 45512df9..479d0264 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -108,7 +108,7 @@ The streaming protocol is defined in `sitestream.proto` (`src/ZB.MOM.WW.ScadaBri - **Service**: `SiteStreamService` — hosted on each site node by `SiteStreamGrpcServer` — exposes six RPCs. Two are real-time **server-streaming** subscriptions (`SubscribeInstance`, `SubscribeSite`); the other four are **unary request/response** calls added by the Audit Log (#23) and Site Call Audit (#22) components. These are distinct from the command/control gRPC services (`CentralControlService` on central, `SiteCommandService` on the site) — `SiteStreamService` is the streaming + audit-pull surface, not the command/control channel: - `SubscribeInstance(InstanceStreamRequest) returns (stream SiteStreamEvent)` — the per-instance real-time debug stream (§6). - - `SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent)` — the **site-wide, alarm-only** aggregated stream (§6.1) added for the operator Alarm Summary live cache. Additive (new RPC + new `SiteStreamRequest { correlation_id }` message; no field renumbering). The server handler mirrors `SubscribeInstance` — but with **its own, larger send channel** (`GrpcSiteAlarmStreamChannelCapacity`, default 20 000, `DropOldest`; WP2.3): sharing the Debug View's 1000-slot channel meant an alarm burst during a WAN stall silently evicted operator-visible transitions to make room for diagnostics traffic. Drops are tagged `stream=site-alarms` on `scadabridge.site.stream.events_dropped`. Otherwise identical — same `GrpcMaxConcurrentStreams`/`GrpcMaxStreamLifetime` limits, `SiteConnectionOpened/Closed` telemetry, and `StreamRelayActor` mapping — but subscribes via `SiteStreamManager.SubscribeSiteAlarms` (all instances, `AlarmStateChanged` only; attribute events dropped, no per-instance filter). + - `SubscribeSite(SiteStreamRequest) returns (stream SiteStreamEvent)` — the **site-wide, alarm-only** aggregated stream (§6.1) added for the operator Alarm Summary live cache. Additive (new RPC + new `SiteStreamRequest { correlation_id }` message; no field renumbering). The server handler mirrors `SubscribeInstance` — but with **its own, larger send channel** (`GrpcSiteAlarmStreamChannelCapacity`, default 20 000, `DropOldest`; WP2.3): sharing the Debug View's 1000-slot channel meant an alarm burst during a WAN stall silently evicted operator-visible transitions to make room for diagnostics traffic. Drops are tagged `stream=site-alarms` on `scadabridge.site.stream.events_dropped`. Otherwise identical — same `GrpcMaxConcurrentStreams`/`GrpcMaxStreamLifetime` limits, `SiteStreamSubscriberOpened/Closed` telemetry, and `StreamRelayActor` mapping — but subscribes via `SiteStreamManager.SubscribeSiteAlarms` (all instances, `AlarmStateChanged` only; attribute events dropped, no per-instance filter). - `IngestAuditEvents(AuditEventBatch) returns (IngestAck)` — a legacy central-side ingest surface on the *site*-hosted service; it is **dead in the shipped topology** because no site ever dials a site for ingest. The production audit-telemetry *push* path is site→central gRPC to `CentralControlService`, which routes the batch to the central `AuditLogIngestActor` proxy and returns the accepted `EventId`s. - `IngestCachedTelemetry(CachedTelemetryBatch) returns (IngestAck)` — ingest receiving surface for the combined cached-call telemetry packet (audit row + `SiteCalls` operational upsert written in one transaction). - `PullAuditEvents(PullAuditEventsRequest) returns (PullAuditEventsResponse)` — central→site **reconciliation pull** for the Audit Log self-heal feed; the site serves `Pending`/`Forwarded` rows from its `ISiteAuditQueue`. **At-least-once (WP2.3):** rows are NOT retired when served. The site retires (`Reconciled`) everything at or before the cursor of the **next** pull, because that cursor is central's only proof of receipt — a fault between the response leaving the site and central committing it re-serves the rows instead of losing them (central dedups on `EventId`). The request carries an additive composite-keyset `after_id` (field 3) mirroring `PullSiteCallsRequest`; with it the read is a strict `(OccurredAtUtc, EventId)` keyset and the retirement is exact, without it the legacy inclusive `>=` read applies and only rows strictly older than the cursor instant are provably received. diff --git a/docs/requirements/Component-HealthMonitoring.md b/docs/requirements/Component-HealthMonitoring.md index ecc8f198..931007f3 100644 --- a/docs/requirements/Component-HealthMonitoring.md +++ b/docs/requirements/Component-HealthMonitoring.md @@ -101,6 +101,35 @@ These tiles are **point-in-time** like the Notification Outbox and Site Call Aud - Notification Outbox and Site Call Audit KPIs are not stored by Health Monitoring; they are computed point-in-time from the central `Notifications` and `SiteCalls` tables respectively each time the dashboard refreshes — consistent with the current-status-only philosophy. - Site connectivity history (online/offline transitions) may optionally be logged via the Audit Log or a separate mechanism if needed in the future. +## Prometheus Metrics (site node) + +Distinct from the health *report* above: these are scraped from the site node's `/metrics` +endpoint (`MetricsPort`, default 8084) rather than sent to central. + +| Series | Meaning | +| --- | --- | +| `scadabridge_site_data_connections_connected` | Data connections on this node in the `Connected` state | +| `scadabridge_site_data_connections_total` | Data connections known to this node, in any state | +| `scadabridge_site_stream_subscribers_active` | Open central→site gRPC subscriber streams | + +- The two `data_connections` gauges (register row 54) are the **only exported data-connection + signal**; before them the health report's `DataConnectionStatuses` was visible in the UI and + the CLI but never as a metric. `connected < total` is the alert condition — neither number + alone distinguishes "a device is down" from "nothing is configured". They are fed directly by + the Data Connection Layer at each connection state transition, deliberately **not** derived + from the health-report path, so the metric survives a dead report pipeline (the exact failure + the *metrics stale* flag exists to catch). +- Both read **0 on the standby node** — data connections are opened by the deployment path on + the active node only. A standby reporting 0 is healthy; alert on the active node, or on + `max()` across the pair. +- `scadabridge_site_stream_subscribers_active` was renamed from `scadabridge_site_connection_up` + by the same row. It has only ever counted open subscriber streams (Debug View + the site-wide + alarm feed), so it correctly reads 0 on a healthy site nobody is watching — the old name + invited reading it as data-connection health, and did cause exactly that misdiagnosis once. +- All three ride the `ZB.MOM.WW.ScadaBridge` meter, already in the `SiteServiceRegistration.ObservedMeters` + allowlist. `ZbTelemetryOptions.Meters` is an **allowlist**: an unlisted meter exports nothing, + with no error and no missing-metric signal. + ## No Alerting - Health monitoring is **display-only** for now — no automated notifications or alerts are triggered by health status changes. diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index 34d3bd1d..cce04d8a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -1,4 +1,5 @@ using System.Diagnostics.Metrics; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.Commons.Observability; @@ -126,8 +127,27 @@ public static class ScadaBridgeTelemetry // ---------------- Observable gauges ---------------- - /// Current count of open site connections, mutated via . - private static long _siteConnectionsUp; + /// + /// Current count of open central→site gRPC subscriber streams, mutated via + /// . See 's gauge for + /// why this is NOT a data-connection count. + /// + private static long _siteStreamSubscriberCount; + + /// + /// Register row 54: last-known state of every data connection on this site node, keyed by + /// connection name. Written by the DCL at its four state-transition points and read by the + /// data-connection gauges below. + /// + /// + /// A static registry, matching 's precedent, rather + /// than a DI-injected holder: the DCL actors already receive ISiteHealthCollector + /// through their constructors, and threading a second dependency through both actor + /// constructors would ripple into every test that builds them. A site node hosts exactly + /// one DCL, so a process-global registry and a per-node one are the same thing here. + /// + private static readonly System.Collections.Concurrent.ConcurrentDictionary + _dataConnectionStates = new(StringComparer.Ordinal); /// Current count of running per-site live-alarm aggregators, mutated via . private static long _liveAlarmAggregatorsActive; @@ -144,11 +164,52 @@ public static class ScadaBridgeTelemetry private static Func? _importSessionCountProvider; #pragma warning disable IDE0052 // Held to keep the observable gauges alive for the meter's lifetime. - /// Gauge reporting the number of currently open site connections. - private static readonly ObservableGauge _siteConnectionUp = - Meter.CreateObservableGauge("scadabridge.site.connection.up", - () => Interlocked.Read(ref _siteConnectionsUp), - description: "Number of currently open site connections."); + /// + /// Gauge reporting the number of open central→site gRPC subscriber streams. + /// + /// + /// Register row 54 RENAMED this from scadabridge.site.connection.up + /// (scadabridge_site_connection_up in Prometheus). The old name read as a + /// data-connection health signal, but it has only ever counted open subscriber streams + /// on SiteStreamGrpcServer — Debug View and the site-wide alarm feed. It therefore + /// sits at 0 on a perfectly healthy site that nobody is watching, and an alert written + /// against the old name fired on an idle operator console rather than a broken device. + /// For real data-connection health use the scadabridge.site.data_connections.* + /// gauges below. Counts both stream kinds together (no stream tag). + /// + private static readonly ObservableGauge _siteStreamSubscribersActive = + Meter.CreateObservableGauge("scadabridge.site.stream.subscribers.active", + () => Interlocked.Read(ref _siteStreamSubscriberCount), + unit: "1", + description: "Open central→site gRPC subscriber streams (Debug View + site-wide alarm feed). NOT data connections."); + + /// + /// Register row 54: gauge reporting how many of this site node's data connections are + /// currently in — the real data-connection + /// signal the misnamed stream-subscriber gauge was being mistaken for. + /// + /// + /// Read together with : connected < total is the + /// alertable condition, and neither number alone distinguishes "one device down" from + /// "nothing configured". Expect **0 on the standby node** — connections are opened by the + /// deployment path on the active node only, so a standby reporting 0 is healthy, not a + /// fault. Alert on the active node, or on max() across the pair. + /// + private static readonly ObservableGauge _dataConnectionsConnected = + Meter.CreateObservableGauge("scadabridge.site.data_connections.connected", + CountDataConnections, + unit: "1", + description: "Data connections on this site node currently in the Connected state."); + + /// + /// Register row 54: gauge reporting how many data connections this site node knows about, + /// in any state. The denominator for . + /// + private static readonly ObservableGauge _dataConnectionsTotal = + Meter.CreateObservableGauge("scadabridge.site.data_connections.total", + () => (long)_dataConnectionStates.Count, + unit: "1", + description: "Data connections known to this site node, in any state."); /// Gauge reporting the current StoreAndForward queue depth via the registered provider. private static readonly ObservableGauge _storeAndForwardQueueDepth = @@ -211,11 +272,68 @@ public static class ScadaBridgeTelemetry /// public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1); - /// Records that a site connection opened (increments the up-count gauge). - public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); + /// + /// Records that a central→site gRPC subscriber stream opened (increments + /// scadabridge.site.stream.subscribers.active). Renamed from + /// SiteConnectionOpened by register row 54 — it never counted data connections. + /// + public static void SiteStreamSubscriberOpened() => Interlocked.Increment(ref _siteStreamSubscriberCount); - /// Records that a site connection closed (decrements the up-count gauge). - public static void SiteConnectionClosed() => Interlocked.Decrement(ref _siteConnectionsUp); + /// + /// Records that a central→site gRPC subscriber stream closed (decrements + /// scadabridge.site.stream.subscribers.active). Renamed from + /// SiteConnectionClosed by register row 54. + /// + public static void SiteStreamSubscriberClosed() => Interlocked.Decrement(ref _siteStreamSubscriberCount); + + /// + /// Records the current state of one data connection, feeding the + /// scadabridge.site.data_connections.* gauges (register row 54). Called by the DCL + /// at each connection state transition; safe to call repeatedly with the same value. + /// + /// The data connection's name (unique per site node). + /// The connection's new state. + public static void RecordDataConnectionState(string connectionName, ConnectionHealth state) + { + if (string.IsNullOrEmpty(connectionName)) + { + return; + } + + _dataConnectionStates[connectionName] = state; + } + + /// + /// Drops a data connection from the gauges (register row 54). Called when the DCL removes + /// a connection, so an undeployed connection stops counting toward the total rather than + /// lingering as a permanently-not-connected row that would alert forever. + /// + /// The data connection's name. + public static void RemoveDataConnection(string connectionName) + { + if (string.IsNullOrEmpty(connectionName)) + { + return; + } + + _dataConnectionStates.TryRemove(connectionName, out _); + } + + /// Counts data connections currently in . + /// The connected count. + private static long CountDataConnections() + { + long connected = 0; + foreach (var state in _dataConnectionStates.Values) + { + if (state == ConnectionHealth.Connected) + { + connected++; + } + } + + return connected; + } /// Records that a per-site live-alarm aggregator started (increments the active-aggregator gauge). public static void LiveAlarmAggregatorStarted() => Interlocked.Increment(ref _liveAlarmAggregatorsActive); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index 2acb2a91..9a257585 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -467,12 +467,12 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase // Telemetry follow-on: the connection is now fully established (Subscribe // succeeded, so no leak via the catch above). Count it up here and balance - // it in the finally below so the scadabridge.site.connection.up gauge is - // decremented on EVERY exit path — normal completion, client-cancel / + // it in the finally below so the scadabridge.site.stream.subscribers.active + // gauge is decremented on EVERY exit path — normal completion, client-cancel / // duplicate-replacement (OperationCanceledException), server shutdown // (CancelAllStreams -> Cts.Cancel), and any other exception — guaranteeing // exactly one Closed per Opened and a gauge that never drifts up. - ScadaBridgeTelemetry.SiteConnectionOpened(); + ScadaBridgeTelemetry.SiteStreamSubscriberOpened(); try { // R2 — event batching. The pump replaces the old per-event @@ -500,7 +500,7 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase } finally { - ScadaBridgeTelemetry.SiteConnectionClosed(); + ScadaBridgeTelemetry.SiteStreamSubscriberClosed(); _streamSubscriber.RemoveSubscriber(relayActor); _actorSystem!.Stop(relayActor); channel.Writer.TryComplete(); diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs index 9cd2c598..659b7f8a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionActor.cs @@ -3,6 +3,7 @@ using Akka.Event; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Types.Alarms; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; using ZB.MOM.WW.ScadaBridge.HealthMonitoring; @@ -312,6 +313,10 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _log.Info("[{0}] Entering Connecting state", _connectionName); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Connecting); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Connecting"); + // Register row 54 — mirror the state into the data-connection gauges. Kept alongside + // the health-collector write (rather than derived from it) so the metric never + // depends on the health-report path being alive. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connecting); Become(Connecting); Self.Tell(new AttemptConnect()); } @@ -394,6 +399,8 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _healthCollector.UpdateTagResolution(_connectionName, TotalSubscribedTagCount, ResolvedTagCount); var endpointLabel = _backupConfig == null ? "Connected" : $"Connected to {_activeEndpoint.ToString().ToLower()}"; _healthCollector.UpdateConnectionEndpoint(_connectionName, endpointLabel); + // Register row 54 — the only transition that raises the connected gauge. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Connected); Become(Connected); Stash.UnstashAll(); } @@ -475,6 +482,9 @@ public class DataConnectionActor : UntypedActor, IWithStash, IWithTimers _log.Warning("[{0}] Entering Reconnecting state", _connectionName); _healthCollector.UpdateConnectionHealth(_connectionName, ConnectionHealth.Disconnected); _healthCollector.UpdateConnectionEndpoint(_connectionName, "Disconnected"); + // Register row 54 — drops this connection out of the connected gauge while the + // total holds, which is exactly the connected < total alert condition. + ScadaBridgeTelemetry.RecordDataConnectionState(_connectionName, ConnectionHealth.Disconnected); // Track unstable connections toward failover. // If we were connected for less than the stability threshold, this counts diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs index 14de697d..b1467468 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Actors/DataConnectionManagerActor.cs @@ -4,6 +4,7 @@ using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Commons.Serialization; using ZB.MOM.WW.ScadaBridge.Commons.Types.DataConnections; using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; @@ -307,6 +308,10 @@ public class DataConnectionManagerActor : ReceiveActor Context.Stop(actor); _connectionActors.Remove(command.ConnectionName); _healthCollector.RemoveConnection(command.ConnectionName); + // Register row 54 — drop it from the gauges too, so an undeployed connection + // stops counting toward the total instead of lingering as a permanently + // not-connected row that would hold the connected < total alert on forever. + ScadaBridgeTelemetry.RemoveDataConnection(command.ConnectionName); _log.Info("Removed DataConnectionActor for {0}", command.ConnectionName); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs new file mode 100644 index 00000000..98f1de8c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Commons.Tests/Observability/DataConnectionGaugeTests.cs @@ -0,0 +1,190 @@ +using System.Diagnostics.Metrics; +using ZB.MOM.WW.ScadaBridge.Commons.Observability; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.Commons.Tests.Observability; + +/// +/// Register row 54: the real per-site data-connection gauges. Before these existed the only +/// connection-ish series was scadabridge_site_connection_up, which counted open +/// central→site gRPC subscriber streams and sat at 0 on a healthy unwatched site — so there +/// was no metric that answered "are this site's devices actually connected?" at all. +/// +/// +/// Every assertion here is a DELTA against a baseline taken inside the test, and every +/// connection name is unique to the test. That is deliberate: the registry behind these +/// gauges is a process-global static, so absolute assertions would inherit the register +/// row 32 flake that makes QueueDepthGaugeTests pass alone and fail under the full +/// concurrent suite. Each test also removes its own connections in a finally so it +/// leaves the shared registry exactly as it found it. +/// +public sealed class DataConnectionGaugeTests +{ + private const string ConnectedGauge = "scadabridge.site.data_connections.connected"; + private const string TotalGauge = "scadabridge.site.data_connections.total"; + + [Fact] + public void ConnectedState_RaisesConnectedAndTotal() + { + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + + Assert.Equal(connectedBefore + 1, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Theory] + [InlineData(ConnectionHealth.Disconnected)] + [InlineData(ConnectionHealth.Connecting)] + [InlineData(ConnectionHealth.Error)] + public void NonConnectedState_RaisesTotalOnly(ConnectionHealth state) + { + // The load-bearing shape of the pair: a device that is configured but not talking + // must show up in the denominator and NOT in the numerator, because connected < total + // is the whole alert condition. A gauge that only counted connected rows would read + // identically for "one device down" and "device deleted". + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, state); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Fact] + public void TransitionFromConnectedToDisconnected_DropsConnectedButKeepsTotal() + { + // The DCL calls Record* once per state transition on the SAME name, so a transition + // must overwrite rather than accumulate — otherwise a flapping connection would + // inflate the total without bound. + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + try + { + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Disconnected); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore + 1, ReadGauge(TotalGauge)); + } + finally + { + ScadaBridgeTelemetry.RemoveDataConnection(name); + } + } + + [Fact] + public void RemoveDataConnection_DropsItFromBothGauges() + { + // An undeployed connection must leave the denominator. If it lingered as a + // permanently not-connected row, connected < total would latch on forever and the + // alert would be useless after the first undeploy. + var name = UniqueName(); + var connectedBefore = ReadGauge(ConnectedGauge); + var totalBefore = ReadGauge(TotalGauge); + + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RemoveDataConnection(name); + + Assert.Equal(connectedBefore, ReadGauge(ConnectedGauge)); + Assert.Equal(totalBefore, ReadGauge(TotalGauge)); + } + + [Fact] + public void RemoveDataConnection_IsIdempotent() + { + var name = UniqueName(); + ScadaBridgeTelemetry.RecordDataConnectionState(name, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RemoveDataConnection(name); + + var ex = Record.Exception(() => ScadaBridgeTelemetry.RemoveDataConnection(name)); + + Assert.Null(ex); + } + + [Fact] + public void NullOrEmptyConnectionName_IsIgnored() + { + // Guards the gauges against a malformed caller creating a phantom row that could + // never be removed by name. + var totalBefore = ReadGauge(TotalGauge); + + ScadaBridgeTelemetry.RecordDataConnectionState(string.Empty, ConnectionHealth.Connected); + ScadaBridgeTelemetry.RecordDataConnectionState(null!, ConnectionHealth.Connected); + + Assert.Equal(totalBefore, ReadGauge(TotalGauge)); + } + + [Fact] + public void StreamSubscriberGauge_IsPublishedUnderTheRenamedName() + { + // Register row 54's rename half. Pins the honest name so a future edit cannot + // silently restore scadabridge.site.connection.up, whose meaning (open gRPC + // subscriber streams) never matched what operators read it as. + var names = PublishedInstrumentNames(); + + Assert.Contains("scadabridge.site.stream.subscribers.active", names); + Assert.DoesNotContain("scadabridge.site.connection.up", names); + } + + private static string UniqueName() => $"conn-{Guid.NewGuid():N}"; + + /// Forces one observation of the named gauge and returns its value. + /// Instrument name to observe. + /// The observed value, or 0 if the instrument never reported. + private static long ReadGauge(string instrumentName) + { + long observed = 0; + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, l) => + { + if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName + && instrument.Name == instrumentName) + { + l.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback((_, measurement, _, _) => observed = measurement); + listener.Start(); + listener.RecordObservableInstruments(); + return observed; + } + + /// Collects every instrument name published on the ScadaBridge meter. + /// The published instrument names. + private static IReadOnlyCollection PublishedInstrumentNames() + { + var names = new List(); + using var listener = new MeterListener(); + listener.InstrumentPublished = (instrument, _) => + { + if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName) + { + names.Add(instrument.Name); + } + }; + listener.Start(); + return names; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs index 43a3c331..84c870e1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs @@ -443,12 +443,12 @@ public class SiteStreamGrpcServerTests : TestKit } [Fact] - public async Task SiteConnectionUpGauge_GoesToOneOnConnect_AndBackToZeroOnCancel() + public async Task SiteStreamSubscribersActiveGauge_GoesToOneOnConnect_AndBackToZeroOnCancel() { - // Telemetry follow-on: the scadabridge.site.connection.up gauge must read + // Telemetry follow-on: the scadabridge.site.stream.subscribers.active gauge must read // exactly 1 while a site stream is established and return to 0 once the - // stream terminates on the cancel path — proving SiteConnectionOpened() is - // matched by exactly one SiteConnectionClosed() in the handler's finally. + // stream terminates on the cancel path — proving SiteStreamSubscriberOpened() is + // matched by exactly one SiteStreamSubscriberClosed() in the handler's finally. var server = CreateServer(); server.SetReady(Sys); @@ -459,7 +459,7 @@ public class SiteStreamGrpcServerTests : TestKit listener.InstrumentPublished = (instrument, l) => { if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName && - instrument.Name == "scadabridge.site.connection.up") + instrument.Name == "scadabridge.site.stream.subscribers.active") { l.EnableMeasurementEvents(instrument); } @@ -485,7 +485,7 @@ public class SiteStreamGrpcServerTests : TestKit // (possibly parallel) tests left behind — read relative so the assertion // is robust to test interleaving on the process-wide static counter. // - // SiteConnectionOpened() runs AFTER the _activeStreams insertion that + // SiteStreamSubscriberOpened() runs AFTER the _activeStreams insertion that // WaitForConditionAsync(ActiveStreamCount == 1) keys off (see // SiteStreamGrpcServer.SubscribeInstance), so a one-shot ReadGauge() here // can race the increment under full-suite CPU oversubscription. Poll the @@ -499,7 +499,7 @@ public class SiteStreamGrpcServerTests : TestKit // After the cancel path runs the finally, the gauge is balanced back to // the baseline — no leaked "up" count. Poll here too: the gauge decrement // and the stream-count drop are independent observations, so under load - // give the finally's SiteConnectionClosed() time to land. + // give the finally's SiteStreamSubscriberClosed() time to land. AwaitAssert(() => Assert.Equal(baseline, ReadGauge()), TimeSpan.FromSeconds(5)); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs index 87549372..67729234 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheTelemetryTests.cs @@ -7,7 +7,7 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Tests; /// Telemetry guards for the aggregated live alarm cache (plan #10, Task 6): the /// active-aggregator observable gauge reflects Started/Stopped, and the reconnect /// counter increments on each recorded reconnect. Mirrors the MeterListener read -/// pattern used by SiteStreamGrpcServerTests.SiteConnectionUpGauge_*; reads are +/// pattern used by SiteStreamGrpcServerTests.SiteStreamSubscribersActiveGauge_*; reads are /// relative to a baseline so the process-wide static instruments are robust to any /// parallel test interleaving. /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs index 6c89f559..c17b290b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/ScadaBridgeTelemetryTests.cs @@ -22,8 +22,8 @@ public class ScadaBridgeTelemetryTests { ScadaBridgeTelemetry.RecordDeploymentApplied(); ScadaBridgeTelemetry.RecordInboundApiRequest("X"); - ScadaBridgeTelemetry.SiteConnectionOpened(); - ScadaBridgeTelemetry.SiteConnectionClosed(); + ScadaBridgeTelemetry.SiteStreamSubscriberOpened(); + ScadaBridgeTelemetry.SiteStreamSubscriberClosed(); ScadaBridgeTelemetry.LiveAlarmAggregatorStarted(); ScadaBridgeTelemetry.LiveAlarmAggregatorStopped(); ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect();