feat: signed releases and a host-side updater for container nodes
CI signs the image digest of each agent and probe release; the panel only stores and forwards it. Nodes write the manifest to their data directory and a host updater verifies the signature with a local public key, pulls by digest, restarts the unit and rolls back when it is unhealthy. The panel can list releases, ask a node to update and show the reported update state.
This commit is contained in:
1 parent
48072737c9
commit
0a66f5e07a
26 files changed
+1384
-215
No files matched your search
@@ -25,6 +25,7 @@ public sealed class AgentApp : IDisposable
|
||||
private readonly TaskCompletionSource _ready = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
private readonly AgentAdmission _admission;
|
||||
private readonly Outbox _outbox = new(64 << 20);
|
||||
private readonly UpdateBridge _update;
|
||||
private readonly BackendUplink _uplink;
|
||||
private readonly UdpMux _udp;
|
||||
private readonly TcpAcceptor _tcp;
|
||||
@@ -34,8 +35,10 @@ public sealed class AgentApp : IDisposable
|
||||
private readonly TestRunner _tests;
|
||||
private readonly FleetLink _link;
|
||||
|
||||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, TimeProvider time, INodeLog log)
|
||||
/// <param name="dataDirectory">配置目录(容器里的 <c>/etc/monitor</c>):宿主机更新器和节点在这里交换清单、请求和状态。</param>
|
||||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log)
|
||||
{
|
||||
_update = new UpdateBridge(dataDirectory, new ConfigStore(dataDirectory).Write, log);
|
||||
_config = config;
|
||||
_identity = identity;
|
||||
_time = time;
|
||||
@@ -200,6 +203,24 @@ public sealed class AgentApp : IDisposable
|
||||
(_, cancel, _) => ValueTask.FromResult(Result<TestRunResult>.Ok(new TestRunResult { Accepted = _tests.Cancel(cancel.TestId) })));
|
||||
}
|
||||
|
||||
if (Allow(MonitorMethods.UpdateOffer))
|
||||
{
|
||||
router.OnNotify(MonitorMethods.UpdateOffer, MonitorJsonContext.Default.UpdateOfferParams, (_, offer, _) =>
|
||||
{
|
||||
_update.WriteOffer(offer);
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
}
|
||||
|
||||
if (Allow(MonitorMethods.UpdateApply))
|
||||
{
|
||||
router.OnNotify(MonitorMethods.UpdateApply, MonitorJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||||
{
|
||||
_update.RequestApply();
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
}
|
||||
|
||||
if (Allow(MonitorMethods.ProbeDisconnect))
|
||||
{
|
||||
router.On(
|
||||
@@ -270,6 +291,7 @@ public sealed class AgentApp : IDisposable
|
||||
DroppedInvalid = _counters.DroppedInvalid,
|
||||
DroppedRateLimited = _counters.DroppedRateLimited,
|
||||
QueueBytes = _outbox.PendingBytes,
|
||||
Update = _update.ReadStatus(),
|
||||
};
|
||||
_lastIn = packetsIn;
|
||||
_lastOut = packetsOut;
|
||||
|
||||
@@ -35,7 +35,7 @@ catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidO
|
||||
|
||||
using (identity)
|
||||
{
|
||||
using var app = new AgentApp(config, identity, time, log);
|
||||
using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log);
|
||||
await ServiceShell.RunAsync(app.RunAsync, log);
|
||||
}
|
||||
|
||||
|
||||
+297
-1
@@ -3696,7 +3696,7 @@
|
||||
"tags": [
|
||||
"admin-build"
|
||||
],
|
||||
"summary": "Version, tag, commit and build time of the backend and every agent and probe",
|
||||
"summary": "Version, tag, commit, build time and update state of the backend and every agent and probe",
|
||||
"operationId": "build_info_get",
|
||||
"responses": {
|
||||
"200": {
|
||||
@@ -3732,6 +3732,175 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/admin/releases": {
|
||||
"get": {
|
||||
"tags": [
|
||||
"admin-build"
|
||||
],
|
||||
"summary": "The latest signed release manifest of each component",
|
||||
"operationId": "releases_list",
|
||||
"responses": {
|
||||
"200": {
|
||||
"description": "OK",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/PageDtoOfReleaseDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"401": {
|
||||
"description": "Unauthorized",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"403": {
|
||||
"description": "Forbidden",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"post": {
|
||||
"tags": [
|
||||
"admin-build"
|
||||
],
|
||||
"summary": "Register a CI-signed release manifest; online nodes get it right away (their updaters verify the signature)",
|
||||
"operationId": "releases_publish",
|
||||
"requestBody": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/PublishReleaseRequest"
|
||||
}
|
||||
}
|
||||
},
|
||||
"required": true
|
||||
},
|
||||
"responses": {
|
||||
"201": {
|
||||
"description": "Created",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ReleaseDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"400": {
|
||||
"description": "Bad Request",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"401": {
|
||||
"description": "Unauthorized",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"403": {
|
||||
"description": "Forbidden",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"409": {
|
||||
"description": "Conflict",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/api/admin/nodes/{id}/update": {
|
||||
"post": {
|
||||
"tags": [
|
||||
"admin-build"
|
||||
],
|
||||
"summary": "Ask an online agent or probe to install the release it has been offered",
|
||||
"operationId": "nodes_update",
|
||||
"parameters": [
|
||||
{
|
||||
"name": "id",
|
||||
"in": "path",
|
||||
"required": true,
|
||||
"schema": {
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"204": {
|
||||
"description": "No Content",
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": { }
|
||||
}
|
||||
}
|
||||
},
|
||||
"401": {
|
||||
"description": "Unauthorized",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"403": {
|
||||
"description": "Forbidden",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"409": {
|
||||
"description": "Conflict",
|
||||
"content": {
|
||||
"application/problem+json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/ProblemDto"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"/healthz": {
|
||||
"get": {
|
||||
"tags": [
|
||||
@@ -5540,6 +5709,16 @@
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"update": {
|
||||
"oneOf": [
|
||||
{
|
||||
"type": "null"
|
||||
},
|
||||
{
|
||||
"$ref": "#/components/schemas/NodeUpdateDto"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
@@ -5612,6 +5791,49 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"NodeUpdateDto": {
|
||||
"required": [
|
||||
"state",
|
||||
"current_digest",
|
||||
"available_tag",
|
||||
"available_version",
|
||||
"error",
|
||||
"at"
|
||||
],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"state": {
|
||||
"type": "string"
|
||||
},
|
||||
"current_digest": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"available_tag": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"available_version": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"error": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"at": {
|
||||
"format": "date-time"
|
||||
}
|
||||
}
|
||||
},
|
||||
"OverviewDto": {
|
||||
"required": [
|
||||
"range",
|
||||
@@ -5989,6 +6211,27 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"PageDtoOfReleaseDto": {
|
||||
"required": [
|
||||
"items",
|
||||
"next_cursor"
|
||||
],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"items": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"$ref": "#/components/schemas/ReleaseDto"
|
||||
}
|
||||
},
|
||||
"next_cursor": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"PageDtoOfSettingDto": {
|
||||
"required": [
|
||||
"items",
|
||||
@@ -6432,6 +6675,27 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"PublishReleaseRequest": {
|
||||
"required": [
|
||||
"manifest",
|
||||
"signature"
|
||||
],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"manifest": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
},
|
||||
"signature": {
|
||||
"type": [
|
||||
"null",
|
||||
"string"
|
||||
]
|
||||
}
|
||||
}
|
||||
},
|
||||
"RawChunkMetaDto": {
|
||||
"required": [
|
||||
"chunk_id",
|
||||
@@ -6573,6 +6837,38 @@
|
||||
}
|
||||
}
|
||||
},
|
||||
"ReleaseDto": {
|
||||
"required": [
|
||||
"component",
|
||||
"seq",
|
||||
"tag",
|
||||
"version",
|
||||
"digest"
|
||||
],
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"component": {
|
||||
"type": "string"
|
||||
},
|
||||
"seq": {
|
||||
"pattern": "^-?(?:0|[1-9]\\d*)$",
|
||||
"type": [
|
||||
"integer",
|
||||
"string"
|
||||
],
|
||||
"format": "int64"
|
||||
},
|
||||
"tag": {
|
||||
"type": "string"
|
||||
},
|
||||
"version": {
|
||||
"type": "string"
|
||||
},
|
||||
"digest": {
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
},
|
||||
"ScoreDto": {
|
||||
"required": [
|
||||
"score",
|
||||
|
||||
@@ -31,6 +31,7 @@ public sealed class ProbeApp : IDisposable
|
||||
_env = new ConnectionEnv(identity, file.ProbeId, file.Limits, directory, new MonoClock(time), time, log);
|
||||
directory.Changed += () => _changed.Release();
|
||||
|
||||
var update = new UpdateBridge(configDirectory, new ConfigStore(configDirectory).Write, log);
|
||||
var options = new FleetLinkOptions
|
||||
{
|
||||
Server = new Uri(file.Server),
|
||||
@@ -38,6 +39,7 @@ public sealed class ProbeApp : IDisposable
|
||||
NodeId = new NodeId(file.ProbeId),
|
||||
NodeVersion = Version,
|
||||
Capabilities = [MonitorCapabilities.Base],
|
||||
HeartbeatExtra = () => JsonSerializer.SerializeToElement(new ProbeHeartbeatExtra { Update = update.ReadStatus() }, MonitorJsonContext.Default.ProbeHeartbeatExtra),
|
||||
HelloMeta = () => JsonSerializer.SerializeToElement(new ProbeHelloMeta { Platform = PlatformDetector.Detect() }, MonitorJsonContext.Default.ProbeHelloMeta),
|
||||
};
|
||||
var list = new StateHandler<ProbeAgentList>(
|
||||
@@ -48,7 +50,18 @@ public sealed class ProbeApp : IDisposable
|
||||
directory.Apply(doc);
|
||||
return ValueTask.FromResult(new StateSyncResult { AppliedVersion = version });
|
||||
});
|
||||
_link = new FleetLink(options, new RpcRouter<LinkContext>(), [list], new Outbox(1 << 20), time, log);
|
||||
var router = new RpcRouter<LinkContext>()
|
||||
.OnNotify(MonitorMethods.UpdateOffer, MonitorJsonContext.Default.UpdateOfferParams, (_, offer, _) =>
|
||||
{
|
||||
update.WriteOffer(offer);
|
||||
return ValueTask.CompletedTask;
|
||||
})
|
||||
.OnNotify(MonitorMethods.UpdateApply, MonitorJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||||
{
|
||||
update.RequestApply();
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
_link = new FleetLink(options, router, [list], new Outbox(1 << 20), time, log);
|
||||
}
|
||||
|
||||
/// <summary>到面板的连接(测试和状态查看用)。</summary>
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
using Monitor.Protocol;
|
||||
|
||||
namespace Monitor.Measurement;
|
||||
|
||||
/// <summary>
|
||||
/// 节点进程和宿主机更新器之间的文件交换(容器里 <c>/etc/monitor</c>,也就是数据目录):
|
||||
/// 节点把面板发来的清单和签名写进 <c>update/</c>,把「更新」请求写成 <c>update/apply.request</c>;更新器验签并更新,状态写在 root 拥有的 <c>update-status/status.json</c>。
|
||||
/// 节点进程不执行任何命令,也不信任清单的内容(验签在宿主机上)。没有安装更新器时目录写不了或没人读,调用只是返回 false。
|
||||
/// </summary>
|
||||
public sealed class UpdateBridge
|
||||
{
|
||||
private const int MaxManifestBytes = 4096;
|
||||
private const int MaxSignatureBytes = 512;
|
||||
private const int MaxStatusBytes = 4096;
|
||||
|
||||
private readonly string _updateDirectory;
|
||||
private readonly string _statusFile;
|
||||
private readonly Action<string, byte[]> _write;
|
||||
private readonly INodeLog _log;
|
||||
|
||||
/// <param name="write">写数据目录里的文件(相对路径,原子替换):由节点传入 <c>ConfigStore.Write</c>——常驻进程里只有它可以写文件。</param>
|
||||
public UpdateBridge(string dataDirectory, Action<string, byte[]> write, INodeLog log)
|
||||
{
|
||||
_updateDirectory = Path.Combine(dataDirectory, "update");
|
||||
_statusFile = Path.Combine(dataDirectory, "update-status", "status.json");
|
||||
_write = write;
|
||||
_log = log;
|
||||
}
|
||||
|
||||
/// <summary>放下面板发来的清单和签名(和已有的一样就不动,免得每次重连都惊动更新器)。先写清单、后写签名:更新器监视的是签名文件。</summary>
|
||||
public bool WriteOffer(UpdateOfferParams offer)
|
||||
{
|
||||
byte[] manifest = Encoding.UTF8.GetBytes(offer.Manifest);
|
||||
byte[] signature = Encoding.UTF8.GetBytes(offer.Signature);
|
||||
if (manifest.Length is 0 or > MaxManifestBytes || signature.Length is 0 or > MaxSignatureBytes)
|
||||
{
|
||||
_log.Warn("ignoring an update offer of an unexpected size");
|
||||
return false;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
string manifestPath = Path.Combine(_updateDirectory, "offer.manifest");
|
||||
string signaturePath = Path.Combine(_updateDirectory, "offer.sig");
|
||||
if (Same(manifestPath, manifest) && Same(signaturePath, signature))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
_write("update/offer.manifest", manifest);
|
||||
_write("update/offer.sig", signature);
|
||||
_log.Info("update offer received");
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
||||
{
|
||||
_log.Warn("could not store the update offer (is the updater installed?)", ("err", ex.Message));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>请求更新:写一个空的 <c>apply.request</c>(更新器验签通过、清单更新才会真的动手)。</summary>
|
||||
public bool RequestApply()
|
||||
{
|
||||
try
|
||||
{
|
||||
_write("update/apply.request", []);
|
||||
_log.Info("update requested");
|
||||
return true;
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
||||
{
|
||||
_log.Warn("could not request the update (is the updater installed?)", ("err", ex.Message));
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>更新器的状态;没有安装更新器、文件不存在或内容不对时返回 null。</summary>
|
||||
public UpdateStatus? ReadStatus()
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!File.Exists(_statusFile) || new FileInfo(_statusFile).Length > MaxStatusBytes)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return JsonSerializer.Deserialize(File.ReadAllBytes(_statusFile), MonitorJsonContext.Default.UpdateStatus);
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or JsonException)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private static bool Same(string path, byte[] content) => File.Exists(path) && File.ReadAllBytes(path).AsSpan().SequenceEqual(content);
|
||||
}
|
||||
@@ -13,6 +13,8 @@ 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 UpdateOffer = "monitor.update.offer";
|
||||
public const string UpdateApply = "monitor.update.apply";
|
||||
|
||||
// Probe ↔ Agent(点对点会话)
|
||||
public const string Clock = "monitor.clock";
|
||||
@@ -217,6 +219,66 @@ public sealed class PlatformInfo
|
||||
public string? BuiltAt { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 面板给节点的更新清单(<c>monitor.update.offer</c>):<c>manifest</c> 是发布清单的原始文本,<c>signature</c> 是 base64 的签名。
|
||||
/// 节点只是把它们原样放进数据目录;验签、拉取、重启都由宿主机上的更新器完成(节点进程不执行任何命令,也不信任清单的内容)。
|
||||
/// </summary>
|
||||
public sealed class UpdateOfferParams
|
||||
{
|
||||
[JsonPropertyName("manifest")]
|
||||
public string Manifest { get; set; } = "";
|
||||
|
||||
[JsonPropertyName("signature")]
|
||||
public string Signature { get; set; } = "";
|
||||
}
|
||||
|
||||
/// <summary>宿主机更新器已经验证过的可用更新。</summary>
|
||||
public sealed class UpdateAvailable
|
||||
{
|
||||
[JsonPropertyName("version")]
|
||||
public string? Version { get; set; }
|
||||
|
||||
[JsonPropertyName("tag")]
|
||||
public string? Tag { get; set; }
|
||||
|
||||
[JsonPropertyName("digest")]
|
||||
public string? Digest { get; set; }
|
||||
|
||||
[JsonPropertyName("seq")]
|
||||
public long Seq { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 宿主机更新器的状态(它写在 <c>update-status/status.json</c>,节点读出来随心跳上报):
|
||||
/// <c>up_to_date</c>、<c>available</c>、<c>applying</c>、<c>failed</c>、<c>rolled_back</c>、<c>invalid</c>(清单验签或校验没过)。
|
||||
/// 没有安装更新器的节点没有这个字段。
|
||||
/// </summary>
|
||||
public sealed class UpdateStatus
|
||||
{
|
||||
[JsonPropertyName("state")]
|
||||
public string State { get; set; } = "";
|
||||
|
||||
/// <summary>正在运行的镜像 digest(更新器记录的;还没更新过时为空)。</summary>
|
||||
[JsonPropertyName("current")]
|
||||
public string? Current { get; set; }
|
||||
|
||||
[JsonPropertyName("available")]
|
||||
public UpdateAvailable? Available { get; set; }
|
||||
|
||||
[JsonPropertyName("error")]
|
||||
public string? Error { get; set; }
|
||||
|
||||
[JsonPropertyName("at")]
|
||||
public long At { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Probe 的 <c>fleet.heartbeat</c> 附加字段:更新器的状态。</summary>
|
||||
public sealed class ProbeHeartbeatExtra
|
||||
{
|
||||
[JsonPropertyName("update")]
|
||||
public UpdateStatus? Update { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Probe 的 <c>fleet.hello</c> 附加字段:系统、架构和构建信息(只用于展示和更新)。</summary>
|
||||
public sealed class ProbeHelloMeta
|
||||
{
|
||||
@@ -326,6 +388,9 @@ public sealed class AgentHeartbeatExtra
|
||||
|
||||
[JsonPropertyName("queue_bytes")]
|
||||
public long QueueBytes { get; set; }
|
||||
|
||||
[JsonPropertyName("update")]
|
||||
public UpdateStatus? Update { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>Agent 下发给 Probe 的一条线路任务(期望状态类型 <c>line_task_set</c> 里的一项)。</summary>
|
||||
@@ -465,6 +530,9 @@ public static class FragmentKeys
|
||||
[JsonSerializable(typeof(TestStatusParams))]
|
||||
[JsonSerializable(typeof(AgentEventParams))]
|
||||
[JsonSerializable(typeof(ProbeAgentList))]
|
||||
[JsonSerializable(typeof(UpdateOfferParams))]
|
||||
[JsonSerializable(typeof(UpdateStatus))]
|
||||
[JsonSerializable(typeof(ProbeHeartbeatExtra))]
|
||||
[JsonSerializable(typeof(ProbeHelloMeta))]
|
||||
[JsonSerializable(typeof(AgentEnrollMeta))]
|
||||
[JsonSerializable(typeof(PlatformInfo))]
|
||||
|
||||
@@ -1,11 +1,13 @@
|
||||
using Fleet.Core;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
using Microsoft.AspNetCore.Routing;
|
||||
using Monitor.Backend.Application;
|
||||
using Monitor.Protocol;
|
||||
|
||||
namespace Monitor.Backend.Api;
|
||||
|
||||
/// <summary>管理端的构建信息:后端自己和每个 Agent、Probe 的版本、tag、提交、构建时间(节点上报的,不含任何地址)。</summary>
|
||||
/// <summary>构建信息与更新:后端和每个节点的版本、tag、构建时间,CI 签名的发布清单,以及「更新这个节点」的请求。更新本身由节点宿主机上的更新器验签并执行。</summary>
|
||||
public static class BuildEndpoints
|
||||
{
|
||||
private const string Tag = "admin-build";
|
||||
@@ -13,18 +15,44 @@ public static class BuildEndpoints
|
||||
public static void Map(RouteGroupBuilder group)
|
||||
{
|
||||
group.MapGet("/build-info", GetAsync)
|
||||
.Api("build_info_get", "Version, tag, commit and build time of the backend and every agent and probe", Tag)
|
||||
.Api("build_info_get", "Version, tag, commit, build time and update state of the backend and every agent and probe", Tag)
|
||||
.Returns<BuildInfoDto>().ProducesProblems(401, 403);
|
||||
group.MapGet("/releases", ListReleasesAsync)
|
||||
.Api("releases_list", "The latest signed release manifest of each component", Tag).Returns<PageDto<ReleaseDto>>().ProducesProblems(401, 403);
|
||||
group.MapPost("/releases", PublishReleaseAsync)
|
||||
.Api("releases_publish", "Register a CI-signed release manifest; online nodes get it right away (their updaters verify the signature)", Tag)
|
||||
.Returns<ReleaseDto>(StatusCodes.Status201Created).ProducesProblems(400, 401, 403, 409);
|
||||
group.MapPost("/nodes/{id}/update", RequestUpdateAsync)
|
||||
.Api("nodes_update", "Ask an online agent or probe to install the release it has been offered", Tag).Returns<object>(StatusCodes.Status204NoContent).ProducesProblems(401, 403, 409);
|
||||
}
|
||||
|
||||
private static async Task<IResult> GetAsync(HttpContext http, NodeInfoService nodes, CancellationToken ct)
|
||||
private static async Task<IResult> GetAsync(HttpContext http, NodeInfoService nodes, ReleaseService releases, CancellationToken ct)
|
||||
{
|
||||
var items = (await nodes.ListAsync(ct))
|
||||
.OrderBy(n => n.Kind, StringComparer.Ordinal)
|
||||
.ThenBy(n => n.NodeId, StringComparer.Ordinal)
|
||||
.Select(n => new NodeBuildDto(n.NodeId, n.Kind, n.Os, n.OsName, n.Arch, n.Version, n.Tag, n.Commit, n.BuiltAt))
|
||||
.Select(n => new NodeBuildDto(n.NodeId, n.Kind, n.Os, n.OsName, n.Arch, n.Version, n.Tag, n.Commit, n.BuiltAt, ToDto(releases.StatusOf(new NodeId(n.NodeId)))))
|
||||
.ToList();
|
||||
http.Response.Headers.CacheControl = "no-store";
|
||||
return TypedResults.Json(new BuildInfoDto(NodeEndpoints.Backend(withCommit: true), items), ApiJsonContext.Default.BuildInfoDto);
|
||||
}
|
||||
|
||||
private static async Task<IResult> ListReleasesAsync(ReleaseService releases, CancellationToken ct) =>
|
||||
TypedResults.Json(new PageDto<ReleaseDto>([.. (await releases.ListAsync(ct)).Select(ToDto)], null), ApiJsonContext.Default.PageDtoReleaseDto);
|
||||
|
||||
private static async Task<IResult> PublishReleaseAsync(HttpContext http, ReleaseService releases, PublishReleaseRequest request)
|
||||
{
|
||||
Result<ReleaseOffer> result = await releases.PublishAsync(request.Manifest ?? "", request.Signature ?? "", http.RequestAborted);
|
||||
return result.TryGet(out ReleaseOffer? offer, out FleetError? error)
|
||||
? TypedResults.Json(ToDto(offer), ApiJsonContext.Default.ReleaseDto, statusCode: StatusCodes.Status201Created)
|
||||
: Problems.From(http, error);
|
||||
}
|
||||
|
||||
private static async Task<IResult> RequestUpdateAsync(HttpContext http, ReleaseService releases, string id, CancellationToken ct) =>
|
||||
await releases.RequestUpdateAsync(new NodeId(id), ct) ? TypedResults.NoContent() : Problems.Create(http, MonitorErrors.AgentOffline, "the node is not connected");
|
||||
|
||||
private static ReleaseDto ToDto(ReleaseOffer offer) => new(offer.Component, offer.Seq, offer.Tag, offer.Version, offer.Digest);
|
||||
|
||||
private static NodeUpdateDto? ToDto(UpdateStatus? status) =>
|
||||
status is null ? null : new NodeUpdateDto(status.State, status.Current, status.Available?.Tag, status.Available?.Version, status.Error, status.At > 0 ? DateTimeOffset.FromUnixTimeSeconds(status.At) : null);
|
||||
}
|
||||
@@ -124,6 +124,14 @@ public sealed record ChangePasswordRequest(string Password);
|
||||
public sealed record RecomputeDto(string State, IReadOnlyList<string> Metrics, long WindowsDone, long WindowsTotal, DateTimeOffset? From, DateTimeOffset? To, int QueueLength);
|
||||
|
||||
/// <summary>管理端看的构建信息:每个节点自报的版本、tag、提交和构建时间,以及系统。</summary>
|
||||
public sealed record NodeBuildDto(string Id, string Kind, string? Os, string? OsName, string? Arch, string? Version, string? Tag, string? Commit, string? BuiltAt);
|
||||
public sealed record NodeBuildDto(string Id, string Kind, string? Os, string? OsName, string? Arch, string? Version, string? Tag, string? Commit, string? BuiltAt, NodeUpdateDto? Update = null);
|
||||
|
||||
/// <summary>节点宿主机上的更新器报告的状态(<c>up_to_date</c>、<c>available</c>、<c>applying</c>、<c>failed</c>、<c>rolled_back</c>、<c>invalid</c>);没装更新器的节点没有这个字段。</summary>
|
||||
public sealed record NodeUpdateDto(string State, string? CurrentDigest, string? AvailableTag, string? AvailableVersion, string? Error, DateTimeOffset? At);
|
||||
|
||||
/// <summary>面板登记的最新发布清单(CI 签名的;验签在节点上)。</summary>
|
||||
public sealed record ReleaseDto(string Component, long Seq, string Tag, string Version, string Digest);
|
||||
|
||||
public sealed record PublishReleaseRequest(string? Manifest, string? Signature);
|
||||
|
||||
public sealed record BuildInfoDto(BuildDto Backend, IReadOnlyList<NodeBuildDto> Nodes);
|
||||
@@ -38,6 +38,9 @@ namespace Monitor.Backend.Api;
|
||||
[JsonSerializable(typeof(PageDto<AgentDto>))]
|
||||
[JsonSerializable(typeof(AgentPatchRequest))]
|
||||
[JsonSerializable(typeof(BuildDto))]
|
||||
[JsonSerializable(typeof(ReleaseDto))]
|
||||
[JsonSerializable(typeof(PageDto<ReleaseDto>))]
|
||||
[JsonSerializable(typeof(PublishReleaseRequest))]
|
||||
[JsonSerializable(typeof(AboutDto))]
|
||||
[JsonSerializable(typeof(NodeBuildDto))]
|
||||
[JsonSerializable(typeof(BuildInfoDto))]
|
||||
|
||||
@@ -52,9 +52,11 @@ public sealed class MonitorNodeEventSink : INodeEventSink
|
||||
private readonly LiveHub _live;
|
||||
private readonly AgentService _service;
|
||||
private readonly NodeInfoService _nodes;
|
||||
private readonly ReleaseService _releases;
|
||||
|
||||
public MonitorNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes)
|
||||
public MonitorNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases)
|
||||
{
|
||||
_releases = releases;
|
||||
_nodes = nodes;
|
||||
_agents = agents;
|
||||
_runtime = runtime;
|
||||
@@ -80,6 +82,7 @@ public sealed class MonitorNodeEventSink : INodeEventSink
|
||||
break;
|
||||
case NodeEventKind.Offline:
|
||||
_runtime.MarkOffline(id);
|
||||
_releases.RecordStatus(id, null);
|
||||
_live.Publish(new NodeSignal("agent", id.Value, false, nodeEvent.AtMs));
|
||||
await _events.AddAsync(id, "agent_offline", nodeEvent.AtMs, null, ct);
|
||||
break;
|
||||
@@ -93,6 +96,7 @@ public sealed class MonitorNodeEventSink : INodeEventSink
|
||||
AgentHeartbeatExtra? beat = extra.Deserialize(MonitorJsonContext.Default.AgentHeartbeatExtra);
|
||||
state.LastHeartbeat = beat;
|
||||
state.LastHeartbeatAtMs = nodeEvent.AtMs;
|
||||
_releases.RecordStatus(id, beat?.Update);
|
||||
if (beat is not null)
|
||||
{
|
||||
state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs)));
|
||||
@@ -116,18 +120,37 @@ public sealed class MonitorNodeEventSink : INodeEventSink
|
||||
{
|
||||
switch (nodeEvent.Kind)
|
||||
{
|
||||
case NodeEventKind.Online or NodeEventKind.Offline:
|
||||
_live.Publish(new NodeSignal("probe", id.Value, nodeEvent.Kind == NodeEventKind.Online, nodeEvent.AtMs));
|
||||
case NodeEventKind.Online:
|
||||
_live.Publish(new NodeSignal("probe", id.Value, true, nodeEvent.AtMs));
|
||||
break;
|
||||
case NodeEventKind.Hello when nodeEvent.Hello?.Meta is { } raw:
|
||||
case NodeEventKind.Hello:
|
||||
if (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)
|
||||
{
|
||||
// 格式不对就忽略
|
||||
}
|
||||
}
|
||||
|
||||
await _releases.OfferToNodeAsync(id, ct);
|
||||
break;
|
||||
case NodeEventKind.Offline:
|
||||
_live.Publish(new NodeSignal("probe", id.Value, false, nodeEvent.AtMs));
|
||||
_releases.RecordStatus(id, null);
|
||||
break;
|
||||
case NodeEventKind.Heartbeat when nodeEvent.Extra is { } extra:
|
||||
try
|
||||
{
|
||||
ProbeHelloMeta? meta = raw.Deserialize(MonitorJsonContext.Default.ProbeHelloMeta);
|
||||
await _nodes.SetPlatformAsync(id.Value, NodeKinds.Probe, meta?.Platform, ct);
|
||||
_releases.RecordStatus(id, extra.Deserialize(MonitorJsonContext.Default.ProbeHeartbeatExtra)?.Update);
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
// 格式不对就忽略
|
||||
// 忽略格式不正确的附加字段
|
||||
}
|
||||
|
||||
break;
|
||||
@@ -178,6 +201,7 @@ public sealed class MonitorNodeEventSink : INodeEventSink
|
||||
// 端口或证书指纹变了,Probe 需要新的列表
|
||||
AgentRecord? after = await _agents.GetAsync(id, ct);
|
||||
_nodes.ObserveAgentInBackground(id.Value, after?.PublicHost, meta.Platform);
|
||||
await _releases.OfferToNodeAsync(id, ct);
|
||||
if (before is not null && after is not null
|
||||
&& (!before.TlsSpki.AsSpan().SequenceEqual(after.TlsSpki) || before.ProbeTcpPort != after.ProbeTcpPort || !before.ProbeUdpPorts.SequenceEqual(after.ProbeUdpPorts))
|
||||
&& after.Status == NodeStatus.Active)
|
||||
|
||||
@@ -35,6 +35,7 @@ public static class ApplicationServiceCollectionExtensions
|
||||
services.AddSingleton<WindowComputer>();
|
||||
services.AddSingleton<AuthService>();
|
||||
services.AddSingleton<SettingsService>();
|
||||
services.AddSingleton<ReleaseService>();
|
||||
services.AddSingleton<JoinTokenService>();
|
||||
services.AddSingleton<RawBrowserService>();
|
||||
services.AddSingleton<EventService>();
|
||||
|
||||
@@ -10,6 +10,9 @@ public interface IAgentGateway
|
||||
{
|
||||
IReadOnlyCollection<NodeId> OnlineAgents { get; }
|
||||
|
||||
/// <summary>所有在线的节点(Agent 和 Probe)。</summary>
|
||||
IReadOnlyCollection<NodeId> OnlineNodes { get; }
|
||||
|
||||
bool IsOnline(NodeId agent);
|
||||
|
||||
/// <summary>某个 Agent 的期望状态变了:Gateway 重新计算并推送。</summary>
|
||||
@@ -26,4 +29,10 @@ public interface IAgentGateway
|
||||
ValueTask<Result<bool>> DisconnectProbeAsync(NodeId agent, ProbeId probe, CancellationToken ct);
|
||||
|
||||
ValueTask DisconnectAsync(NodeId agent, string reason, CancellationToken ct);
|
||||
|
||||
/// <summary>把发布清单和签名原样交给节点(通知,不等回复)。</summary>
|
||||
ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct);
|
||||
|
||||
/// <summary>请求节点更新到它手里的清单。节点不在线返回 false。</summary>
|
||||
ValueTask<bool> NotifyUpdateApplyAsync(NodeId node, CancellationToken ct);
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Text.Json;
|
||||
using System.Text.RegularExpressions;
|
||||
using Fleet.Core;
|
||||
using Monitor.Protocol;
|
||||
|
||||
namespace Monitor.Backend.Application;
|
||||
|
||||
/// <summary>面板记下的、CI 签名的发布清单(只留每个组件最新的一份)。<c>Manifest</c> 是签名对象的原始文本,面板一个字节也不改。</summary>
|
||||
public sealed record ReleaseOffer(string Component, long Seq, string Tag, string Version, string Digest, string Manifest, string Signature);
|
||||
|
||||
/// <summary>
|
||||
/// 发布清单的分发:CI 签名之后推给面板,面板存下来并原样转给在线的 Agent、Probe(节点连上时再补发一次)。
|
||||
/// <b>面板不验签、也不决定拉什么</b>:节点宿主机上的更新器用本机的发布公钥验签,并且只认本机配置的镜像仓库;
|
||||
/// 所以面板即使被攻破,也推不出一个能被装上的假镜像。面板只负责「通知」和「请求更新」,状态由节点随心跳报回来。
|
||||
/// </summary>
|
||||
public sealed partial class ReleaseService
|
||||
{
|
||||
public const int MaxManifestBytes = 4096;
|
||||
public const int MaxSignatureBytes = 512;
|
||||
private static readonly string[] s_components = [NodeKinds.Agent, NodeKinds.Probe];
|
||||
|
||||
private readonly ISettingsStore _settings;
|
||||
private readonly IAgentGateway _gateway;
|
||||
private readonly ConcurrentDictionary<string, UpdateStatus> _status = new();
|
||||
|
||||
public ReleaseService(ISettingsStore settings, IAgentGateway gateway)
|
||||
{
|
||||
_settings = settings;
|
||||
_gateway = gateway;
|
||||
}
|
||||
|
||||
/// <summary>登记一份新清单并推给在线的同类节点。只要格式对、<c>seq</c> 比现有的新就收下(签名由节点验证)。</summary>
|
||||
public async ValueTask<Result<ReleaseOffer>> PublishAsync(string manifest, string signature, CancellationToken ct)
|
||||
{
|
||||
if (!TryParse(manifest, signature, out ReleaseOffer? offer, out string? problem))
|
||||
{
|
||||
return ValidationError.Fail<ReleaseOffer>("manifest", problem);
|
||||
}
|
||||
|
||||
ReleaseOffer? current = await GetAsync(offer.Component, ct);
|
||||
if (current is not null && offer.Seq <= current.Seq)
|
||||
{
|
||||
return Result<ReleaseOffer>.Fail(MonitorErrors.NameConflict, $"a release with seq {current.Seq} or newer is already registered");
|
||||
}
|
||||
|
||||
await _settings.SetAsync(Key(offer.Component, "manifest"), offer.Manifest, ct);
|
||||
await _settings.SetAsync(Key(offer.Component, "signature"), offer.Signature, ct);
|
||||
foreach (NodeId node in OnlineOf(offer.Component))
|
||||
{
|
||||
await _gateway.NotifyUpdateOfferAsync(node, offer.Manifest, offer.Signature, ct);
|
||||
}
|
||||
|
||||
return Result<ReleaseOffer>.Ok(offer);
|
||||
}
|
||||
|
||||
public async ValueTask<ReleaseOffer?> GetAsync(string component, CancellationToken ct)
|
||||
{
|
||||
string? manifest = await _settings.GetAsync(Key(component, "manifest"), ct);
|
||||
string? signature = await _settings.GetAsync(Key(component, "signature"), ct);
|
||||
return manifest is not null && signature is not null && TryParse(manifest, signature, out ReleaseOffer? offer, out _) && offer.Component == component ? offer : null;
|
||||
}
|
||||
|
||||
public async ValueTask<IReadOnlyList<ReleaseOffer>> ListAsync(CancellationToken ct)
|
||||
{
|
||||
var list = new List<ReleaseOffer>();
|
||||
foreach (string component in s_components)
|
||||
{
|
||||
if (await GetAsync(component, ct) is { } offer)
|
||||
{
|
||||
list.Add(offer);
|
||||
}
|
||||
}
|
||||
|
||||
return list;
|
||||
}
|
||||
|
||||
/// <summary>节点刚连上:把它这个组件最新的清单补发一份(节点离线时发布的也不会错过)。</summary>
|
||||
public async ValueTask OfferToNodeAsync(NodeId node, CancellationToken ct)
|
||||
{
|
||||
if (ComponentOf(node) is { } component && await GetAsync(component, ct) is { } offer)
|
||||
{
|
||||
await _gateway.NotifyUpdateOfferAsync(node, offer.Manifest, offer.Signature, ct);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>请求某个节点更新到它手里那份(验过签的)清单。节点不在线返回 false。</summary>
|
||||
public ValueTask<bool> RequestUpdateAsync(NodeId node, CancellationToken ct) => _gateway.NotifyUpdateApplyAsync(node, ct);
|
||||
|
||||
public void RecordStatus(NodeId node, UpdateStatus? status)
|
||||
{
|
||||
if (status is null)
|
||||
{
|
||||
_status.TryRemove(node.Value, out _);
|
||||
}
|
||||
else
|
||||
{
|
||||
_status[node.Value] = status;
|
||||
}
|
||||
}
|
||||
|
||||
public UpdateStatus? StatusOf(NodeId node) => _status.TryGetValue(node.Value, out UpdateStatus? status) ? status : null;
|
||||
|
||||
public void Forget(NodeId node) => _status.TryRemove(node.Value, out _);
|
||||
|
||||
private IEnumerable<NodeId> OnlineOf(string component) => _gateway.OnlineNodes.Where(n => ComponentOf(n) == component);
|
||||
|
||||
private static string? ComponentOf(NodeId node) =>
|
||||
node.Value.StartsWith(AgentIds.Prefix, StringComparison.Ordinal) ? NodeKinds.Agent
|
||||
: node.Value.StartsWith(ProbeId.Prefix, StringComparison.Ordinal) ? NodeKinds.Probe
|
||||
: null;
|
||||
|
||||
private static string Key(string component, string part) => $"release.{component}.{part}";
|
||||
|
||||
private static bool TryParse(string manifest, string signature, [NotNullWhen(true)] out ReleaseOffer? offer, [NotNullWhen(false)] out string? problem)
|
||||
{
|
||||
offer = null;
|
||||
problem = null;
|
||||
if (manifest.Length is 0 or > MaxManifestBytes || signature.Length is 0 or > MaxSignatureBytes || !Base64().IsMatch(signature))
|
||||
{
|
||||
problem = "manifest or signature has an unexpected size or encoding";
|
||||
return false;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using JsonDocument document = JsonDocument.Parse(manifest);
|
||||
JsonElement root = document.RootElement;
|
||||
string component = Text(root, "component");
|
||||
string digest = Text(root, "digest");
|
||||
if (root.ValueKind != JsonValueKind.Object || Array.IndexOf(s_components, component) < 0
|
||||
|| !root.TryGetProperty("seq", out JsonElement seq) || !seq.TryGetInt64(out long seqValue) || seqValue <= 0
|
||||
|| !Digest().IsMatch(digest) || Text(root, "image").Length is 0 or > 200)
|
||||
{
|
||||
problem = "manifest needs a known component, a positive seq, an image and a sha256 digest";
|
||||
return false;
|
||||
}
|
||||
|
||||
offer = new ReleaseOffer(component, seqValue, Text(root, "tag"), Text(root, "version"), digest, manifest, signature);
|
||||
return true;
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
problem = "manifest is not valid JSON";
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static string Text(JsonElement root, string name) =>
|
||||
root.ValueKind == JsonValueKind.Object && root.TryGetProperty(name, out JsonElement value) && value.ValueKind == JsonValueKind.String ? value.GetString() ?? "" : "";
|
||||
|
||||
[GeneratedRegex("^[A-Za-z0-9+/=]+$")]
|
||||
private static partial Regex Base64();
|
||||
|
||||
[GeneratedRegex("^sha256:[0-9a-f]{64}$")]
|
||||
private static partial Regex Digest();
|
||||
}
|
||||
@@ -26,6 +26,8 @@ public sealed class FleetGateway : IAgentGateway
|
||||
|
||||
public IReadOnlyCollection<NodeId> OnlineAgents => [.. Connections.OnlineNodes.Where(n => n.Value.StartsWith(AgentIds.Prefix, StringComparison.Ordinal))];
|
||||
|
||||
public IReadOnlyCollection<NodeId> OnlineNodes => Connections.OnlineNodes;
|
||||
|
||||
public bool IsOnline(NodeId agent) => Connections.IsOnline(agent);
|
||||
|
||||
public void NotifyTaskSetChanged(NodeId agent) => Notifier.NotifyChanged(agent);
|
||||
@@ -61,4 +63,10 @@ public sealed class FleetGateway : IAgentGateway
|
||||
}
|
||||
|
||||
public ValueTask DisconnectAsync(NodeId agent, string reason, CancellationToken ct) => Connections.DisconnectAsync(agent, reason, ct);
|
||||
|
||||
public async ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct) =>
|
||||
await Connections.NotifyAsync(node, MonitorMethods.UpdateOffer, new UpdateOfferParams { Manifest = manifest, Signature = signature }, MonitorJsonContext.Default.UpdateOfferParams, ct);
|
||||
|
||||
public async ValueTask<bool> NotifyUpdateApplyAsync(NodeId node, CancellationToken ct) =>
|
||||
await Connections.NotifyAsync(node, MonitorMethods.UpdateApply, new UpdateOfferParams(), MonitorJsonContext.Default.UpdateOfferParams, ct);
|
||||
}
|
||||
@@ -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");
|
||||
|
||||
@@ -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;
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -48,6 +48,13 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
|
||||
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; } = [];
|
||||
@@ -188,6 +195,13 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
new TestStatusParams { TestId = test, Status = status, Phase = phase, ErrorCode = errorCode },
|
||||
MonitorJsonContext.Default.TestStatusParams));
|
||||
|
||||
/// <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);
|
||||
|
||||
public async Task<TaskSet> WaitForTaskSetAsync(Func<TaskSet, bool> predicate, TimeSpan? timeout = null)
|
||||
@@ -389,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);
|
||||
|
||||
Reference in new issue
Block a user