mxaccessgw/src/ZB.MOM.WW.MxGateway.Server/Metrics/GatewayMetrics.cs

using System.Collections.Concurrent;
using System.Diagnostics.Metrics;
using System.Globalization;

namespace ZB.MOM.WW.MxGateway.Server.Metrics;

public sealed class GatewayMetrics : IDisposable
{
    public const string MeterName = "ZB.MOM.WW.MxGateway";

    private readonly object _syncRoot = new();
    private readonly Meter _meter;
    private readonly Counter<long> _sessionsOpenedCounter;
    private readonly Counter<long> _sessionsClosedCounter;
    private readonly Counter<long> _commandsStartedCounter;
    private readonly Counter<long> _commandsSucceededCounter;
    private readonly Counter<long> _commandsFailedCounter;
    private readonly Counter<long> _eventsReceivedCounter;
    private readonly Counter<long> _queueOverflowsCounter;
    private readonly Counter<long> _faultsCounter;
    private readonly Counter<long> _workerKillsCounter;
    private readonly Counter<long> _workerExitsCounter;
    private readonly Counter<long> _heartbeatFailuresCounter;
    private readonly Counter<long> _streamDisconnectsCounter;
    private readonly Counter<long> _retryAttemptsCounter;
    private readonly Counter<long> _alarmProviderSwitchesCounter;
    private readonly Counter<long> _authThrottledCounter;
    private readonly Histogram<double> _workerStartupLatencyHistogram;
    private readonly Histogram<double> _commandLatencyHistogram;
    private readonly Histogram<double> _eventStreamSendLatencyHistogram;
    // 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<string, long> _commandFailuresByMethod = new(StringComparer.OrdinalIgnoreCase);
    private readonly ConcurrentDictionary<string, long> _eventsByFamily = new(StringComparer.OrdinalIgnoreCase);
    private readonly ConcurrentDictionary<string, long> _eventsBySession = new(StringComparer.Ordinal);
    private readonly Dictionary<string, long> _retryAttemptsByArea = new(StringComparer.OrdinalIgnoreCase);

    // GWC-15: live backlog sources for the gRPC event-stream queue-depth gauge. Each active
    // StreamEvents subscriber registers a delegate returning its channel's current backlog; the
    // gauge sums them ONLY when scraped or projected (GetGrpcEventStreamQueueDepth), so nothing is
    // read on the per-event streaming hot path. Concurrent because subscribers register/unregister
    // from arbitrary gRPC threads while the gauge callback / GetSnapshot enumerate.
    private readonly ConcurrentDictionary<long, Func<int>> _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<long, Func<int>> _workerEventQueueDepthSources = new();
    private long _nextWorkerEventQueueDepthSourceId;

    private int _openSessions;
    private int _workersRunning;
    private int _alarmProviderMode;
    private long _sessionsOpened;
    private long _sessionsClosed;
    private long _commandsStarted;
    private long _commandsSucceeded;
    private long _commandsFailed;
    private long _eventsReceived;
    private long _queueOverflows;
    private long _faults;
    private long _workerKills;
    private long _workerExits;
    private long _heartbeatFailures;
    private long _streamDisconnects;
    private long _retryAttempts;
    private long _alarmProviderSwitches;
    private bool _disposed;

