diff --git a/docs/Metrics.md b/docs/Metrics.md index 191c7a8..c13ea91 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -72,7 +72,7 @@ Observable gauges are pull-based; the `Meter` invokes the supplied callback when |------------|--------------|-------------| | `mxgateway.sessions.open` | `_openSessions` | Currently open sessions tracked by `SessionManager`. | | `mxgateway.workers.running` | `_workersRunning` | Worker clients in a running state. | -| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepth` | Undelivered worker events held by `WorkerClient` — staged *and* queued (GWC-24). Incremented when the read loop stages an event, decremented when the consumer reads it, so a backlog stuck in the staging channel is visible rather than invisible. | +| `mxgateway.events.worker_queue.depth` | `_workerEventQueueDepthSources` (summed on demand) | Undelivered worker events held by `WorkerClient` — staged *and* queued (GWC-24) — summed across every live client at collection time (GWC-30). Each client owns an interlocked counter incremented when the read loop stages an event and decremented when the consumer reads it, and registers it as a gauge source for its lifetime, so a backlog stuck in a staging channel is visible and concurrent sessions add up instead of overwriting one another. | | `mxgateway.events.grpc_stream_queue.depth` | `_eventStreamBacklogSources` (summed on demand) | Live backlog buffered across every active `EventStreamService` subscriber, summed from the subscribers' channel `Count` at collection time. | ## Snapshot Shape @@ -111,7 +111,7 @@ The scalar fields mirror the counters and gauges. The four dictionaries provide - `EventsBySession` keys by `sessionId`; entries are removed via `RemoveSessionEvents` when a session closes so the map does not grow without bound. - `RetryAttemptsByArea` keys by the resilience `area` tag, e.g. `worker_startup`. -`EventsReceived` is read with `Interlocked.Read(ref _eventsReceived)` because `EventReceived` increments it via `Interlocked.Increment` outside the lock to keep the event-ingestion path non-blocking. +`EventsReceived` is read with `Interlocked.Read(ref _eventsReceived)` because `EventReceived` increments it via `Interlocked.Increment` outside the lock to keep the event-ingestion path non-blocking. `CommandsStarted`, `CommandsSucceeded`, `CommandsFailed`, and `CommandFailuresByMethod` are read the same way: the command counters run two-to-three times per gRPC call, so they are recorded with `Interlocked` and a `ConcurrentDictionary` rather than under `_syncRoot` (GWC-30). The two queue depths are pulled from their registered sources before the lock is taken, since those delegates reach into subscriber channels and worker clients. ## Recording Sites @@ -146,7 +146,7 @@ _metrics.RemoveSessionEvents(session.SessionId); - `RecordWorkerStoppedOnce` calls `WorkerStopped(reason)` exactly once per worker, guarding against double-counting on simultaneous fault and exit signals. - `WorkerKilled(reason)` when the client forcibly terminates the worker. - `HeartbeatFailed(SessionId)` per missed heartbeat. -- `SetWorkerEventQueueDepth(queueDepth)` when the read loop stages an event and when the consumer reads one, so the gauge tracks staged + queued events. +- `RegisterWorkerEventQueueDepthSource(...)` once at construction, disposed in `DisposeAsync`. The client's own `_eventQueueDepth` is incremented when the read loop stages an event and decremented when the consumer reads one, so the gauge tracks staged + queued events without either hot-path step calling into `GatewayMetrics`. - `EventReceived(SessionId, workerEvent.Event.Family.ToString())` for each worker event. - `QueueOverflow("worker-events")` when the timed write into the bounded consumer channel exceeds `EventChannelFullModeTimeout`, and `QueueOverflow("worker-event-staging")` when the staging channel is full at its `2 × EventChannelCapacity` bound. The two labels distinguish a stalled consumer from one that merely drains too slowly; both fault the session with `ProtocolViolation`. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs b/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs index 5f520cc..36e8938 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs @@ -28,7 +28,9 @@ public sealed class GatewayMetrics : IDisposable private readonly Histogram _workerStartupLatencyHistogram; private readonly Histogram _commandLatencyHistogram; private readonly Histogram _eventStreamSendLatencyHistogram; - private readonly Dictionary _commandFailuresByMethod = new(StringComparer.OrdinalIgnoreCase); + // Concurrent (not Dictionary + _syncRoot) because CommandFailed runs on every failing gRPC call: + // the command counters are recorded outside the lock, so their breakdown map must be too. + private readonly ConcurrentDictionary _commandFailuresByMethod = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _eventsByFamily = new(StringComparer.OrdinalIgnoreCase); private readonly ConcurrentDictionary _eventsBySession = new(StringComparer.Ordinal); private readonly Dictionary _retryAttemptsByArea = new(StringComparer.OrdinalIgnoreCase); @@ -41,9 +43,16 @@ public sealed class GatewayMetrics : IDisposable private readonly ConcurrentDictionary> _eventStreamBacklogSources = new(); private long _nextEventStreamBacklogSourceId; + // GWC-30: the same pull model for the worker event queue depth. It replaces a pushed scalar that + // every WorkerClient wrote twice per event (staged, consumed) under _syncRoot — a process-wide + // lock on the hottest path, and last-writer-wins across sessions, so the gauge reported one + // arbitrary session's backlog instead of the gateway's. Each client registers a source returning + // its own undelivered depth; the gauge sums them at collection time only. + private readonly ConcurrentDictionary> _workerEventQueueDepthSources = new(); + private long _nextWorkerEventQueueDepthSourceId; + private int _openSessions; private int _workersRunning; - private int _workerEventQueueDepth; private int _alarmProviderMode; private long _sessionsOpened; private long _sessionsClosed; @@ -201,10 +210,10 @@ public sealed class GatewayMetrics : IDisposable /// Name of the command method. public void CommandStarted(string method) { - lock (_syncRoot) - { - _commandsStarted++; - } + // GWC-30: the three command counters run two-to-three times per gRPC call, so they use + // Interlocked rather than _syncRoot — the same idiom as EventReceived. Nothing here needs a + // consistent multi-field view; GetSnapshot reads each with Interlocked.Read. + Interlocked.Increment(ref _commandsStarted); _commandsStartedCounter.Add(1, new KeyValuePair("method", method)); } @@ -216,10 +225,7 @@ public sealed class GatewayMetrics : IDisposable /// Elapsed time to complete the command. public void CommandSucceeded(string method, TimeSpan duration) { - lock (_syncRoot) - { - _commandsSucceeded++; - } + Interlocked.Increment(ref _commandsSucceeded); KeyValuePair methodTag = new("method", method); _commandsSucceededCounter.Add(1, methodTag); @@ -234,11 +240,8 @@ public sealed class GatewayMetrics : IDisposable /// Elapsed time before command failed. public void CommandFailed(string method, string category, TimeSpan duration) { - lock (_syncRoot) - { - _commandsFailed++; - Increment(_commandFailuresByMethod, method); - } + Interlocked.Increment(ref _commandsFailed); + Increment(_commandFailuresByMethod, method); KeyValuePair methodTag = new("method", method); KeyValuePair categoryTag = new("category", category); @@ -275,29 +278,24 @@ public sealed class GatewayMetrics : IDisposable } /// - /// Sets the worker event queue depth; delegates to SetWorkerEventQueueDepth. + /// Registers a live depth source for the worker event queue-depth gauge and returns a handle + /// that removes it when disposed. Each WorkerClient registers once and reports its own + /// undelivered (staged + queued) event count, so the gauge is the gateway-wide sum instead of + /// the last value any one session happened to push (GWC-30). /// - /// Queue depth value. - public void SetEventQueueDepth(int depth) + /// + /// Returns this worker client's current undelivered event count. Invoked only at collection + /// time; must be cheap and non-blocking (a of an + /// interlocked counter). Negative readings — which a racing decrement can produce — are + /// clamped to zero when summed. + /// + /// A handle whose disposal unregisters the source. Safe to dispose more than once. + public IDisposable RegisterWorkerEventQueueDepthSource(Func depth) { - SetWorkerEventQueueDepth(depth); - } - - /// - /// Sets the worker event queue depth to the given value. - /// - /// Queue depth value. - public void SetWorkerEventQueueDepth(int depth) - { - if (depth < 0) - { - throw new ArgumentOutOfRangeException(nameof(depth), depth, "Queue depth cannot be negative."); - } - - lock (_syncRoot) - { - _workerEventQueueDepth = depth; - } + ArgumentNullException.ThrowIfNull(depth); + long id = Interlocked.Increment(ref _nextWorkerEventQueueDepthSourceId); + _workerEventQueueDepthSources[id] = depth; + return new GaugeSourceRegistration(_workerEventQueueDepthSources, id); } /// @@ -318,7 +316,7 @@ public sealed class GatewayMetrics : IDisposable ArgumentNullException.ThrowIfNull(backlog); long id = Interlocked.Increment(ref _nextEventStreamBacklogSourceId); _eventStreamBacklogSources[id] = backlog; - return new EventStreamBacklogRegistration(this, id); + return new GaugeSourceRegistration(_eventStreamBacklogSources, id); } /// @@ -460,21 +458,23 @@ public sealed class GatewayMetrics : IDisposable /// The current metrics snapshot. public GatewayMetricsSnapshot GetSnapshot() { - // Compute the live gRPC stream backlog outside _syncRoot: the sources are the subscriber - // channels' Count (their own locks) and must not run under this lock. GWC-15. + // Compute the live queue depths outside _syncRoot: the sources are the subscriber channels' + // Count (their own locks) and the worker clients' interlocked counters, neither of which may + // run under this lock. GWC-15, GWC-30. + int workerEventQueueDepth = GetWorkerEventQueueDepth(); int grpcEventStreamQueueDepth = GetGrpcEventStreamQueueDepth(); lock (_syncRoot) { return new GatewayMetricsSnapshot( OpenSessions: _openSessions, WorkersRunning: _workersRunning, - WorkerEventQueueDepth: _workerEventQueueDepth, + WorkerEventQueueDepth: workerEventQueueDepth, GrpcEventStreamQueueDepth: grpcEventStreamQueueDepth, SessionsOpened: _sessionsOpened, SessionsClosed: _sessionsClosed, - CommandsStarted: _commandsStarted, - CommandsSucceeded: _commandsSucceeded, - CommandsFailed: _commandsFailed, + CommandsStarted: Interlocked.Read(ref _commandsStarted), + CommandsSucceeded: Interlocked.Read(ref _commandsSucceeded), + CommandsFailed: Interlocked.Read(ref _commandsFailed), EventsReceived: Interlocked.Read(ref _eventsReceived), QueueOverflows: _queueOverflows, Faults: _faults, @@ -521,22 +521,19 @@ public sealed class GatewayMetrics : IDisposable } } - private int GetWorkerEventQueueDepth() - { - lock (_syncRoot) - { - return _workerEventQueueDepth; - } - } + // Sums the undelivered event backlog across every live worker client (GWC-30). + private int GetWorkerEventQueueDepth() => SumSources(_workerEventQueueDepthSources); - // Sums the live backlog across every registered event-stream subscriber. Runs at collection - // time (ObservableGauge scrape) or when GetSnapshot projects the value — never on the - // per-event path. Enumerating ConcurrentDictionary.Values never throws on concurrent + // Sums the live backlog across every registered event-stream subscriber. + private int GetGrpcEventStreamQueueDepth() => SumSources(_eventStreamBacklogSources); + + // Runs at collection time (ObservableGauge scrape) or when GetSnapshot projects the value — + // never on a per-event path. Enumerating ConcurrentDictionary.Values never throws on concurrent // register/unregister; a source removed mid-enumeration simply drops from this sample. - private int GetGrpcEventStreamQueueDepth() + private static int SumSources(ConcurrentDictionary> sources) { int total = 0; - foreach (Func source in _eventStreamBacklogSources.Values) + foreach (Func source in sources.Values) { int value = source(); if (value > 0) @@ -548,11 +545,6 @@ public sealed class GatewayMetrics : IDisposable return total; } - private void UnregisterEventStreamBacklogSource(long id) - { - _eventStreamBacklogSources.TryRemove(id, out _); - } - private int GetAlarmProviderMode() { lock (_syncRoot) @@ -572,9 +564,10 @@ public sealed class GatewayMetrics : IDisposable values.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1); } - // Handle returned by RegisterEventStreamBacklogSource. Disposal (once) removes the source - // from the gauge's live sum. Idempotent so a double dispose from a stream teardown is safe. - private sealed class EventStreamBacklogRegistration(GatewayMetrics metrics, long id) : IDisposable + // Handle returned by the pull-model gauge registrations. Disposal (once) removes the source from + // that gauge's live sum. Idempotent so a double dispose from a stream or worker-client teardown + // is safe, and shared by both gauges so the two registrations cannot drift apart. + private sealed class GaugeSourceRegistration(ConcurrentDictionary> sources, long id) : IDisposable { private int _disposed; @@ -582,7 +575,7 @@ public sealed class GatewayMetrics : IDisposable { if (Interlocked.Exchange(ref _disposed, 1) == 0) { - metrics.UnregisterEventStreamBacklogSource(id); + sources.TryRemove(id, out _); } } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index 3ad808c..be74d2b 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -40,6 +40,12 @@ public sealed class WorkerClient : IWorkerClient private readonly ConcurrentDictionary _pendingCommands = new(StringComparer.Ordinal); private readonly SemaphoreSlim _pendingCommandSlots; private readonly CancellationTokenSource _stopCts = new(); + + // GWC-30: this client's contribution to the gateway-wide worker queue-depth gauge. Registered + // once here and read only when the gauge is scraped, so staging and consuming an event cost an + // Interlocked on _eventQueueDepth and nothing else — previously each of those two hot-path steps + // called into GatewayMetrics and took its process-wide lock. Null when metrics are disabled. + private readonly IDisposable? _eventQueueDepthRegistration; // Touched only by WriteLoopAsync — the single consumer of _outboundEnvelopes — so it needs no // interlocking. See WriteLoopAsync for why the stamp happens there rather than at construction. private ulong _nextSequence; @@ -111,6 +117,8 @@ public sealed class WorkerClient : IWorkerClient FullMode = BoundedChannelFullMode.Wait, AllowSynchronousContinuations = false, }); + _eventQueueDepthRegistration = _metrics?.RegisterWorkerEventQueueDepthSource( + () => Volatile.Read(ref _eventQueueDepth)); _lastHeartbeatAt = _timeProvider.GetUtcNow(); } @@ -299,8 +307,8 @@ public sealed class WorkerClient : IWorkerClient { await foreach (WorkerEvent workerEvent in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) { - int queueDepth = Math.Max(0, Interlocked.Decrement(ref _eventQueueDepth)); - _metrics?.SetWorkerEventQueueDepth(queueDepth); + // No metrics call on the hot path: the gauge pulls _eventQueueDepth when scraped (GWC-30). + Interlocked.Decrement(ref _eventQueueDepth); yield return workerEvent; } } @@ -371,6 +379,9 @@ public sealed class WorkerClient : IWorkerClient } _disposed = true; + // Drop out of the worker queue-depth gauge before teardown: whatever this client still holds + // is about to be discarded, and the sum must not keep counting a client that is going away. + _eventQueueDepthRegistration?.Dispose(); KillOwnedProcess("Dispose"); _stopCts.Cancel(); _outboundEnvelopes.Writer.TryComplete(); @@ -598,8 +609,7 @@ public sealed class WorkerClient : IWorkerClient { // Counted here rather than at the _events write so the single gauge reports total // undelivered events (staged + queued). ReadEventsCoreAsync decrements on consumer read. - int queueDepth = Interlocked.Increment(ref _eventQueueDepth); - _metrics?.SetWorkerEventQueueDepth(queueDepth); + Interlocked.Increment(ref _eventQueueDepth); return; } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs index 374ff6f..5d67920 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Metrics/GatewayMetricsTests.cs @@ -19,7 +19,10 @@ public sealed class GatewayMetricsTests metrics.CommandFailed("WriteSecured", "AuthorizationFailed", TimeSpan.FromMilliseconds(12)); metrics.EventReceived("session-1", "OnDataChange"); metrics.EventReceived("session-1", "OnDataChange"); - metrics.SetWorkerEventQueueDepth(7); + // GWC-30: the worker queue-depth gauge sums one live source per worker client, so the two + // registrations below stand in for two concurrent sessions holding 3 and 4 events. + using IDisposable workerDepthSourceA = metrics.RegisterWorkerEventQueueDepthSource(static () => 3); + using IDisposable workerDepthSourceB = metrics.RegisterWorkerEventQueueDepthSource(static () => 4); // GWC-15: the gRPC stream queue-depth gauge sums live backlog sources at collection time // rather than tracking a pushed running total. Register a source reporting 3. using IDisposable backlogSource = metrics.RegisterEventStreamBacklogSource(static () => 3); @@ -54,16 +57,111 @@ public sealed class GatewayMetricsTests Assert.Equal(2, snapshot.EventsBySession["session-1"]); } - /// Verifies that negative queue depth is rejected. + /// + /// GWC-30: the worker queue-depth gauge sums every registered source rather than holding a + /// single pushed scalar, so concurrent sessions add up instead of overwriting one another, + /// and a disposed registration (a worker client going away) drops out of the sum. Disposal + /// is idempotent because a client's dispose path can run twice. + /// [Fact] - public void SetEventQueueDepth_RejectsNegativeDepth() + public void WorkerEventQueueDepthSources_SumAcrossRegistrationsAndDropOnDispose() { using GatewayMetrics metrics = new(); - ArgumentOutOfRangeException exception = Assert.Throws( - () => metrics.SetWorkerEventQueueDepth(-1)); + IDisposable firstSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 3); + using IDisposable secondSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 4); - Assert.Equal("depth", exception.ParamName); + Assert.Equal(7, metrics.GetSnapshot().WorkerEventQueueDepth); + + firstSource.Dispose(); + Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth); + + firstSource.Dispose(); + Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth); + } + + /// + /// A depth source reads a lock-free counter that a racing decrement can momentarily push + /// below zero, so the sum clamps each reading instead of rejecting it — the pull model has + /// no caller to throw back at. + /// + [Fact] + public void WorkerEventQueueDepthSources_ClampNegativeReadingsToZero() + { + using GatewayMetrics metrics = new(); + + using IDisposable negativeSource = metrics.RegisterWorkerEventQueueDepthSource(static () => -5); + using IDisposable positiveSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 4); + + Assert.Equal(4, metrics.GetSnapshot().WorkerEventQueueDepth); + } + + /// + /// Verifies the exported gauge keeps the name mxgateway.events.worker_queue.depth and + /// reports the summed sources, so the pull-model rework is invisible to exporters. + /// + [Fact] + public void WorkerEventQueueDepthGauge_ReportsSummedSources() + { + using GatewayMetrics metrics = new(); + using MeterListener listener = new(); + + int? capturedDepth = null; + + listener.InstrumentPublished = (instrument, meterListener) => + { + if (ReferenceEquals(instrument.Meter, metrics.Meter) + && instrument.Name == "mxgateway.events.worker_queue.depth") + { + meterListener.EnableMeasurementEvents(instrument); + } + }; + listener.SetMeasurementEventCallback( + (instrument, measurement, _, _) => + { + if (ReferenceEquals(instrument.Meter, metrics.Meter) + && instrument.Name == "mxgateway.events.worker_queue.depth") + { + capturedDepth = measurement; + } + }); + listener.Start(); + + using IDisposable firstSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 5); + using IDisposable secondSource = metrics.RegisterWorkerEventQueueDepthSource(static () => 6); + listener.RecordObservableInstruments(); + + Assert.Equal(11, capturedDepth); + } + + /// + /// The command counters are incremented with rather than under the + /// process-wide metrics lock, so this asserts no increment is lost when every gRPC thread + /// records at once. + /// + [Fact] + public void CommandCounters_CountEveryConcurrentInvocation() + { + const int workers = 8; + const int perWorker = 500; + using GatewayMetrics metrics = new(); + + Parallel.For(0, workers, _ => + { + for (int index = 0; index < perWorker; index++) + { + metrics.CommandStarted("Register"); + metrics.CommandSucceeded("Register", TimeSpan.FromMilliseconds(1)); + metrics.CommandFailed("WriteSecured", "AuthorizationFailed", TimeSpan.FromMilliseconds(1)); + } + }); + + GatewayMetricsSnapshot snapshot = metrics.GetSnapshot(); + + Assert.Equal(workers * perWorker, snapshot.CommandsStarted); + Assert.Equal(workers * perWorker, snapshot.CommandsSucceeded); + Assert.Equal(workers * perWorker, snapshot.CommandsFailed); + Assert.Equal(workers * perWorker, snapshot.CommandFailuresByMethod["WriteSecured"]); } ///