docs: merge the Fleet docs into backend/fleet/README.md, add TODO.md, unify the Backend compose, fix stale references [CI SKIP]
This commit is contained in:
justme committed
1 parent
a286a453f4
commit
c974a7dddb
24 files changed
+218
-525
No files matched your search
@@ -0,0 +1,215 @@
|
||||
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;
|
||||
|
||||
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases)
|
||||
{
|
||||
_releases = releases;
|
||||
_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)));
|
||||
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);
|
||||
}
|
||||
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
|
||||
{
|
||||
_releases.RecordStatus(id, extra.Deserialize(MoShiroJsonContext.Default.ProbeHeartbeatExtra)?.Update);
|
||||
}
|
||||
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);
|
||||
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts), 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))
|
||||
&& 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;
|
||||
Reference in new issue
Block a user