    /// <summary>
    /// Initializes the gateway metrics with OpenTelemetry counters and histograms.
    /// </summary>
    public GatewayMetrics()
    {
        _meter = new Meter(MeterName, typeof(GatewayMetrics).Assembly.GetName().Version?.ToString());
        _sessionsOpenedCounter = _meter.CreateCounter<long>("mxgateway.sessions.opened");
        _sessionsClosedCounter = _meter.CreateCounter<long>("mxgateway.sessions.closed");
        _commandsStartedCounter = _meter.CreateCounter<long>("mxgateway.commands.started");
        _commandsSucceededCounter = _meter.CreateCounter<long>("mxgateway.commands.succeeded");
        _commandsFailedCounter = _meter.CreateCounter<long>("mxgateway.commands.failed");
        _eventsReceivedCounter = _meter.CreateCounter<long>("mxgateway.events.received");
        _queueOverflowsCounter = _meter.CreateCounter<long>("mxgateway.queues.overflows");
        _faultsCounter = _meter.CreateCounter<long>("mxgateway.faults");
        _workerKillsCounter = _meter.CreateCounter<long>("mxgateway.workers.killed");
        _workerExitsCounter = _meter.CreateCounter<long>("mxgateway.workers.exited");
        _heartbeatFailuresCounter = _meter.CreateCounter<long>("mxgateway.heartbeats.failed");
        _streamDisconnectsCounter = _meter.CreateCounter<long>("mxgateway.grpc.streams.disconnected");
        _retryAttemptsCounter = _meter.CreateCounter<long>("mxgateway.retries.attempted");
        _alarmProviderSwitchesCounter = _meter.CreateCounter<long>("mxgateway.alarms.provider_switches");
        _authThrottledCounter = _meter.CreateCounter<long>("mxgateway.auth.throttled");
        _workerStartupLatencyHistogram = _meter.CreateHistogram<double>("mxgateway.workers.startup.duration", "s");
        _commandLatencyHistogram = _meter.CreateHistogram<double>("mxgateway.commands.duration", "s");
        _eventStreamSendLatencyHistogram = _meter.CreateHistogram<double>("mxgateway.events.stream_send.duration", "s");

        _meter.CreateObservableGauge("mxgateway.sessions.open", GetOpenSessions);
        _meter.CreateObservableGauge("mxgateway.workers.running", GetWorkersRunning);
        _meter.CreateObservableGauge("mxgateway.events.worker_queue.depth", GetWorkerEventQueueDepth);
        _meter.CreateObservableGauge("mxgateway.events.grpc_stream_queue.depth", GetGrpcEventStreamQueueDepth);
        _meter.CreateObservableGauge("mxgateway.alarms.provider_mode", GetAlarmProviderMode);
    }

    /// <summary>
    /// Gets the underlying <see cref="Meter"/> instance backing this metrics object. Exposed to tests
    /// (via <c>InternalsVisibleTo</c>) so a <see cref="MeterListener"/> can filter measurements by
    /// <see cref="object.ReferenceEquals"/> against this specific instance rather than by the
    /// process-shared <see cref="MeterName"/>, which would cross-talk between parallel tests that
    /// each build their own <see cref="GatewayMetrics"/>.
    /// </summary>
    internal Meter Meter => _meter;

    /// <summary>
    /// Records that a session has been opened.
    /// </summary>
    public void SessionOpened()
    {
        lock (_syncRoot)
        {
            _openSessions++;
            _sessionsOpened++;
        }

        _sessionsOpenedCounter.Add(1);
    }

    /// <summary>
    /// Records that a session has been closed.
    /// </summary>
    public void SessionClosed()
    {
        lock (_syncRoot)
        {
            if (_openSessions > 0)
            {
                _openSessions--;
            }

            _sessionsClosed++;
        }

        _sessionsClosedCounter.Add(1);
    }

    /// <summary>
    /// Records that a session has been removed from registry.
    /// </summary>
    public void SessionRemoved()
    {
        lock (_syncRoot)
        {
            if (_openSessions > 0)
            {
                _openSessions--;
            }
        }
    }

    /// <summary>
    /// Records that a worker process has started and its startup latency.
    /// </summary>
    /// <param name="startupDuration">Duration elapsed while starting the worker.</param>
    public void WorkerStarted(TimeSpan startupDuration)
    {
        lock (_syncRoot)
        {
            _workersRunning++;
        }

        _workerStartupLatencyHistogram.Record(startupDuration.TotalSeconds);
    }

    /// <summary>
    /// Records that a worker process has stopped with the given reason.
    /// </summary>
    /// <param name="reason">Cause of the worker stopping.</param>
    public void WorkerStopped(string reason)
    {
        lock (_syncRoot)
        {
            if (_workersRunning > 0)
            {
                _workersRunning--;
            }

            _workerExits++;
        }

        _workerExitsCounter.Add(1, new KeyValuePair<string, object?>("reason", reason));
    }

    /// <summary>
    /// Records that a worker process was killed with the given reason.
    /// </summary>
    /// <param name="reason">Cause of the worker termination.</param>
    public void WorkerKilled(string reason)
    {
        lock (_syncRoot)
        {
            _workerKills++;
        }

        _workerKillsCounter.Add(1, new KeyValuePair<string, object?>("reason", reason));
    }

