feat: node updates by restart, auto by default; Podman Quadlet install from the panel alongside compose [CI]
ci/woodpecker/push/main-amd64-build Pipeline was successful
ci/woodpecker/push/main-amd64-build Pipeline was successful
This commit is contained in:
justme committed
1 parent
88b1a0a9bb
commit
535bb3517d
32 files changed
+751
-152
No files matched your search
@@ -0,0 +1,153 @@
|
||||
using System.Globalization;
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
using MoShiro.Protocol;
|
||||
|
||||
namespace MoShiro.Measurement;
|
||||
|
||||
/// <summary>
|
||||
/// 节点的「更新」：面板建议、请求，节点<b>只负责有序退出</b>，换新版本由外面的服务管理器完成——
|
||||
/// 例如 Podman 的 Quadlet 单元写了 <c>Pull=always</c>，每次启动都先从镜像仓库拉最新的镜像，所以退出再被拉起就是更新。
|
||||
/// 节点自己不下载、不安装、不执行任何东西，所以不需要验证什么：最多让面板（管理员）请求一次重启，和定时重启是同一种能力。
|
||||
/// <list type="bullet">
|
||||
/// <item>只有设了环境变量 <c>MOSHIRO_UPDATE=restart</c> 才启用（Quadlet 模板里写好了）：别的部署方式退出之后不会重新拉取，开了也不会更新，所以没有启用就不报告更新状态。</item>
|
||||
/// <item>策略由环境变量 <c>MOSHIRO_UPDATE_POLICY</c> 决定：<c>auto</c>（默认）发现新版本就自己退出重启，<c>manual</c> 等管理端请求。自动更新有保护：
|
||||
/// 进程至少运行 <see cref="MinimumUptime"/> 才会自动退出，新镜像没拉到（标签没变、仓库还没更新）最多每个这么久重启一次，不会陷入重启循环。</item>
|
||||
/// <item>面板转来的发布清单（面板登记时已验过 CI 签名）只用来显示「有没有新版本」：组件对、比这个程序新（seq 晚于构建时间）、tag 和提交不同。</item>
|
||||
/// </list>
|
||||
/// </summary>
|
||||
public sealed class RestartUpdater
|
||||
{
|
||||
public const string EnvironmentVariable = "MOSHIRO_UPDATE";
|
||||
public const string PolicyVariable = "MOSHIRO_UPDATE_POLICY";
|
||||
|
||||
/// <summary>自动更新之前，进程至少要运行这么久。</summary>
|
||||
public static readonly TimeSpan MinimumUptime = TimeSpan.FromMinutes(10);
|
||||
|
||||
private const int MaxManifestBytes = 4096;
|
||||
|
||||
private readonly string _component;
|
||||
private readonly Action<string> _requestRestart;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly INodeLog _log;
|
||||
private readonly string _tag;
|
||||
private readonly string _commit;
|
||||
private readonly long _builtAt;
|
||||
private readonly object _gate = new();
|
||||
private readonly bool _auto;
|
||||
private int _autoScheduled;
|
||||
private readonly long _startedAt;
|
||||
private UpdateStatus _status;
|
||||
|
||||
/// <param name="tag">这个程序的构建标签、提交和构建时间（秒）：默认取 <see cref="BuildInfo"/>，测试可以传入。</param>
|
||||
public RestartUpdater(string component, Action<string> requestRestart, TimeProvider time, INodeLog log, string? tag = null, string? commit = null, long? builtAtEpoch = null, bool auto = false)
|
||||
{
|
||||
_auto = auto;
|
||||
_startedAt = time.GetTimestamp();
|
||||
_component = component;
|
||||
_requestRestart = requestRestart;
|
||||
_time = time;
|
||||
_log = log;
|
||||
_tag = tag ?? BuildInfo.Tag;
|
||||
_commit = commit ?? BuildInfo.Commit;
|
||||
_builtAt = builtAtEpoch ?? (DateTimeOffset.TryParse(BuildInfo.BuiltAt, CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal, out DateTimeOffset built) ? built.ToUnixTimeSeconds() : 0);
|
||||
_status = new UpdateStatus { State = "up_to_date", At = Now() };
|
||||
}
|
||||
|
||||
/// <summary>环境变量 <c>MOSHIRO_UPDATE=restart</c> 才启用；否则返回 null，节点不报告更新状态。</summary>
|
||||
public static RestartUpdater? FromEnvironment(string component, Action<string> requestRestart, TimeProvider time, INodeLog log) =>
|
||||
string.Equals(Environment.GetEnvironmentVariable(EnvironmentVariable), "restart", StringComparison.Ordinal)
|
||||
? new RestartUpdater(component, requestRestart, time, log, auto: !string.Equals(Environment.GetEnvironmentVariable(PolicyVariable), "manual", StringComparison.OrdinalIgnoreCase))
|
||||
: null;
|
||||
|
||||
/// <summary>面板转来的发布清单：只用来判断有没有新版本。</summary>
|
||||
public void Offer(UpdateOfferParams offer)
|
||||
{
|
||||
if (offer.Manifest.Length is 0 or > MaxManifestBytes)
|
||||
{
|
||||
_log.Warn("ignoring an update offer of an unexpected size");
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using JsonDocument document = JsonDocument.Parse(offer.Manifest);
|
||||
JsonElement root = document.RootElement;
|
||||
if (root.ValueKind != JsonValueKind.Object || Text(root, "component") != _component || !root.TryGetProperty("seq", out JsonElement seqValue) || !seqValue.TryGetInt64(out long seq))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
string tag = Text(root, "tag");
|
||||
string commit = Text(root, "commit");
|
||||
bool same = (_tag.Length > 0 && tag == _tag) || (_commit.Length > 0 && commit.Length > 0 && commit == _commit);
|
||||
bool available;
|
||||
lock (_gate)
|
||||
{
|
||||
_status = same || seq <= _builtAt
|
||||
? new UpdateStatus { State = "up_to_date", At = Now() }
|
||||
: new UpdateStatus
|
||||
{
|
||||
State = "available",
|
||||
Available = new UpdateAvailable { Tag = tag, Version = Text(root, "version"), Digest = Text(root, "digest"), Seq = seq },
|
||||
At = Now(),
|
||||
};
|
||||
available = _status.State == "available";
|
||||
}
|
||||
|
||||
if (available && _auto && Interlocked.Exchange(ref _autoScheduled, 1) == 0)
|
||||
{
|
||||
_ = ApplyAfterMinimumUptimeAsync();
|
||||
}
|
||||
}
|
||||
catch (JsonException)
|
||||
{
|
||||
_log.Warn("ignoring an update offer that is not valid JSON");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>自动更新：等进程运行够久再退出（刚启动就发现新版本，通常是重启之后标签还没变，立刻再退出会变成重启循环）。</summary>
|
||||
private async Task ApplyAfterMinimumUptimeAsync()
|
||||
{
|
||||
TimeSpan wait = MinimumUptime - _time.GetElapsedTime(_startedAt);
|
||||
if (wait > TimeSpan.Zero)
|
||||
{
|
||||
_log.Info("update available, restarting automatically later", ("in", wait.ToString("c", CultureInfo.InvariantCulture)));
|
||||
await Task.Delay(wait, _time);
|
||||
}
|
||||
|
||||
RequestApply();
|
||||
}
|
||||
|
||||
/// <summary>面板请求更新：有新版本才退出（由外面的服务管理器拉起并拉取最新镜像）。</summary>
|
||||
public bool RequestApply()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_status.State != "available")
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
_status = new UpdateStatus { State = "applying", Available = _status.Available, At = Now() };
|
||||
}
|
||||
|
||||
_log.Info("update requested, restarting so the supervisor pulls the latest image");
|
||||
_requestRestart("update");
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>随心跳上报的状态。</summary>
|
||||
public UpdateStatus Status()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
return new UpdateStatus { State = _status.State, Current = _status.Current, Available = _status.Available, Error = _status.Error, At = _status.At };
|
||||
}
|
||||
}
|
||||
|
||||
private long Now() => _time.GetUtcNow().ToUnixTimeSeconds();
|
||||
|
||||
private static string Text(JsonElement root, string name) =>
|
||||
root.TryGetProperty(name, out JsonElement value) && value.ValueKind == JsonValueKind.String ? value.GetString() ?? "" : "";
|
||||
}
|
||||
@@ -1,100 +0,0 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
using MoShiro.Protocol;
|
||||
|
||||
namespace MoShiro.Measurement;
|
||||
|
||||
/// <summary>
|
||||
/// 节点进程和宿主机更新器之间的文件交换（容器里 <c>/etc/moshiro</c>，也就是数据目录）：
|
||||
/// 节点把面板发来的清单和签名写进 <c>update/</c>，把「更新」请求写成 <c>update/apply.request</c>；更新器验签并更新，状态由宿主更新器写在只读挂载的 <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), MoShiroJsonContext.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);
|
||||
}
|
||||
@@ -5,6 +5,7 @@
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageReference Include="xunit.v3" />
|
||||
<PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" />
|
||||
<ProjectReference Include="..\..\MoShiro.Measurement\MoShiro.Measurement.csproj" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
using Fleet.Core;
|
||||
using Microsoft.Extensions.Time.Testing;
|
||||
using MoShiro.Protocol;
|
||||
|
||||
namespace MoShiro.Measurement.Tests;
|
||||
|
||||
public sealed class RestartUpdaterTests
|
||||
{
|
||||
private readonly List<string> _restarts = [];
|
||||
|
||||
private RestartUpdater Create(string tag = "dev-aaa", string commit = "aaa", long builtAt = 1_000) =>
|
||||
new("agent", _restarts.Add, TimeProvider.System, NullNodeLog.Instance, tag, commit, builtAt);
|
||||
|
||||
private static UpdateOfferParams Offer(string tag, long seq, string component = "agent", string commit = "bbb") =>
|
||||
new() { Manifest = $"{{\"component\":\"{component}\",\"tag\":\"{tag}\",\"version\":\"0.4.0\",\"commit\":\"{commit}\",\"digest\":\"sha256:{new string('a', 64)}\",\"seq\":{seq}}}", Signature = "ignored" };
|
||||
|
||||
[Fact]
|
||||
public void ANewerRelease_IsAvailable_AndOnlyARequestRestartsTheNode()
|
||||
{
|
||||
RestartUpdater updater = Create();
|
||||
updater.Offer(Offer("dev-bbb", 2_000));
|
||||
UpdateStatus status = updater.Status();
|
||||
Assert.Equal("available", status.State);
|
||||
Assert.Equal("dev-bbb", status.Available!.Tag);
|
||||
Assert.Equal(2_000, status.Available.Seq);
|
||||
Assert.Empty(_restarts);
|
||||
|
||||
Assert.True(updater.RequestApply());
|
||||
Assert.Equal(["update"], _restarts);
|
||||
Assert.Equal("applying", updater.Status().State);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void TheReleaseThatIsAlreadyRunning_IsUpToDate_ByTagOrByCommit()
|
||||
{
|
||||
RestartUpdater updater = Create();
|
||||
updater.Offer(Offer("dev-aaa", 2_000));
|
||||
Assert.Equal("up_to_date", updater.Status().State);
|
||||
updater.Offer(Offer("dev-other-tag", 2_000, commit: "aaa"));
|
||||
Assert.Equal("up_to_date", updater.Status().State);
|
||||
Assert.False(updater.RequestApply());
|
||||
Assert.Empty(_restarts);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void AnOfferOlderThanThisBuild_OrForAnotherComponent_IsNotAnUpdate()
|
||||
{
|
||||
RestartUpdater updater = Create(builtAt: 5_000);
|
||||
updater.Offer(Offer("dev-bbb", 2_000));
|
||||
Assert.Equal("up_to_date", updater.Status().State);
|
||||
updater.Offer(Offer("dev-ccc", 9_000, component: "probe"));
|
||||
Assert.Equal("up_to_date", updater.Status().State);
|
||||
Assert.False(updater.RequestApply());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void GarbageOffers_AreIgnored()
|
||||
{
|
||||
RestartUpdater updater = Create();
|
||||
updater.Offer(new UpdateOfferParams { Manifest = "not json", Signature = "x" });
|
||||
updater.Offer(new UpdateOfferParams { Manifest = new string('x', 5000), Signature = "x" });
|
||||
updater.Offer(new UpdateOfferParams { Manifest = "{\"component\":\"agent\"}", Signature = "x" });
|
||||
Assert.Equal("up_to_date", updater.Status().State);
|
||||
Assert.False(updater.RequestApply());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TheAutoPolicy_RestartsByItself_ButNotBeforeTheMinimumUptime()
|
||||
{
|
||||
var time = new FakeTimeProvider(DateTimeOffset.FromUnixTimeSeconds(2_000_000_000));
|
||||
var updater = new RestartUpdater("agent", _restarts.Add, time, NullNodeLog.Instance, "dev-aaa", "aaa", 1_000, auto: true);
|
||||
updater.Offer(Offer("dev-bbb", 2_000));
|
||||
updater.Offer(Offer("dev-bbb", 2_000)); // 重连时面板会再发一次：不能排两个
|
||||
Assert.Empty(_restarts);
|
||||
time.Advance(TimeSpan.FromMinutes(9));
|
||||
await Task.Delay(50);
|
||||
Assert.Empty(_restarts);
|
||||
time.Advance(TimeSpan.FromMinutes(2));
|
||||
await WaitAsync(() => _restarts.Count == 1);
|
||||
Assert.Equal("applying", updater.Status().State);
|
||||
await Task.Delay(50);
|
||||
Assert.Single(_restarts);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task TheManualPolicy_NeverRestartsByItself()
|
||||
{
|
||||
var time = new FakeTimeProvider(DateTimeOffset.FromUnixTimeSeconds(2_000_000_000));
|
||||
var updater = new RestartUpdater("agent", _restarts.Add, time, NullNodeLog.Instance, "dev-aaa", "aaa", 1_000, auto: false);
|
||||
updater.Offer(Offer("dev-bbb", 2_000));
|
||||
time.Advance(TimeSpan.FromHours(1));
|
||||
await Task.Delay(50);
|
||||
Assert.Empty(_restarts);
|
||||
Assert.Equal("available", updater.Status().State);
|
||||
}
|
||||
|
||||
private static async Task WaitAsync(Func<bool> condition)
|
||||
{
|
||||
for (int i = 0; i < 100 && !condition(); i++)
|
||||
{
|
||||
await Task.Delay(20);
|
||||
}
|
||||
|
||||
Assert.True(condition());
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void WithoutTheEnvironmentVariable_ThereIsNoUpdater()
|
||||
{
|
||||
Assert.Null(Environment.GetEnvironmentVariable(RestartUpdater.EnvironmentVariable));
|
||||
Assert.Null(RestartUpdater.FromEnvironment("agent", _restarts.Add, TimeProvider.System, NullNodeLog.Instance));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user