using System.Text.RegularExpressions; using Fleet.Core; using Fleet.Server; using Monitor.Protocol; namespace Monitor.Backend.Application; public sealed record AgentView(AgentRecord Agent, bool Online, IReadOnlyList Probes, AgentHeartbeatExtra? Heartbeat, int EndpointChanges, long? HeartbeatAtMs = null, HostSnapshot? Host = null) { /// 至少有一个 Probe 连着。 public bool ProbeConnected => Probes.Count > 0; } /// Agent 的管理:列表、修改、审批、启停、接管、删除,以及 Probe 需要的 Agent 列表。 public sealed partial class AgentService { private readonly IAgentStore _agents; private readonly AgentRuntime _runtime; private readonly IAgentGateway _gateway; private readonly TaskPlanner _planner; private readonly ISettingsStore _settings; private readonly NodeInfoService _nodes; private readonly LinkService _links; public AgentService(IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TaskPlanner planner, ISettingsStore settings, NodeInfoService nodes, LinkService links) { _links = links; _nodes = nodes; _agents = agents; _runtime = runtime; _gateway = gateway; _planner = planner; _settings = settings; } public async ValueTask> ListAsync(CancellationToken ct) { IReadOnlyList agents = await _agents.ListAsync(ct); return [.. agents.Select(View)]; } public async ValueTask GetAsync(NodeId id, CancellationToken ct) => await _agents.GetAsync(id, ct) is { } agent ? View(agent) : null; public AgentView View(AgentRecord agent) { _runtime.TryGet(agent.Id, out AgentRuntimeState? state); return new AgentView(agent, _gateway.IsOnline(agent.Id), state?.Probes ?? [], state?.LastHeartbeat, state?.EndpointChanges ?? 0, state?.LastHeartbeatAtMs, state?.Host); } public async ValueTask> PatchAsync(NodeId id, AgentPatch patch, long? expectedRowVersion, CancellationToken ct) { var errors = new Dictionary(); if (patch.Name is not null && !NamePattern().IsMatch(patch.Name)) { errors["name"] = "must be 1-64 characters: letters (any language), digits, '.', '_' or '-'"; } if (patch.PublicHost is not null && !HostPattern().IsMatch(patch.PublicHost)) { errors["public_host"] = "must be a host name or IP address"; } if (patch.Labels is not null && (patch.Labels.Count > 16 || patch.Labels.Any(l => l.Length is 0 or > 32))) { errors["labels"] = "at most 16 labels of 1-32 characters"; } if (errors.Count > 0) { return ValidationError.Fail(errors); } Result result = await _agents.PatchAsync(id, patch, expectedRowVersion, ct); if (!result.TryGet(out AgentRecord? agent, out FleetError? error)) { return Result.Fail(error); } if (patch.Name is not null || patch.PublicHost is not null) { await AfterAgentSetChangedAsync(ct); } return Result.Ok(View(agent)); } public ValueTask> ApproveAsync(NodeId id, CancellationToken ct) => ChangeStatusAsync(id, NodeStatus.Active, ct); public ValueTask> EnableAsync(NodeId id, CancellationToken ct) => ChangeStatusAsync(id, NodeStatus.Active, ct); public async ValueTask> DisableAsync(NodeId id, CancellationToken ct) { Result result = await ChangeStatusAsync(id, NodeStatus.Disabled, ct); if (result.IsSuccess) { await _gateway.DisconnectAsync(id, "disabled", ct); } return result; } public async ValueTask> TakeoverAsync(NodeId pendingId, CancellationToken ct) { Result result = await _agents.TakeoverAsync(pendingId, ct); if (!result.TryGet(out AgentRecord? agent, out FleetError? error)) { return Result.Fail(error); } // 公钥换了,旧连接使用的是旧密钥 await _gateway.DisconnectAsync(agent.Id, "key replaced", ct); await AfterAgentSetChangedAsync(ct); return Result.Ok(View(agent)); } public async ValueTask DeleteAsync(NodeId id, CancellationToken ct) { await _gateway.DisconnectAsync(id, "deleted", ct); await _agents.DeleteAsync(id, ct); await _nodes.ForgetAsync(id.Value, ct); await _links.ForgetNodeAsync(id.Value, ct); _runtime.Remove(id); await AfterAgentSetChangedAsync(ct); } /// 这个 Probe 要连接的 Agent 列表(已激活、有公开地址,并且线路是开着的)。版本号在列表变化时递增。 public async ValueTask ProbeAgentsAsync(string probeId, CancellationToken ct) { LinkSnapshot links = await _links.SnapshotAsync(ct); IReadOnlyList agents = await _agents.ListAsync(ct); long version = long.TryParse(await _settings.GetAsync(TaskPlanner.ProbeAgentsVersionKey, ct), out long v) ? v : 0; var result = new ProbeAgentList { Version = version }; foreach (AgentRecord agent in agents.Where(a => a.Status == NodeStatus.Active && a.PublicHost.Length > 0 && links.IsLinked(probeId, a.Id.Value))) { result.Agents.Add(new AgentEndpoint { AgentId = agent.Id, Name = agent.Name, Host = agent.PublicHost, TcpPort = agent.ProbeTcpPort, UdpPorts = [.. agent.ProbeUdpPorts], Spki = Convert.ToBase64String(agent.TlsSpki), }); } return result; } /// Agent 集合变了:递增列表版本,重算各 Agent 的期望状态,并把新的 Agent 列表推给 Probe(都是期望状态,一起通知)。 public async ValueTask AfterAgentSetChangedAsync(CancellationToken ct) { await _planner.BumpProbeAgentsVersionAsync(ct); _gateway.NotifyAllTaskSetsChanged(); } private async ValueTask> ChangeStatusAsync(NodeId id, NodeStatus status, CancellationToken ct) { Result result = await _agents.SetStatusAsync(id, status, ct); if (!result.TryGet(out AgentRecord? agent, out FleetError? error)) { return Result.Fail(error); } await AfterAgentSetChangedAsync(ct); _gateway.NotifyTaskSetChanged(id); return Result.Ok(View(agent)); } [GeneratedRegex(@"^[\p{L}\p{M}\p{N}._-]{1,64}$")] private static partial Regex NamePattern(); [GeneratedRegex(@"^[A-Za-z0-9.:\-\[\]]{1,253}$")] private static partial Regex HostPattern(); }