using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ScadaLink.Commons.Messages.Health; using ScadaLink.StoreAndForward; namespace ScadaLink.HealthMonitoring; /// /// Periodically collects a SiteHealthReport and sends it to central via Akka remoting. /// Sequence numbers are monotonic and reset on service restart. They are not /// zero/one-based: the per-process counter is seeded with the current Unix epoch /// (milliseconds) at construction so that, after a failover, reports from a /// freshly-active node always sort after reports from any prior active node for the /// same site — otherwise the central aggregator's sequence-number guard would /// silently reject the new active's first reports as stale. /// public class HealthReportSender : BackgroundService { private readonly ISiteHealthCollector _collector; private readonly IHealthReportTransport _transport; private readonly HealthMonitoringOptions _options; private readonly ILogger _logger; private readonly string _siteId; private readonly StoreAndForwardStorage? _sfStorage; private readonly IClusterNodeProvider? _clusterNodeProvider; // Seeded with Unix-ms at construction so reports from a freshly-active // node always sort after reports from any prior active node for the same // site. Without this seeding, failover would silently drop the new // active's first reports because their per-process counter starts below // the prior active's last sequence number. The clock is read through the // injected TimeProvider so the seeding is deterministically testable. private long _sequenceNumber; public HealthReportSender( ISiteHealthCollector collector, IHealthReportTransport transport, IOptions options, ILogger logger, ISiteIdentityProvider siteIdentityProvider, StoreAndForwardStorage? sfStorage = null, IClusterNodeProvider? clusterNodeProvider = null, TimeProvider? timeProvider = null) { _collector = collector; _transport = transport; _options = options.Value; _logger = logger; _siteId = siteIdentityProvider.SiteId; _sfStorage = sfStorage; _clusterNodeProvider = clusterNodeProvider; _sequenceNumber = (timeProvider ?? TimeProvider.System).GetUtcNow().ToUnixTimeMilliseconds(); } /// /// Current sequence number (for testing). /// public long CurrentSequenceNumber => Interlocked.Read(ref _sequenceNumber); protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _logger.LogInformation( "Health report sender starting for site {SiteId}, interval {Interval}s", _siteId, _options.ReportInterval.TotalSeconds); using var timer = new PeriodicTimer(_options.ReportInterval); while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) { try { // Only the active node (running the DeploymentManager singleton) sends health reports. // The standby node has no instance/connection data and would overwrite the active's report. if (!_collector.IsActiveNode) continue; if (_clusterNodeProvider != null) { try { _collector.SetClusterNodes(_clusterNodeProvider.GetClusterNodes()); } catch (Exception ex) { // Non-fatal — the report ships with the previous cluster // node list. Logged so a persistent failure is diagnosable. _logger.LogWarning(ex, "Failed to refresh cluster nodes for health report (site {SiteId}); using stale list", _siteId); } } if (_sfStorage != null) { try { var parkedCount = await _sfStorage.GetParkedMessageCountAsync(); _collector.SetParkedMessageCount(parkedCount); } catch (Exception ex) { // Non-fatal — parked count will be 0 in this report. _logger.LogWarning(ex, "Failed to query parked message count for health report (site {SiteId})", _siteId); } try { // Per-category pending-message buffer depths (the documented // "store-and-forward buffer depth" triage metric). Keyed by // StoreAndForwardCategory name so the central dashboard can // render external/notification/DB-write depths separately. var depthsByCategory = await _sfStorage.GetBufferDepthByCategoryAsync(); var depths = depthsByCategory.ToDictionary( kvp => kvp.Key.ToString(), kvp => kvp.Value); _collector.SetStoreAndForwardDepths(depths); } catch (Exception ex) { // Non-fatal — buffer depths will be empty in this report. _logger.LogWarning(ex, "Failed to query store-and-forward buffer depths for health report (site {SiteId})", _siteId); } } var seq = Interlocked.Increment(ref _sequenceNumber); var report = _collector.CollectReport(_siteId); // Replace the placeholder sequence number with our monotonic one var reportWithSeq = report with { SequenceNumber = seq }; _transport.Send(reportWithSeq); _logger.LogInformation("Sent health report #{Seq} for site {SiteId}", seq, _siteId); } catch (Exception ex) { _logger.LogError(ex, "Failed to send health report for site {SiteId}", _siteId); // Continue sending — don't let a single failure stop reporting } } } }