    /// <summary>
    /// Records that a command has started for the given method.
    /// </summary>
    /// <param name="method">Name of the command method.</param>
    public void CommandStarted(string method)
    {
        // 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<string, object?>("method", method));
    }

    /// <summary>
    /// Records that a command succeeded for the given method and duration.
    /// </summary>
    /// <param name="method">Name of the command method.</param>
    /// <param name="duration">Elapsed time to complete the command.</param>
    public void CommandSucceeded(string method, TimeSpan duration)
    {
        Interlocked.Increment(ref _commandsSucceeded);

        KeyValuePair<string, object?> methodTag = new("method", method);
        _commandsSucceededCounter.Add(1, methodTag);
        _commandLatencyHistogram.Record(duration.TotalSeconds, methodTag);
    }

    /// <summary>
    /// Records that a command failed for the given method, category, and duration.
    /// </summary>
    /// <param name="method">Name of the command method.</param>
    /// <param name="category">Classification of the failure.</param>
    /// <param name="duration">Elapsed time before command failed.</param>
    public void CommandFailed(string method, string category, TimeSpan duration)
    {
        Interlocked.Increment(ref _commandsFailed);
        Increment(_commandFailuresByMethod, method);

        KeyValuePair<string, object?> methodTag = new("method", method);
        KeyValuePair<string, object?> categoryTag = new("category", category);
        _commandsFailedCounter.Add(1, methodTag, categoryTag);
        _commandLatencyHistogram.Record(duration.TotalSeconds, methodTag, categoryTag);
    }

    /// <summary>
    /// Records that an event was received for the given session and family.
    /// </summary>
    /// <param name="sessionId">Identifier of the session receiving the event.</param>
    /// <param name="family">Event family classification.</param>
    public void EventReceived(string sessionId, string family)
    {
        Interlocked.Increment(ref _eventsReceived);
        Increment(_eventsByFamily, family);
        Increment(_eventsBySession, sessionId);

        _eventsReceivedCounter.Add(
            1,
            new KeyValuePair<string, object?>("family", family));
    }

    /// <summary>
    /// Records the latency of sending an event to a client stream.
    /// </summary>
    /// <param name="family">Event family name.</param>
    /// <param name="duration">Time taken to send the event.</param>
    public void RecordEventStreamSend(string family, TimeSpan duration)
    {
        _eventStreamSendLatencyHistogram.Record(
            duration.TotalSeconds,
            new KeyValuePair<string, object?>("family", family));
    }

    /// <summary>
    ///     Registers a live depth source for the worker event queue-depth gauge and returns a handle
    ///     that removes it when disposed. Each <c>WorkerClient</c> 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).
    /// </summary>
    /// <param name="depth">
    ///     Returns this worker client's current undelivered event count. Invoked only at collection
    ///     time; must be cheap and non-blocking (a <see cref="Volatile.Read(ref int)"/> of an
    ///     interlocked counter). Negative readings — which a racing decrement can produce — are
    ///     clamped to zero when summed.
    /// </param>
    /// <returns>A handle whose disposal unregisters the source. Safe to dispose more than once.</returns>
    public IDisposable RegisterWorkerEventQueueDepthSource(Func<int> depth)
    {
        ArgumentNullException.ThrowIfNull(depth);
        long id = Interlocked.Increment(ref _nextWorkerEventQueueDepthSourceId);
        _workerEventQueueDepthSources[id] = depth;
        return new GaugeSourceRegistration(_workerEventQueueDepthSources, id);
    }

    /// <summary>
    ///     Registers a live backlog source for the gRPC event-stream queue-depth gauge and
    ///     returns a handle that removes it when disposed. The gauge sums every registered
    ///     source's current value on demand (when scraped or via <see cref="GetSnapshot"/>),
    ///     rather than tracking a pushed running total, so the streaming hot path does no
    ///     per-event gauge bookkeeping (GWC-15).
    /// </summary>
    /// <param name="backlog">
    ///     Returns this subscriber's current channel backlog. Invoked only at collection time;
    ///     must be cheap and non-blocking (a bounded channel's <c>Count</c>). Negative returns
    ///     are clamped to zero when summed.
    /// </param>
    /// <returns>A handle whose disposal unregisters the source. Safe to dispose more than once.</returns>
    public IDisposable RegisterEventStreamBacklogSource(Func<int> backlog)
    {
        ArgumentNullException.ThrowIfNull(backlog);
        long id = Interlocked.Increment(ref _nextEventStreamBacklogSourceId);
        _eventStreamBacklogSources[id] = backlog;
        return new GaugeSourceRegistration(_eventStreamBacklogSources, id);
    }

