新增 OTA WatchDog 编排与车队健康/报警后端。
覆盖包库回传、任务下发、CDM 任务同步与报警采集，并为包/任务 ID 与上传文件名加上路径安全校验。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zhaowei.huang
Cursor
@@ -0,0 +1,169 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Options;
|
||||
using MiGu.Server.Auth;
|
||||
using MiGu.Server.Launcher;
|
||||
using MiGu.Server.Ota;
|
||||
|
||||
namespace MiGu.Server.Fleet;
|
||||
|
||||
/// <summary>
|
||||
/// 车队健康：保留 SimpleLite 的 CPU/故障率等，延迟改为对 WatchDog(:9776) 做 TCP RTT。
|
||||
/// SimpleLite /fleet/health 探测的是车载 HTTP :8081，多数现场未开该端口会假超时 2000ms。
|
||||
/// </summary>
|
||||
public sealed class FleetHealthService
|
||||
{
|
||||
private readonly IHttpClientFactory _httpFactory;
|
||||
private readonly SimpleLiteOptions _sl;
|
||||
private readonly InternalTokenStore _token;
|
||||
private readonly WatchDogClient _wd;
|
||||
private readonly ILogger<FleetHealthService> _log;
|
||||
|
||||
private static readonly JsonSerializerOptions JsonOpt = new()
|
||||
{
|
||||
PropertyNameCaseInsensitive = true
|
||||
};
|
||||
|
||||
public FleetHealthService(
|
||||
IHttpClientFactory httpFactory,
|
||||
IOptions<SimpleLiteOptions> sl,
|
||||
InternalTokenStore token,
|
||||
WatchDogClient wd,
|
||||
ILogger<FleetHealthService> log)
|
||||
{
|
||||
_httpFactory = httpFactory;
|
||||
_sl = sl.Value;
|
||||
_token = token;
|
||||
_wd = wd;
|
||||
_log = log;
|
||||
}
|
||||
|
||||
public async Task<List<FleetHealthRowDto>> GetAsync(CancellationToken ct)
|
||||
{
|
||||
var rows = await FetchSimpleLiteHealthAsync(ct);
|
||||
if (rows.Count == 0)
|
||||
rows = await BuildRowsFromCarsAsync(ct);
|
||||
|
||||
await Parallel.ForEachAsync(
|
||||
rows,
|
||||
new ParallelOptions { MaxDegreeOfParallelism = 8, CancellationToken = ct },
|
||||
async (row, token) =>
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(row.Ip))
|
||||
{
|
||||
// 无 IP 时保留 SimpleLite 原探测结果
|
||||
row.LatencySource ??= row.LatencyMs != null ? "onboard" : "none";
|
||||
return;
|
||||
}
|
||||
|
||||
// 取两次 TCP 连接的较小值，降低偶发握手抖动
|
||||
var a = await _wd.MeasureRttMsAsync(row.Ip, token);
|
||||
var b = await _wd.MeasureRttMsAsync(row.Ip, token);
|
||||
int? rtt = (a, b) switch
|
||||
{
|
||||
(null, null) => null,
|
||||
(int x, null) => x,
|
||||
(null, int y) => y,
|
||||
(int x, int y) => Math.Min(x, y)
|
||||
};
|
||||
|
||||
if (rtt != null)
|
||||
{
|
||||
row.LatencyMs = rtt;
|
||||
row.Reachable = true;
|
||||
row.LatencySource = "watchdog";
|
||||
row.ProbedAt = DateTimeOffset.UtcNow.ToString("O");
|
||||
}
|
||||
else
|
||||
{
|
||||
// WatchDog 不通：若 SimpleLite 车载探测成功则保留，否则标不可达
|
||||
if (row.Reachable == true && row.LatencyMs is > 0 and < 2000)
|
||||
{
|
||||
row.LatencySource = "onboard";
|
||||
}
|
||||
else
|
||||
{
|
||||
row.Reachable = false;
|
||||
row.LatencyMs = null;
|
||||
row.LatencySource = "watchdog";
|
||||
row.ProbedAt = DateTimeOffset.UtcNow.ToString("O");
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
return rows;
|
||||
}
|
||||
|
||||
private async Task<List<FleetHealthRowDto>> FetchSimpleLiteHealthAsync(CancellationToken ct)
|
||||
{
|
||||
var port = _sl.ProjectionPort > 0 ? _sl.ProjectionPort : 8222;
|
||||
try
|
||||
{
|
||||
using var client = _httpFactory.CreateClient();
|
||||
client.Timeout = TimeSpan.FromSeconds(20);
|
||||
using var req = new HttpRequestMessage(HttpMethod.Get, $"http://127.0.0.1:{port}/projection/fleet/health");
|
||||
var token = _token.Token;
|
||||
if (!string.IsNullOrEmpty(token))
|
||||
req.Headers.TryAddWithoutValidation("X-Platform-Internal-Token", token);
|
||||
|
||||
using var resp = await client.SendAsync(req, ct);
|
||||
if (!resp.IsSuccessStatusCode) return new();
|
||||
|
||||
var text = await resp.Content.ReadAsStringAsync(ct);
|
||||
var list = JsonSerializer.Deserialize<List<FleetHealthRowDto>>(text, JsonOpt);
|
||||
return list ?? new();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogDebug(ex, "fleet/health from SimpleLite failed");
|
||||
return new();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<List<FleetHealthRowDto>> BuildRowsFromCarsAsync(CancellationToken ct)
|
||||
{
|
||||
var port = _sl.ProjectionPort > 0 ? _sl.ProjectionPort : 8222;
|
||||
try
|
||||
{
|
||||
using var client = _httpFactory.CreateClient();
|
||||
client.Timeout = TimeSpan.FromSeconds(10);
|
||||
using var req = new HttpRequestMessage(HttpMethod.Get, $"http://127.0.0.1:{port}/projection/cars");
|
||||
var token = _token.Token;
|
||||
if (!string.IsNullOrEmpty(token))
|
||||
req.Headers.TryAddWithoutValidation("X-Platform-Internal-Token", token);
|
||||
|
||||
using var resp = await client.SendAsync(req, ct);
|
||||
if (!resp.IsSuccessStatusCode) return new();
|
||||
|
||||
var text = await resp.Content.ReadAsStringAsync(ct);
|
||||
using var doc = JsonDocument.Parse(text);
|
||||
if (doc.RootElement.ValueKind != JsonValueKind.Array) return new();
|
||||
|
||||
var list = new List<FleetHealthRowDto>();
|
||||
foreach (var el in doc.RootElement.EnumerateArray())
|
||||
{
|
||||
var rawId = el.TryGetProperty("rawId", out var rid) && rid.TryGetInt32(out var id)
|
||||
? id
|
||||
: 0;
|
||||
if (rawId <= 0) continue;
|
||||
var ip = el.TryGetProperty("ip", out var ipEl) ? ipEl.GetString() : null;
|
||||
var name = el.TryGetProperty("name", out var nEl) ? nEl.GetString() : null;
|
||||
var onboard = el.TryGetProperty("onboardUrl", out var oEl) ? oEl.GetString() : null;
|
||||
if (string.IsNullOrEmpty(onboard) && !string.IsNullOrEmpty(ip))
|
||||
onboard = $"http://{ip}:8081";
|
||||
list.Add(new FleetHealthRowDto
|
||||
{
|
||||
CarId = rawId,
|
||||
CarName = name,
|
||||
Ip = ip,
|
||||
OnboardUrl = onboard
|
||||
});
|
||||
}
|
||||
return list;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.LogDebug(ex, "projection/cars fallback for fleet health failed");
|
||||
return new();
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user