using Admin.NET.Plugin.AiDOP.DataPlatform; using Admin.NET.Plugin.AiDOP.DataPlatform.Inbound; using Admin.NET.Plugin.AiDOP.Entity.DataPlatform; using Furion.Schedule; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; namespace Admin.NET.Plugin.AiDOP.Job; /// /// 定时探活 mdp_source:DB SELECT 1 / API GET baseUrl(或 /health)。 /// [JobDetail("job_mdp_source_health", Description = "MDP 数据源健康检查", GroupName = "default", Concurrent = false)] [PeriodSeconds(300, TriggerId = "trigger_mdp_source_health", Description = "每 5 分钟探活数据源", RunOnStart = false)] public class MdpSourceHealthCheckJob : IJob { private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; public MdpSourceHealthCheckJob(IServiceScopeFactory scopeFactory, ILoggerFactory loggerFactory) { _scopeFactory = scopeFactory; _logger = loggerFactory.CreateLogger(nameof(MdpSourceHealthCheckJob)); } public async Task ExecuteAsync(JobExecutingContext context, CancellationToken stoppingToken) { using var scope = _scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var scopeFactory = scope.ServiceProvider.GetRequiredService(); using var http = new HttpClient { Timeout = TimeSpan.FromSeconds(15) }; var sources = await db.Queryable().Where(x => x.Status == 1).ToListAsync(stoppingToken); var ok = 0; var fail = 0; foreach (var src in sources) { stoppingToken.ThrowIfCancellationRequested(); var now = DateTime.Now; try { if (string.Equals(src.SourceType, "DB", StringComparison.OrdinalIgnoreCase)) { var scopeDb = await scopeFactory.GetScopeAsync(src.SourceCode, stoppingToken); await scopeDb.Ado.GetIntAsync("SELECT 1"); src.HealthStatus = 1; src.HealthMsg = "OK"; ok++; } else if (string.Equals(src.SourceType, "API", StringComparison.OrdinalIgnoreCase)) { if (string.IsNullOrWhiteSpace(src.ApiBaseUrl)) throw new InvalidOperationException("api_base_url 为空"); var url = src.ApiBaseUrl.TrimEnd('/') + "/"; using var resp = await http.GetAsync(url, stoppingToken); src.HealthStatus = (int)resp.StatusCode is >= 200 and < 500 ? 1 : 0; src.HealthMsg = $"HTTP {(int)resp.StatusCode}"; if (src.HealthStatus == 1) ok++; else fail++; } else if (string.Equals(src.SourceType, "API_INBOUND", StringComparison.OrdinalIgnoreCase)) { src.HealthStatus = 1; src.HealthMsg = "OK"; ok++; } else { src.HealthStatus = 0; src.HealthMsg = $"未知 source_type={src.SourceType}"; fail++; } } catch (Exception ex) { src.HealthStatus = 0; src.HealthMsg = ex.Message.Length > 480 ? ex.Message[..480] : ex.Message; fail++; } src.LastHealthCheck = now; src.UpdateTime = now; await db.Updateable(src) .UpdateColumns(x => new { x.HealthStatus, x.HealthMsg, x.LastHealthCheck, x.UpdateTime }) .ExecuteCommandAsync(stoppingToken); } if (sources.Count > 0) _logger.LogInformation("[MdpSourceHealthCheckJob] total={Total} ok={Ok} fail={Fail}", sources.Count, ok, fail); await ExpireInboundSnapshotsAsync(scope, stoppingToken); await CheckInboundSilenceAsync(db, sources, stoppingToken); } /// 顺带把 OPEN 且已过期的快照置 EXPIRED;未 commit 不产生删除语义。 private async Task ExpireInboundSnapshotsAsync(IServiceScope scope, CancellationToken ct) { try { var snaps = scope.ServiceProvider.GetRequiredService(); await snaps.ExpireStaleOpenAsync(ct); } catch (Exception ex) { _logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound snapshot expire failed"); } } /// /// 入站实体沉默:inbound_enabled=1 且配置了 silence_alert_hours, /// 最近 COMMITTED 超过阈值且当天命中 silence_calendar(空=每天)则写回源健康告警。 /// 从未推送过的实体以 update_time/create_time 为基线。 /// private async Task CheckInboundSilenceAsync( ISqlSugarClient db, List sources, CancellationToken ct) { List entities; try { entities = await db.Queryable() .Where(e => e.InboundEnabled == 1 && e.SilenceAlertHours != null) .ToListAsync(ct); } catch (Exception ex) { _logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound silence query entities failed"); return; } if (entities.Count == 0) return; var today = DateTime.Now; var isoDow = today.DayOfWeek == DayOfWeek.Sunday ? 7 : (int)today.DayOfWeek; var alertsBySource = new Dictionary>(); foreach (var entity in entities) { ct.ThrowIfCancellationRequested(); if (!HitsSilenceCalendar(entity.SilenceCalendar, isoDow)) continue; DateTime? lastCommitted = null; try { var last = await db.Queryable() .Where(r => r.Status == "COMMITTED") .Where("UPPER(entity_code) = @code", new SugarParameter("@code", entity.EntityCode.ToUpperInvariant())) .OrderBy(r => r.CreateTime, OrderByType.Desc) .FirstAsync(ct); lastCommitted = last?.CreateTime; } catch (Exception ex) { _logger.LogWarning(ex, "[MdpSourceHealthCheckJob] inbound silence query requests failed entity={Entity}", entity.EntityCode); continue; } var baseline = lastCommitted ?? (entity.UpdateTime != default ? entity.UpdateTime : entity.CreateTime); var hours = entity.SilenceAlertHours.GetValueOrDefault(); if (hours <= 0) continue; if ((today - baseline).TotalHours <= hours) continue; var lastText = lastCommitted.HasValue ? lastCommitted.Value.ToString("yyyy-MM-dd HH:mm:ss") : "never"; var msg = $"INBOUND silence: {entity.EntityCode} last COMMITTED {lastText} (threshold {hours}h)"; _logger.LogWarning("[MdpSourceHealthCheckJob] {Message}", msg); if (!alertsBySource.TryGetValue(entity.SourceId, out var list)) { list = []; alertsBySource[entity.SourceId] = list; } list.Add(msg); } var sourceById = sources.ToDictionary(s => s.Id); foreach (var (sourceId, messages) in alertsBySource) { if (!sourceById.TryGetValue(sourceId, out var src)) { src = await db.Queryable().Where(s => s.Id == sourceId).FirstAsync(ct); if (src == null) continue; } var now = DateTime.Now; src.HealthStatus = 0; src.HealthMsg = Truncate(string.Join("; ", messages), 480); src.LastHealthCheck = now; src.UpdateTime = now; await db.Updateable(src) .UpdateColumns(x => new { x.HealthStatus, x.HealthMsg, x.LastHealthCheck, x.UpdateTime }) .ExecuteCommandAsync(ct); } } private static bool HitsSilenceCalendar(string calendar, int isoDow) { if (string.IsNullOrWhiteSpace(calendar)) return true; var days = calendar.Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries); return days.Any(d => int.TryParse(d, out var n) && n == isoDow); } private static string Truncate(string s, int max) => string.IsNullOrEmpty(s) || s.Length <= max ? s : s[..max]; }