    /// <summary>
    /// Removes event counters for the given session.
    /// </summary>
    /// <param name="sessionId">Identifier of the session.</param>
    public void RemoveSessionEvents(string sessionId)
    {
        _eventsBySession.TryRemove(sessionId, out _);
    }

    /// <summary>
    /// Records that a queue overflow occurred for the given queue name.
    /// </summary>
    /// <param name="queueName">Name of the queue that overflowed.</param>
    public void QueueOverflow(string queueName)
    {
        lock (_syncRoot)
        {
            _queueOverflows++;
        }

        _queueOverflowsCounter.Add(1, new KeyValuePair<string, object?>("queue", queueName));
    }

    /// <summary>
    /// Records that an API-key authentication attempt was short-circuited by the failure limiter
    /// before verification.
    /// </summary>
    /// <param name="stage">Limiter layer that refused the attempt: <c>peer</c> or <c>aggregate</c>.</param>
    public void RecordAuthThrottled(string stage)
    {
        // Tagged by stage only. Neither the key id nor the peer address may become a tag: key ids
        // are credential material and peer addresses are unbounded cardinality, and /metrics is not
        // authenticated (open SEC-14), so anything tagged here is world-readable.
        _authThrottledCounter.Add(1, new KeyValuePair<string, object?>("stage", stage));
    }

    /// <summary>
    /// Records that a fault occurred in the given category.
    /// </summary>
    /// <param name="category">Category of the fault.</param>
    public void Fault(string category)
    {
        lock (_syncRoot)
        {
            _faults++;
        }

        _faultsCounter.Add(1, new KeyValuePair<string, object?>("category", category));
    }

    /// <summary>
    /// Records that a heartbeat failed for the given session.
    /// </summary>
    /// <param name="sessionId">Identifier of the session.</param>
    public void HeartbeatFailed(string sessionId)
    {
        lock (_syncRoot)
        {
            _heartbeatFailures++;
        }

        // Exported without a session_id tag: per-session attribution is unbounded cardinality
        // (every session ever opened mints a new exporter time series, bloating Prometheus/OTLP
        // storage on long-running gateways). Per-session detail stays available via the dashboard
        // snapshot and the log scope; the exported counter tracks only the aggregate.
        _heartbeatFailuresCounter.Add(1);
    }

    /// <summary>
    /// Records that an event stream was disconnected with the given reason.
    /// </summary>
    /// <param name="reason">Reason for the disconnection.</param>
    public void StreamDisconnected(string reason)
    {
        lock (_syncRoot)
        {
            _streamDisconnects++;
        }

        _streamDisconnectsCounter.Add(1, new KeyValuePair<string, object?>("reason", reason));
    }

    /// <summary>
    /// Records that a retry was attempted in the given area.
    /// </summary>
    /// <param name="area">Area in which the retry was attempted.</param>
    public void RetryAttempted(string area)
    {
        lock (_syncRoot)
        {
            _retryAttempts++;
            Increment(_retryAttemptsByArea, area);
        }

        _retryAttemptsCounter.Add(1, new KeyValuePair<string, object?>("area", area));
    }

