228 lines
9.2 KiB
C#
228 lines
9.2 KiB
C#
using System.Text.Json;
|
|
using Fleet.Core;
|
|
using Fleet.Rpc;
|
|
using Fleet.Server;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Backend.Application;
|
|
|
|
/// <summary>MoShiro 在 Fleet 之上的接入点：注册之后补全公开地址与端口。</summary>
|
|
public sealed class MoShiroEnrollmentHook : IEnrollmentHook
|
|
{
|
|
private readonly IAgentStore _agents;
|
|
private readonly AgentService _service;
|
|
|
|
public MoShiroEnrollmentHook(IAgentStore agents, AgentService service)
|
|
{
|
|
_agents = agents;
|
|
_service = service;
|
|
}
|
|
|
|
public async ValueTask OnEnrolledAsync(NodeRecord node, EnrollmentContext context, CancellationToken ct)
|
|
{
|
|
AgentEnrollMeta meta = new();
|
|
if (context.ProductMeta is { } raw)
|
|
{
|
|
try
|
|
{
|
|
meta = raw.Deserialize(MoShiroJsonContext.Default.AgentEnrollMeta) ?? meta;
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 元数据格式不对就用默认值，节点之后可以通过 hello 补全
|
|
}
|
|
}
|
|
|
|
// public_host 未提供时，取注册请求的来源 IP
|
|
string host = !string.IsNullOrWhiteSpace(meta.PublicHost) ? meta.PublicHost : context.RemoteIp ?? "";
|
|
await _agents.ApplyEnrollmentAsync(node.Id, new AgentEnrollment(host, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Kinds), ct);
|
|
if (node.Status == NodeStatus.Active)
|
|
{
|
|
await _service.AfterAgentSetChangedAsync(ct);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>节点上线、下线、hello、心跳：更新运行时状态，补全 Agent 信息，并推送给 SSE。</summary>
|
|
public sealed class MoShiroNodeEventSink : INodeEventSink
|
|
{
|
|
private readonly IAgentStore _agents;
|
|
private readonly AgentRuntime _runtime;
|
|
private readonly IEventStore _events;
|
|
private readonly LiveHub _live;
|
|
private readonly AgentService _service;
|
|
private readonly NodeInfoService _nodes;
|
|
private readonly ReleaseService _releases;
|
|
private readonly IProbeStore _probes;
|
|
|
|
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, IProbeStore probes)
|
|
{
|
|
_releases = releases;
|
|
_probes = probes;
|
|
_nodes = nodes;
|
|
_agents = agents;
|
|
_runtime = runtime;
|
|
_events = events;
|
|
_live = live;
|
|
_service = service;
|
|
}
|
|
|
|
public async ValueTask OnEventAsync(NodeEvent nodeEvent, CancellationToken ct)
|
|
{
|
|
NodeId id = nodeEvent.Node.Id;
|
|
if (id.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal))
|
|
{
|
|
await OnProbeEventAsync(id, nodeEvent, ct);
|
|
return;
|
|
}
|
|
|
|
switch (nodeEvent.Kind)
|
|
{
|
|
case NodeEventKind.Online:
|
|
_live.Publish(new NodeSignal("agent", id.Value, true, nodeEvent.AtMs));
|
|
await _events.AddAsync(id, "agent_online", nodeEvent.AtMs, null, ct);
|
|
break;
|
|
case NodeEventKind.Offline:
|
|
_runtime.MarkOffline(id);
|
|
_releases.RecordStatus(id, null);
|
|
_live.Publish(new NodeSignal("agent", id.Value, false, nodeEvent.AtMs));
|
|
await _events.AddAsync(id, "agent_offline", nodeEvent.AtMs, null, ct);
|
|
break;
|
|
case NodeEventKind.Hello when nodeEvent.Hello is { } hello:
|
|
await ApplyHelloAsync(id, hello, ct);
|
|
break;
|
|
case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra:
|
|
try
|
|
{
|
|
AgentRuntimeState state = _runtime.Get(id);
|
|
AgentHeartbeatExtra? beat = extra.Deserialize(MoShiroJsonContext.Default.AgentHeartbeatExtra);
|
|
state.LastHeartbeat = beat;
|
|
state.LastHeartbeatAtMs = nodeEvent.AtMs;
|
|
_releases.RecordStatus(id, beat?.Update);
|
|
if (beat is not null)
|
|
{
|
|
state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs, p.IpVersion)));
|
|
foreach (ConnectedProbe p in beat.Probes)
|
|
{
|
|
_nodes.ObserveProbeInBackground(p.Id.Value, p.Remote);
|
|
}
|
|
}
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 忽略格式不正确的附加字段
|
|
}
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
/// <summary>Probe 自己连上来了：hello 里带着它的系统、架构和构建信息；上线下线只推给 SSE，不记事件。</summary>
|
|
private async ValueTask OnProbeEventAsync(NodeId id, NodeEvent nodeEvent, CancellationToken ct)
|
|
{
|
|
switch (nodeEvent.Kind)
|
|
{
|
|
case NodeEventKind.Online:
|
|
_live.Publish(new NodeSignal("probe", id.Value, true, nodeEvent.AtMs));
|
|
break;
|
|
case NodeEventKind.Hello:
|
|
if (nodeEvent.Hello?.Meta is { } raw)
|
|
{
|
|
try
|
|
{
|
|
ProbeHelloMeta? meta = raw.Deserialize(MoShiroJsonContext.Default.ProbeHelloMeta);
|
|
await _nodes.SetPlatformAsync(id.Value, NodeKinds.Probe, meta?.Platform, ct);
|
|
if (meta?.Ipv6 is { } ipv6)
|
|
{
|
|
await _probes.SetIpv6DiagnosticsAsync(id.Value, JsonSerializer.Serialize(ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic), ct);
|
|
}
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 格式不对就忽略
|
|
}
|
|
}
|
|
|
|
await _releases.OfferToNodeAsync(id, ct);
|
|
break;
|
|
case NodeEventKind.Offline:
|
|
_live.Publish(new NodeSignal("probe", id.Value, false, nodeEvent.AtMs));
|
|
_releases.RecordStatus(id, null);
|
|
break;
|
|
case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra:
|
|
try
|
|
{
|
|
ProbeHeartbeatExtra? heartbeat = extra.Deserialize(MoShiroJsonContext.Default.ProbeHeartbeatExtra);
|
|
_releases.RecordStatus(id, heartbeat?.Update);
|
|
if (heartbeat?.Ipv6 is { } ipv6)
|
|
{
|
|
await _probes.SetIpv6DiagnosticsAsync(id.Value, JsonSerializer.Serialize(ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic), ct);
|
|
}
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 忽略格式不正确的附加字段
|
|
}
|
|
|
|
break;
|
|
}
|
|
}
|
|
|
|
private async ValueTask ApplyHelloAsync(NodeId id, HelloParams hello, CancellationToken ct)
|
|
{
|
|
var kinds = new List<string>();
|
|
foreach (string capability in hello.Capabilities)
|
|
{
|
|
if (capability.StartsWith(MoShiroCapabilities.KindsPrefix, StringComparison.Ordinal))
|
|
{
|
|
kinds.AddRange(capability[MoShiroCapabilities.KindsPrefix.Length..].Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries));
|
|
}
|
|
}
|
|
|
|
AgentHelloMeta meta = new();
|
|
if (hello.Meta is { } raw)
|
|
{
|
|
try
|
|
{
|
|
meta = raw.Deserialize(MoShiroJsonContext.Default.AgentHelloMeta) ?? meta;
|
|
}
|
|
catch (JsonException)
|
|
{
|
|
// 忽略
|
|
}
|
|
}
|
|
|
|
byte[]? spki = null;
|
|
if (meta.TlsSpki is not null)
|
|
{
|
|
try
|
|
{
|
|
spki = Convert.FromBase64String(meta.TlsSpki);
|
|
}
|
|
catch (FormatException)
|
|
{
|
|
// 忽略
|
|
}
|
|
}
|
|
|
|
AgentRecord? before = await _agents.GetAsync(id, ct);
|
|
string? limitsJson = meta.LocalLimits is null ? null : JsonSerializer.Serialize(meta.LocalLimits, LocalLimitsJson.Default.LocalLimits);
|
|
string? ipv6Json = meta.Ipv6 is null ? null : JsonSerializer.Serialize(meta.Ipv6, MoShiroJsonContext.Default.Ipv6Diagnostic);
|
|
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Ipv6Supported, ipv6Json), ct);
|
|
|
|
// 端口或证书指纹变了，Probe 需要新的列表
|
|
AgentRecord? after = await _agents.GetAsync(id, ct);
|
|
_nodes.ObserveAgentInBackground(id.Value, after?.PublicHost, meta.Platform);
|
|
await _releases.OfferToNodeAsync(id, ct);
|
|
if (before is not null && after is not null
|
|
&& (!before.TlsSpki.AsSpan().SequenceEqual(after.TlsSpki) || before.ProbeTcpPort != after.ProbeTcpPort || !before.ProbeUdpPorts.SequenceEqual(after.ProbeUdpPorts) || before.Ipv6Supported != after.Ipv6Supported)
|
|
&& after.Status == NodeStatus.Active)
|
|
{
|
|
await _service.AfterAgentSetChangedAsync(ct);
|
|
}
|
|
}
|
|
}
|
|
|
|
[System.Text.Json.Serialization.JsonSerializable(typeof(LocalLimits))]
|
|
internal sealed partial class LocalLimitsJson : System.Text.Json.Serialization.JsonSerializerContext;
|