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}