    /// <summary>
    /// Records that the alarm provider switched modes, increments the switch count, and updates the
    /// current provider mode gauge.
    /// </summary>
    /// <param name="fromMode">Provider mode before the switch (1=alarmmgr, 2=subtag, 0=unknown).</param>
    /// <param name="toMode">Provider mode after the switch (1=alarmmgr, 2=subtag, 0=unknown).</param>
    /// <param name="reason">Bounded switch classification used as the counter's <c>reason</c> tag.</param>
    public void AlarmProviderSwitched(int fromMode, int toMode, AlarmProviderSwitchReason reason)
    {
        lock (_syncRoot)
        {
            _alarmProviderMode = toMode;
            _alarmProviderSwitches++;
        }

        _alarmProviderSwitchesCounter.Add(
            1,
            new KeyValuePair<string, object?>("from", fromMode.ToString(CultureInfo.InvariantCulture)),
            new KeyValuePair<string, object?>("to", toMode.ToString(CultureInfo.InvariantCulture)),
            new KeyValuePair<string, object?>("reason", ReasonTag(reason)));
    }

    private static string ReasonTag(AlarmProviderSwitchReason reason) => reason switch
    {
        AlarmProviderSwitchReason.Failover => "failover",
        AlarmProviderSwitchReason.Failback => "failback",
        _ => "unknown",
    };

    /// <summary>Sets the current alarm provider-mode gauge without recording a switch (e.g. startup baseline).</summary>
    /// <param name="mode">The alarm provider mode value to record.</param>
    public void SetAlarmProviderMode(int mode)
    {
        lock (_syncRoot) { _alarmProviderMode = mode; }
    }

    /// <summary>
    /// Returns a snapshot of all current metric values.
    /// </summary>
    /// <returns>The current metrics snapshot.</returns>
    public GatewayMetricsSnapshot GetSnapshot()
    {
        // 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,
                GrpcEventStreamQueueDepth: grpcEventStreamQueueDepth,
                SessionsOpened: _sessionsOpened,
                SessionsClosed: _sessionsClosed,
                CommandsStarted: Interlocked.Read(ref _commandsStarted),
                CommandsSucceeded: Interlocked.Read(ref _commandsSucceeded),
                CommandsFailed: Interlocked.Read(ref _commandsFailed),
                EventsReceived: Interlocked.Read(ref _eventsReceived),
                QueueOverflows: _queueOverflows,
                Faults: _faults,
                WorkerKills: _workerKills,
                WorkerExits: _workerExits,
                HeartbeatFailures: _heartbeatFailures,
                StreamDisconnects: _streamDisconnects,
                RetryAttempts: _retryAttempts,
                AlarmProviderSwitchCount: _alarmProviderSwitches,
                CommandFailuresByMethod: new Dictionary<string, long>(_commandFailuresByMethod, StringComparer.OrdinalIgnoreCase),
                EventsByFamily: new Dictionary<string, long>(_eventsByFamily, StringComparer.OrdinalIgnoreCase),
                EventsBySession: new Dictionary<string, long>(_eventsBySession, StringComparer.Ordinal),
                RetryAttemptsByArea: new Dictionary<string, long>(_retryAttemptsByArea, StringComparer.OrdinalIgnoreCase));
        }
    }

    /// <summary>
    /// Disposes the underlying OpenTelemetry meter.
    /// </summary>
    public void Dispose()
    {
        if (_disposed)
        {
            return;
        }

        _meter.Dispose();
        _disposed = true;
    }

    private int GetOpenSessions()
    {
        lock (_syncRoot)
        {
            return _openSessions;
        }
    }

    private int GetWorkersRunning()
    {
        lock (_syncRoot)
        {
            return _workersRunning;
        }
    }

    // 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.
    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 static int SumSources(ConcurrentDictionary<long, Func<int>> sources)
    {
        int total = 0;
        foreach (Func<int> source in sources.Values)
        {
            int value = source();
            if (value > 0)
            {
                total += value;
            }
        }

        return total;
    }

    private int GetAlarmProviderMode()
    {
        lock (_syncRoot)
        {
            return _alarmProviderMode;
        }
    }

    private static void Increment(Dictionary<string, long> values, string key)
    {
        values.TryGetValue(key, out long currentValue);
        values[key] = currentValue + 1;
    }

    private static void Increment(ConcurrentDictionary<string, long> values, string key)
    {
        values.AddOrUpdate(key, 1, static (_, currentValue) => currentValue + 1);
    }

    // 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<long, Func<int>> sources, long id) : IDisposable
    {
        private int _disposed;

        public void Dispose()
        {
            if (Interlocked.Exchange(ref _disposed, 1) == 0)
            {
                sources.TryRemove(id, out _);
            }
        }
    }
}