justme
ad5efc62ff
Replace Probe with Agent across node runtime, Fleet contracts, backend state, API and storage. Report incoming sessions and outgoing link health through each Agent uplink, and retain directed link controls and retry. Update protocol fixtures and end-to-end coverage for the unified model; accept requested cancellation during the Fleet timeout test cleanup.
378 lines
16 KiB
C#
378 lines
16 KiB
C#
using System.Text;
|
|
using System.Text.Json;
|
|
using System.Text.Json.Serialization.Metadata;
|
|
using Fleet.Core;
|
|
using Microsoft.AspNetCore.Builder;
|
|
using Microsoft.AspNetCore.Http;
|
|
using Microsoft.AspNetCore.Routing;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Primitives;
|
|
using MoShiro.Backend.Application;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Backend.Api;
|
|
|
|
/// <summary>
|
|
/// 实时推送（SSE）：先推送当前快照，再推送增量（不补发历史）。SSE 只负责“更快地获得变化”，
|
|
/// 推送的内容都能由对应的 REST 查询重建。实时延迟取决于原始块的切块节奏（10 s）加上处理时间，典型值 10～12 s。
|
|
/// </summary>
|
|
public static class LiveEndpoints
|
|
{
|
|
private static readonly TimeSpan s_tick = TimeSpan.FromMilliseconds(500);
|
|
private static readonly TimeSpan s_debounce = TimeSpan.FromSeconds(1);
|
|
private static readonly TimeSpan s_heartbeat = TimeSpan.FromSeconds(15);
|
|
private static readonly TimeSpan s_periodicRefresh = TimeSpan.FromSeconds(10);
|
|
|
|
public static void Map(RouteGroupBuilder group)
|
|
{
|
|
group.MapGet("/live", StreamAsync)
|
|
.Api("live_stream", "Server-sent events: nodes, overview, series and test topics", "public")
|
|
.Produces(StatusCodes.Status200OK, contentType: "text/event-stream")
|
|
.ProducesProblems(400, 403, 404, 422, 429);
|
|
}
|
|
|
|
private static async Task<IResult> StreamAsync(
|
|
HttpContext http, AgentService agents, HostStatusService hosts, LineService lineService, TestService tests, QueryService query, LiveHub hub, SseLimiter limiter,
|
|
BackendOptions options, BackendCounters counters, TimeProvider time, IHostApplicationLifetime lifetime, string? ip_version)
|
|
{
|
|
string ipVersion = ip_version ?? IpVersions.V4;
|
|
if (!IpVersions.IsValid(ipVersion))
|
|
{
|
|
return Problems.From(http, ValidationError.Single("ip_version", "must be ipv4 or ipv6"));
|
|
}
|
|
|
|
StringValues raw = http.Request.Query["topics"];
|
|
Result<IReadOnlyList<LiveTopic>> parsed = LiveTopic.ParseAll(raw.Where(v => v is not null)!, options.MaxSseTopics);
|
|
if (!parsed.TryGet(out IReadOnlyList<LiveTopic>? topics, out FleetError? error))
|
|
{
|
|
return Problems.From(http, error);
|
|
}
|
|
|
|
// 订阅的线路必须存在并且对访问者可见
|
|
var lineKeys = new Dictionary<string, LineKey>(StringComparer.Ordinal);
|
|
foreach (LiveTopic topic in topics.Where(t => t.Line is not null))
|
|
{
|
|
Result<LineView> found = await QueryHelpers.FindVisibleLineAsync(http, lineService, topic.Line!, http.RequestAborted);
|
|
if (!found.TryGet(out LineView? resolved, out FleetError? lineError))
|
|
{
|
|
return Problems.From(http, lineError);
|
|
}
|
|
|
|
lineKeys[topic.Raw] = resolved.Key;
|
|
}
|
|
|
|
foreach (LiveTopic topic in topics.Where(t => t.Preset is not null))
|
|
{
|
|
Result<MetricSelection> selection = MetricSelection.Resolve(null, topic.Preset, options.MaxMetricsPerQuery);
|
|
if (!selection.IsSuccess)
|
|
{
|
|
return Problems.From(http, selection.Error);
|
|
}
|
|
}
|
|
|
|
string client = http.ClientAddress();
|
|
if (!limiter.TryAcquire(client))
|
|
{
|
|
return Problems.Create(http, MoShiroErrors.TooManyStreams, $"at most {options.MaxSsePerClient} live connections per client");
|
|
}
|
|
|
|
counters.SseOpened();
|
|
try
|
|
{
|
|
http.Response.ContentType = "text/event-stream";
|
|
http.Response.Headers.CacheControl = "no-store";
|
|
http.Response.Headers["X-Accel-Buffering"] = "no";
|
|
await http.Response.Body.FlushAsync(http.RequestAborted);
|
|
|
|
var session = new LiveSession(http, agents, hosts, lineService, tests, query, hub, options, time, topics, lineKeys, ipVersion);
|
|
using var linked = CancellationTokenSource.CreateLinkedTokenSource(http.RequestAborted, lifetime.ApplicationStopping);
|
|
await session.RunAsync(linked.Token);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// 客户端断开或服务正在停止
|
|
}
|
|
finally
|
|
{
|
|
counters.SseClosed();
|
|
limiter.Release(client);
|
|
}
|
|
|
|
return TypedResults.Empty;
|
|
}
|
|
|
|
private sealed class LiveSession
|
|
{
|
|
private readonly HttpContext _http;
|
|
private readonly AgentService _agents;
|
|
private readonly HostStatusService _hosts;
|
|
private readonly LineService _lines;
|
|
private readonly IReadOnlyDictionary<string, LineKey> _lineKeys;
|
|
private readonly string _ipVersion;
|
|
private readonly TestService _tests;
|
|
private readonly QueryService _query;
|
|
private readonly LiveHub _hub;
|
|
private readonly BackendOptions _options;
|
|
private readonly TimeProvider _time;
|
|
private readonly IReadOnlyList<LiveTopic> _topics;
|
|
private readonly object _gate = new();
|
|
private readonly Dictionary<string, long> _lastSent = new(StringComparer.Ordinal);
|
|
private readonly HashSet<(NodeId Agent, string Series)> _dirtyLines = [];
|
|
private readonly HashSet<TestId> _dirtyTests = [];
|
|
private readonly Queue<NodeSignal> _nodeSignals = new();
|
|
private readonly Dictionary<NodeId, HostSignal> _hostSignals = [];
|
|
private long _pushAtMs;
|
|
private long _lastWriteMs;
|
|
private long _lastPeriodicMs;
|
|
|
|
public LiveSession(HttpContext http, AgentService agents, HostStatusService hosts, LineService lines, TestService tests, QueryService query, LiveHub hub, BackendOptions options, TimeProvider time, IReadOnlyList<LiveTopic> topics, IReadOnlyDictionary<string, LineKey> lineKeys, string ipVersion)
|
|
{
|
|
_http = http;
|
|
_agents = agents;
|
|
_hosts = hosts;
|
|
_lines = lines;
|
|
_lineKeys = lineKeys;
|
|
_ipVersion = ipVersion;
|
|
_tests = tests;
|
|
_query = query;
|
|
_hub = hub;
|
|
_options = options;
|
|
_time = time;
|
|
_topics = topics;
|
|
}
|
|
|
|
public async Task RunAsync(CancellationToken ct)
|
|
{
|
|
using LiveSubscription subscription = _hub.Subscribe();
|
|
Task drain = DrainAsync(subscription, ct);
|
|
long now = Now();
|
|
_lastWriteMs = now;
|
|
_lastPeriodicMs = now;
|
|
await PushSnapshotAsync(ct);
|
|
|
|
using var timer = new PeriodicTimer(s_tick, _time);
|
|
while (await timer.WaitForNextTickAsync(ct))
|
|
{
|
|
await FlushPendingAsync(ct);
|
|
}
|
|
|
|
await drain;
|
|
}
|
|
|
|
private async Task DrainAsync(LiveSubscription subscription, CancellationToken ct)
|
|
{
|
|
try
|
|
{
|
|
await foreach (LiveSignal signal in subscription.Reader.ReadAllAsync(ct))
|
|
{
|
|
lock (_gate)
|
|
{
|
|
switch (signal)
|
|
{
|
|
case NodeSignal node when _topics.Any(t => t.Kind == LiveTopicKind.Nodes):
|
|
_nodeSignals.Enqueue(node);
|
|
break;
|
|
case HostSignal host when _topics.Any(t => t.Kind == LiveTopicKind.Hosts):
|
|
_hostSignals[host.Node] = host;
|
|
break;
|
|
case ChunkSignal chunk:
|
|
_dirtyLines.Add((chunk.AgentId, chunk.Series));
|
|
_pushAtMs = _pushAtMs == 0 ? Now() + (long)s_debounce.TotalMilliseconds : _pushAtMs;
|
|
break;
|
|
case TestSignal test when _topics.Any(t => t.Test == test.TestId):
|
|
_dirtyTests.Add(test.TestId);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// 连接结束
|
|
}
|
|
}
|
|
|
|
private async Task FlushPendingAsync(CancellationToken ct)
|
|
{
|
|
long now = Now();
|
|
NodeSignal[] nodes;
|
|
HostSignal[] hostSignals;
|
|
TestId[] testIds;
|
|
(NodeId Agent, string Series)[] dirtyLines = [];
|
|
bool refreshAll = now - _lastPeriodicMs >= (long)s_periodicRefresh.TotalMilliseconds;
|
|
lock (_gate)
|
|
{
|
|
nodes = [.. _nodeSignals];
|
|
_nodeSignals.Clear();
|
|
hostSignals = [.. _hostSignals.Values];
|
|
_hostSignals.Clear();
|
|
testIds = [.. _dirtyTests];
|
|
_dirtyTests.Clear();
|
|
if (_pushAtMs != 0 && now >= _pushAtMs)
|
|
{
|
|
dirtyLines = [.. _dirtyLines];
|
|
_dirtyLines.Clear();
|
|
_pushAtMs = 0;
|
|
}
|
|
}
|
|
|
|
foreach (HostSignal host in hostSignals)
|
|
{
|
|
AgentView? view = await _agents.GetAsync(host.Node, ct);
|
|
if (view is not null && IsVisible(view) && view.Agent.Status == Fleet.Server.NodeStatus.Active)
|
|
{
|
|
await WriteAsync("host", new LiveHostDto(host.Node.Value, view.Online, Mappers.ToDto(host.Host)), ApiJsonContext.Default.LiveHostDto, ct);
|
|
}
|
|
}
|
|
|
|
foreach (NodeSignal node in nodes)
|
|
{
|
|
AgentView? view = await _agents.GetAsync(new NodeId(node.Id), ct);
|
|
if (view is not null && IsVisible(view))
|
|
{
|
|
await WriteAsync("node", new LiveNodeDto(node.Kind, node.Id, node.Online, Mappers.Time(node.AtMs)), ApiJsonContext.Default.LiveNodeDto, ct);
|
|
}
|
|
}
|
|
|
|
foreach (TestId testId in testIds)
|
|
{
|
|
await PushTestAsync(testId, ct);
|
|
}
|
|
|
|
if (dirtyLines.Length > 0)
|
|
{
|
|
foreach (LiveTopic topic in _topics.Where(t => t.Kind == LiveTopicKind.Series && dirtyLines.Contains((_lineKeys[t.Raw].Listener, _lineKeys[t.Raw].Series))))
|
|
{
|
|
await PushSeriesAsync(topic, ct);
|
|
}
|
|
|
|
foreach (LiveTopic topic in _topics.Where(t => t.Kind == LiveTopicKind.Overview))
|
|
{
|
|
await PushOverviewAsync(topic, ct);
|
|
}
|
|
|
|
_lastPeriodicMs = now;
|
|
}
|
|
else if (refreshAll)
|
|
{
|
|
// 没有新块也定期刷新（例如线路断了之后汇总需要更新）
|
|
foreach (LiveTopic topic in _topics.Where(t => t.Kind is LiveTopicKind.Overview or LiveTopicKind.Series))
|
|
{
|
|
await (topic.Kind == LiveTopicKind.Overview ? PushOverviewAsync(topic, ct) : PushSeriesAsync(topic, ct));
|
|
}
|
|
|
|
_lastPeriodicMs = now;
|
|
}
|
|
|
|
if (Now() - _lastWriteMs >= (long)s_heartbeat.TotalMilliseconds)
|
|
{
|
|
await WriteRawAsync(": keepalive\n\n", ct);
|
|
}
|
|
}
|
|
|
|
private async Task PushSnapshotAsync(CancellationToken ct)
|
|
{
|
|
foreach (LiveTopic topic in _topics)
|
|
{
|
|
switch (topic.Kind)
|
|
{
|
|
case LiveTopicKind.Nodes:
|
|
foreach (AgentView view in (await _agents.ListAsync(ct)).Where(v => v.Agent.Status == Fleet.Server.NodeStatus.Active && IsVisible(v)))
|
|
{
|
|
await WriteAsync("node", new LiveNodeDto("agent", view.Agent.Id.Value, view.Online, Mappers.Time(Now())), ApiJsonContext.Default.LiveNodeDto, ct);
|
|
}
|
|
|
|
break;
|
|
case LiveTopicKind.Hosts:
|
|
foreach (AgentView view in (await _agents.ListAsync(ct)).Where(v => v.Agent.Status == Fleet.Server.NodeStatus.Active && IsVisible(v) && v.Host is not null))
|
|
{
|
|
await WriteAsync("host", new LiveHostDto(view.Agent.Id.Value, view.Online, Mappers.ToDto(view.Host!)), ApiJsonContext.Default.LiveHostDto, ct);
|
|
}
|
|
|
|
break;
|
|
case LiveTopicKind.Overview:
|
|
await PushOverviewAsync(topic, ct);
|
|
break;
|
|
case LiveTopicKind.Series:
|
|
await PushSeriesAsync(topic, ct);
|
|
break;
|
|
case LiveTopicKind.Test:
|
|
await PushTestAsync(topic.Test!.Value, ct);
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task PushOverviewAsync(LiveTopic topic, CancellationToken ct)
|
|
{
|
|
MetricSelection selection = MetricSelection.Resolve(null, topic.Preset, _options.MaxMetricsPerQuery).Value;
|
|
IReadOnlyList<LineView> visible = await _lines.ListAsync(a => a.Agent.Status == Fleet.Server.NodeStatus.Active && IsVisible(a), ct);
|
|
Result<OverviewResult> result = await _query.OverviewAsync([.. visible.Where(l => l.IpVersion == _ipVersion)], new RangeRequest(topic.Range, null, null, null, null), selection, ct);
|
|
if (result.TryGet(out OverviewResult? overview, out _))
|
|
{
|
|
var dto = new OverviewDto(
|
|
topic.Range!, Mappers.Time(overview.Range.FromMs), Mappers.Time(overview.Range.ToMs), Steps.Format(overview.Range.StepMs),
|
|
overview.Plan.Name, overview.Plan.Degraded, [.. overview.Lines.Select(Mappers.ToDto)]);
|
|
await WriteAsync("overview", dto, ApiJsonContext.Default.OverviewDto, ct);
|
|
}
|
|
}
|
|
|
|
private async Task PushSeriesAsync(LiveTopic topic, CancellationToken ct)
|
|
{
|
|
MetricSelection selection = MetricSelection.Resolve(null, topic.Preset, _options.MaxMetricsPerQuery).Value;
|
|
string step = Steps.Format(topic.StepMs);
|
|
long now = Now();
|
|
long from = _lastSent.TryGetValue(topic.Raw, out long last) ? last : now - 20_000;
|
|
|
|
Result<SeriesResult> latest = await _query.SeriesAsync(_lineKeys[topic.Raw], new RangeRequest(null, DateTimeOffset.FromUnixTimeMilliseconds(from), null, step, null), selection, ct);
|
|
Result<SeriesResult> summary = await _query.SeriesAsync(_lineKeys[topic.Raw], new RangeRequest("15m", null, null, step, null), selection, ct);
|
|
if (!latest.TryGet(out SeriesResult? series, out _) || !summary.TryGet(out SeriesResult? whole, out _))
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (series.T.Count > 0)
|
|
{
|
|
_lastSent[topic.Raw] = series.T[^1];
|
|
}
|
|
|
|
SeriesDto dto = Mappers.ToDto(series, live: true) with { Summary = Mappers.ToDto(whole.Summary) };
|
|
await WriteAsync("series", dto, ApiJsonContext.Default.SeriesDto, ct);
|
|
}
|
|
|
|
private async Task PushTestAsync(TestId testId, CancellationToken ct)
|
|
{
|
|
TestDetail? detail = await _tests.GetAsync(testId, ct);
|
|
if (detail is null)
|
|
{
|
|
return;
|
|
}
|
|
|
|
AgentView? view = await _agents.GetAsync(detail.Test.AgentId, ct);
|
|
if (view is null || !IsVisible(view))
|
|
{
|
|
return;
|
|
}
|
|
|
|
await WriteAsync("test", new LiveTestDto(testId.Value, detail.Test.Status, detail.Test.Phase, detail.Test.PhaseCount), ApiJsonContext.Default.LiveTestDto, ct);
|
|
}
|
|
|
|
private bool IsVisible(AgentView view) => view.Agent.PublicVisible || PublicGuard.CanSeeHidden(_http);
|
|
|
|
private async Task WriteAsync<T>(string name, T payload, JsonTypeInfo<T> typeInfo, CancellationToken ct)
|
|
{
|
|
string json = JsonSerializer.Serialize(payload, typeInfo);
|
|
await WriteRawAsync($"event: {name}\ndata: {json}\n\n", ct);
|
|
}
|
|
|
|
private async Task WriteRawAsync(string text, CancellationToken ct)
|
|
{
|
|
await _http.Response.Body.WriteAsync(Encoding.UTF8.GetBytes(text), ct);
|
|
await _http.Response.Body.FlushAsync(ct);
|
|
_lastWriteMs = Now();
|
|
}
|
|
|
|
private long Now() => _time.GetUtcNow().ToUnixTimeMilliseconds();
|
|
}
|
|
}
|