AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Infrastructure/EtlInstanceRegistrar.cs

using Admin.NET.Plugin.AiDOP.Entity.SmartOps;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace Admin.NET.Plugin.AiDOP.Infrastructure;

/// <summary>
/// ETL 执行机实例注册器：把本进程登记进 <c>ado_etl_instance</c> 并定期心跳，
/// 同时把本实例的指派位发布到 <see cref="AidopRunnerState"/> 供 <see cref="AidopJobGate"/> 读取。
///
/// <para><b>只读指派、从不自行指派</b>：本注册器插入新行时 <c>is_runner</c> 恒为 0。
/// 指派是人工动作（超管在 MDP 运行监控页操作，或直接改库）。
/// 禁止在这里实现「抢锁自荐」或「心跳超时自动接管」——
/// 那会让开发机在某次重启后意外变成执行机，正是本次治理要消除的情形。</para>
///
/// <para><b>本服务不受 <see cref="AidopJobGate"/> 管辖</b>，必须常开。
/// 它是闸门的数据来源，被闸门关掉就成了死锁：没人心跳 → 没人发布指派 → 永远不是执行机。</para>
/// </summary>
public sealed class EtlInstanceRegistrar : BackgroundService
{
    /// <summary>心跳周期。<see cref="AidopRunnerState.FreshnessWindow"/> 按其 3 倍取值。</summary>
    public static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(20);

    /// <summary>启动延时：等宿主配置与建表脚本就绪，避免首拍必然失败刷一条 warn。</summary>
    private static readonly TimeSpan StartupDelay = TimeSpan.FromSeconds(5);

    /// <summary>下线实例行的保留期。超期且未被指派的行会被清理，防止表随重启无限增长。</summary>
    private static readonly TimeSpan DeadInstanceRetention = TimeSpan.FromDays(7);

    private readonly IServiceScopeFactory _scopeFactory;
    private readonly ILogger _logger;
    private int _consecutiveFailures;
    private int _guardrailsLogged;

    public EtlInstanceRegistrar(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory)
    {
        _scopeFactory = scopeFactory;
        _logger = loggerFactory.CreateLogger(nameof(EtlInstanceRegistrar));
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try { await Task.Delay(StartupDelay, stoppingToken); }
        catch (OperationCanceledException) { return; }

        _logger.LogInformation(
            "[EtlInstanceRegistrar] 本实例 instanceId={InstanceId} machine={Machine} pid={Pid} version={Version}",
            AidopInstanceIdentity.InstanceId, AidopInstanceIdentity.MachineName,
            AidopInstanceIdentity.ProcessId, AidopInstanceIdentity.AppVersion);

        var nextCleanupAt = DateTimeOffset.UtcNow;
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var assigned = await HeartbeatAsync(stoppingToken);
                AidopRunnerState.Publish(assigned);

                if (Interlocked.Exchange(ref _guardrailsLogged, 1) == 0)
                    await LogMySqlGuardrailsAsync(assigned, stoppingToken);

                if (_consecutiveFailures > 0)
                {
                    _logger.LogInformation(
                        "[EtlInstanceRegistrar] 心跳恢复，连续失败 {Count} 次后已重新发布指派 isRunner={IsRunner}",
                        _consecutiveFailures, assigned);
                    _consecutiveFailures = 0;
                }

                if (DateTimeOffset.UtcNow >= nextCleanupAt)
                {
                    await CleanupDeadAsync(stoppingToken);
                    nextCleanupAt = DateTimeOffset.UtcNow + TimeSpan.FromHours(6);
                }
            }
            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
            {
                break;
            }
            catch (Exception ex)
            {
                // 查库失败必须打回未知，让闸门回退到配置值：
                // 指派可能已被运维改掉，继续沿用旧值会让两台实例同时自认执行机。
                AidopRunnerState.Invalidate();

                // 库不可用时本循环每 20 秒一次，不做节流会刷屏淹没真正的错误。
                _consecutiveFailures++;
                if (_consecutiveFailures == 1 || _consecutiveFailures % 30 == 0)
                    _logger.LogWarning(ex,
                        "[EtlInstanceRegistrar] 心跳失败（连续第 {Count} 次），指派状态已置未知，闸门回退到配置值",
                        _consecutiveFailures);
            }

