scadaproj/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/LocalDbMetrics.cs

using System.Diagnostics.Metrics;

namespace ZB.MOM.WW.LocalDb.Replication;

/// <summary>
/// The replication engine's OpenTelemetry instruments, under the single
/// <c>ZB.MOM.WW.LocalDb.Replication</c> meter (family METRIC-CONVENTIONS §1: meters are named
/// after their root namespace). Optional throughout: the sessions and adapters hold a nullable
/// reference and every call is null-guarded, so replication runs identically whether or not this
/// is registered. Instrument names follow the family metric conventions (lower-case,
/// dot-separated, seconds for durations).
/// </summary>
public sealed class LocalDbMetrics : IDisposable
{
    /// <summary>The single meter name every replication instrument is published under.</summary>
    public const string MeterName = "ZB.MOM.WW.LocalDb.Replication";

    private readonly Meter _meter;
    private readonly Counter<long> _applied;
    private readonly Counter<long> _conflictsDiscarded;
    private readonly Counter<long> _deadLettered;
    private readonly Counter<long> _reconnects;
    private readonly Counter<long> _snapshots;
    private readonly Func<DateTimeOffset> _utcNow;
    private bool _disposed;

    /// <summary>Creates the meter and its instruments. <paramref name="utcNow"/> is a test seam for the lag gauge's clock; null uses <see cref="DateTimeOffset.UtcNow"/>.</summary>
    public LocalDbMetrics(Func<DateTimeOffset>? utcNow = null)
    {
        _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
        _meter = new Meter(MeterName, typeof(LocalDbMetrics).Assembly.GetName().Version?.ToString(3));
        _applied = _meter.CreateCounter<long>("localdb.sync.applied");
        _conflictsDiscarded = _meter.CreateCounter<long>("localdb.sync.conflicts_discarded");
        _deadLettered = _meter.CreateCounter<long>("localdb.sync.dead_lettered");
        _reconnects = _meter.CreateCounter<long>("localdb.sync.reconnects");
        _snapshots = _meter.CreateCounter<long>("localdb.sync.snapshots");
        _meter.CreateObservableGauge("localdb.oplog.depth", ObserveOplogDepth, "1");
        _meter.CreateObservableGauge("localdb.sync.lag.seconds", ObserveSyncLag, "s");
    }

    /// <summary>
    /// Supplies the current unacked oplog backlog for the <c>localdb.oplog.depth</c> gauge. Invoked
    /// only at collection time; a cheap synchronous SELECT is acceptable at that frequency.
    /// </summary>
    public Func<long>? OplogDepthProvider { get; internal set; }

    /// <summary>Supplies the last successful sync time for the <c>localdb.sync.lag.seconds</c> gauge; null means no sync yet, and the gauge then reports no measurement.</summary>
    public Func<DateTimeOffset?>? LastSyncUtcProvider { get; internal set; }

    /// <summary>The backing meter — exposed so a test <see cref="MeterListener"/> can filter by instance reference rather than the process-shared meter name.</summary>
    internal Meter Meter => _meter;

    /// <summary>Adds <paramref name="count"/> rows applied from inbound delta batches to <c>localdb.sync.applied</c> (no-op when not positive).</summary>
    public void RecordApplied(long count)
    {
        if (count > 0) _applied.Add(count);
    }

    /// <summary>Adds <paramref name="count"/> inbound rows that lost LWW to <c>localdb.sync.conflicts_discarded</c> (no-op when not positive).</summary>
    public void RecordConflictsDiscarded(long count)
    {
        if (count > 0) _conflictsDiscarded.Add(count);
    }

    /// <summary>Adds <paramref name="count"/> poison / over-drift rows routed to the dead-letter table to <c>localdb.sync.dead_lettered</c> (no-op when not positive).</summary>
    public void RecordDeadLettered(long count)
    {
        if (count > 0) _deadLettered.Add(count);
    }

    /// <summary>Increments <c>localdb.sync.reconnects</c> for one initiator reconnect attempt.</summary>
    public void RecordReconnect() => _reconnects.Add(1);

    /// <summary>Increments <c>localdb.sync.snapshots</c> with <c>direction=sent</c> for one completed outbound snapshot.</summary>
    public void RecordSnapshotSent() =>
        _snapshots.Add(1, new KeyValuePair<string, object?>("direction", "sent"));

    /// <summary>Increments <c>localdb.sync.snapshots</c> with <c>direction=received</c> for one completed inbound snapshot.</summary>
    public void RecordSnapshotReceived() =>
        _snapshots.Add(1, new KeyValuePair<string, object?>("direction", "received"));

    private IEnumerable<Measurement<long>> ObserveOplogDepth()
    {
        var provider = OplogDepthProvider;
        if (provider is null)
            return [];
        try
        {
            return [new Measurement<long>(provider())];
        }
        catch
        {
            // The observation SELECT can race a disposed db / closing connection; a scrape must
            // never throw, so drop this sample rather than surface a partial value.
            return [];
        }
    }

    private IEnumerable<Measurement<double>> ObserveSyncLag()
    {
        var last = LastSyncUtcProvider?.Invoke();
        if (last is null)
            return [];
        var lag = (_utcNow() - last.Value).TotalSeconds;
        if (lag < 0) lag = 0;
        return [new Measurement<double>(lag)];
    }

    /// <summary>Disposes the backing meter; idempotent.</summary>
    public void Dispose()
    {
        if (_disposed)
            return;
        _meter.Dispose();
        _disposed = true;
    }
}