feat!: probes link to the panel; signed release updates for container nodes [CI]

Probes connect to the panel directly and get the agent list as desired state.
CI signs the image digest of agent and probe releases, the panel forwards the
manifest, and a host updater verifies it with a local public key, pulls by
digest and rolls back when the new version is unhealthy.
This commit is contained in:
justme committed 2026-10-06 22:42:12 +08:00
1 parent c645851c0e
commit 8dd3209a65
57 files changed
+1640 -559

No files matched your search

@@ -56,7 +56,7 @@ public sealed class AgentHarness : IAsyncDisposable
};
configure?.Invoke(config);
LoadedNodeIdentity identity = LoadedNodeIdentity.Load(directory);
var app = new AgentApp(config, identity, TimeProvider.System, new ConsoleNodeLog(TimeProvider.System, Console.Out));
var app = new AgentApp(config, identity, directory, TimeProvider.System, new ConsoleNodeLog(TimeProvider.System, Console.Out));
var harness = new AgentHarness(backend, directory, identity, app);
await app.Ready.WaitAsync(TimeSpan.FromSeconds(10));
await BackendFixture.EventuallyAsync(() => Task.FromResult(app.Link.IsConnected), because: "the agent should connect to the backend");
@@ -338,7 +338,7 @@ public sealed class EnrollmentAndTaskSetTests
JsonNode identity = JsonNode.Parse(await created.Content.ReadAsStringAsync())!;
string id = identity["probe_id"]!.GetValue<string>();
using var key1 = Fleet.Identity.NodeIdentity.FromPkcs8(Convert.FromBase64String(identity["private_key"]!.GetValue<string>()));
Assert.Contains(identity["bootstrap"]!.AsArray(), b => b!["agent_id"]!.GetValue<string>() == agent.NodeId.Value);
Assert.True(Uri.TryCreate(identity["server"]!.GetValue<string>(), 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<ProbeAgentsResult> WaitForAsync(AgentSimulator agent, Func<ProbeAgentsResult, bool> predicate)
{
ProbeAgentsResult? found = null;
await BackendFixture.EventuallyAsync(() =>
{
found = agent.ProbeAgentLists.ToArray().LastOrDefault(predicate);
return Task.FromResult(found is not null);
});
return found!;
}
private Task<HttpResponseMessage> PostEnrollAsync(JsonNode body) =>
_backend.Anonymous.PostAsync("/api/agent/v1/enroll", BackendFixture.Json(body));
}
@@ -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);
@@ -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<string>() == 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<bool>());
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<bool>(), 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<string>() == 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<bool>(), because: "a revoked probe is disconnected and cannot come back");
Assert.False(probe.App.Link.IsConnected);
}
[Fact]
public async Task Probe_FollowsTheAgentListThePanelKnows_AndSavesItForTheNextStart()
{
@@ -0,0 +1,96 @@
using System.Net;
using System.Text.Json.Nodes;
using Monitor.Protocol;
using Monitor.TestSupport;
namespace Monitor.Backend.Tests;
[Collection("probe-e2e")]
public sealed class UpdateTests
{
private const string Digest = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
private readonly BackendFixture _backend;
public UpdateTests(BackendFixture backend)
{
_backend = backend;
}
// 面板不验签(节点宿主机上的更新器验),所以测试里签名随便写一个格式对的
private static string Manifest(string component, long seq, string digest = Digest) =>
$$"""{"schema":1,"component":"{{component}}","channel":"dev","image":"registry.example/monitor-dev/monitor-{{component}}","digest":"{{digest}}","tag":"dev-{{seq}}","version":"0.1.1","commit":"abc","built_at":"2026-10-06T00:00:00Z","seq":{{seq}}}""";
private async Task<HttpResponseMessage> PublishAsync(string manifest, string signature = "MEUCIQDc") =>
await _backend.Admin.PostAsync("/api/admin/releases", BackendFixture.Json(new JsonObject { ["manifest"] = manifest, ["signature"] = signature }));
[Fact]
public async Task Release_IsForwardedByteForByte_ToOnlineNodesOfThatComponent_AndToNodesThatConnectLater()
{
long seq = TimeProvider.System.GetUtcNow().ToUnixTimeSeconds() * 1000;
await using AgentSimulator online = await _backend.ConnectAgentAsync();
string manifest = Manifest("agent", seq);
using HttpResponseMessage published = await PublishAsync(manifest);
Assert.Equal(HttpStatusCode.Created, published.StatusCode);
JsonNode release = JsonNode.Parse(await published.Content.ReadAsStringAsync())!;
Assert.Equal("agent", release["component"]!.GetValue<string>());
Assert.Equal(Digest, release["digest"]!.GetValue<string>());
await BackendFixture.EventuallyAsync(() => Task.FromResult(online.UpdateOffers.ToArray().Any(o => o.Manifest == manifest && o.Signature == "MEUCIQDc")), because: "the offer reaches online agents unchanged");
// 之后才连上的节点,hello 之后补发
await using AgentSimulator late = await _backend.ConnectAgentAsync();
await BackendFixture.EventuallyAsync(() => Task.FromResult(late.UpdateOffers.ToArray().Any(o => o.Manifest == manifest)), because: "a node that connects later still gets the latest offer");
// 另一个组件(probe)的清单不会发给 Agent
using HttpResponseMessage probeRelease = await PublishAsync(Manifest("probe", seq));
Assert.Equal(HttpStatusCode.Created, probeRelease.StatusCode);
await Task.Delay(300);
Assert.DoesNotContain(online.UpdateOffers.ToArray(), o => o.Manifest.Contains("\"component\":\"probe\"", StringComparison.Ordinal));
}
[Fact]
public async Task Release_Registration_RefusesBadManifests_StaleSeq_AndNeedsAnAdmin()
{
long seq = TimeProvider.System.GetUtcNow().ToUnixTimeSeconds() * 1000 + 5;
Assert.Equal(HttpStatusCode.BadRequest, (await PublishAsync("{not json")).StatusCode);
Assert.Equal(HttpStatusCode.BadRequest, (await PublishAsync(Manifest("agent", seq, digest: "sha256:short"))).StatusCode);
Assert.Equal(HttpStatusCode.BadRequest, (await PublishAsync(Manifest("backend", seq))).StatusCode);
Assert.Equal(HttpStatusCode.BadRequest, (await PublishAsync(Manifest("agent", seq), signature: "not base64!")).StatusCode);
Assert.Equal(HttpStatusCode.Created, (await PublishAsync(Manifest("agent", seq))).StatusCode);
Assert.Equal(HttpStatusCode.Conflict, (await PublishAsync(Manifest("agent", seq))).StatusCode); // 同一个 seq:重放
Assert.Equal(HttpStatusCode.Conflict, (await PublishAsync(Manifest("agent", seq - 1))).StatusCode); // 更旧的 seq
using HttpResponseMessage anonymous = await _backend.Anonymous.PostAsync("/api/admin/releases", BackendFixture.Json(new JsonObject { ["manifest"] = Manifest("agent", seq + 1), ["signature"] = "MEUC" }));
Assert.Contains(anonymous.StatusCode, new[] { HttpStatusCode.Unauthorized, HttpStatusCode.Forbidden });
}
[Fact]
public async Task UpdateRequest_ReachesTheNode_AndItsReportedStatusShowsInBuildInfo()
{
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
using HttpResponseMessage request = await _backend.Admin.PostAsync($"/api/admin/nodes/{agent.NodeId.Value}/update", null);
Assert.Equal(HttpStatusCode.NoContent, request.StatusCode);
await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.UpdateApplies == 1));
// 节点(宿主机更新器)报告状态:面板在构建信息里显示
await agent.SendUpdateStatusAsync(new UpdateStatus { State = "available", Available = new UpdateAvailable { Tag = "dev-9", Version = "0.1.1", Digest = Digest, Seq = 9 }, At = TimeProvider.System.GetUtcNow().ToUnixTimeSeconds() });
await BackendFixture.EventuallyAsync(async () =>
{
JsonNode info = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/build-info");
JsonNode? node = info["nodes"]!.AsArray().FirstOrDefault(n => n!["id"]!.GetValue<string>() == agent.NodeId.Value);
return node?["update"]?["state"]?.GetValue<string>() == "available" && node["update"]!["available_tag"]!.GetValue<string>() == "dev-9";
}, because: "the update state comes from the node's heartbeat");
// 节点离线之后不能请求更新
string id = agent.NodeId.Value;
await agent.DisposeAsync();
await BackendFixture.EventuallyAsync(async () =>
{
using HttpResponseMessage offline = await _backend.Admin.PostAsync($"/api/admin/nodes/{id}/update", null);
return offline.StatusCode == HttpStatusCode.Conflict || offline.StatusCode == HttpStatusCode.NotFound;
});
}
}
@@ -46,10 +46,15 @@ public sealed class AgentSimulator : IAsyncDisposable
public ConcurrentQueue<TaskSet> TaskSets { get; } = new();
public ConcurrentQueue<ProbeAgentsResult> ProbeAgentLists { get; } = new();
public ConcurrentQueue<TestRunParams> TestRuns { get; } = new();
/// <summary>面板发来的发布清单(<c>monitor.update.offer</c>)和更新请求(<c>monitor.update.apply</c>)。</summary>
public ConcurrentQueue<UpdateOfferParams> UpdateOffers { get; } = new();
private int _updateApplies;
public int UpdateApplies => Volatile.Read(ref _updateApplies);
public ConcurrentQueue<TestId> TestCancels { get; } = new();
public IReadOnlyList<RejectedItem> RejectNextState { get; set; } = [];
@@ -190,11 +195,12 @@ public sealed class AgentSimulator : IAsyncDisposable
new TestStatusParams { TestId = test, Status = status, Phase = phase, ErrorCode = errorCode },
MonitorJsonContext.Default.TestStatusParams));
public async Task<ProbeAgentsResult> RequestProbeAgentsAsync()
{
RpcMessage reply = await RequestAsync(MonitorMethods.ProbeAgents, JsonSerializer.SerializeToElement(new ProbeAgentsParams(), MonitorJsonContext.Default.ProbeAgentsParams));
return reply.Result!.Value.Deserialize(MonitorJsonContext.Default.ProbeAgentsResult)!;
}
/// <summary>像节点的更新器报告的那样,随心跳上报更新状态。</summary>
public Task SendUpdateStatusAsync(UpdateStatus status) =>
SendAsync(RpcMessage.Notification(
FleetMethods.Heartbeat,
new HeartbeatParams { Extra = JsonSerializer.SerializeToElement(new AgentHeartbeatExtra { Update = status }, MonitorJsonContext.Default.AgentHeartbeatExtra) },
FleetRpcContext.Default.HeartbeatParams));
public async Task<RpcMessage> RequestRawAsync(string method, JsonElement? parameters = null) => await RequestAsync(method, parameters);
@@ -397,6 +403,12 @@ public sealed class AgentSimulator : IAsyncDisposable
case FleetMethods.Revoked:
Revoked = true;
break;
case MonitorMethods.UpdateOffer:
UpdateOffers.Enqueue(message.Params!.Value.Deserialize(MonitorJsonContext.Default.UpdateOfferParams)!);
break;
case MonitorMethods.UpdateApply:
Interlocked.Increment(ref _updateApplies);
break;
case MonitorMethods.TestRun:
TestRunParams run = message.Params!.Value.Deserialize(MonitorJsonContext.Default.TestRunParams)!;
TestRuns.Enqueue(run);
@@ -416,9 +428,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;
}
}
}
@@ -61,8 +61,6 @@ public sealed class ProbeSimulator : IAsyncDisposable
public ConcurrentQueue<TestId> Aborts { get; } = new();
public ConcurrentQueue<ProbeAgentsResult> 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,