            try { await Task.Delay(HeartbeatInterval, stoppingToken); }
            catch (OperationCanceledException) { break; }
        }
    }

    /// <summary>
    /// 心跳一拍：更新本实例行（不存在则插入），回读指派位。
    /// 返回本实例是否被指派为执行机。
    /// </summary>
    private async Task<bool> HeartbeatAsync(CancellationToken ct)
    {
        using var scope = _scopeFactory.CreateScope();
        var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
        var instanceId = AidopInstanceIdentity.InstanceId;
        var now = DateTime.Now;

        // 刻意不更新 is_runner：指派归人工，注册器只报到。
        var affected = await db.Updateable<AdoEtlInstance>()
            .SetColumns(x => new AdoEtlInstance
            {
                LastHeartbeatAt = now,
                AppVersion = AidopInstanceIdentity.AppVersion,
                UpdateTime = now
            })
            .Where(x => x.InstanceId == instanceId)
            .ExecuteCommandAsync(ct);

        if (affected <= 0)
        {
            await db.Insertable(new AdoEtlInstance
            {
                InstanceId = instanceId,
                MachineName = AidopInstanceIdentity.MachineName,
                ProcessId = AidopInstanceIdentity.ProcessId,
                AppVersion = AidopInstanceIdentity.AppVersion,
                IsRunner = false,
                StartedAt = AidopInstanceIdentity.StartedAt,
                LastHeartbeatAt = now,
                CreateTime = now,
                UpdateTime = now
            }).ExecuteCommandAsync(ct);
        }

        return await db.Queryable<AdoEtlInstance>()
            .Where(x => x.InstanceId == instanceId)
            .Select(x => x.IsRunner)
            .FirstAsync(ct);
    }

    /// <summary>
    /// 首次心跳成功后打一条现场参数，全进程只打一次。
    ///
    /// <para><b>为什么要打</b>：2026-09-24 的 1040 事故里，165 的
    /// <c>innodb_buffer_pool_size</c> 一直是默认 128MB、<c>max_connections</c> 是默认 151，
    /// 而没有任何人能从日志看出这一点——只能事后 SSH 上去查。
    /// 把它连同执行机身份打进启动日志，下次排障第一眼就能看到。</para>
    ///
    /// <para>不打连接串、不打任何口令。失败只 warn，不影响心跳。</para>
    /// </summary>
    private async Task LogMySqlGuardrailsAsync(bool assigned, CancellationToken ct)
    {
        try
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var row = await db.Ado.SqlQuerySingleAsync<MySqlGuardrails>(
                "SELECT @@max_connections AS MaxConnections, "
                + "@@innodb_buffer_pool_size AS BufferPoolBytes, "
                + "DATABASE() AS DbName");

            _logger.LogInformation(
                "[EtlInstanceRegistrar] 现场参数 db={Db} maxConnections={MaxConn} bufferPoolMB={PoolMb} "
                + "isRunner={IsRunner} instanceId={InstanceId}",
                row?.DbName, row?.MaxConnections, row?.BufferPoolBytes / 1024 / 1024,
                assigned, AidopInstanceIdentity.InstanceId);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "[EtlInstanceRegistrar] 读取现场 MySQL 参数失败，跳过");
        }
    }

    private sealed class MySqlGuardrails
    {
        public long MaxConnections { get; set; }
        public long BufferPoolBytes { get; set; }
        public string DbName { get; set; }
    }

    /// <summary>
    /// 清理早已下线的实例行。被指派的那行即使心跳超期也保留——
    /// 运维需要看到「指派的机器挂了」，而不是记录凭空消失。
    /// </summary>
    private async Task CleanupDeadAsync(CancellationToken ct)
    {
        try
        {
            using var scope = _scopeFactory.CreateScope();
            var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
            var cutoff = DateTime.Now - DeadInstanceRetention;
            var removed = await db.Deleteable<AdoEtlInstance>()
                .Where(x => !x.IsRunner && x.LastHeartbeatAt < cutoff)
                .ExecuteCommandAsync(ct);
            if (removed > 0)
                _logger.LogInformation("[EtlInstanceRegistrar] 已清理下线实例 {Count} 行", removed);
        }
        catch (Exception ex)
        {
            // 清理失败不影响心跳与指派，吞掉即可
            _logger.LogWarning(ex, "[EtlInstanceRegistrar] 清理下线实例失败");
        }
    }
}