|
|
@@ -7,13 +7,24 @@ namespace Admin.NET.Plugin.AiDOP.Infrastructure;
|
|
|
|
|
|
/// <summary>
|
|
|
/// ETL 执行机实例注册器:把本进程登记进 <c>ado_etl_instance</c> 并定期心跳,
|
|
|
-/// 同时把本实例的指派位发布到 <see cref="AidopRunnerState"/> 供 <see cref="AidopJobGate"/> 读取。
|
|
|
+/// 同时把本实例是否为执行机发布到 <see cref="AidopRunnerState"/> 供 <see cref="AidopJobGate"/> 读取。
|
|
|
///
|
|
|
-/// <para><b>只读指派、从不自行指派</b>:本注册器插入新行时 <c>is_runner</c> 恒为 0。
|
|
|
-/// 指派是人工动作(超管在 MDP 运行监控页操作,或直接改库)。
|
|
|
+/// <para><b>判定方式:槽位比对</b>。本进程声明自己属于哪个部署槽位
|
|
|
+/// (<c>AidopInstanceIdentity.SlotCode</c>),指派记在 <c>ado_etl_runner_designation</c>;
|
|
|
+/// 两者相等即为执行机。<b>未声明槽位者永不是执行机</b>。
|
|
|
+/// 之所以不拿 <c>instance_id</c> 比对,是因为它含进程ID与启动时间,
|
|
|
+/// 指派挂上去每次重启都会丢——2026-09-28 的 28 条积压即由此而来。</para>
|
|
|
+///
|
|
|
+/// <para><b>只读指派、从不自行指派</b>:本注册器插入新行时 <c>is_runner</c> 恒为 0,
|
|
|
+/// 也绝不写指派表。指派是人工动作(超管在 MDP 运行监控页操作,或直接改库)。
|
|
|
/// 禁止在这里实现「抢锁自荐」或「心跳超时自动接管」——
|
|
|
/// 那会让开发机在某次重启后意外变成执行机,正是本次治理要消除的情形。</para>
|
|
|
///
|
|
|
+/// <para><b>同槽位撞车则两边都不跑</b>:若同一槽位另有存活实例(滚动重启的重叠窗口,
|
|
|
+/// 或有人把两个进程配成同一槽位),本进程发布 false 而不是照常跑。
|
|
|
+/// 没有租约就没有仲裁者,此时抢着跑等于双跑全量,那正是 2026-09-24 连接耗尽的成因;
|
|
|
+/// 代价只是重启重叠的几十秒空窗,定时作业不受影响。</para>
|
|
|
+///
|
|
|
/// <para><b>本服务不受 <see cref="AidopJobGate"/> 管辖</b>,必须常开。
|
|
|
/// 它是闸门的数据来源,被闸门关掉就成了死锁:没人心跳 → 没人发布指派 → 永远不是执行机。</para>
|
|
|
/// </summary>
|
|
|
@@ -25,9 +36,12 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
/// <summary>启动延时:等宿主配置与建表脚本就绪,避免首拍必然失败刷一条 warn。</summary>
|
|
|
private static readonly TimeSpan StartupDelay = TimeSpan.FromSeconds(5);
|
|
|
|
|
|
- /// <summary>下线实例行的保留期。超期且未被指派的行会被清理,防止表随重启无限增长。</summary>
|
|
|
+ /// <summary>下线实例行的保留期。超期且不在被指派槽位上的行会被清理,防止表随重启无限增长。</summary>
|
|
|
private static readonly TimeSpan DeadInstanceRetention = TimeSpan.FromDays(7);
|
|
|
|
|
|
+ /// <summary>存活判据:连续三拍没心跳即视为下线,与 <see cref="AidopRunnerState.FreshnessWindow"/> 同宽。</summary>
|
|
|
+ private static readonly TimeSpan LivenessWindow = TimeSpan.FromSeconds(60);
|
|
|
+
|
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
|
private readonly ILogger _logger;
|
|
|
private int _consecutiveFailures;
|
|
|
@@ -45,26 +59,44 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
catch (OperationCanceledException) { return; }
|
|
|
|
|
|
_logger.LogInformation(
|
|
|
- "[EtlInstanceRegistrar] 本实例 instanceId={InstanceId} machine={Machine} pid={Pid} version={Version}",
|
|
|
+ "[EtlInstanceRegistrar] 本实例 instanceId={InstanceId} machine={Machine} pid={Pid} version={Version} slot={Slot}",
|
|
|
AidopInstanceIdentity.InstanceId, AidopInstanceIdentity.MachineName,
|
|
|
- AidopInstanceIdentity.ProcessId, AidopInstanceIdentity.AppVersion);
|
|
|
+ AidopInstanceIdentity.ProcessId, AidopInstanceIdentity.AppVersion,
|
|
|
+ AidopInstanceIdentity.SlotCode ?? "(未声明)");
|
|
|
|
|
|
var nextCleanupAt = DateTimeOffset.UtcNow;
|
|
|
+ var lastConflictLogged = false;
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
|
{
|
|
|
try
|
|
|
{
|
|
|
- var assigned = await HeartbeatAsync(stoppingToken);
|
|
|
- AidopRunnerState.Publish(assigned);
|
|
|
+ var beat = await HeartbeatAsync(stoppingToken);
|
|
|
+ AidopRunnerState.Publish(beat.IsRunner);
|
|
|
+
|
|
|
+ // 撞车是配置事故或滚动重启重叠,两者都必须让人看见;
|
|
|
+ // 但每 20 秒一条会刷屏,故只在状态翻转时打。
|
|
|
+ if (beat.SlotConflict != lastConflictLogged)
|
|
|
+ {
|
|
|
+ if (beat.SlotConflict)
|
|
|
+ _logger.LogWarning(
|
|
|
+ "[EtlInstanceRegistrar] 槽位撞车:slot={Slot} 另有 {Count} 个存活实例,"
|
|
|
+ + "本实例暂不作为执行机(避免双跑全量)",
|
|
|
+ AidopInstanceIdentity.SlotCode, beat.SameSlotLiveOthers);
|
|
|
+ else
|
|
|
+ _logger.LogInformation(
|
|
|
+ "[EtlInstanceRegistrar] 槽位撞车已解除 slot={Slot} isRunner={IsRunner}",
|
|
|
+ AidopInstanceIdentity.SlotCode, beat.IsRunner);
|
|
|
+ lastConflictLogged = beat.SlotConflict;
|
|
|
+ }
|
|
|
|
|
|
if (Interlocked.Exchange(ref _guardrailsLogged, 1) == 0)
|
|
|
- await LogMySqlGuardrailsAsync(assigned, stoppingToken);
|
|
|
+ await LogMySqlGuardrailsAsync(beat.IsRunner, stoppingToken);
|
|
|
|
|
|
if (_consecutiveFailures > 0)
|
|
|
{
|
|
|
_logger.LogInformation(
|
|
|
"[EtlInstanceRegistrar] 心跳恢复,连续失败 {Count} 次后已重新发布指派 isRunner={IsRunner}",
|
|
|
- _consecutiveFailures, assigned);
|
|
|
+ _consecutiveFailures, beat.IsRunner);
|
|
|
_consecutiveFailures = 0;
|
|
|
}
|
|
|
|
|
|
@@ -97,23 +129,31 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /// <summary>一拍心跳的结果。</summary>
|
|
|
+ private readonly record struct HeartbeatResult(bool IsRunner, string DesignatedSlot, int SameSlotLiveOthers)
|
|
|
+ {
|
|
|
+ public bool SlotConflict => SameSlotLiveOthers > 0;
|
|
|
+ }
|
|
|
+
|
|
|
/// <summary>
|
|
|
- /// 心跳一拍:更新本实例行(不存在则插入),回读指派位。
|
|
|
- /// 返回本实例是否被指派为执行机。
|
|
|
+ /// 心跳一拍:更新本实例行(不存在则插入),读取指派槽位并与本进程槽位比对。
|
|
|
/// </summary>
|
|
|
- private async Task<bool> HeartbeatAsync(CancellationToken ct)
|
|
|
+ private async Task<HeartbeatResult> HeartbeatAsync(CancellationToken ct)
|
|
|
{
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
|
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
|
|
|
+ var instances = scope.ServiceProvider.GetRequiredService<IEtlInstanceStore>();
|
|
|
var instanceId = AidopInstanceIdentity.InstanceId;
|
|
|
+ var slot = AidopInstanceIdentity.SlotCode;
|
|
|
var now = DateTime.Now;
|
|
|
|
|
|
- // 刻意不更新 is_runner:指派归人工,注册器只报到。
|
|
|
+ // 刻意不更新 is_runner:那已是遗留列,指派在 ado_etl_runner_designation。
|
|
|
var affected = await db.Updateable<AdoEtlInstance>()
|
|
|
.SetColumns(x => new AdoEtlInstance
|
|
|
{
|
|
|
LastHeartbeatAt = now,
|
|
|
AppVersion = AidopInstanceIdentity.AppVersion,
|
|
|
+ SlotCode = slot,
|
|
|
UpdateTime = now
|
|
|
})
|
|
|
.Where(x => x.InstanceId == instanceId)
|
|
|
@@ -127,6 +167,7 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
MachineName = AidopInstanceIdentity.MachineName,
|
|
|
ProcessId = AidopInstanceIdentity.ProcessId,
|
|
|
AppVersion = AidopInstanceIdentity.AppVersion,
|
|
|
+ SlotCode = slot,
|
|
|
IsRunner = false,
|
|
|
StartedAt = AidopInstanceIdentity.StartedAt,
|
|
|
LastHeartbeatAt = now,
|
|
|
@@ -135,10 +176,19 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
}).ExecuteCommandAsync(ct);
|
|
|
}
|
|
|
|
|
|
- return await db.Queryable<AdoEtlInstance>()
|
|
|
- .Where(x => x.InstanceId == instanceId)
|
|
|
- .Select(x => x.IsRunner)
|
|
|
- .FirstAsync(ct);
|
|
|
+ var designated = await instances.GetDesignatedSlotAsync(ct);
|
|
|
+
|
|
|
+ // 没声明槽位就不是候选,连撞车都不必查——省一次查询,也省得把「未声明」
|
|
|
+ // 误报成撞车(多台未配置的开发机 slot 全是 NULL,NULL 之间不构成同槽位)。
|
|
|
+ if (string.IsNullOrEmpty(slot) || !string.Equals(slot, designated, StringComparison.Ordinal))
|
|
|
+ return new HeartbeatResult(false, designated, 0);
|
|
|
+
|
|
|
+ var liveCutoff = now - LivenessWindow;
|
|
|
+ var others = await db.Queryable<AdoEtlInstance>()
|
|
|
+ .Where(x => x.SlotCode == slot && x.InstanceId != instanceId && x.LastHeartbeatAt >= liveCutoff)
|
|
|
+ .CountAsync(ct);
|
|
|
+
|
|
|
+ return new HeartbeatResult(others == 0, designated, others);
|
|
|
}
|
|
|
|
|
|
/// <summary>
|
|
|
@@ -182,8 +232,12 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
}
|
|
|
|
|
|
/// <summary>
|
|
|
- /// 清理早已下线的实例行。被指派的那行即使心跳超期也保留——
|
|
|
- /// 运维需要看到「指派的机器挂了」,而不是记录凭空消失。
|
|
|
+ /// 清理早已下线的实例行。**被指派槽位上的行即使心跳超期也保留**——
|
|
|
+ /// 运维需要看到「指派槽位上那台机器挂了」,而不是记录凭空消失。
|
|
|
+ ///
|
|
|
+ /// <para>注意这里保留的只是**观测线索**:指派本身存在
|
|
|
+ /// <c>ado_etl_runner_designation</c>,即使这些行被清掉也不会丢指派。
|
|
|
+ /// 旧模型正相反——指派就寄生在实例行上,所以那时必须靠这个过滤条件兜着。</para>
|
|
|
/// </summary>
|
|
|
private async Task CleanupDeadAsync(CancellationToken ct)
|
|
|
{
|
|
|
@@ -191,9 +245,14 @@ public sealed class EtlInstanceRegistrar : BackgroundService
|
|
|
{
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
|
var db = scope.ServiceProvider.GetRequiredService<ISqlSugarClient>();
|
|
|
+ var instances = scope.ServiceProvider.GetRequiredService<IEtlInstanceStore>();
|
|
|
+ var designated = await instances.GetDesignatedSlotAsync(ct);
|
|
|
var cutoff = DateTime.Now - DeadInstanceRetention;
|
|
|
var removed = await db.Deleteable<AdoEtlInstance>()
|
|
|
- .Where(x => !x.IsRunner && x.LastHeartbeatAt < cutoff)
|
|
|
+ .Where(x => x.LastHeartbeatAt < cutoff)
|
|
|
+ // 必须显式放行 slot_code IS NULL:SQL 里 NULL <> 'x' 求值为 NULL 而非真,
|
|
|
+ // 少了这一支,未声明槽位的开发机死行会永远清不掉。
|
|
|
+ .WhereIF(designated != null, x => x.SlotCode == null || x.SlotCode != designated)
|
|
|
.ExecuteCommandAsync(ct);
|
|
|
if (removed > 0)
|
|
|
_logger.LogInformation("[EtlInstanceRegistrar] 已清理下线实例 {Count} 行", removed);
|