From 48072737c90c65ed5b8c5278f1e0ce32447021a2 Mon Sep 17 00:00:00 2001 From: JustMe Date: Tue, 6 Oct 2026 22:25:16 +0800 Subject: [PATCH] feat!: probes connect to the backend directly A probe now links to the backend like an agent (probe.json carries the server address). The backend sends the agent list as desired state and the probe reports its platform and build in its own hello. The agent no longer relays the list or the probe platform. Existing probes must be re-created. --- backend/agent/src/Monitor.Agent/AgentApp.cs | 8 --- .../src/Monitor.Agent/DataPlane/Registries.cs | 3 -- .../agent/src/Monitor.Agent/Peer/PeerHub.cs | 52 ++----------------- backend/openapi/v1.json | 41 +++------------ backend/probe/README.md | 11 ++-- .../src/Monitor.Probe/Link/AgentConnection.cs | 34 ++---------- .../src/Monitor.Probe/Link/AgentDirectory.cs | 29 +++++------ backend/probe/src/Monitor.Probe/ProbeApp.cs | 43 ++++++++++++--- .../probe/src/Monitor.Probe/ProbeConfig.cs | 12 ++--- backend/probe/src/Monitor.Probe/Program.cs | 4 +- .../tests/Monitor.Probe.Tests/ProbeTests.cs | 20 ++++--- .../shared/Monitor.Protocol/Rpc/MonitorRpc.cs | 30 +++++------ .../shared/Monitor.Protocol/Tasks/TaskSet.cs | 3 -- .../Admin/AgentEndpoints.cs | 15 +++--- .../src/Monitor.Backend.Api/Common/Mappers.cs | 2 +- .../Contracts/AdminDtos.cs | 6 +-- .../Agents/AgentEventService.cs | 4 +- .../Agents/AgentRuntime.cs | 2 +- .../Agents/AgentService.cs | 7 ++- .../Agents/MonitorFleetHooks.cs | 33 +++++++++++- .../Agents/ProbeService.cs | 35 +++++-------- .../Nodes/NodeInfoService.cs | 5 +- .../Ports/IAgentGateway.cs | 5 +- .../Tasks/TaskPlanner.cs | 3 +- .../Monitor.Backend.Gateway/FleetGateway.cs | 11 +--- .../GatewayServiceCollectionExtensions.cs | 8 +-- .../ProbeAgentsStateProvider.cs | 33 ++++++++++++ .../TaskSetStateProvider.cs | 5 ++ .../SqliteAgentStore.cs | 11 +++- .../EnrollmentAndTaskSetTests.cs | 30 +---------- .../OperationsAndSecurityTests.cs | 8 +-- .../ProbeEndToEndTests.cs | 24 ++++++++- .../Monitor.TestSupport/AgentSimulator.cs | 11 ---- .../Monitor.TestSupport/ProbeSimulator.cs | 10 ---- docs/api.md | 2 +- docs/deployment.md | 3 +- docs/enrollment.md | 23 ++++---- docs/project-layout.md | 3 +- docs/protocol.md | 6 +-- docs/roadmap.md | 4 +- web/admin/src/pages/Probes.tsx | 11 +++- web/shared/api/schema.ts | 10 +--- 42 files changed, 266 insertions(+), 354 deletions(-) create mode 100644 backend/src/Monitor.Backend.Gateway/ProbeAgentsStateProvider.cs diff --git a/backend/agent/src/Monitor.Agent/AgentApp.cs b/backend/agent/src/Monitor.Agent/AgentApp.cs index f2bda34..181ce9b 100644 --- a/backend/agent/src/Monitor.Agent/AgentApp.cs +++ b/backend/agent/src/Monitor.Agent/AgentApp.cs @@ -209,14 +209,6 @@ public sealed class AgentApp : IDisposable async (_, request, _) => Result.Ok(new TestRunResult { Accepted = await _hub.DisconnectAsync(request.ProbeId.Value) })); } - if (Allow(MonitorMethods.ProbeAgentsChanged)) - { - router.OnNotify( - MonitorMethods.ProbeAgentsChanged, - MonitorJsonContext.Default.ProbeAgentsResult, - (_, list, _) => _hub.RelayAgentsAsync(list)); - } - var options = new FleetLinkOptions { Server = new Uri(_config.Fleet.Server), diff --git a/backend/agent/src/Monitor.Agent/DataPlane/Registries.cs b/backend/agent/src/Monitor.Agent/DataPlane/Registries.cs index 4f97f85..fff6753 100644 --- a/backend/agent/src/Monitor.Agent/DataPlane/Registries.cs +++ b/backend/agent/src/Monitor.Agent/DataPlane/Registries.cs @@ -64,9 +64,6 @@ public sealed class ProbeSession : DataSession public long ConnectedAtUtcMs { get; init; } - /// Probe 自报的系统和架构（随列表同步一起来，只用于展示）；还没报过就是 null。 - public PlatformInfo? Platform { get; set; } - public RpcSession? Rpc { get; set; } /// 最近一次认证通过的 UDP 包的源地址（Agent 把单向数据发往这里）。 diff --git a/backend/agent/src/Monitor.Agent/Peer/PeerHub.cs b/backend/agent/src/Monitor.Agent/Peer/PeerHub.cs index 1d8bf66..4392242 100644 --- a/backend/agent/src/Monitor.Agent/Peer/PeerHub.cs +++ b/backend/agent/src/Monitor.Agent/Peer/PeerHub.cs @@ -57,18 +57,6 @@ public sealed class BackendUplink await SafeAsync(() => session.NotifyAsync(MonitorMethods.TestStatus, parameters, MonitorJsonContext.Default.TestStatusParams, CancellationToken.None)); } - /// 代 Probe 向 Backend 取 Agent 列表（Probe 不认识 Backend）。链路断开时返回 fleet_node_offline。 - public async ValueTask> RequestProbeAgentsAsync(ProbeAgentsParams request, CancellationToken ct) - { - RpcSession? session = _link?.CurrentSession; - if (session is null) - { - return Result.Fail(FleetErrors.NodeOffline, "the agent has no link to the backend"); - } - - return await session.RequestAsync(MonitorMethods.ProbeAgents, request, MonitorJsonContext.Default.ProbeAgentsParams, MonitorJsonContext.Default.ProbeAgentsResult, ct); - } - private static async ValueTask SafeAsync(Func send) { try @@ -104,7 +92,7 @@ public sealed class PeerHub public PeerHub(SessionRegistry sessions, MonoClock clock, ChunkAssembler chunks, TaskManager tasks, BackendUplink uplink, TimeProvider time, INodeLog log) { - _router = BuildRouter(uplink); + _router = BuildRouter(); _sessions = sessions; _clock = clock; _chunks = chunks; @@ -220,7 +208,7 @@ public sealed class PeerHub /// 当前连着的 Probe（心跳和 Backend 重连之后的补报都用它）。 public IReadOnlyList ConnectedProbes() => [.. _sessions.All.Select(Describe)]; - private static ConnectedProbe Describe(ProbeSession p) => new() { Id = new ProbeId(p.ProbeId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs, Platform = p.Platform }; + private static ConnectedProbe Describe(ProbeSession p) => new() { Id = new ProbeId(p.ProbeId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs }; public IReadOnlyList ConnectedProbeIds() => [.. _sessions.All.Select(p => p.ProbeId)]; @@ -252,27 +240,6 @@ public sealed class PeerHub } } - /// Backend 的 Agent 列表变化时转交给所有 Probe（Probe 据此调整它连接的 Agent）。 - public async ValueTask RelayAgentsAsync(ProbeAgentsResult list) - { - foreach (ProbeSession probe in _sessions.All) - { - if (probe.Rpc is not { } rpc) - { - continue; - } - - try - { - await rpc.NotifyAsync(MonitorMethods.ProbeAgentsChanged, list, MonitorJsonContext.Default.ProbeAgentsResult, CancellationToken.None); - } - catch (Exception ex) when (ex is ObjectDisposedException or InvalidOperationException or IOException) - { - // 会话刚好结束 - } - } - } - private async ValueTask PushAsync(ProbeSession probe, LineTaskSet set, CancellationToken ct) { if (probe.Rpc is null) @@ -300,21 +267,8 @@ public sealed class PeerHub } } - private static RpcRouter BuildRouter(BackendUplink uplink) => + private static RpcRouter BuildRouter() => new RpcRouter() - .On( - MonitorMethods.ProbeAgents, - MonitorJsonContext.Default.ProbeAgentsParams, - MonitorJsonContext.Default.ProbeAgentsResult, - (context, request, ct) => - { - if (request.Platform is { } platform) - { - context.Session.Platform = platform; - } - - return uplink.RequestProbeAgentsAsync(request, ct); - }) .On( MonitorMethods.Clock, MonitorJsonContext.Default.ClockParams, diff --git a/backend/openapi/v1.json b/backend/openapi/v1.json index eb08ea3..336e303 100644 --- a/backend/openapi/v1.json +++ b/backend/openapi/v1.json @@ -4136,34 +4136,6 @@ } } }, - "BootstrapAgentDto": { - "required": [ - "agent_id", - "host", - "tcp_port", - "spki" - ], - "type": "object", - "properties": { - "agent_id": { - "type": "string" - }, - "host": { - "type": "string" - }, - "tcp_port": { - "pattern": "^-?(?:0|[1-9]\\d*)$", - "type": [ - "integer", - "string" - ], - "format": "int32" - }, - "spki": { - "type": "string" - } - } - }, "BuildDto": { "required": [ "version", @@ -6326,6 +6298,7 @@ "created_at", "revoked_at", "revoked", + "online", "connections" ], "type": "object", @@ -6345,6 +6318,9 @@ "revoked": { "type": "boolean" }, + "online": { + "type": "boolean" + }, "connections": { "type": "array", "items": { @@ -6357,7 +6333,7 @@ "required": [ "probe_id", "private_key", - "bootstrap", + "server", "limits" ], "type": "object", @@ -6368,11 +6344,8 @@ "private_key": { "type": "string" }, - "bootstrap": { - "type": "array", - "items": { - "$ref": "#/components/schemas/BootstrapAgentDto" - } + "server": { + "type": "string" }, "limits": { "$ref": "#/components/schemas/ProbeLimitsDto" diff --git a/backend/probe/README.md b/backend/probe/README.md index cfeb861..07805f0 100644 --- a/backend/probe/README.md +++ b/backend/probe/README.md @@ -1,16 +1,17 @@ # Monitor Probe -Probe 是网络测量的发起端，通常运行在家宽或其他需要观测的网络出口。它连接 Agent、执行线路任务，并把本端采集的原始数据交给 Agent 汇总上报。Probe 不直接连接 Backend，也不提供远程命令或文件管理功能。 +Probe 是网络测量的发起端，通常运行在家宽或其他需要观测的网络出口。它连接 Backend（身份、Agent 列表、状态）和 Agent（测量），执行线路任务，并把本端采集的原始数据交给 Agent 汇总上报。Probe 不提供远程命令或文件管理功能。 ## 工作流程 ```text -Backend ──任务与节点列表──> Agent <──测量会话──> Probe - │ │ - └──汇总并上报数据 <───┘ +Backend ──任务──> Agent <──测量会话──> Probe + │ ▲ │ │ + │ └──汇总并上报数据 <───┘ │ + └─────Agent 列表、状态（Fleet 长连接）───────────┘ ``` -Probe 使用面板生成的 `probe.json` 读取自身身份、首次连接用的 Agent 地址和本地测试限制。成功连接后，Agent 会同步可用节点列表；Probe 将列表缓存到同目录的 `agents.json`。`probe.json` 含私钥，请限制文件访问权限，不要提交到版本控制。 +Probe 使用面板生成的 `probe.json` 读取自身身份、面板地址（`server`）和本地测试限制。连上面板之后，面板下发要连接的 Agent 列表；Probe 将列表缓存到同目录的 `agents.json`，面板暂时连不上时靠它继续测量。`probe.json` 含私钥，请限制文件访问权限，不要提交到版本控制。 ## 构建 diff --git a/backend/probe/src/Monitor.Probe/Link/AgentConnection.cs b/backend/probe/src/Monitor.Probe/Link/AgentConnection.cs index 2a7661c..f01ad3d 100644 --- a/backend/probe/src/Monitor.Probe/Link/AgentConnection.cs +++ b/backend/probe/src/Monitor.Probe/Link/AgentConnection.cs @@ -15,7 +15,7 @@ namespace Monitor.Probe; public sealed record ConnectionEnv(NodeIdentity Identity, string ProbeId, ProbeLimits Limits, AgentDirectory Directory, MonoClock Clock, TimeProvider Time, INodeLog Log); /// -/// 与一个 Agent 的点对点会话：认证握手 → 同步时钟 → 取 Agent 列表 → 按 Agent 下发的 line_task_set 运行角色 → +/// 与一个 Agent 的点对点会话：认证握手 → 同步时钟 → 按 Agent 下发的 line_task_set 运行角色 → /// 每 10 秒把已完成的行切成列块放进发件箱 → 响应按需测试的阶段。会话断开时所有角色停止（凭据属于会话），按退避重连。 /// 会话内的所有状态（套接字、发件箱、角色）随会话一起创建和释放。 /// @@ -28,13 +28,13 @@ public sealed class AgentConnection private readonly ConnectionEnv _env; private volatile ConnectionSession? _session; - public AgentConnection(BootstrapAgent endpoint, ConnectionEnv env) + public AgentConnection(KnownAgent endpoint, ConnectionEnv env) { Endpoint = endpoint; _env = env; } - public BootstrapAgent Endpoint { get; } + public KnownAgent Endpoint { get; } public bool IsConnected => _session is not null; @@ -105,12 +105,11 @@ public sealed class AgentConnection Task pump = PumpAsync(rpc, outbox, cts.Token); try { - // 先同步时钟（块头里要带偏移），再取 Agent 列表；Agent 随后推送线路任务集 + // 先同步时钟（块头里要带偏移）；Agent 随后推送线路任务集 await SyncClockAsync(rpc, clock, rounds: 5, cts.Token); backoff.Reset(); _session = session; _env.Log.Info("agent session up", ("agent", Endpoint.AgentId), ("offset_us", clock.OffsetUs)); - await RefreshDirectoryAsync(rpc, cts.Token); Task maintenance = MaintainAsync(rpc, session, clock, cts.Token); Task finished = await Task.WhenAny(run, receive, pump, maintenance); await cts.CancelAsync(); @@ -163,23 +162,6 @@ public sealed class AgentConnection } } - /// 本机的系统和架构：随列表同步一起告诉 Agent，只用于在面板上展示。 - private static readonly PlatformInfo s_platform = PlatformDetector.Detect(); - - private async Task RefreshDirectoryAsync(RpcSession rpc, CancellationToken ct) - { - Result list = await rpc.RequestAsync( - MonitorMethods.ProbeAgents, new ProbeAgentsParams { HaveVersion = _env.Directory.Version, Platform = s_platform }, MonitorJsonContext.Default.ProbeAgentsParams, MonitorJsonContext.Default.ProbeAgentsResult, ct); - if (list.TryGet(out ProbeAgentsResult? agents, out FleetError? error)) - { - _env.Directory.Apply(agents); - } - else - { - _env.Log.Warn("could not fetch the agent list", ("agent", Endpoint.AgentId), ("err", error.Code)); - } - } - /// 切块（每 10 秒）和时钟同步（每 30 秒）。 private async Task MaintainAsync(RpcSession rpc, ConnectionSession session, ClockEstimator clock, CancellationToken ct) { @@ -232,14 +214,6 @@ public sealed class AgentConnection MonitorMethods.TestAbort, MonitorJsonContext.Default.TestAbortParams, async (session, abort, _) => await session.AbortAsync(abort.TestId)) - .OnNotify( - MonitorMethods.ProbeAgentsChanged, - MonitorJsonContext.Default.ProbeAgentsResult, - (session, list, _) => - { - session.Env.Directory.Apply(list); - return ValueTask.CompletedTask; - }) .OnNotify( FleetMethods.OutboxAck, FleetRpcContext.Default.OutboxAckParams, diff --git a/backend/probe/src/Monitor.Probe/Link/AgentDirectory.cs b/backend/probe/src/Monitor.Probe/Link/AgentDirectory.cs index ad06d12..4c74202 100644 --- a/backend/probe/src/Monitor.Probe/Link/AgentDirectory.cs +++ b/backend/probe/src/Monitor.Probe/Link/AgentDirectory.cs @@ -6,8 +6,7 @@ using Monitor.Protocol; namespace Monitor.Probe; /// -/// Probe 要连接的 Agent 列表。Probe 不认识 Backend：列表先取自 probe.json 的 bootstrap（或上一次保存的 agents.json）， -/// 连上任意一个 Agent 之后，由它代为向 Backend 查询（monitor.probe.agents），之后的变化由 Agent 推送（agents_changed）。 +/// Probe 要连接的 Agent 列表：由面板作为期望状态（probe_agents）直接下发；面板连不上时用上一次保存的 agents.json。 /// 列表变化时用 写回 agents.json（这是配置，不是测量数据；只在变化时写，保护路由器的闪存）。 /// public sealed class AgentDirectory @@ -17,29 +16,25 @@ public sealed class AgentDirectory private readonly object _gate = new(); private readonly ConfigStore _store; private readonly INodeLog _log; - private List _agents; + private List _agents = []; private long _version; - public AgentDirectory(ConfigStore store, string configDirectory, ProbeFile file, INodeLog log) + public AgentDirectory(ConfigStore store, string configDirectory, INodeLog log) { _store = store; _log = log; - _agents = [.. file.Bootstrap]; string cache = Path.Combine(configDirectory, FileName); if (File.Exists(cache)) { try { AgentsCache saved = NodeConfig.Load(cache, ProbeConfigContext.Default.AgentsCache); - if (saved.Agents.Count > 0) - { - _agents = saved.Agents; - _version = saved.Version; - } + _agents = saved.Agents; + _version = saved.Version; } catch (Exception ex) when (ex is IOException or InvalidDataException or JsonException or InvalidOperationException) { - _log.Warn("agents.json is unreadable, using the bootstrap list", ("err", ex.Message)); + _log.Warn("agents.json is unreadable, waiting for the panel", ("err", ex.Message)); } } } @@ -58,7 +53,7 @@ public sealed class AgentDirectory } } - public IReadOnlyList Agents + public IReadOnlyList Agents { get { @@ -69,10 +64,10 @@ public sealed class AgentDirectory } } - /// 应用 Agent 转来的列表。版本不比当前新的忽略（乱序到达的旧推送）。 - public void Apply(ProbeAgentsResult result) + /// 应用面板下发的列表。版本不比当前新的忽略（乱序到达的旧推送）。 + public void Apply(ProbeAgentList result) { - var next = result.Agents.Select(a => new BootstrapAgent { AgentId = a.AgentId.Value, Host = a.Host, TcpPort = a.TcpPort, Spki = a.Spki }).ToList(); + var next = result.Agents.Select(a => new KnownAgent { AgentId = a.AgentId.Value, Host = a.Host, TcpPort = a.TcpPort, Spki = a.Spki }).ToList(); lock (_gate) { if (result.Version < _version || (result.Version == _version && Same(next, _agents))) @@ -97,9 +92,9 @@ public sealed class AgentDirectory Changed?.Invoke(); } - private static bool Same(List a, List b) => + private static bool Same(List a, List b) => a.Count == b.Count && a.Zip(b).All(p => Equal(p.First, p.Second)); - public static bool Equal(BootstrapAgent a, BootstrapAgent b) => + public static bool Equal(KnownAgent a, KnownAgent b) => a.AgentId == b.AgentId && a.Host == b.Host && a.TcpPort == b.TcpPort && a.Spki == b.Spki; } diff --git a/backend/probe/src/Monitor.Probe/ProbeApp.cs b/backend/probe/src/Monitor.Probe/ProbeApp.cs index 24ece9a..11866ef 100644 --- a/backend/probe/src/Monitor.Probe/ProbeApp.cs +++ b/backend/probe/src/Monitor.Probe/ProbeApp.cs @@ -1,13 +1,17 @@ using Fleet.Core; using Fleet.Hosting; +using System.Text.Json; using Fleet.Identity; +using Fleet.Link; +using Fleet.Rpc; using Monitor.Measurement; +using Monitor.Protocol; namespace Monitor.Probe; /// -/// Probe 的组装点：Agent 列表 → 每个 Agent 一个连接（认证、重连、角色）。没有全局状态，测试可以在同一进程里启动多个。 -/// 列表变化时只调和连接：新增的建立，移除的断开，地址或证书指纹变化的重建，其余不受影响。 +/// Probe 的组装点：到面板的长连接（身份、Agent 列表、状态）+ 每个 Agent 一个测量连接（认证、重连、角色）。没有全局状态，测试可以在同一进程里启动多个。 +/// Agent 列表变化时只调和连接：新增的建立，移除的断开，地址或证书指纹变化的重建，其余不受影响。 /// public sealed class ProbeApp : IDisposable { @@ -17,16 +21,39 @@ public sealed class ProbeApp : IDisposable private readonly Dictionary _connections = []; private readonly SemaphoreSlim _changed = new(0); private readonly INodeLog _log; + private readonly FleetLink _link; public ProbeApp(ProbeFile file, string configDirectory, TimeProvider time, INodeLog log) { _log = log; - var directory = new AgentDirectory(new ConfigStore(configDirectory), configDirectory, file, log); + var directory = new AgentDirectory(new ConfigStore(configDirectory), configDirectory, log); NodeIdentity identity = NodeIdentity.FromPkcs8(Convert.FromBase64String(file.PrivateKey)); _env = new ConnectionEnv(identity, file.ProbeId, file.Limits, directory, new MonoClock(time), time, log); directory.Changed += () => _changed.Release(); + + var options = new FleetLinkOptions + { + Server = new Uri(file.Server), + Identity = identity, + NodeId = new NodeId(file.ProbeId), + NodeVersion = Version, + Capabilities = [MonitorCapabilities.Base], + HelloMeta = () => JsonSerializer.SerializeToElement(new ProbeHelloMeta { Platform = PlatformDetector.Detect() }, MonitorJsonContext.Default.ProbeHelloMeta), + }; + var list = new StateHandler( + ProbeAgentList.StateType, + MonitorJsonContext.Default.ProbeAgentList, + (version, doc, _) => + { + directory.Apply(doc); + return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version }); + }); + _link = new FleetLink(options, new RpcRouter(), [list], new Outbox(1 << 20), time, log); } + /// 到面板的连接（测试和状态查看用）。 + public FleetLink Link => _link; + public AgentDirectory Directory => _env.Directory; /// 某个 Agent 的会话是否已经建立（测试和状态查看用）。 @@ -50,9 +77,10 @@ public sealed class ProbeApp : IDisposable { if (_env.Directory.Agents.Count == 0) { - _log.Warn("no agents known yet: probe.json has an empty bootstrap list"); + _log.Info("no agents known yet, waiting for the panel to send the list"); } + Task link = _link.RunAsync(ct); try { while (!ct.IsCancellationRequested) @@ -68,6 +96,7 @@ public sealed class ProbeApp : IDisposable finally { await ReconcileAsync([], CancellationToken.None); + await Task.WhenAny(link, Task.Delay(TimeSpan.FromSeconds(5), CancellationToken.None)); } } @@ -77,7 +106,7 @@ public sealed class ProbeApp : IDisposable _changed.Dispose(); } - private async Task ReconcileAsync(IReadOnlyList agents, CancellationToken ct) + private async Task ReconcileAsync(IReadOnlyList agents, CancellationToken ct) { var wanted = agents.ToDictionary(a => a.AgentId); var stopped = new List(); @@ -85,7 +114,7 @@ public sealed class ProbeApp : IDisposable { foreach ((string id, Running running) in _connections.ToList()) { - if (!wanted.TryGetValue(id, out BootstrapAgent? endpoint) || !AgentDirectory.Equal(endpoint, running.Connection.Endpoint)) + if (!wanted.TryGetValue(id, out KnownAgent? endpoint) || !AgentDirectory.Equal(endpoint, running.Connection.Endpoint)) { stopped.Add(running); _connections.Remove(id); @@ -105,7 +134,7 @@ public sealed class ProbeApp : IDisposable lock (_connections) { - foreach ((string id, BootstrapAgent endpoint) in wanted) + foreach ((string id, KnownAgent endpoint) in wanted) { if (!_connections.ContainsKey(id)) { diff --git a/backend/probe/src/Monitor.Probe/ProbeConfig.cs b/backend/probe/src/Monitor.Probe/ProbeConfig.cs index 9b72feb..bcaf11e 100644 --- a/backend/probe/src/Monitor.Probe/ProbeConfig.cs +++ b/backend/probe/src/Monitor.Probe/ProbeConfig.cs @@ -15,15 +15,15 @@ public sealed class ProbeFile [JsonPropertyName("private_key")] public string PrivateKey { get; set; } = ""; - /// 首次连接用的 Agent；连上之后以 agents.json（最新的列表）为准。 - [JsonPropertyName("bootstrap")] - public List Bootstrap { get; set; } = []; + /// 面板的地址（http 或 https）。Probe 直接连它：身份、Agent 列表、更新都从这里来。 + [JsonPropertyName("server")] + public string Server { get; set; } = ""; [JsonPropertyName("limits")] public ProbeLimits Limits { get; set; } = new(); } -public sealed class BootstrapAgent +public sealed class KnownAgent { [JsonPropertyName("agent_id")] public string AgentId { get; set; } = ""; @@ -54,12 +54,12 @@ public sealed class ProbeLimits [JsonSerializable(typeof(AgentsCache))] public sealed partial class ProbeConfigContext : JsonSerializerContext; -/// agents.json：最近一次从 Agent 同步到的列表（配置，不是测量数据）。 +/// agents.json：最近一次从面板同步到的 Agent 列表（配置，不是测量数据）；面板连不上时 Probe 靠它继续测量。 public sealed class AgentsCache { [JsonPropertyName("version")] public long Version { get; set; } [JsonPropertyName("agents")] - public List Agents { get; set; } = []; + public List Agents { get; set; } = []; } diff --git a/backend/probe/src/Monitor.Probe/Program.cs b/backend/probe/src/Monitor.Probe/Program.cs index 4892703..2a025ac 100644 --- a/backend/probe/src/Monitor.Probe/Program.cs +++ b/backend/probe/src/Monitor.Probe/Program.cs @@ -24,9 +24,9 @@ ProbeFile file; try { file = NodeConfig.Load(configPath, ProbeConfigContext.Default.ProbeFile); - if (file.ProbeId.Length == 0 || file.PrivateKey.Length == 0) + if (file.ProbeId.Length == 0 || file.PrivateKey.Length == 0 || !Uri.TryCreate(file.Server, UriKind.Absolute, out _)) { - throw new InvalidOperationException("probe.json needs probe_id and private_key (download it from the panel: POST /api/admin/probe/identity)"); + throw new InvalidOperationException("probe.json needs probe_id, private_key and server (download it from the panel: Probes → Create)"); } } catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidOperationException or System.Text.Json.JsonException) diff --git a/backend/probe/tests/Monitor.Probe.Tests/ProbeTests.cs b/backend/probe/tests/Monitor.Probe.Tests/ProbeTests.cs index 60412c3..89d3f20 100644 --- a/backend/probe/tests/Monitor.Probe.Tests/ProbeTests.cs +++ b/backend/probe/tests/Monitor.Probe.Tests/ProbeTests.cs @@ -30,23 +30,22 @@ public sealed class ProbeTests return path; } - private static ProbeAgentsResult List(long version, params string[] ids) => new() + private static ProbeAgentList List(long version, params string[] ids) => new() { Version = version, Agents = [.. ids.Select(id => new AgentEndpoint { AgentId = new NodeId(id), Host = id + ".example", TcpPort = 7443, Spki = "c3BraQ==" })], }; [Fact] - public void AgentDirectory_StartsFromBootstrap_IgnoresStaleLists_PersistsChanges_AndReloadsThem() + public void AgentDirectory_StartsEmpty_IgnoresStaleLists_PersistsChanges_AndReloadsThem() { string directory = NewDirectory(); try { - var file = new ProbeFile { ProbeId = "prb_x", Bootstrap = [new BootstrapAgent { AgentId = "agt_boot", Host = "boot.example", TcpPort = 7443, Spki = "c3BraQ==" }] }; - var directoryA = new AgentDirectory(new ConfigStore(directory), directory, file, NullNodeLog.Instance); + var directoryA = new AgentDirectory(new ConfigStore(directory), directory, NullNodeLog.Instance); int changes = 0; directoryA.Changed += () => changes++; - Assert.Equal(["agt_boot"], directoryA.Agents.Select(a => a.AgentId)); + Assert.Empty(directoryA.Agents); directoryA.Apply(List(5, "agt_1", "agt_2")); Assert.Equal(1, changes); @@ -57,8 +56,8 @@ public sealed class ProbeTests Assert.Equal(1, changes); Assert.Equal(5, directoryA.Version); - // The next start uses the saved list, not the bootstrap one - var directoryB = new AgentDirectory(new ConfigStore(directory), directory, file, NullNodeLog.Instance); + // The next start uses the saved list while the panel is not reachable yet + var directoryB = new AgentDirectory(new ConfigStore(directory), directory, NullNodeLog.Instance); Assert.Equal(["agt_1", "agt_2"], directoryB.Agents.Select(a => a.AgentId)); Assert.Equal(5, directoryB.Version); } @@ -69,15 +68,14 @@ public sealed class ProbeTests } [Fact] - public void AgentDirectory_FallsBackToBootstrap_WhenTheSavedFileIsCorrupt() + public void AgentDirectory_StartsEmpty_WhenTheSavedFileIsCorrupt() { string directory = NewDirectory(); try { new ConfigStore(directory).Write(AgentDirectory.FileName, "{ not json"u8.ToArray()); - var file = new ProbeFile { Bootstrap = [new BootstrapAgent { AgentId = "agt_boot", Host = "h", TcpPort = 1, Spki = "AA==" }] }; - var loaded = new AgentDirectory(new ConfigStore(directory), directory, file, NullNodeLog.Instance); - Assert.Equal(["agt_boot"], loaded.Agents.Select(a => a.AgentId)); + var loaded = new AgentDirectory(new ConfigStore(directory), directory, NullNodeLog.Instance); + Assert.Empty(loaded.Agents); } finally { diff --git a/backend/shared/Monitor.Protocol/Rpc/MonitorRpc.cs b/backend/shared/Monitor.Protocol/Rpc/MonitorRpc.cs index 356b373..6d5db07 100644 --- a/backend/shared/Monitor.Protocol/Rpc/MonitorRpc.cs +++ b/backend/shared/Monitor.Protocol/Rpc/MonitorRpc.cs @@ -13,8 +13,6 @@ public static class MonitorMethods public const string TestStatus = "monitor.test.status"; public const string Event = "monitor.event"; public const string Host = "monitor.host"; - public const string ProbeAgents = "monitor.probe.agents"; - public const string ProbeAgentsChanged = "monitor.probe.agents_changed"; // Probe ↔ Agent（点对点会话） public const string Clock = "monitor.clock"; @@ -147,19 +145,11 @@ public sealed class AgentEndpoint public string Spki { get; set; } = ""; } -public sealed class ProbeAgentsParams +/// 期望状态文档（Fleet state type = probe_agents）：Backend 直接下发给每个 Probe 的 Agent 列表。 +public sealed class ProbeAgentList { - /// Probe 目前持有的列表版本（用于调试；服务端总是返回当前列表）。 - [JsonPropertyName("have_version")] - public long HaveVersion { get; set; } + public const string StateType = "probe_agents"; - /// Probe 的系统和架构（只用于展示；Agent 记在会话上，随连接状态报给 Backend）。 - [JsonPropertyName("platform")] - public PlatformInfo? Platform { get; set; } -} - -public sealed class ProbeAgentsResult -{ [JsonPropertyName("version")] public long Version { get; set; } @@ -227,6 +217,13 @@ public sealed class PlatformInfo public string? BuiltAt { get; set; } } +/// Probe 的 fleet.hello 附加字段：系统、架构和构建信息（只用于展示和更新）。 +public sealed class ProbeHelloMeta +{ + [JsonPropertyName("platform")] + public PlatformInfo? Platform { get; set; } +} + /// fleet.hello 的附加字段。 public sealed class AgentHelloMeta { @@ -302,9 +299,6 @@ public sealed class ConnectedProbe [JsonPropertyName("since_utc_ms")] public long SinceUtcMs { get; set; } - - [JsonPropertyName("platform")] - public PlatformInfo? Platform { get; set; } } /// fleet.heartbeat 的附加字段：只有进程内的计数，没有主机信息。 @@ -470,8 +464,8 @@ public static class FragmentKeys [JsonSerializable(typeof(HostSnapshot))] [JsonSerializable(typeof(TestStatusParams))] [JsonSerializable(typeof(AgentEventParams))] -[JsonSerializable(typeof(ProbeAgentsParams))] -[JsonSerializable(typeof(ProbeAgentsResult))] +[JsonSerializable(typeof(ProbeAgentList))] +[JsonSerializable(typeof(ProbeHelloMeta))] [JsonSerializable(typeof(AgentEnrollMeta))] [JsonSerializable(typeof(PlatformInfo))] [JsonSerializable(typeof(AgentHelloMeta))] diff --git a/backend/shared/Monitor.Protocol/Tasks/TaskSet.cs b/backend/shared/Monitor.Protocol/Tasks/TaskSet.cs index 89476d0..2c320a5 100644 --- a/backend/shared/Monitor.Protocol/Tasks/TaskSet.cs +++ b/backend/shared/Monitor.Protocol/Tasks/TaskSet.cs @@ -56,9 +56,6 @@ public sealed class TaskSet /// 允许连接到这个 Agent 的 Probe 公钥。 [JsonPropertyName("peers")] public List Peers { get; set; } = []; - - [JsonPropertyName("probe_agents_version")] - public long ProbeAgentsVersion { get; set; } } public static class TaskSchedule diff --git a/backend/src/Monitor.Backend.Api/Admin/AgentEndpoints.cs b/backend/src/Monitor.Backend.Api/Admin/AgentEndpoints.cs index a4f0d18..a9ff75d 100644 --- a/backend/src/Monitor.Backend.Api/Admin/AgentEndpoints.cs +++ b/backend/src/Monitor.Backend.Api/Admin/AgentEndpoints.cs @@ -150,9 +150,9 @@ public static class AgentEndpoints private static async Task ListProbesAsync(ProbeService probes, CancellationToken ct) => TypedResults.Json(new PageDto([.. (await probes.ListAsync(ct)).Select(Mappers.ToDto)], null), ApiJsonContext.Default.PageDtoProbeDto); - private static async Task CreateProbeAsync(HttpContext http, ProbeService probes, CreateProbeRequest? request, CancellationToken ct) + private static async Task CreateProbeAsync(HttpContext http, ProbeService probes, BackendOptions options, CreateProbeRequest? request, CancellationToken ct) { - Result result = await probes.CreateAsync(request?.Name, ct); + Result result = await probes.CreateAsync(request?.Name, PanelUrl(http, options), ct); if (!result.TryGet(out ProbeIdentityFile? file, out FleetError? error)) { return Problems.From(http, error); @@ -179,9 +179,9 @@ public static class AgentEndpoints return result.TryGet(out ProbeView? view, out FleetError? error) ? TypedResults.Json(Mappers.ToDto(view), ApiJsonContext.Default.ProbeDto) : Problems.From(http, error); } - private static async Task RotateProbeAsync(HttpContext http, ProbeService probes, string id, CancellationToken ct) + private static async Task RotateProbeAsync(HttpContext http, ProbeService probes, BackendOptions options, string id, CancellationToken ct) { - Result result = await probes.RotateAsync(id, ct); + Result result = await probes.RotateAsync(id, PanelUrl(http, options), ct); if (!result.TryGet(out ProbeIdentityFile? file, out FleetError? error)) { return Problems.From(http, error); @@ -214,8 +214,9 @@ public static class AgentEndpoints return dropped ? TypedResults.NoContent() : Problems.Create(http, MonitorErrors.ProbeOffline, "this probe has no session with this agent"); } + /// Probe 要连的面板地址：配置的公开地址，没有就用这次请求的地址。 + private static string PanelUrl(HttpContext http, BackendOptions options) => (options.PublicUrl ?? $"{http.Request.Scheme}://{http.Request.Host}").TrimEnd('/'); + private static ProbeIdentityDto ToIdentityDto(ProbeIdentityFile file) => - new( - file.ProbeId.Value, file.PrivateKey, [.. file.Bootstrap.Select(b => new BootstrapAgentDto(b.AgentId.Value, b.Host, b.TcpPort, b.Spki))], - new ProbeLimitsDto(file.Limits.MaxTestMbps, file.Limits.MaxTotalPps)); + new(file.ProbeId.Value, file.PrivateKey, file.Server, new ProbeLimitsDto(file.Limits.MaxTestMbps, file.Limits.MaxTotalPps)); } diff --git a/backend/src/Monitor.Backend.Api/Common/Mappers.cs b/backend/src/Monitor.Backend.Api/Common/Mappers.cs index 79e3d7a..198d99a 100644 --- a/backend/src/Monitor.Backend.Api/Common/Mappers.cs +++ b/backend/src/Monitor.Backend.Api/Common/Mappers.cs @@ -40,7 +40,7 @@ public static class Mappers public static ProbeDto ToDto(ProbeView v) => new( - v.Probe.Id, v.Probe.Name, Time(v.Probe.CreatedAtMs), Time(v.Probe.RevokedAtMs), v.Revoked, + v.Probe.Id, v.Probe.Name, Time(v.Probe.CreatedAtMs), Time(v.Probe.RevokedAtMs), v.Revoked, v.Online, [.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs)))]); public static SeriesDto ToDto(SeriesResult r, bool live) => diff --git a/backend/src/Monitor.Backend.Api/Contracts/AdminDtos.cs b/backend/src/Monitor.Backend.Api/Contracts/AdminDtos.cs index 83ce68b..129b260 100644 --- a/backend/src/Monitor.Backend.Api/Contracts/AdminDtos.cs +++ b/backend/src/Monitor.Backend.Api/Contracts/AdminDtos.cs @@ -57,14 +57,12 @@ public sealed record ConnectedProbeDto(string ProbeId, string? Version, string? /// 某个 Probe 现在连着的一个 Agent。 public sealed record ProbeConnectionDto(string AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, DateTimeOffset ConnectedSince); -public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, IReadOnlyList Connections); - -public sealed record BootstrapAgentDto(string AgentId, string Host, int TcpPort, string Spki); +public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList Connections); public sealed record ProbeLimitsDto(double MaxTestMbps, long MaxTotalPps); /// probe.json：含私钥，只在生成时返回一次。 -public sealed record ProbeIdentityDto(string ProbeId, string PrivateKey, IReadOnlyList Bootstrap, ProbeLimitsDto Limits); +public sealed record ProbeIdentityDto(string ProbeId, string PrivateKey, string Server, ProbeLimitsDto Limits); public sealed record CreateProbeRequest(string? Name); diff --git a/backend/src/Monitor.Backend.Application/Agents/AgentEventService.cs b/backend/src/Monitor.Backend.Application/Agents/AgentEventService.cs index 1afbb61..957624c 100644 --- a/backend/src/Monitor.Backend.Application/Agents/AgentEventService.cs +++ b/backend/src/Monitor.Backend.Application/Agents/AgentEventService.cs @@ -34,8 +34,8 @@ public sealed class AgentEventService switch (e.Kind) { case AgentEventKinds.ProbeConnected when !string.IsNullOrEmpty(e.Detail): - state.AddProbe(e.Probe is { } p ? new ConnectedProbeInfo(e.Detail, p.Version, p.Remote, p.SinceUtcMs > 0 ? p.SinceUtcMs : at, p.Platform) : new ConnectedProbeInfo(e.Detail, null, null, at)); - _nodes.ObserveProbeInBackground(e.Detail, e.Probe?.Remote, e.Probe?.Platform); + state.AddProbe(e.Probe is { } p ? new ConnectedProbeInfo(e.Detail, p.Version, p.Remote, p.SinceUtcMs > 0 ? p.SinceUtcMs : at) : new ConnectedProbeInfo(e.Detail, null, null, at)); + _nodes.ObserveProbeInBackground(e.Detail, e.Probe?.Remote); break; case AgentEventKinds.ProbeDisconnected when !string.IsNullOrEmpty(e.Detail): state.RemoveProbe(e.Detail); diff --git a/backend/src/Monitor.Backend.Application/Agents/AgentRuntime.cs b/backend/src/Monitor.Backend.Application/Agents/AgentRuntime.cs index fd1d2ba..98e7738 100644 --- a/backend/src/Monitor.Backend.Application/Agents/AgentRuntime.cs +++ b/backend/src/Monitor.Backend.Application/Agents/AgentRuntime.cs @@ -5,7 +5,7 @@ using Monitor.Protocol; namespace Monitor.Backend.Application; /// Agent 眼中的一个已连接 Probe（Agent 在事件和心跳里上报）。 -public sealed record ConnectedProbeInfo(string Id, string? Version, string? Remote, long SinceMs, PlatformInfo? Platform = null); +public sealed record ConnectedProbeInfo(string Id, string? Version, string? Remote, long SinceMs); /// Agent 的运行时状态（只在内存中）：连着哪些 Probe、最近的进程内计数、端点变化次数。 public sealed class AgentRuntimeState diff --git a/backend/src/Monitor.Backend.Application/Agents/AgentService.cs b/backend/src/Monitor.Backend.Application/Agents/AgentService.cs index c3418b7..61550f5 100644 --- a/backend/src/Monitor.Backend.Application/Agents/AgentService.cs +++ b/backend/src/Monitor.Backend.Application/Agents/AgentService.cs @@ -123,11 +123,11 @@ public sealed partial class AgentService } /// Probe 需要连接的 Agent 列表（所有已激活的 Agent）。版本号在列表变化时递增。 - public async ValueTask ProbeAgentsAsync(CancellationToken ct) + public async ValueTask ProbeAgentsAsync(CancellationToken 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 ProbeAgentsResult { Version = version }; + var result = new ProbeAgentList { Version = version }; foreach (AgentRecord agent in agents.Where(a => a.Status == NodeStatus.Active && a.PublicHost.Length > 0)) { result.Agents.Add(new AgentEndpoint @@ -144,12 +144,11 @@ public sealed partial class AgentService return result; } - /// Agent 集合变了：递增列表版本、重算各 Agent 的期望状态、通知 Probe。 + /// Agent 集合变了：递增列表版本，重算各 Agent 的期望状态，并把新的 Agent 列表推给 Probe（都是期望状态，一起通知）。 public async ValueTask AfterAgentSetChangedAsync(CancellationToken ct) { await _planner.BumpProbeAgentsVersionAsync(ct); _gateway.NotifyAllTaskSetsChanged(); - await _gateway.NotifyProbeAgentsChangedAsync(await ProbeAgentsAsync(ct), ct); } private async ValueTask> ChangeStatusAsync(NodeId id, NodeStatus status, CancellationToken ct) diff --git a/backend/src/Monitor.Backend.Application/Agents/MonitorFleetHooks.cs b/backend/src/Monitor.Backend.Application/Agents/MonitorFleetHooks.cs index d8f6b27..d032553 100644 --- a/backend/src/Monitor.Backend.Application/Agents/MonitorFleetHooks.cs +++ b/backend/src/Monitor.Backend.Application/Agents/MonitorFleetHooks.cs @@ -66,6 +66,12 @@ public sealed class MonitorNodeEventSink : INodeEventSink 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: @@ -89,10 +95,10 @@ public sealed class MonitorNodeEventSink : INodeEventSink state.LastHeartbeatAtMs = nodeEvent.AtMs; if (beat is not null) { - state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs, p.Platform))); + 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, p.Platform); + _nodes.ObserveProbeInBackground(p.Id.Value, p.Remote); } } } @@ -105,6 +111,29 @@ public sealed class MonitorNodeEventSink : INodeEventSink } } + /// Probe 自己连上来了：hello 里带着它的系统、架构和构建信息；上线下线只推给 SSE，不记事件。 + private async ValueTask OnProbeEventAsync(NodeId id, NodeEvent nodeEvent, CancellationToken ct) + { + switch (nodeEvent.Kind) + { + case NodeEventKind.Online or NodeEventKind.Offline: + _live.Publish(new NodeSignal("probe", id.Value, nodeEvent.Kind == NodeEventKind.Online, nodeEvent.AtMs)); + break; + case NodeEventKind.Hello when nodeEvent.Hello?.Meta is { } raw: + try + { + ProbeHelloMeta? meta = raw.Deserialize(MonitorJsonContext.Default.ProbeHelloMeta); + await _nodes.SetPlatformAsync(id.Value, NodeKinds.Probe, meta?.Platform, ct); + } + catch (JsonException) + { + // 格式不对就忽略 + } + + break; + } + } + private async ValueTask ApplyHelloAsync(NodeId id, HelloParams hello, CancellationToken ct) { var kinds = new List(); diff --git a/backend/src/Monitor.Backend.Application/Agents/ProbeService.cs b/backend/src/Monitor.Backend.Application/Agents/ProbeService.cs index 263db9f..c5db8cf 100644 --- a/backend/src/Monitor.Backend.Application/Agents/ProbeService.cs +++ b/backend/src/Monitor.Backend.Application/Agents/ProbeService.cs @@ -6,49 +6,45 @@ using Monitor.Protocol; namespace Monitor.Backend.Application; -public sealed record BootstrapAgent(NodeId AgentId, string Host, int TcpPort, string Spki); - /// 本地硬上限：写在 Probe 本机的配置里，面板无法修改（在路由器上建议调低）。 public sealed record ProbeLimits(double MaxTestMbps, long MaxTotalPps); -/// probe.json：只在生成（或轮换密钥）时返回一次，含私钥。 -public sealed record ProbeIdentityFile(ProbeId ProbeId, string PrivateKey, IReadOnlyList Bootstrap, ProbeLimits Limits); +/// probe.json：只在生成（或轮换密钥）时返回一次，含私钥；Server 是 Probe 要连的面板地址。 +public sealed record ProbeIdentityFile(ProbeId ProbeId, string PrivateKey, string Server, ProbeLimits Limits); /// 某个 Probe 现在连着的一个 Agent（Agent 上报）。 public sealed record ProbeConnection(NodeId AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, long SinceMs); -public sealed record ProbeView(ProbeRecord Probe, IReadOnlyList Connections) +public sealed record ProbeView(ProbeRecord Probe, bool Online, IReadOnlyList Connections) { public bool Revoked => Probe.RevokedAtMs is not null; } /// /// Probe 身份（Backend 生成，私钥只返回一次）：可以有多个，每个都能改名、轮换密钥、吊销、删除。 -/// 所有未吊销的 Probe 的公钥都会写进每个 Agent 的期望状态（peers）；吊销、删除、轮换密钥之后，Agent 立刻断开对应会话。 +/// 所有未吊销的 Probe 的公钥都会写进每个 Agent 的期望状态（peers）；吊销、删除、轮换密钥之后，Agent 立刻断开对应会话，Probe 到面板的连接也会被拒绝。 /// 一条线路是一对（Probe，Agent）。 /// public sealed partial class ProbeService { private readonly IProbeStore _probes; private readonly IAgentStore _agents; - private readonly AgentService _agentService; private readonly AgentRuntime _runtime; private readonly IAgentGateway _gateway; private readonly TimeProvider _time; private readonly NodeInfoService _nodes; - public ProbeService(IProbeStore probes, IAgentStore agents, AgentService agentService, AgentRuntime runtime, IAgentGateway gateway, TimeProvider time, NodeInfoService nodes) + public ProbeService(IProbeStore probes, IAgentStore agents, AgentRuntime runtime, IAgentGateway gateway, TimeProvider time, NodeInfoService nodes) { _nodes = nodes; _probes = probes; _agents = agents; - _agentService = agentService; _runtime = runtime; _gateway = gateway; _time = time; } - public async ValueTask> CreateAsync(string? name, CancellationToken ct) + public async ValueTask> CreateAsync(string? name, string server, CancellationToken ct) { string finalName = string.IsNullOrWhiteSpace(name) ? "probe" : name.Trim(); if (!NamePattern().IsMatch(finalName)) @@ -65,7 +61,7 @@ public sealed partial class ProbeService var probe = new ProbeRecord(Ulid.NewPrefixed(ProbeId.Prefix, _time), finalName, identity.PublicKeySpki, _time.GetUtcNow().ToUnixTimeMilliseconds(), null); await _probes.SaveAsync(probe, ct); _gateway.NotifyAllTaskSetsChanged(); - return Result.Ok(await BuildFileAsync(probe, identity, ct)); + return Result.Ok(BuildFile(probe, identity, server)); } public async ValueTask> ListAsync(CancellationToken ct) @@ -101,7 +97,7 @@ public sealed partial class ProbeService } /// 换一把新密钥（id 不变）：旧私钥立即失效，所有 Agent 上这个 Probe 的会话被断开。返回新的 probe.json（只此一次）。 - public async ValueTask> RotateAsync(string id, CancellationToken ct) + public async ValueTask> RotateAsync(string id, string server, CancellationToken ct) { if (await _probes.GetAsync(id, ct) is not { } probe) { @@ -118,7 +114,7 @@ public sealed partial class ProbeService await _probes.SaveAsync(updated, ct); _gateway.NotifyAllTaskSetsChanged(); await DisconnectEverywhereAsync(id, ct); - return Result.Ok(await BuildFileAsync(updated, identity, ct)); + return Result.Ok(BuildFile(updated, identity, server)); } /// 吊销：保留记录和历史数据，公钥从所有 Agent 的名单里移除，Agent 立即断开它的会话。 @@ -183,18 +179,11 @@ public sealed partial class ProbeService } } - return new ProbeView(probe, [.. connections.OrderBy(c => c.AgentName, StringComparer.Ordinal)]); + return new ProbeView(probe, _gateway.IsOnline(new NodeId(probe.Id)), [.. connections.OrderBy(c => c.AgentName, StringComparer.Ordinal)]); } - private async ValueTask BuildFileAsync(ProbeRecord probe, NodeIdentity identity, CancellationToken ct) - { - ProbeAgentsResult agents = await _agentService.ProbeAgentsAsync(ct); - return new ProbeIdentityFile( - new ProbeId(probe.Id), - Convert.ToBase64String(identity.ExportPkcs8()), - [.. agents.Agents.Select(a => new BootstrapAgent(a.AgentId, a.Host, a.TcpPort, a.Spki))], - new ProbeLimits(MaxTestMbps: 100, MaxTotalPps: 20_000)); - } + private static ProbeIdentityFile BuildFile(ProbeRecord probe, NodeIdentity identity, string server) => + new(new ProbeId(probe.Id), Convert.ToBase64String(identity.ExportPkcs8()), server, new ProbeLimits(MaxTestMbps: 100, MaxTotalPps: 20_000)); [GeneratedRegex(@"^[\p{L}\p{M}\p{N}._-]{1,64}$")] private static partial Regex NamePattern(); diff --git a/backend/src/Monitor.Backend.Application/Nodes/NodeInfoService.cs b/backend/src/Monitor.Backend.Application/Nodes/NodeInfoService.cs index f3c3e23..9a9d3ea 100644 --- a/backend/src/Monitor.Backend.Application/Nodes/NodeInfoService.cs +++ b/backend/src/Monitor.Backend.Application/Nodes/NodeInfoService.cs @@ -111,14 +111,13 @@ public sealed class NodeInfoService } } - /// Probe 连上来或心跳报告它还连着：记下它的系统信息，并在后台查它的国家和运营商。出错只吞掉，不影响调用方。 - public void ObserveProbeInBackground(string probeId, string? remote, PlatformInfo? platform) => + /// Agent 报告某个 Probe 连着它：在后台用 Probe 的来源地址查国家和运营商（系统信息是 Probe 自己连上面板时报的）。出错只吞掉。 + public void ObserveProbeInBackground(string probeId, string? remote) => _ = Task.Run(async () => { try { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); - await SetPlatformAsync(probeId, NodeKinds.Probe, platform, cts.Token); await ObserveAddressAsync(probeId, NodeKinds.Probe, remote, cts.Token); } catch (Exception ex) when (ex is OperationCanceledException or InvalidOperationException or IOException) diff --git a/backend/src/Monitor.Backend.Application/Ports/IAgentGateway.cs b/backend/src/Monitor.Backend.Application/Ports/IAgentGateway.cs index 443a795..ee8627a 100644 --- a/backend/src/Monitor.Backend.Application/Ports/IAgentGateway.cs +++ b/backend/src/Monitor.Backend.Application/Ports/IAgentGateway.cs @@ -4,7 +4,7 @@ using Monitor.Protocol; namespace Monitor.Backend.Application; /// -/// Application 通往 Agent 连接的端口（由 Gateway 实现）：下发状态变化、发起按需测试、通知 Probe 列表变化。 +/// Application 通往 Agent 连接的端口（由 Gateway 实现）：下发状态变化、发起按需测试。 /// public interface IAgentGateway { @@ -17,9 +17,6 @@ public interface IAgentGateway void NotifyAllTaskSetsChanged(); - /// Agent 列表变了：通知所有在线 Agent 转发给 Probe。 - ValueTask NotifyProbeAgentsChangedAsync(ProbeAgentsResult agents, CancellationToken ct); - /// 失败时错误码为 probe_offline、test_busy、agent_offline 等。 ValueTask> RunTestAsync(NodeId agent, TestRunParams run, CancellationToken ct); diff --git a/backend/src/Monitor.Backend.Application/Tasks/TaskPlanner.cs b/backend/src/Monitor.Backend.Application/Tasks/TaskPlanner.cs index c3021fb..11ee07d 100644 --- a/backend/src/Monitor.Backend.Application/Tasks/TaskPlanner.cs +++ b/backend/src/Monitor.Backend.Application/Tasks/TaskPlanner.cs @@ -41,9 +41,8 @@ public sealed class TaskPlanner : IDisposable IReadOnlyList all = await _tasks.ListAsync(ct); IReadOnlyList probes = await _probes.ListAsync(ct); - long listVersion = long.TryParse(await _settings.GetAsync(ProbeAgentsVersionKey, ct), out long v) ? v : 0; - var document = new TaskSet { ProbeAgentsVersion = listVersion }; + var document = new TaskSet(); foreach (EffectiveTask task in EffectiveTasks.Compute(all, agent, filterByAgentKinds: true)) { document.Tasks.Add(new TaskSpec diff --git a/backend/src/Monitor.Backend.Gateway/FleetGateway.cs b/backend/src/Monitor.Backend.Gateway/FleetGateway.cs index 84847fd..4ba5927 100644 --- a/backend/src/Monitor.Backend.Gateway/FleetGateway.cs +++ b/backend/src/Monitor.Backend.Gateway/FleetGateway.cs @@ -24,7 +24,7 @@ public sealed class FleetGateway : IAgentGateway private IStateNotifier Notifier => _services.GetRequiredService(); - public IReadOnlyCollection OnlineAgents => Connections.OnlineNodes; + public IReadOnlyCollection OnlineAgents => [.. Connections.OnlineNodes.Where(n => n.Value.StartsWith(AgentIds.Prefix, StringComparison.Ordinal))]; public bool IsOnline(NodeId agent) => Connections.IsOnline(agent); @@ -32,15 +32,6 @@ public sealed class FleetGateway : IAgentGateway public void NotifyAllTaskSetsChanged() => Notifier.NotifyAllChanged(); - public async ValueTask NotifyProbeAgentsChangedAsync(ProbeAgentsResult agents, CancellationToken ct) - { - IFleetConnections connections = Connections; - foreach (NodeId node in connections.OnlineNodes) - { - await connections.NotifyAsync(node, MonitorMethods.ProbeAgentsChanged, agents, MonitorJsonContext.Default.ProbeAgentsResult, ct); - } - } - public async ValueTask> RunTestAsync(NodeId agent, TestRunParams run, CancellationToken ct) { Result result = await Connections.RequestAsync( diff --git a/backend/src/Monitor.Backend.Gateway/GatewayServiceCollectionExtensions.cs b/backend/src/Monitor.Backend.Gateway/GatewayServiceCollectionExtensions.cs index 450dde1..b8701aa 100644 --- a/backend/src/Monitor.Backend.Gateway/GatewayServiceCollectionExtensions.cs +++ b/backend/src/Monitor.Backend.Gateway/GatewayServiceCollectionExtensions.cs @@ -20,6 +20,7 @@ public static class GatewayServiceCollectionExtensions services.AddFleetEnrollment(); services.AddFleetServer(options => options.NodeIdPrefix = AgentIds.Prefix) .AddState() + .AddState() .AddInbox() .MapMethods(MapMonitorMethods); return services; @@ -39,11 +40,6 @@ public static class GatewayServiceCollectionExtensions .OnNotify( MonitorMethods.TestStatus, MonitorJsonContext.Default.TestStatusParams, - (context, parameters, ct) => services.GetRequiredService().OnStatusAsync(context.Node.Id, parameters, ct)) - .On( - MonitorMethods.ProbeAgents, - MonitorJsonContext.Default.ProbeAgentsParams, - MonitorJsonContext.Default.ProbeAgentsResult, - async (context, parameters, ct) => Result.Ok(await services.GetRequiredService().ProbeAgentsAsync(ct))); + (context, parameters, ct) => services.GetRequiredService().OnStatusAsync(context.Node.Id, parameters, ct)); } } diff --git a/backend/src/Monitor.Backend.Gateway/ProbeAgentsStateProvider.cs b/backend/src/Monitor.Backend.Gateway/ProbeAgentsStateProvider.cs new file mode 100644 index 0000000..e832faf --- /dev/null +++ b/backend/src/Monitor.Backend.Gateway/ProbeAgentsStateProvider.cs @@ -0,0 +1,33 @@ +using System.Text.Json; +using Fleet.Rpc; +using Fleet.Server; +using Monitor.Backend.Application; +using Monitor.Protocol; + +namespace Monitor.Backend.Gateway; + +/// 期望状态：为每个 Probe 下发要连接的 Agent 列表（地址、端口、证书指纹）。Agent 集合变化时由 NotifyAllChanged 一起重算推送。 +public sealed class ProbeAgentsStateProvider : IStateProvider +{ + private readonly AgentService _agents; + + public ProbeAgentsStateProvider(AgentService agents) + { + _agents = agents; + } + + public string Type => ProbeAgentList.StateType; + + public async ValueTask GetAsync(NodeRecord node, CancellationToken ct) + { + if (!node.Id.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal)) + { + return null; + } + + ProbeAgentList list = await _agents.ProbeAgentsAsync(ct); + return new StateDocument(list.Version, JsonSerializer.SerializeToElement(list, MonitorJsonContext.Default.ProbeAgentList)); + } + + public ValueTask OnAppliedAsync(NodeRecord node, long version, StateSyncResult result, CancellationToken ct) => ValueTask.CompletedTask; +} diff --git a/backend/src/Monitor.Backend.Gateway/TaskSetStateProvider.cs b/backend/src/Monitor.Backend.Gateway/TaskSetStateProvider.cs index f72669f..6f7bfcc 100644 --- a/backend/src/Monitor.Backend.Gateway/TaskSetStateProvider.cs +++ b/backend/src/Monitor.Backend.Gateway/TaskSetStateProvider.cs @@ -26,6 +26,11 @@ public sealed class TaskSetStateProvider : IStateProvider public async ValueTask GetAsync(NodeRecord node, CancellationToken ct) { + if (node.Id.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal)) + { + return null; // Probe 不运行任务集 + } + DesiredTaskSet? desired = await _planner.GetDesiredAsync(node.Id, ct); return desired is null ? null : new StateDocument(desired.Version, JsonSerializer.SerializeToElement(desired.Document, MonitorJsonContext.Default.TaskSet)); } diff --git a/backend/src/Monitor.Backend.Storage.Sqlite/SqliteAgentStore.cs b/backend/src/Monitor.Backend.Storage.Sqlite/SqliteAgentStore.cs index 13454bd..9685b25 100644 --- a/backend/src/Monitor.Backend.Storage.Sqlite/SqliteAgentStore.cs +++ b/backend/src/Monitor.Backend.Storage.Sqlite/SqliteAgentStore.cs @@ -166,8 +166,17 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore return Result.Ok(ToNode(GetInternal(c, node.Id)!, node.MetaJson)); }, ct); + /// Agent 在 agents 表里；Probe 的身份在 probes 表里（吊销之后按「禁用」处理，连不上）。 async ValueTask IFleetStore.GetNodeAsync(NodeId id, CancellationToken ct) => - await GetAsync(id, ct) is { } agent ? ToNode(agent, null) : null; + await GetAsync(id, ct) is { } agent + ? ToNode(agent, null) + : await _db.ReadAsync( + c => c.QueryOne( + "SELECT id, name, pubkey, created_at, revoked_at FROM probes WHERE id = @id", + r => new NodeRecord( + new NodeId(r.GetString(0)), r.GetString(1), r.IsDBNull(4) ? NodeStatus.Active : NodeStatus.Disabled, r.Bytes(2), [], [], null, null, r.GetInt64(3), null), + ("@id", id.Value)), + ct); public ValueTask FindEstablishedNodeByNameAsync(string name, CancellationToken ct) => _db.ReadAsync( diff --git a/backend/tests/Monitor.Backend.Tests/EnrollmentAndTaskSetTests.cs b/backend/tests/Monitor.Backend.Tests/EnrollmentAndTaskSetTests.cs index e383a97..6bb7d7e 100644 --- a/backend/tests/Monitor.Backend.Tests/EnrollmentAndTaskSetTests.cs +++ b/backend/tests/Monitor.Backend.Tests/EnrollmentAndTaskSetTests.cs @@ -338,7 +338,7 @@ public sealed class EnrollmentAndTaskSetTests JsonNode identity = JsonNode.Parse(await created.Content.ReadAsStringAsync())!; string id = identity["probe_id"]!.GetValue(); using var key1 = Fleet.Identity.NodeIdentity.FromPkcs8(Convert.FromBase64String(identity["private_key"]!.GetValue())); - Assert.Contains(identity["bootstrap"]!.AsArray(), b => b!["agent_id"]!.GetValue() == agent.NodeId.Value); + Assert.True(Uri.TryCreate(identity["server"]!.GetValue(), UriKind.Absolute, out _)); // Probe 直接连面板：probe.json 里是面板的地址 await agent.WaitForTaskSetAsync(s => s.Peers.Any(p => p.ProbeId.Value == id && p.PublicKey == Convert.ToBase64String(key1.PublicKeySpki))); // The panel lists it; two probes can coexist and a duplicate name is refused @@ -374,34 +374,6 @@ public sealed class EnrollmentAndTaskSetTests Assert.Equal(HttpStatusCode.NotFound, gone.StatusCode); } - [Fact] - public async Task ProbeAgentList_IsServedToAgents_AndPushedWhenTheSetOfAgentsChanges() - { - await using AgentSimulator first = await _backend.ConnectAgentAsync(); - ProbeAgentsResult list = await first.RequestProbeAgentsAsync(); - Assert.Contains(list.Agents, a => a.AgentId == first.NodeId && a.Host == "203.0.113.10" && a.TcpPort == 7443); - Assert.All(list.Agents, a => Assert.Equal(32, Convert.FromBase64String(a.Spki).Length)); - - await using AgentSimulator second = await _backend.ConnectAgentAsync(); - ProbeAgentsResult pushed = await WaitForAsync(first, l => l.Agents.Any(a => a.AgentId == second.NodeId)); - Assert.True(pushed.Version > list.Version); - - using HttpResponseMessage disable = await _backend.Admin.PostAsync($"/api/admin/agents/{second.NodeId.Value}/disable", null); - Assert.Equal(HttpStatusCode.OK, disable.StatusCode); - await WaitForAsync(first, l => l.Agents.All(a => a.AgentId != second.NodeId)); - } - - private static async Task WaitForAsync(AgentSimulator agent, Func predicate) - { - ProbeAgentsResult? found = null; - await BackendFixture.EventuallyAsync(() => - { - found = agent.ProbeAgentLists.ToArray().LastOrDefault(predicate); - return Task.FromResult(found is not null); - }); - return found!; - } - private Task PostEnrollAsync(JsonNode body) => _backend.Anonymous.PostAsync("/api/agent/v1/enroll", BackendFixture.Json(body)); } diff --git a/backend/tests/Monitor.Backend.Tests/OperationsAndSecurityTests.cs b/backend/tests/Monitor.Backend.Tests/OperationsAndSecurityTests.cs index a5526ae..4485688 100644 --- a/backend/tests/Monitor.Backend.Tests/OperationsAndSecurityTests.cs +++ b/backend/tests/Monitor.Backend.Tests/OperationsAndSecurityTests.cs @@ -175,9 +175,8 @@ public sealed class OperationsAndSecurityTests Assert.Equal("fleet_unsupported", reply.Error!.Data!.Code); } - // 连接仍然可用 - ProbeAgentsResult list = await agent.RequestProbeAgentsAsync(); - Assert.Contains(list.Agents, a => a.AgentId == agent.NodeId); + // 连接仍然可用：路由器照常回复 + Assert.Equal("fleet_unsupported", (await agent.RequestRawAsync("fleet.exec")).Error!.Data!.Code); } [Fact] @@ -195,9 +194,6 @@ public sealed class OperationsAndSecurityTests public async Task PendingNodes_CannotCallProductMethods() { await using AgentSimulator agent = await _backend.ConnectAgentAsync(autoApprove: false); - RpcMessage reply = await agent.RequestRawAsync(MonitorMethods.ProbeAgents, JsonSerializer.SerializeToElement(new ProbeAgentsParams(), MonitorJsonContext.Default.ProbeAgentsParams)); - Assert.Equal("fleet_node_pending", reply.Error!.Data!.Code); - // 状态与事件也不会被接受：监控数据里没有它 await agent.SendEventAsync(AgentEventKinds.ProbeConnected, _backend.DefaultProbeId); await Task.Delay(200); diff --git a/backend/tests/Monitor.Backend.Tests/ProbeEndToEndTests.cs b/backend/tests/Monitor.Backend.Tests/ProbeEndToEndTests.cs index 2bc74b1..21562d1 100644 --- a/backend/tests/Monitor.Backend.Tests/ProbeEndToEndTests.cs +++ b/backend/tests/Monitor.Backend.Tests/ProbeEndToEndTests.cs @@ -19,13 +19,13 @@ public sealed class ProbeEndToEndTests (await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/raw/chunks?agent_id={agent.Id}"))["items"]!.AsArray().Count(c => c!["kind"]!.GetValue() == kind); [Fact] - public async Task Probe_ConnectsThroughTheBootstrapList_RunsTheLineTasks_AndTheBackendStoresMergedChunks() + public async Task Probe_GetsTheAgentListFromThePanel_RunsTheLineTasks_AndTheBackendStoresMergedChunks() { await using AgentHarness agent = await AgentHarness.StartAsync(_backend); await agent.OverrideLineTasksAsync(rateHz: 30); await using ProbeHarness probe = await ProbeHarness.StartAsync(_backend); - await BackendFixture.EventuallyAsync(() => Task.FromResult(probe.App.IsConnected(agent.Id)), because: "the probe should connect to the bootstrap agent"); + await BackendFixture.EventuallyAsync(() => Task.FromResult(probe.App.IsConnected(agent.Id)), because: "the panel sends the agent list to the probe, which then connects"); await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["probe_connected"]!.GetValue()); await BackendFixture.EventuallyAsync(() => Task.FromResult(probe.App.Connection(agent.Id)!.RoleCount == 2), because: "udp_echo and tcp_echo"); @@ -38,6 +38,26 @@ public sealed class ProbeEndToEndTests Assert.InRange(probe.App.Connection(agent.Id)!.ClockOffsetUs, -20_000, 20_000); } + [Fact] + public async Task Probe_IntroducesItselfToThePanel_AndARevokedProbeIsLockedOut() + { + await _backend.SiteMustBePublicAsync(); + await using ProbeHarness probe = await ProbeHarness.StartAsync(_backend); + string path = "/api/admin/probes/" + probe.ProbeId; + + // 直连面板：在线，hello 里的系统、架构和版本都被记下来了 + await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, path))["online"]!.GetValue(), because: "the probe links to the panel"); + await BackendFixture.EventuallyAsync( + async () => (await _backend.GetJsonAsync(_backend.Anonymous, "/api/public/nodes"))["items"]!.AsArray().Any(n => n!["id"]!.GetValue() == probe.ProbeId && n["os"] is not null && n["arch"] is not null), + because: "the platform comes from the probe's own hello"); + + // 吊销：面板立刻断开它，之后连不上 + using HttpResponseMessage revoked = await _backend.Admin.PostAsync(path + "/revoke", null); + revoked.EnsureSuccessStatusCode(); + await BackendFixture.EventuallyAsync(async () => !(await _backend.GetJsonAsync(_backend.Admin, path))["online"]!.GetValue(), because: "a revoked probe is disconnected and cannot come back"); + Assert.False(probe.App.Link.IsConnected); + } + [Fact] public async Task Probe_FollowsTheAgentListThePanelKnows_AndSavesItForTheNextStart() { diff --git a/backend/tests/Monitor.TestSupport/AgentSimulator.cs b/backend/tests/Monitor.TestSupport/AgentSimulator.cs index 9510343..7e26f4e 100644 --- a/backend/tests/Monitor.TestSupport/AgentSimulator.cs +++ b/backend/tests/Monitor.TestSupport/AgentSimulator.cs @@ -46,8 +46,6 @@ public sealed class AgentSimulator : IAsyncDisposable public ConcurrentQueue TaskSets { get; } = new(); - public ConcurrentQueue ProbeAgentLists { get; } = new(); - public ConcurrentQueue TestRuns { get; } = new(); public ConcurrentQueue TestCancels { get; } = new(); @@ -190,12 +188,6 @@ public sealed class AgentSimulator : IAsyncDisposable new TestStatusParams { TestId = test, Status = status, Phase = phase, ErrorCode = errorCode }, MonitorJsonContext.Default.TestStatusParams)); - public async Task RequestProbeAgentsAsync() - { - RpcMessage reply = await RequestAsync(MonitorMethods.ProbeAgents, JsonSerializer.SerializeToElement(new ProbeAgentsParams(), MonitorJsonContext.Default.ProbeAgentsParams)); - return reply.Result!.Value.Deserialize(MonitorJsonContext.Default.ProbeAgentsResult)!; - } - public async Task RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters); public async Task WaitForTaskSetAsync(Func predicate, TimeSpan? timeout = null) @@ -416,9 +408,6 @@ public sealed class AgentSimulator : IAsyncDisposable TestCancels.Enqueue(cancel.TestId); await SendAsync(RpcMessage.Success(message.Id!.Value, new TestRunResult { Accepted = true }, MonitorJsonContext.Default.TestRunResult)); break; - case MonitorMethods.ProbeAgentsChanged: - ProbeAgentLists.Enqueue(message.Params!.Value.Deserialize(MonitorJsonContext.Default.ProbeAgentsResult)!); - break; } } } diff --git a/backend/tests/Monitor.TestSupport/ProbeSimulator.cs b/backend/tests/Monitor.TestSupport/ProbeSimulator.cs index 8e3c5df..d2734a1 100644 --- a/backend/tests/Monitor.TestSupport/ProbeSimulator.cs +++ b/backend/tests/Monitor.TestSupport/ProbeSimulator.cs @@ -61,8 +61,6 @@ public sealed class ProbeSimulator : IAsyncDisposable public ConcurrentQueue Aborts { get; } = new(); - public ConcurrentQueue AgentLists { get; } = new(); - public bool IsConnected => _rpc is not null && !_stop.IsCancellationRequested; public long NowUs() => (TimeProvider.System.GetTimestamp() - _origin) * 1_000_000 / TimeProvider.System.TimestampFrequency; @@ -347,14 +345,6 @@ public sealed class ProbeSimulator : IAsyncDisposable probe.Aborts.Enqueue(abort.TestId); return ValueTask.CompletedTask; }) - .OnNotify( - MonitorMethods.ProbeAgentsChanged, - MonitorJsonContext.Default.ProbeAgentsResult, - (probe, list, _) => - { - probe.AgentLists.Enqueue(list); - return ValueTask.CompletedTask; - }) .OnNotify( FleetMethods.OutboxAck, FleetRpcContext.Default.OutboxAckParams, diff --git a/docs/api.md b/docs/api.md index 1e847ce..f8ffdc8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -91,7 +91,7 @@ `GET /api/public/about`：`{show_build, backend}`。`show_build` 关着时 `backend` 是 `null`；打开后 `backend` 是 `{version, tag, built_at}`（后端自己的版本、tag、构建时间，不含提交号）。 -- 系统和架构是节点自报的：Agent 在 `fleet.hello` 里，Probe 随列表同步（`monitor.probe.agents` 的 `platform`）告诉 Agent，Agent 再放进心跳的 `probes[]`。只有粗粒度的名字，没有主机名、内核版本、CPU 型号。 +- 系统和架构是节点自报的：Agent 和 Probe 都在各自连面板时的 `fleet.hello` 里自报。只有粗粒度的名字，没有主机名、内核版本、CPU 型号。 - 国家和运营商由 **Backend** 用节点的来源地址向 IP 归属服务查一次（`Monitor__GeoLookupUrl`，缺省 `https://ip.007995.xyz`，留空就不查）：Probe 用 Agent 看到的来源 IP，Agent 用它的公开地址。**Probe 和 Agent 不访问任何外部服务。** 只查公网地址（回环、内网、CGNAT、链路本地一律不发出去）；同一个地址一周内只查一次，失败 10 分钟后再试；查到的只留国家代码和运营商，地址只存在 Backend 内部的 `node_meta.geo_host`，不对外。 ### 2.4 测试结果与目录 diff --git a/docs/deployment.md b/docs/deployment.md index 0a621fd..a834159 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -157,6 +157,7 @@ OpenWrt 默认不带 OpenSSL：`opkg install libopenssl3`（较新的快照和 2 ```text 浏览器 / 本机前端 ──► https://<公网域名> ──隧道（Cloudflare Tunnel 之类）──► 家宽机器: Backend (:29181) + 家宽机器: Probe ──WSS 到 https://<公网域名>──► Backend（身份、Agent 列表） 家宽机器: Probe ──TCP/UDP 7443、TCP 7444──► VPS: Agent VPS: Agent ──WSS 到 https://<公网域名>──► Backend ``` @@ -165,7 +166,7 @@ OpenWrt 默认不带 OpenSSL：`opkg install libopenssl3`（较新的快照和 2 - **Backend 通过隧道暴露**：隧道把一个公网域名转到家宽机器的 `29181`；Backend 监听 `http://0.0.0.0:29181`（只写一个监听地址，同时写 `0.0.0.0` 和 `[::]` 会报地址已被占用），家宽路由器的防火墙不放行这个端口，公网只能经隧道进来。Agent 用 `fleet-enroll --server https://<公网域名>` 注册，之后用 WSS 连回 Backend。 - **经隧道注册之后，检查 Agent 的 `public_host`**：Backend 看到的来源地址是隧道的出口，不是 VPS 的地址，要改成 VPS 的真实地址，Probe 才连得上。 -- **先注册 Agent，再创建 Probe**：`probe.json` 的 `bootstrap` 里才有这个 Agent；Probe 启动后会通过 Agent 同步列表。 +- **Probe 直连面板**：`probe.json` 里的 `server` 就是面板地址（取自 `Monitor__PublicUrl`，没配就用创建它时的请求地址），Probe 用它连面板，Agent 列表由面板下发；先注册还是先创建 Probe 都可以。 - **小 VPS 把 `limits.max_total_mbps` 调低**（试跑时用 100）；Agent 常驻约 23 MB，VPS 上原有的服务不受影响，结束后删除 `/opt/…`、停掉进程即可，没有留下任何东西。 - 家宽的 IPv6 前缀会在路由器重新拨号（比如停电）后变化，SSH 地址要跟着换；面板走隧道的域名，不受影响。 - 本地电脑没有直连 VPS 的路径时，用家宽机器当跳板（`SSH_ASKPASS` + `SSH_ASKPASS_REQUIRE=force`，密码放环境变量，不写进文件）。 diff --git a/docs/enrollment.md b/docs/enrollment.md index eeb4b72..dbf043a 100644 --- a/docs/enrollment.md +++ b/docs/enrollment.md @@ -7,7 +7,7 @@ | 主体 | 凭据 | 生成方 | Backend 保存 | | --- | --- | --- | --- | | Agent | Fleet 节点身份（ECDSA P-256）+ 面向 Probe 的 TLS 证书 | `fleet-enroll` | 公钥、SPKI 指纹 | -| Probe | Fleet 节点身份 | 面板生成 `probe.json`（可以有多个 Probe，每个独立吊销） | 公钥 | +| Probe | Fleet 节点身份（同时用于连面板和连 Agent） | 面板生成 `probe.json`（可以有多个 Probe，每个独立吊销） | 公钥 | | 面板用户 | 账号 + 密码（PBKDF2-SHA256） | 面板 | 密码哈希 | | API 令牌 | 随机高熵令牌，作用域为 `read` 或 `admin` | 面板 | SHA-256 | | 发布签名 | 离线的 ECDSA 发布密钥 | CI 或离线环境 | 只在节点的 `fleet-updater` 中配置公钥，**面板不保存** | @@ -30,7 +30,7 @@ install.sh（只做安装，不包含任何远程控制功能） 面板收到 OnEnrolled（Monitor 挂钩） ├─ public_host 未提供时，取注册请求的来源 IP ├─ 按“线路任务模板”为该 Agent 计算有效任务集（TaskSet） - └─ 通知 Probe（通过 monitor.probe.agents_changed）→ Probe 自动连接新 Agent + └─ 面板把新的 Agent 列表作为期望状态推给所有 Probe → Probe 自动连接新 Agent 从安装到开始出数据，全程不需要手工配置 ``` @@ -38,21 +38,20 @@ install.sh（只做安装，不包含任何远程控制功能） ## 3. Probe 身份与 Agent 列表同步 -Probe 不认识 Backend，它需要的 Agent 列表经由任意一个已连接的 Agent 获取。 +Probe 直接连面板（Fleet 长连接，身份是下面生成的密钥），Agent 列表由面板作为期望状态下发；测量数据仍然只走 Probe → Agent。 ```text 面板：POST /api/admin/probes {name} - └─ 生成 Fleet 节点身份，返回 probe.json（只能下载一次；轮换密钥 /rotate 会使旧密钥失效，吊销 /revoke 会让所有 Agent 断开它） - { probe_id, private_key, bootstrap: [ {agent_id, host, tcp_port, spki}, ... ], + └─ 生成 Fleet 节点身份，返回 probe.json（只能下载一次；轮换密钥 /rotate 会使旧密钥失效，吊销 /revoke 会让面板和所有 Agent 断开它） + { probe_id, private_key, server: "https://panel...", limits: { max_test_mbps, max_total_pps } } ← 本地硬上限，面板无法修改；在路由器上建议调低 Probe 启动 - ├─ 依次尝试 bootstrap 中的 Agent，与其中一个建立 Fleet 点对点会话（peer 用途签名） - ├─ monitor.probe.agents ──> Agent ──> Backend - │ <── { version, agents: [ {agent_id, host, tcp_port, udp_ports, spki}... ] } - ├─ 调和：新增的 Agent → 建立会话；被移除或禁用的 → 断开 - └─ 之后列表的任何变化，都通过已连接的 Agent 推送 monitor.probe.agents_changed； - Probe 用 Fleet.Hosting.ConfigStore 把最新列表写回 bootstrap（这是配置，不是测量数据） + ├─ 连接 server：Fleet 长连接（用 probe_id 和私钥认证），fleet.hello 带上系统、架构和构建信息 + ├─ 面板推送期望状态 probe_agents：{ version, agents: [ {agent_id, host, tcp_port, udp_ports, spki}... ] } + ├─ 调和：新增的 Agent → 建立点对点会话（peer 用途签名）；被移除或禁用的 → 断开 + └─ 之后列表的任何变化（Agent 注册、禁用、地址或证书变化），面板都重新推送同一种期望状态； + Probe 用 Fleet.Hosting.ConfigStore 把最新列表写回 agents.json（这是配置，不是测量数据；面板暂时连不上时，Probe 靠它继续测量） ``` - 每个 Agent 通过自己的期望状态（`TaskSet.peers`）获得 Probe 的公钥，用来验证点对点会话的签名。 @@ -70,7 +69,7 @@ Probe 启动 "allowed_target_cidrs": [], "max_concurrent_tests": 1 }, - "fleet_policy": { "allow_methods": ["monitor.test.run", "monitor.test.cancel", "monitor.probe.agents_changed"] } + "fleet_policy": { "allow_methods": ["monitor.test.run", "monitor.test.cancel"] } } ``` diff --git a/docs/project-layout.md b/docs/project-layout.md index b3a7c83..01485cf 100644 --- a/docs/project-layout.md +++ b/docs/project-layout.md @@ -155,13 +155,12 @@ Agent Program.cs peers = FleetLink.PeerListener(cfg.Listen, identity, ProbeRpcContext.Default) Probe 会话 host = new ModuleHost(log) ├─ link、peers （Fleet） - ├─ ProbeDirectoryRelay 转发 monitor.probe.agents / agents_changed ├─ UdpMux / TcpAcceptor UDP/TCP/TLS 数据端口（Measurement 共享），按 (session, stream) 分发给监听方积木；AgentAdmission 提供重放、限速 ├─ ChunkModule 合并 Probe 与本端的列 → 切块 → link.Outbox └─ TestModule 按需测试的阶段编排 Probe Program.cs - ├─ AgentDirectoryModule 维护 Agent 列表（bootstrap → monitor.probe.agents → agents_changed） + ├─ FleetLink（到面板） 身份、hello、期望状态 probe_agents（Agent 列表） ├─ PeerPool（Fleet） 每个 Agent 一个 FleetLink.ToPeer：认证、重连 ├─ TaskReconciler 每个会话各自的期望状态（与 Agent 相同的代码） ├─ ColumnShipper 本端的列通过会话的 Outbox 发送（二进制通道 17） diff --git a/docs/protocol.md b/docs/protocol.md index ba981f2..a4b4486 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -11,7 +11,7 @@ Monitor 的两条控制链路都建立在 Fleet 之上： | Monitor 需要的能力 | 使用的 Fleet 机制 | Monitor 提供的内容 | | --- | --- | --- | -| 下发任务 | 期望状态 `fleet.state.sync` | 文档类型 `TaskSet`，包括 `tasks[]`、`peers`（Probe 公钥），以及 `probe_agents` 的版本 | +| 下发任务 | 期望状态 `fleet.state.sync` | 文档类型 `TaskSet`，包括 `tasks[]`、`peers`（Probe 公钥），（Probe 的 Agent 列表是另一种期望状态，见下） | | 上报原始块 | 发件箱，二进制通道 **16** | `RawChunk`（自描述，见 [measurement.md](measurement.md) 第 4 节），`msg_id = chunk_id` | | Probe 交付列块 | 点对点会话的发件箱，二进制通道 **17** | 列块：`u32 stream_id \| u32 first_seq \| u32 count \| 列数据(Brotli)` | | 节点自身状态 | `fleet.heartbeat` 的附加字段 | `sessions`、`pps_in/out`、`dropped_invalid`、`dropped_rate_limited`、`queue_bytes` | @@ -30,8 +30,6 @@ Monitor 的两条控制链路都建立在 Fleet 之上： | `monitor.event` | A→B | 通知 | `kind`（`probe_connected`（`detail` 是 `probe_id`，`probe` 里带版本、远端地址和连接时间）、`probe_disconnected`、`endpoint_changed`、`session_rejected`、`clock_step`、`gap`、`cpu_saturated`（节点进程 CPU 饱和，可能影响测量，由 Probe 或 Agent 上报））、`at_utc`、`detail` | 心跳的附加字段里有 `probes[]`：Agent 当前连着的每个 Probe（`id`、`version`、`remote`、`since_utc_ms`），每次都是完整快照，Backend 靠它纠正错过的连接事件。点对点握手的认证消息里可以带 Probe 的 `version`（只用于展示）。块头里的 `probe_id` 以 Agent 认证过的会话为准，Probe 自报的值会被覆盖。 -| `monitor.probe.agents` | A→B | 请求 | 代理 Probe 获取 Agent 列表 → `version`、`agents[]` | -| `monitor.probe.agents_changed` | B→A | 通知 | Agent 列表发生变化，由 Agent 转发给 Probe | ### Probe ↔ Agent（点对点会话） @@ -42,8 +40,6 @@ Monitor 的两条控制链路都建立在 Fleet 之上： | `monitor.test.phase` | A→P | 请求 | 测试的一个阶段：`test_id`、`phase`、`kind`、`params`、`stream_id`、`duration_s`、`port` → `accepted`；Probe 在回复之前准备好自己一侧的角色（单向数据流：先发几个探活包让 Agent 学到源地址） | | `monitor.test.phase_end` | A→P | 请求 | `sent_count`（Agent 发出的包数）→ `fragments`：Probe 收尾，把列块交给 Agent，回复交出的个数；Agent 等全部列块合并上传之后才进入下一阶段 | | `monitor.test.abort` | A→P | 通知 | `test_id` | -| `monitor.probe.agents` | P→A | 请求 | `have_version` → `version`、`agents[]`：Agent 代 Probe 向 Backend 查询（链路断开时 `fleet_node_offline`） | -| `monitor.probe.agents_changed` | A→P | 通知 | `version`、`agents[]`（Agent 从 Backend 转发） | | `fleet.outbox.ack` | 双向 | 通知 | Probe 的列块（通道 17）是至少一次投递，Agent 每收到一个就确认，重复到达的只处理一次 | 会话密钥不需要单独的方法：点对点握手（TLS 固定证书 → `peer.challenge` → Probe 用身份私钥签 `(peer_id, nonce, client_pub, server_pub)` → 两端各自 ECDH + HKDF）直接得到 `session_id` 和 `session_key`，`peer.auth` 的应答只回显 `session_id` 用来互相确认。 diff --git a/docs/roadmap.md b/docs/roadmap.md index 5aaf7d2..2ade908 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -40,7 +40,7 @@ ## M3 Backend：注册、RPC、存储 -- 在 Fleet 之上接入：`IFleetStore` 实现、`TaskSet` 状态提供者（有效任务集的计算与下发）、`RawChunk` 收件箱、`monitor.*` 方法、`monitor.probe.agents` 同步。 +- 在 Fleet 之上接入：`IFleetStore` 实现、`TaskSet` 状态提供者（有效任务集的计算与下发）、`RawChunk` 收件箱、`monitor.*` 方法、`probe_agents` 期望状态（Probe 直连面板）同步。 - 段文件与索引、发件箱 ACK、去重、缺口；管理员账号、Cookie 会话、API 令牌；`/api/admin` 中的节点、令牌、Probe 身份、任务管理。 **验收**：用一行安装命令部署一台新 VPS，自动注册 → 自动生成线路任务 → Probe 自动连接 → 1 分钟内在 `/api/admin` 中看到它的原始块。整个过程不需要手工配置。 @@ -95,7 +95,7 @@ - **M5 的一部分**：保留清理。 - **F1 的节点部分**：`Fleet.Link`（链路、发件箱、点对点监听与连接）、`Fleet.Hosting`、`Fleet.Enrollment.Client` 与 `fleet-enroll`。 - **Agent（M1 的一部分）**：`Monitor.Agent`（NativeAOT 可发布）与 `Monitor.Measurement`（列块合并、预算守卫）。UDP/TCP 数据面（认证、重放窗口、限速、端点学习）、线路任务调和、目标任务（`icmp_echo`、`connect`）、按需测试编排（`udp_train` 由 Agent 发包）。测试里的假 Probe（`ProbeSimulator`）通过真实的点对点握手连接真实的 Agent 与后端，端到端覆盖。详见 `backend/agent/README.md`。 -- **Probe（M0/M2 的一部分）**：`Monitor.Probe`（NativeAOT，5.6 MB，常驻约 22 MB）——Agent 列表（bootstrap → 由 Agent 代为查询 → 推送，写回 `agents.json`）、每个 Agent 一个自动重连的会话、`udp_echo`/`tcp_echo`/`connect`/`hold` 角色、`udp_train` 接收端（按序号范围切块）、时钟偏移估计、有界发件箱。真实 Probe + 真实 Agent + 真实后端的端到端测试，以及用发布的 AOT 二进制做过的冒烟（内存稳定、CPU < 1%）。详见 `backend/probe/README.md`。 +- **Probe（M0/M2 的一部分）**：`Monitor.Probe`（NativeAOT，5.6 MB，常驻约 22 MB）——Agent 列表（由面板经 Fleet 长连接直接下发，写回 `agents.json`）、每个 Agent 一个自动重连的会话、`udp_echo`/`tcp_echo`/`connect`/`hold` 角色、`udp_train` 接收端（按序号范围切块）、时钟偏移估计、有界发件箱。真实 Probe + 真实 Agent + 真实后端的端到端测试，以及用发布的 AOT 二进制做过的冒烟（内存稳定、CPU < 1%）。详见 `backend/probe/README.md`。 - **采集积木统一（重构）**：Probe 和 Agent 的数据面合并成 `Monitor.Measurement` 里的一份实现（每种采集方法一个文件、拨号方和监听方一对角色，`UdpMux`/`Tcp`/`RowStore` 共用）；合并改成由 Kind 的流声明（`Driver`、`Key`）驱动；测试方案移回 Backend；架构测试禁止采集层出现业务概念。 - **TCP 内核采样与长尾**：`tcp_echo` 两端各自采样套接字的 `TCP_INFO`（Linux，`union` 流拼接）；`tcp.rtt` 增加 `p999`、`max`、长尾比 `tail`，新增 `tcp.slow`（块内基线的慢回显占比）；序列接口增加 `anchor=latest`。 - **打包与部署（M6 的一部分）**：`scripts/publish.sh`（linux-musl-x64、linux-x64 在 NativeAOT SDK 容器里构建）、`scripts/package.sh`、`packaging/`（`install.sh`、systemd 单元、OpenWrt procd 脚本）、节点启动时的 OpenSSL 自检。在 Alpine、Debian、OpenWrt 官方 rootfs、带 systemd 的 Debian 容器里验证过。详见 `docs/deployment.md`。 diff --git a/web/admin/src/pages/Probes.tsx b/web/admin/src/pages/Probes.tsx index 6308d41..852b229 100644 --- a/web/admin/src/pages/Probes.tsx +++ b/web/admin/src/pages/Probes.tsx @@ -30,7 +30,7 @@ function IdentityDialog({ identity, title, onClose }: { identity: ProbeIdentityD - + 
{`monitor-probe --config /etc/monitor/probe.json`}
@@ -126,7 +126,14 @@ export function Probes() {
{p.name} - {p.revoked ? 已吊销 : 0 ? 'ok' : ''}>{p.connections.length > 0 ? `连着 ${p.connections.length} 个 Agent` : '没有连接'}} + {p.revoked ? ( + 已吊销 + ) : ( + <> + {p.online ? '已连面板' : '未连面板'} + 0 ? 'ok' : ''}>{p.connections.length > 0 ? `连着 ${p.connections.length} 个 Agent` : '没有连接'} + + )} {p.id}