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.
This commit is contained in:
justme committed
1 parent
604cbe3123
commit
48072737c9
42 files changed
+266
-354
No files matched your search
@@ -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` 含私钥，请限制文件访问权限，不要提交到版本控制。
|
||||
|
||||
## 构建
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
/// <summary>
|
||||
/// 与一个 Agent 的点对点会话：认证握手 → 同步时钟 → 取 Agent 列表 → 按 Agent 下发的 <c>line_task_set</c> 运行角色 →
|
||||
/// 与一个 Agent 的点对点会话：认证握手 → 同步时钟 → 按 Agent 下发的 <c>line_task_set</c> 运行角色 →
|
||||
/// 每 10 秒把已完成的行切成列块放进发件箱 → 响应按需测试的阶段。会话断开时所有角色停止（凭据属于会话），按退避重连。
|
||||
/// 会话内的所有状态（套接字、发件箱、角色）随会话一起创建和释放。
|
||||
/// </summary>
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>本机的系统和架构：随列表同步一起告诉 Agent，只用于在面板上展示。</summary>
|
||||
private static readonly PlatformInfo s_platform = PlatformDetector.Detect();
|
||||
|
||||
private async Task RefreshDirectoryAsync(RpcSession<ConnectionSession> rpc, CancellationToken ct)
|
||||
{
|
||||
Result<ProbeAgentsResult> 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));
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>切块（每 10 秒）和时钟同步（每 30 秒）。</summary>
|
||||
private async Task MaintainAsync(RpcSession<ConnectionSession> 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,
|
||||
|
||||
@@ -6,8 +6,7 @@ using Monitor.Protocol;
|
||||
namespace Monitor.Probe;
|
||||
|
||||
/// <summary>
|
||||
/// Probe 要连接的 Agent 列表。Probe 不认识 Backend：列表先取自 <c>probe.json</c> 的 bootstrap（或上一次保存的 <c>agents.json</c>），
|
||||
/// 连上任意一个 Agent 之后，由它代为向 Backend 查询（<c>monitor.probe.agents</c>），之后的变化由 Agent 推送（<c>agents_changed</c>）。
|
||||
/// Probe 要连接的 Agent 列表：由面板作为期望状态（<c>probe_agents</c>）直接下发；面板连不上时用上一次保存的 <c>agents.json</c>。
|
||||
/// 列表变化时用 <see cref="ConfigStore"/> 写回 <c>agents.json</c>（这是配置，不是测量数据；只在变化时写，保护路由器的闪存）。
|
||||
/// </summary>
|
||||
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<BootstrapAgent> _agents;
|
||||
private List<KnownAgent> _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<BootstrapAgent> Agents
|
||||
public IReadOnlyList<KnownAgent> Agents
|
||||
{
|
||||
get
|
||||
{
|
||||
@@ -69,10 +64,10 @@ public sealed class AgentDirectory
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>应用 Agent 转来的列表。版本不比当前新的忽略（乱序到达的旧推送）。</summary>
|
||||
public void Apply(ProbeAgentsResult result)
|
||||
/// <summary>应用面板下发的列表。版本不比当前新的忽略（乱序到达的旧推送）。</summary>
|
||||
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<BootstrapAgent> a, List<BootstrapAgent> b) =>
|
||||
private static bool Same(List<KnownAgent> a, List<KnownAgent> 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;
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Probe 的组装点：Agent 列表 → 每个 Agent 一个连接（认证、重连、角色）。没有全局状态，测试可以在同一进程里启动多个。
|
||||
/// 列表变化时只调和连接：新增的建立，移除的断开，地址或证书指纹变化的重建，其余不受影响。
|
||||
/// Probe 的组装点：到面板的长连接（身份、Agent 列表、状态）+ 每个 Agent 一个测量连接（认证、重连、角色）。没有全局状态，测试可以在同一进程里启动多个。
|
||||
/// Agent 列表变化时只调和连接：新增的建立，移除的断开，地址或证书指纹变化的重建，其余不受影响。
|
||||
/// </summary>
|
||||
public sealed class ProbeApp : IDisposable
|
||||
{
|
||||
@@ -17,16 +21,39 @@ public sealed class ProbeApp : IDisposable
|
||||
private readonly Dictionary<string, Running> _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>(
|
||||
ProbeAgentList.StateType,
|
||||
MonitorJsonContext.Default.ProbeAgentList,
|
||||
(version, doc, _) =>
|
||||
{
|
||||
directory.Apply(doc);
|
||||
return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version });
|
||||
});
|
||||
_link = new FleetLink(options, new RpcRouter<LinkContext>(), [list], new Outbox(1 << 20), time, log);
|
||||
}
|
||||
|
||||
/// <summary>到面板的连接（测试和状态查看用）。</summary>
|
||||
public FleetLink Link => _link;
|
||||
|
||||
public AgentDirectory Directory => _env.Directory;
|
||||
|
||||
/// <summary>某个 Agent 的会话是否已经建立（测试和状态查看用）。</summary>
|
||||
@@ -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<BootstrapAgent> agents, CancellationToken ct)
|
||||
private async Task ReconcileAsync(IReadOnlyList<KnownAgent> agents, CancellationToken ct)
|
||||
{
|
||||
var wanted = agents.ToDictionary(a => a.AgentId);
|
||||
var stopped = new List<Running>();
|
||||
@@ -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))
|
||||
{
|
||||
|
||||
@@ -15,15 +15,15 @@ public sealed class ProbeFile
|
||||
[JsonPropertyName("private_key")]
|
||||
public string PrivateKey { get; set; } = "";
|
||||
|
||||
/// <summary>首次连接用的 Agent；连上之后以 <c>agents.json</c>（最新的列表）为准。</summary>
|
||||
[JsonPropertyName("bootstrap")]
|
||||
public List<BootstrapAgent> Bootstrap { get; set; } = [];
|
||||
/// <summary>面板的地址（http 或 https）。Probe 直接连它：身份、Agent 列表、更新都从这里来。</summary>
|
||||
[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;
|
||||
|
||||
/// <summary><c>agents.json</c>：最近一次从 Agent 同步到的列表（配置，不是测量数据）。</summary>
|
||||
/// <summary><c>agents.json</c>：最近一次从面板同步到的 Agent 列表（配置，不是测量数据）；面板连不上时 Probe 靠它继续测量。</summary>
|
||||
public sealed class AgentsCache
|
||||
{
|
||||
[JsonPropertyName("version")]
|
||||
public long Version { get; set; }
|
||||
|
||||
[JsonPropertyName("agents")]
|
||||
public List<BootstrapAgent> Agents { get; set; } = [];
|
||||
public List<KnownAgent> Agents { get; set; } = [];
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user