feat: daily node restart controlled from the admin, connect timings as a default line task [CI]
ci/woodpecker/push/main-amd64-build Pipeline was successful
ci/woodpecker/push/main-amd64-build Pipeline was successful
Agents and Probes exit in order at a configured time (default 04:30 UTC+8, switch and time zone in the admin settings) and the service manager starts them again; a local restart_at or MOSHIRO_RESTART_AT wins. The connect kind (TCP connect and TLS handshake timings) is now a default continuous line template, with a section on the line page.
This commit is contained in:
justme committed
1 parent
388619b8b6
commit
9bcef56497
37 files changed
+928
-101
No files matched your search
@@ -16,6 +16,7 @@ public sealed class AgentApp : IDisposable
|
||||
|
||||
private readonly AgentConfig _config;
|
||||
private readonly bool _ipv6Supported;
|
||||
private readonly RestartController? _restart;
|
||||
private readonly LoadedNodeIdentity _identity;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly INodeLog _log;
|
||||
@@ -37,8 +38,9 @@ public sealed class AgentApp : IDisposable
|
||||
private readonly FleetLink _link;
|
||||
|
||||
/// <param name="dataDirectory">配置目录（容器里的 <c>/etc/moshiro</c>）：宿主机更新器和节点在这里交换清单、请求和状态。</param>
|
||||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log)
|
||||
public AgentApp(AgentConfig config, LoadedNodeIdentity identity, string dataDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
|
||||
{
|
||||
_restart = restart;
|
||||
_update = new UpdateBridge(dataDirectory, new ConfigStore(dataDirectory).Write, log);
|
||||
_config = config;
|
||||
_identity = identity;
|
||||
@@ -219,6 +221,19 @@ public sealed class AgentApp : IDisposable
|
||||
});
|
||||
}
|
||||
|
||||
if (Allow(MoShiroMethods.Schedule) && _restart is not null)
|
||||
{
|
||||
router.OnNotify(MoShiroMethods.Schedule, MoShiroJsonContext.Default.ScheduleParams, (_, schedule, _) =>
|
||||
{
|
||||
if (!_restart.SetRemote(schedule.RestartAt))
|
||||
{
|
||||
_log.Warn("ignoring an invalid restart schedule from the panel");
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
}
|
||||
|
||||
if (Allow(MoShiroMethods.UpdateApply))
|
||||
{
|
||||
router.OnNotify(MoShiroMethods.UpdateApply, MoShiroJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||||
|
||||
@@ -22,6 +22,10 @@ public sealed class AgentConfig
|
||||
|
||||
[JsonPropertyName("reporting")]
|
||||
public ReportingSection Reporting { get; set; } = new();
|
||||
|
||||
/// <summary>每天定时重启一次，例如 <c>04:00+08:00</c>（不写偏移按 UTC）；不写就不重启。环境变量 <c>MOSHIRO_RESTART_AT</c> 优先。</summary>
|
||||
[JsonPropertyName("restart_at")]
|
||||
public string? RestartAt { get; set; }
|
||||
}
|
||||
|
||||
public sealed class ReportingSection
|
||||
|
||||
@@ -22,9 +22,11 @@ for (int i = 0; i < args.Length - 1; i++)
|
||||
|
||||
AgentConfig config;
|
||||
LoadedNodeIdentity identity;
|
||||
RestartController restart;
|
||||
try
|
||||
{
|
||||
config = NodeConfig.Load(configPath, AgentConfigContext.Default.AgentConfig);
|
||||
restart = new RestartController(RestartSchedule.Resolve(config.RestartAt));
|
||||
identity = LoadedNodeIdentity.Load(config.Fleet.IdentityDir);
|
||||
}
|
||||
catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidOperationException or System.Text.Json.JsonException or System.Security.Cryptography.CryptographicException)
|
||||
@@ -33,10 +35,11 @@ catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidO
|
||||
return 1;
|
||||
}
|
||||
|
||||
using var restartScope = restart;
|
||||
using (identity)
|
||||
{
|
||||
using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log);
|
||||
await ServiceShell.RunAsync(app.RunAsync, log);
|
||||
using var app = new AgentApp(config, identity, Path.GetDirectoryName(Path.GetFullPath(configPath))!, time, log, restart);
|
||||
await ServiceShell.RunAsync(app.RunAsync, log, restart, time);
|
||||
}
|
||||
|
||||
return 0;
|
||||
@@ -126,7 +126,8 @@ public static class PlatformCheck
|
||||
/// <summary>处理 Ctrl+C 和 SIGTERM：取消令牌，让各个模块有序退出。</summary>
|
||||
public static class ServiceShell
|
||||
{
|
||||
public static async Task RunAsync(Func<CancellationToken, Task> run, INodeLog log)
|
||||
/// <param name="restart">每天定时有序退出一次（见 <see cref="RestartController"/>），重新拉起由服务管理器负责。</param>
|
||||
public static async Task RunAsync(Func<CancellationToken, Task> run, INodeLog log, RestartController? restart = null, TimeProvider? time = null)
|
||||
{
|
||||
using var cts = new CancellationTokenSource();
|
||||
using PosixSignalRegistration term = PosixSignalRegistration.Create(PosixSignal.SIGTERM, context =>
|
||||
@@ -156,6 +157,7 @@ public static class ServiceShell
|
||||
cts.Cancel();
|
||||
});
|
||||
Task trimmer = TrimHeapAsync(cts.Token);
|
||||
Task scheduled = restart is null ? Task.CompletedTask : WaitForRestartAsync(restart, time ?? TimeProvider.System, log, cts);
|
||||
try
|
||||
{
|
||||
await run(cts.Token);
|
||||
@@ -168,6 +170,38 @@ public static class ServiceShell
|
||||
{
|
||||
await cts.CancelAsync();
|
||||
await trimmer;
|
||||
await scheduled;
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task WaitForRestartAsync(RestartController restart, TimeProvider time, INodeLog log, CancellationTokenSource cts)
|
||||
{
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
// 先取令牌、再取设置：中间被改动的话令牌会被取消，下一轮重新排
|
||||
CancellationToken changed = restart.ChangeToken;
|
||||
RestartSchedule? schedule = restart.Current;
|
||||
using CancellationTokenSource wait = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, changed);
|
||||
try
|
||||
{
|
||||
if (schedule is null)
|
||||
{
|
||||
await Task.Delay(Timeout.Infinite, wait.Token);
|
||||
continue;
|
||||
}
|
||||
|
||||
DateTimeOffset next = schedule.NextAfter(time.GetUtcNow() + RestartSchedule.MinimumUptime);
|
||||
log.Info("restart scheduled", ("at", next.ToString("O", System.Globalization.CultureInfo.InvariantCulture)), ("source", restart.IsLocal ? "local" : "panel"));
|
||||
await Task.Delay(next - time.GetUtcNow(), time, wait.Token);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
continue; // 进程在退出，或者设置被改了：循环条件和下一轮会处理
|
||||
}
|
||||
|
||||
log.Info("stopping", ("result", "scheduled_restart"));
|
||||
await cts.CancelAsync();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
namespace Fleet.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// 当前生效的定时重启：节点本地的设置优先；本地没有写时用面板下发的，面板改了之后立即重新排下一次，不需要重启进程。
|
||||
/// </summary>
|
||||
public sealed class RestartController : IDisposable
|
||||
{
|
||||
private readonly RestartSchedule? _local;
|
||||
private readonly object _gate = new();
|
||||
private RestartSchedule? _remote;
|
||||
private CancellationTokenSource _changed = new();
|
||||
|
||||
public RestartController(RestartSchedule? local = null)
|
||||
{
|
||||
_local = local;
|
||||
}
|
||||
|
||||
/// <summary>本地有设置时为 true：面板下发的不生效。</summary>
|
||||
public bool IsLocal => _local is not null;
|
||||
|
||||
public RestartSchedule? Current
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
return _local ?? _remote;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>当前设置被改动时会被取消的令牌；先取令牌、再取 <see cref="Current"/>，就不会漏掉中间的改动。</summary>
|
||||
public CancellationToken ChangeToken
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
return _changed.Token;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>面板下发的设置；为空表示面板关闭了定时重启。返回 false 表示文本不合法（节点忽略它）。</summary>
|
||||
public bool SetRemote(string? text)
|
||||
{
|
||||
RestartSchedule? schedule = null;
|
||||
if (!string.IsNullOrWhiteSpace(text) && !RestartSchedule.TryParse(text, out schedule))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
CancellationTokenSource previous;
|
||||
lock (_gate)
|
||||
{
|
||||
_remote = schedule;
|
||||
previous = _changed;
|
||||
_changed = new CancellationTokenSource();
|
||||
}
|
||||
|
||||
previous.Cancel();
|
||||
previous.Dispose();
|
||||
return true;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_changed.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
using System.Globalization;
|
||||
|
||||
namespace Fleet.Hosting;
|
||||
|
||||
/// <summary>
|
||||
/// 每天固定时刻有序退出一次，由 systemd、容器引擎或 procd 的「总是重启」拉起。写法 <c>HH:mm</c>（按 UTC）、
|
||||
/// <c>HH:mm±hh:mm</c> 或 <c>HH:mm</c> 加 <c>Z</c>：容器里没有时区数据库，所以偏移要写明，例如 <c>04:00+08:00</c>。
|
||||
/// 进程自己退出，不需要任何执行外部命令的能力。
|
||||
/// </summary>
|
||||
public sealed class RestartSchedule
|
||||
{
|
||||
/// <summary>环境变量，设置了就覆盖配置文件里的值。</summary>
|
||||
public const string EnvironmentVariable = "MOSHIRO_RESTART_AT";
|
||||
|
||||
/// <summary>刚启动的一段时间内不会触发：重启之后不会因为时刻还没过去而再重启一次，也避免重启风暴。</summary>
|
||||
public static readonly TimeSpan MinimumUptime = TimeSpan.FromMinutes(10);
|
||||
|
||||
private readonly TimeSpan _time;
|
||||
private readonly TimeSpan _offset;
|
||||
|
||||
private RestartSchedule(TimeSpan time, TimeSpan offset)
|
||||
{
|
||||
_time = time;
|
||||
_offset = offset;
|
||||
}
|
||||
|
||||
public static bool TryParse(string? text, out RestartSchedule? schedule)
|
||||
{
|
||||
schedule = null;
|
||||
string value = text?.Trim() ?? "";
|
||||
// HH:mm，后面可以跟 Z 或 ±hh:mm
|
||||
if (value.Length < 5 || value[2] != ':' ||
|
||||
!TryTwoDigits(value, 0, out int hour) || !TryTwoDigits(value, 3, out int minute) || hour > 23 || minute > 59)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
TimeSpan offset = TimeSpan.Zero;
|
||||
string rest = value[5..];
|
||||
if (rest.Length > 0 && rest != "Z")
|
||||
{
|
||||
if (rest.Length != 6 || rest[0] is not ('+' or '-') || rest[3] != ':' ||
|
||||
!TryTwoDigits(rest, 1, out int offsetHours) || !TryTwoDigits(rest, 4, out int offsetMinutes) || offsetMinutes > 59)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
offset = new TimeSpan(offsetHours, offsetMinutes, 0);
|
||||
if (offset > TimeSpan.FromHours(14))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if (rest[0] == '-')
|
||||
{
|
||||
offset = -offset;
|
||||
}
|
||||
}
|
||||
|
||||
schedule = new RestartSchedule(new TimeSpan(hour, minute, 0), offset);
|
||||
return true;
|
||||
}
|
||||
|
||||
private static bool TryTwoDigits(string text, int start, out int value)
|
||||
{
|
||||
value = 0;
|
||||
if (!char.IsAsciiDigit(text[start]) || !char.IsAsciiDigit(text[start + 1]))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
value = (text[start] - '0') * 10 + (text[start + 1] - '0');
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 环境变量优先，其次配置文件的值；都没有就不定时重启。写错了抛 <see cref="InvalidOperationException"/>，
|
||||
/// 启动时就能看到，而不是悄悄不生效。
|
||||
/// </summary>
|
||||
public static RestartSchedule? Resolve(string? configured)
|
||||
{
|
||||
string? text = Environment.GetEnvironmentVariable(EnvironmentVariable);
|
||||
if (string.IsNullOrWhiteSpace(text))
|
||||
{
|
||||
text = configured;
|
||||
}
|
||||
|
||||
if (string.IsNullOrWhiteSpace(text))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return TryParse(text, out RestartSchedule? schedule)
|
||||
? schedule
|
||||
: throw new InvalidOperationException($"restart_at must look like 04:00, 04:00Z or 04:00+08:00 (got '{text}')");
|
||||
}
|
||||
|
||||
/// <summary>严格晚于 <paramref name="now"/> 的下一个触发时刻。</summary>
|
||||
public DateTimeOffset NextAfter(DateTimeOffset now)
|
||||
{
|
||||
DateTimeOffset local = now.ToOffset(_offset);
|
||||
DateTimeOffset candidate = new DateTimeOffset(local.Year, local.Month, local.Day, 0, 0, 0, _offset) + _time;
|
||||
return candidate <= local ? candidate.AddDays(1) : candidate;
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@
|
||||
<ProjectReference Include="..\..\src\Fleet.Identity\Fleet.Identity.csproj" />
|
||||
<ProjectReference Include="..\..\src\Fleet.Rpc\Fleet.Rpc.csproj" />
|
||||
<ProjectReference Include="..\..\src\Fleet.Link\Fleet.Link.csproj" />
|
||||
<ProjectReference Include="..\..\src\Fleet.Hosting\Fleet.Hosting.csproj" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<Using Include="Xunit" />
|
||||
|
||||
@@ -0,0 +1,260 @@
|
||||
using Fleet.Core;
|
||||
using Fleet.Hosting;
|
||||
|
||||
namespace Fleet.Tests;
|
||||
|
||||
public sealed class RestartScheduleTests
|
||||
{
|
||||
private static RestartSchedule Parse(string text)
|
||||
{
|
||||
Assert.True(RestartSchedule.TryParse(text, out RestartSchedule? schedule), text);
|
||||
return schedule!;
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData("04:00")]
|
||||
[InlineData("04:00Z")]
|
||||
[InlineData("04:00+08:00")]
|
||||
[InlineData("23:59-05:30")]
|
||||
[InlineData(" 00:00 ")]
|
||||
public void Parse_AcceptsTimeWithOptionalOffset(string text) => Parse(text);
|
||||
|
||||
[Theory]
|
||||
[InlineData("")]
|
||||
[InlineData("4:00")]
|
||||
[InlineData("24:00")]
|
||||
[InlineData("04:60")]
|
||||
[InlineData("04:00+8:00")]
|
||||
[InlineData("04:00+15:00")]
|
||||
[InlineData("04:00 UTC")]
|
||||
[InlineData("0400")]
|
||||
[InlineData("04:00+08:60")]
|
||||
public void Parse_RejectsEverythingElse(string text) => Assert.False(RestartSchedule.TryParse(text, out _));
|
||||
|
||||
[Fact]
|
||||
public void NextAfter_IsStrictlyLater_AndRollsToTomorrow()
|
||||
{
|
||||
RestartSchedule utc = Parse("04:00");
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 7, 4, 0, 0, TimeSpan.Zero), utc.NextAfter(new DateTimeOffset(2026, 10, 7, 3, 59, 59, TimeSpan.Zero)));
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 8, 4, 0, 0, TimeSpan.Zero), utc.NextAfter(new DateTimeOffset(2026, 10, 7, 4, 0, 0, TimeSpan.Zero)));
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 8, 4, 0, 0, TimeSpan.Zero), utc.NextAfter(new DateTimeOffset(2026, 10, 7, 20, 0, 0, TimeSpan.Zero)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NextAfter_UsesTheWrittenOffset()
|
||||
{
|
||||
// 04:00+08:00 就是 UTC 前一天 20:00
|
||||
RestartSchedule china = Parse("04:00+08:00");
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 6, 20, 0, 0, TimeSpan.Zero), china.NextAfter(new DateTimeOffset(2026, 10, 6, 19, 0, 0, TimeSpan.Zero)));
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 7, 20, 0, 0, TimeSpan.Zero), china.NextAfter(new DateTimeOffset(2026, 10, 6, 20, 0, 0, TimeSpan.Zero)));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void NextAfter_WithMinimumUptime_DoesNotRestartAgainRightAfterARestart()
|
||||
{
|
||||
RestartSchedule schedule = Parse("04:00");
|
||||
// 04:00:03 重启后启动：加上最短运行时间再算，下一次是明天，而不是几秒后
|
||||
var started = new DateTimeOffset(2026, 10, 7, 4, 0, 3, TimeSpan.Zero);
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 8, 4, 0, 0, TimeSpan.Zero), schedule.NextAfter(started + RestartSchedule.MinimumUptime));
|
||||
// 03:55 启动：离 04:00 只有 5 分钟，不够最短运行时间，顺延到明天
|
||||
var early = new DateTimeOffset(2026, 10, 7, 3, 55, 0, TimeSpan.Zero);
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 8, 4, 0, 0, TimeSpan.Zero), schedule.NextAfter(early + RestartSchedule.MinimumUptime));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Resolve_ReportsABadValueInsteadOfIgnoringIt()
|
||||
{
|
||||
Assert.Null(RestartSchedule.Resolve(null));
|
||||
Assert.Null(RestartSchedule.Resolve(" "));
|
||||
InvalidOperationException error = Assert.Throws<InvalidOperationException>(() => RestartSchedule.Resolve("tomorrow"));
|
||||
Assert.Contains("restart_at", error.Message);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ServiceShell_StopsAtTheScheduledTime()
|
||||
{
|
||||
var time = new ImmediateTimeProvider(new DateTimeOffset(2026, 10, 7, 12, 0, 0, TimeSpan.Zero));
|
||||
var log = new ListLog();
|
||||
bool stoppedByToken = false;
|
||||
|
||||
// 业务循环只在被取消时返回；定时重启到点以后令牌被取消，RunAsync 正常返回
|
||||
await ServiceShell.RunAsync(async ct =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(Timeout.Infinite, ct);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
stoppedByToken = true;
|
||||
throw;
|
||||
}
|
||||
}, log, new RestartController(Parse("04:00")), time).WaitAsync(TimeSpan.FromSeconds(10));
|
||||
|
||||
Assert.True(stoppedByToken);
|
||||
Assert.Contains(log.Lines, l => l.Contains("scheduled_restart", StringComparison.Ordinal));
|
||||
Assert.Contains(log.Lines, l => l.Contains("restart scheduled", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ServiceShell_PicksUpAScheduleThePanelSendsLater()
|
||||
{
|
||||
var time = new ImmediateTimeProvider(new DateTimeOffset(2026, 10, 7, 12, 0, 0, TimeSpan.Zero));
|
||||
var log = new ListLog();
|
||||
using var controller = new RestartController();
|
||||
|
||||
Task shell = ServiceShell.RunAsync(ct => Task.Delay(Timeout.Infinite, ct), log, controller, time);
|
||||
await Task.Delay(300);
|
||||
Assert.False(shell.IsCompleted, "no schedule yet: the node keeps running");
|
||||
|
||||
Assert.True(controller.SetRemote("04:30+08:00"));
|
||||
await shell.WaitAsync(TimeSpan.FromSeconds(10));
|
||||
Assert.Contains(log.Lines, l => l.Contains("scheduled_restart", StringComparison.Ordinal));
|
||||
Assert.Contains(log.Lines, l => l.Contains("source=panel", StringComparison.Ordinal));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ServiceShell_ClearingTheScheduleCancelsTheRestart()
|
||||
{
|
||||
var time = new ManualTimeProvider(new DateTimeOffset(2026, 10, 7, 12, 0, 0, TimeSpan.Zero));
|
||||
var log = new ListLog();
|
||||
using var controller = new RestartController();
|
||||
|
||||
controller.SetRemote("04:30+08:00");
|
||||
|
||||
Task shell = ServiceShell.RunAsync(ct => Task.Delay(Timeout.Infinite, ct), log, controller, time);
|
||||
await Task.Delay(300);
|
||||
controller.SetRemote(null); // 面板关掉了
|
||||
await Task.Delay(300);
|
||||
Assert.False(shell.IsCompleted);
|
||||
Assert.Contains(log.Lines, l => l.Contains("restart scheduled", StringComparison.Ordinal));
|
||||
Assert.DoesNotContain(log.Lines, l => l.Contains("scheduled_restart", StringComparison.Ordinal));
|
||||
// 清理：把进程停掉
|
||||
Assert.True(controller.SetRemote("04:30+08:00"));
|
||||
time.Advance();
|
||||
await shell.WaitAsync(TimeSpan.FromSeconds(10));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Controller_LocalSettingWinsOverThePanel_AndBadTextIsRejected()
|
||||
{
|
||||
using var controller = new RestartController(Parse("04:00"));
|
||||
Assert.True(controller.IsLocal);
|
||||
Assert.True(controller.SetRemote("05:00"));
|
||||
Assert.Equal(new DateTimeOffset(2026, 10, 7, 4, 0, 0, TimeSpan.Zero), controller.Current!.NextAfter(new DateTimeOffset(2026, 10, 7, 0, 0, 0, TimeSpan.Zero)));
|
||||
|
||||
using var remoteOnly = new RestartController();
|
||||
Assert.Null(remoteOnly.Current);
|
||||
Assert.False(remoteOnly.SetRemote("whenever"));
|
||||
Assert.Null(remoteOnly.Current);
|
||||
Assert.True(remoteOnly.SetRemote("05:00"));
|
||||
Assert.NotNull(remoteOnly.Current);
|
||||
Assert.True(remoteOnly.SetRemote(null));
|
||||
Assert.Null(remoteOnly.Current);
|
||||
}
|
||||
|
||||
/// <summary>计时器马上触发：用来验证「到点」之后的行为，不用真的等。</summary>
|
||||
private sealed class ImmediateTimeProvider : TimeProvider
|
||||
{
|
||||
private readonly DateTimeOffset _now;
|
||||
|
||||
public ImmediateTimeProvider(DateTimeOffset now) => _now = now;
|
||||
|
||||
public override DateTimeOffset GetUtcNow() => _now;
|
||||
|
||||
public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period)
|
||||
{
|
||||
// TrimHeapAsync 这类每 30 秒一次的循环也会走到这里；只让「很长」的延迟（定时重启，至少 10 分钟）马上触发
|
||||
var timer = new ImmediateTimer(callback, state);
|
||||
if (dueTime >= RestartSchedule.MinimumUptime)
|
||||
{
|
||||
ThreadPool.QueueUserWorkItem(_ => timer.Fire());
|
||||
}
|
||||
|
||||
return timer;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>计时器只在 <see cref="Advance"/> 时触发：用来验证「还没到点」时不会退出。</summary>
|
||||
private sealed class ManualTimeProvider : TimeProvider
|
||||
{
|
||||
private readonly DateTimeOffset _now;
|
||||
private readonly List<ImmediateTimer> _timers = [];
|
||||
|
||||
public ManualTimeProvider(DateTimeOffset now) => _now = now;
|
||||
|
||||
public override DateTimeOffset GetUtcNow() => _now;
|
||||
|
||||
public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period)
|
||||
{
|
||||
var timer = new ImmediateTimer(callback, state);
|
||||
lock (_timers)
|
||||
{
|
||||
_timers.Add(timer);
|
||||
}
|
||||
|
||||
return timer;
|
||||
}
|
||||
|
||||
public void Advance()
|
||||
{
|
||||
ImmediateTimer[] due;
|
||||
lock (_timers)
|
||||
{
|
||||
due = [.. _timers];
|
||||
_timers.Clear();
|
||||
}
|
||||
|
||||
foreach (ImmediateTimer timer in due)
|
||||
{
|
||||
timer.Fire();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class ImmediateTimer : ITimer
|
||||
{
|
||||
private readonly TimerCallback _callback;
|
||||
private readonly object? _state;
|
||||
|
||||
public ImmediateTimer(TimerCallback callback, object? state)
|
||||
{
|
||||
_callback = callback;
|
||||
_state = state;
|
||||
}
|
||||
|
||||
public void Fire() => _callback(_state);
|
||||
|
||||
public bool Change(TimeSpan dueTime, TimeSpan period) => true;
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
}
|
||||
|
||||
public ValueTask DisposeAsync() => ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
private sealed class ListLog : INodeLog
|
||||
{
|
||||
private readonly List<string> _lines = [];
|
||||
|
||||
public IReadOnlyList<string> Lines
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_lines)
|
||||
{
|
||||
return [.. _lines];
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields)
|
||||
{
|
||||
lock (_lines)
|
||||
{
|
||||
_lines.Add(message + " " + string.Join(' ', fields.Select(f => f.Key + "=" + f.Value)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -23,7 +23,7 @@ public sealed class ProbeApp : IDisposable
|
||||
private readonly INodeLog _log;
|
||||
private readonly FleetLink _link;
|
||||
|
||||
public ProbeApp(ProbeFile file, string configDirectory, TimeProvider time, INodeLog log)
|
||||
public ProbeApp(ProbeFile file, string configDirectory, TimeProvider time, INodeLog log, RestartController? restart = null)
|
||||
{
|
||||
_log = log;
|
||||
var directory = new AgentDirectory(new ConfigStore(configDirectory), configDirectory, log);
|
||||
@@ -59,6 +59,15 @@ public sealed class ProbeApp : IDisposable
|
||||
.OnNotify(MoShiroMethods.UpdateApply, MoShiroJsonContext.Default.UpdateOfferParams, (_, _, _) =>
|
||||
{
|
||||
update.RequestApply();
|
||||
return ValueTask.CompletedTask;
|
||||
})
|
||||
.OnNotify(MoShiroMethods.Schedule, MoShiroJsonContext.Default.ScheduleParams, (_, schedule, _) =>
|
||||
{
|
||||
if (restart is not null && !restart.SetRemote(schedule.RestartAt))
|
||||
{
|
||||
log.Warn("ignoring an invalid restart schedule from the panel");
|
||||
}
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
});
|
||||
_link = new FleetLink(options, router, [list], new Outbox(1 << 20), time, log);
|
||||
|
||||
@@ -20,6 +20,10 @@ public sealed class ProbeFile
|
||||
[JsonPropertyName("server")]
|
||||
public string Server { get; set; } = "";
|
||||
|
||||
/// <summary>每天定时重启一次，例如 <c>04:00+08:00</c>（不写偏移按 UTC）；不写就不重启。环境变量 <c>MOSHIRO_RESTART_AT</c> 优先；这里是本地配置，从面板重新下载 probe.json 会丢掉它，所以容器里推荐用环境变量。</summary>
|
||||
[JsonPropertyName("restart_at")]
|
||||
public string? RestartAt { get; set; }
|
||||
|
||||
[JsonPropertyName("limits")]
|
||||
public ProbeLimits Limits { get; set; } = new();
|
||||
}
|
||||
|
||||
@@ -21,9 +21,11 @@ for (int i = 0; i < args.Length - 1; i++)
|
||||
}
|
||||
|
||||
ProbeFile file;
|
||||
RestartController restart;
|
||||
try
|
||||
{
|
||||
file = NodeConfig.Load(configPath, ProbeConfigContext.Default.ProbeFile);
|
||||
restart = new RestartController(RestartSchedule.Resolve(file.RestartAt));
|
||||
if (file.ProbeId.Length == 0 || file.PrivateKey.Length == 0 || !Uri.TryCreate(file.Server, UriKind.Absolute, out _))
|
||||
{
|
||||
throw new InvalidOperationException("probe.json needs probe_id, private_key and server (download it from the panel: Probes → Create)");
|
||||
@@ -35,11 +37,12 @@ catch (Exception ex) when (ex is IOException or InvalidDataException or InvalidO
|
||||
return 1;
|
||||
}
|
||||
|
||||
using var restartScope = restart;
|
||||
string directory = Path.GetDirectoryName(Path.GetFullPath(configPath))!;
|
||||
try
|
||||
{
|
||||
using var app = new ProbeApp(file, directory, time, log);
|
||||
await ServiceShell.RunAsync(app.RunAsync, log);
|
||||
using var app = new ProbeApp(file, directory, time, log, restart);
|
||||
await ServiceShell.RunAsync(app.RunAsync, log, restart, time);
|
||||
}
|
||||
catch (Exception ex) when (ex is FormatException or System.Security.Cryptography.CryptographicException)
|
||||
{
|
||||
|
||||
@@ -15,6 +15,7 @@ public static class MoShiroMethods
|
||||
public const string Host = "moshiro.host";
|
||||
public const string UpdateOffer = "moshiro.update.offer";
|
||||
public const string UpdateApply = "moshiro.update.apply";
|
||||
public const string Schedule = "moshiro.schedule";
|
||||
|
||||
// Probe ↔ Agent（点对点会话）
|
||||
public const string Clock = "moshiro.clock";
|
||||
@@ -238,6 +239,17 @@ public sealed class UpdateOfferParams
|
||||
public string Signature { get; set; } = "";
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 面板给节点的定时重启设置（<c>moshiro.schedule</c> 通知）：<c>restart_at</c> 形如 <c>04:30+08:00</c>，为空表示不定时重启。
|
||||
/// 节点连上时补发、设置变化时群发；节点本地写了 <c>restart_at</c> 或环境变量 <c>MOSHIRO_RESTART_AT</c> 时以本地为准。
|
||||
/// 节点只是到点自己有序退出，由服务管理器拉起，不执行任何命令。
|
||||
/// </summary>
|
||||
public sealed class ScheduleParams
|
||||
{
|
||||
[JsonPropertyName("restart_at")]
|
||||
public string? RestartAt { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>宿主机更新器已经验证过的可用更新。</summary>
|
||||
public sealed class UpdateAvailable
|
||||
{
|
||||
@@ -560,6 +572,7 @@ public static class FragmentKeys
|
||||
[JsonSerializable(typeof(AgentEventParams))]
|
||||
[JsonSerializable(typeof(ProbeAgentList))]
|
||||
[JsonSerializable(typeof(UpdateOfferParams))]
|
||||
[JsonSerializable(typeof(ScheduleParams))]
|
||||
[JsonSerializable(typeof(UpdateStatus))]
|
||||
[JsonSerializable(typeof(ProbeHeartbeatExtra))]
|
||||
[JsonSerializable(typeof(ProbeHelloMeta))]
|
||||
|
||||
@@ -63,9 +63,14 @@ public static class SystemEndpoints
|
||||
private static async Task<IResult> ListSettingsAsync(SettingsService settings, CancellationToken ct) =>
|
||||
TypedResults.Json(new PageDto<SettingDto>([.. (await settings.ListAsync(ct)).Select(Mappers.ToDto)], null), ApiJsonContext.Default.PageDtoSettingDto);
|
||||
|
||||
private static async Task<IResult> PatchSettingsAsync(HttpContext http, SettingsService settings, AgentService agents, Dictionary<string, JsonElement> changes)
|
||||
private static async Task<IResult> PatchSettingsAsync(HttpContext http, SettingsService settings, AgentService agents, ScheduleService schedule, Dictionary<string, JsonElement> changes)
|
||||
{
|
||||
Result<IReadOnlyList<SettingView>> result = await settings.PatchAsync(changes, http.RequestAborted);
|
||||
if (result.IsSuccess && changes.Keys.Any(k => k.StartsWith("fleet.restart_", StringComparison.Ordinal)))
|
||||
{
|
||||
await schedule.SendToAllAsync(http.RequestAborted); // 定时重启设置变了：马上通知所有在线节点
|
||||
}
|
||||
|
||||
if (result.IsSuccess && changes.ContainsKey(SettingsService.AutoLink))
|
||||
{
|
||||
await agents.AfterAgentSetChangedAsync(http.RequestAborted); // 自动连接开关变了：重新下发各 Probe 的 Agent 列表和各 Agent 允许的 Probe
|
||||
|
||||
@@ -14,6 +14,10 @@ public sealed class SettingsService
|
||||
public const string SitePublic = "site.public";
|
||||
public const string SiteShowBuild = "site.show_build";
|
||||
public const string AutoLink = "fleet.auto_link";
|
||||
public const string RestartEnabled = "fleet.restart_enabled";
|
||||
public const string RestartHour = "fleet.restart_hour";
|
||||
public const string RestartMinute = "fleet.restart_minute";
|
||||
public const string RestartOffsetMinutes = "fleet.restart_utc_offset_minutes";
|
||||
public const string RawDays = "retention.raw_days";
|
||||
public const string TestRawDays = "retention.test_raw_days";
|
||||
public const string Metric10sHours = "retention.metric_10s_hours";
|
||||
@@ -44,6 +48,10 @@ public sealed class SettingsService
|
||||
new(SitePublic, "bool", "false", 0, 1, "Whether anonymous visitors can read the public API."),
|
||||
new(SiteShowBuild, "bool", "false", 0, 1, "Whether the public pages show the versions, tags and build times of the backend, agents and probes."),
|
||||
new(AutoLink, "bool", "true", 0, 1, "Whether every probe connects to every agent automatically; when off, only the links chosen by hand are connected."),
|
||||
new(RestartEnabled, "bool", "true", 0, 1, "Whether agents and probes restart themselves once a day (they exit in order and the service manager starts them again); a local restart_at on the node wins."),
|
||||
new(RestartHour, "int", "4", 0, 23, "Hour of the daily restart, in the time zone below."),
|
||||
new(RestartMinute, "int", "30", 0, 59, "Minute of the daily restart."),
|
||||
new(RestartOffsetMinutes, "int", "480", -840, 840, "Time zone of the daily restart as minutes east of UTC (480 = Beijing)."),
|
||||
new(RawDays, "int", o.RawRetentionDays.ToString(CultureInfo.InvariantCulture), 1, 365, "Days to keep raw chunks of continuous tasks."),
|
||||
new(TestRawDays, "int", o.TestRawRetentionDays.ToString(CultureInfo.InvariantCulture), 1, 365, "Days to keep raw chunks of on-demand tests."),
|
||||
new(Metric10sHours, "int", o.Metric10sRetentionHours.ToString(CultureInfo.InvariantCulture), 1, 720, "Hours to keep the 10 s metric layer."),
|
||||
@@ -121,6 +129,12 @@ public sealed class SettingsService
|
||||
/// <summary>访客能不能看到版本、tag 和构建时间（站点公开之后的另一道开关，默认关）。</summary>
|
||||
public async ValueTask<bool> IsShowBuildAsync(CancellationToken ct) => await _store.GetAsync(SiteShowBuild, ct) == "true";
|
||||
|
||||
public async ValueTask<bool> GetBoolAsync(string key, CancellationToken ct)
|
||||
{
|
||||
SettingSpec spec = Specs.First(s => s.Key == key);
|
||||
return (await _store.GetAsync(key, ct) ?? spec.Default) == "true";
|
||||
}
|
||||
|
||||
public async ValueTask<int> GetIntAsync(string key, CancellationToken ct)
|
||||
{
|
||||
SettingSpec spec = Specs.First(s => s.Key == key);
|
||||
|
||||
@@ -53,11 +53,13 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
||||
private readonly AgentService _service;
|
||||
private readonly NodeInfoService _nodes;
|
||||
private readonly ReleaseService _releases;
|
||||
private readonly ScheduleService _schedule;
|
||||
private readonly IProbeStore _probes;
|
||||
|
||||
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, IProbeStore probes)
|
||||
public MoShiroNodeEventSink(IAgentStore agents, AgentRuntime runtime, IEventStore events, LiveHub live, AgentService service, NodeInfoService nodes, ReleaseService releases, ScheduleService schedule, IProbeStore probes)
|
||||
{
|
||||
_releases = releases;
|
||||
_schedule = schedule;
|
||||
_probes = probes;
|
||||
_nodes = nodes;
|
||||
_agents = agents;
|
||||
@@ -144,6 +146,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
||||
}
|
||||
|
||||
await _releases.OfferToNodeAsync(id, ct);
|
||||
await _schedule.SendToNodeAsync(id, ct);
|
||||
break;
|
||||
case NodeEventKind.Offline:
|
||||
_live.Publish(new NodeSignal("probe", id.Value, false, nodeEvent.AtMs));
|
||||
@@ -214,6 +217,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
|
||||
AgentRecord? after = await _agents.GetAsync(id, ct);
|
||||
_nodes.ObserveAgentInBackground(id.Value, after?.PublicHost, meta.Platform);
|
||||
await _releases.OfferToNodeAsync(id, ct);
|
||||
await _schedule.SendToNodeAsync(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) || before.Ipv6Supported != after.Ipv6Supported)
|
||||
&& after.Status == NodeStatus.Active)
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
using System.Globalization;
|
||||
using Fleet.Core;
|
||||
|
||||
namespace MoShiro.Backend.Application;
|
||||
|
||||
/// <summary>
|
||||
/// 节点的定时重启：设置在管理端（开关、时刻、时区偏移），节点连上时补发、设置变化时群发给所有在线节点。
|
||||
/// 面板只告诉节点「几点退出」，节点自己到点有序退出、由服务管理器拉起；节点本地写了设置时以本地为准。
|
||||
/// </summary>
|
||||
public sealed class ScheduleService
|
||||
{
|
||||
private readonly SettingsService _settings;
|
||||
private readonly IAgentGateway _gateway;
|
||||
|
||||
public ScheduleService(SettingsService settings, IAgentGateway gateway)
|
||||
{
|
||||
_settings = settings;
|
||||
_gateway = gateway;
|
||||
}
|
||||
|
||||
/// <summary>形如 <c>04:30+08:00</c>；开关关着时为空。</summary>
|
||||
public async ValueTask<string?> RestartAtAsync(CancellationToken ct)
|
||||
{
|
||||
if (!await _settings.GetBoolAsync(SettingsService.RestartEnabled, ct))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return Format(
|
||||
await _settings.GetIntAsync(SettingsService.RestartHour, ct),
|
||||
await _settings.GetIntAsync(SettingsService.RestartMinute, ct),
|
||||
await _settings.GetIntAsync(SettingsService.RestartOffsetMinutes, ct));
|
||||
}
|
||||
|
||||
public static string Format(int hour, int minute, int offsetMinutes)
|
||||
{
|
||||
int abs = Math.Abs(offsetMinutes);
|
||||
return string.Create(CultureInfo.InvariantCulture, $"{hour:00}:{minute:00}{(offsetMinutes < 0 ? '-' : '+')}{abs / 60:00}:{abs % 60:00}");
|
||||
}
|
||||
|
||||
/// <summary>节点刚连上：补发当前设置（节点离线时改的设置也不会错过）。</summary>
|
||||
public async ValueTask SendToNodeAsync(NodeId node, CancellationToken ct) =>
|
||||
await _gateway.NotifyScheduleAsync(node, await RestartAtAsync(ct), ct);
|
||||
|
||||
/// <summary>管理端改了设置：通知所有在线节点。</summary>
|
||||
public async ValueTask SendToAllAsync(CancellationToken ct)
|
||||
{
|
||||
string? restartAt = await RestartAtAsync(ct);
|
||||
foreach (NodeId node in _gateway.OnlineNodes)
|
||||
{
|
||||
await _gateway.NotifyScheduleAsync(node, restartAt, ct);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,7 @@ public static class ApplicationServiceCollectionExtensions
|
||||
services.AddSingleton<AuthService>();
|
||||
services.AddSingleton<SettingsService>();
|
||||
services.AddSingleton<ReleaseService>();
|
||||
services.AddSingleton<ScheduleService>();
|
||||
services.AddSingleton<BackendUpdates>();
|
||||
services.AddHostedService(p => p.GetRequiredService<BackendUpdates>());
|
||||
services.AddSingleton<JoinTokenService>();
|
||||
|
||||
@@ -33,6 +33,9 @@ public interface IAgentGateway
|
||||
/// <summary>把发布清单和签名原样交给节点（通知，不等回复）。</summary>
|
||||
ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct);
|
||||
|
||||
/// <summary>把定时重启设置交给节点（通知，不等回复）；<paramref name="restartAt"/> 为空表示关闭。</summary>
|
||||
ValueTask NotifyScheduleAsync(NodeId node, string? restartAt, CancellationToken ct);
|
||||
|
||||
/// <summary>请求节点更新到它手里的清单。节点不在线返回 false。</summary>
|
||||
ValueTask<bool> NotifyUpdateApplyAsync(NodeId node, CancellationToken ct);
|
||||
}
|
||||
@@ -40,23 +40,35 @@ public sealed class TaskService
|
||||
|
||||
public ValueTask<TaskRecord?> GetAsync(TaskId id, CancellationToken ct) => _tasks.GetAsync(id, ct);
|
||||
|
||||
/// <summary>首次启动时创建默认的线路任务模板（udp_echo、tcp_echo 持续运行），这样新 Agent 注册后自动获得任务。</summary>
|
||||
/// <summary>
|
||||
/// 创建默认的线路任务模板，这样新 Agent 注册后自动获得任务：udp_echo、tcp_echo 持续运行，connect（TCP 建连和 TLS 握手耗时，
|
||||
/// 每分钟 6 次）也持续运行。每一批默认任务只种一次，用设置里的标记记住；你删掉的默认模板不会被重新建出来，
|
||||
/// 后来新增的默认任务（connect）在已有部署升级后会补建一次。
|
||||
/// </summary>
|
||||
public async ValueTask SeedDefaultsAsync(CancellationToken ct)
|
||||
{
|
||||
if (await _settings.GetAsync("tasks.seeded", ct) is not null)
|
||||
if (await _settings.GetAsync("tasks.seeded", ct) is null)
|
||||
{
|
||||
return;
|
||||
await SeedAsync([Kinds.UdpEcho, Kinds.TcpEcho], ct);
|
||||
await _settings.SetAsync("tasks.seeded", "1", ct);
|
||||
}
|
||||
|
||||
if (await _settings.GetAsync("tasks.seeded.connect", ct) is null)
|
||||
{
|
||||
await SeedAsync([Kinds.Connect], ct);
|
||||
await _settings.SetAsync("tasks.seeded.connect", "1", ct);
|
||||
}
|
||||
}
|
||||
|
||||
private async ValueTask SeedAsync(KindSpec[] kinds, CancellationToken ct)
|
||||
{
|
||||
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
|
||||
foreach (KindSpec kind in new[] { Kinds.UdpEcho, Kinds.TcpEcho })
|
||||
foreach (KindSpec kind in kinds)
|
||||
{
|
||||
await _tasks.CreateAsync(
|
||||
new TaskRecord(new TaskId(Ulid.NewPrefixed(TaskId.Prefix, _time)), EffectiveTasks.LineScope, null, null, kind.Id, null, new Dictionary<string, JsonElement>(), TaskSchedule.Continuous, true, now, 0),
|
||||
ct);
|
||||
}
|
||||
|
||||
await _settings.SetAsync("tasks.seeded", "1", ct);
|
||||
}
|
||||
|
||||
public async ValueTask<Result<TaskRecord>> CreateAsync(TaskInput input, CancellationToken ct)
|
||||
|
||||
@@ -67,6 +67,9 @@ public sealed class FleetGateway : IAgentGateway
|
||||
public async ValueTask NotifyUpdateOfferAsync(NodeId node, string manifest, string signature, CancellationToken ct) =>
|
||||
await Connections.NotifyAsync(node, MoShiroMethods.UpdateOffer, new UpdateOfferParams { Manifest = manifest, Signature = signature }, MoShiroJsonContext.Default.UpdateOfferParams, ct);
|
||||
|
||||
public async ValueTask NotifyScheduleAsync(NodeId node, string? restartAt, CancellationToken ct) =>
|
||||
await Connections.NotifyAsync(node, MoShiroMethods.Schedule, new ScheduleParams { RestartAt = restartAt }, MoShiroJsonContext.Default.ScheduleParams, ct);
|
||||
|
||||
public async ValueTask<bool> NotifyUpdateApplyAsync(NodeId node, CancellationToken ct) =>
|
||||
await Connections.NotifyAsync(node, MoShiroMethods.UpdateApply, new UpdateOfferParams(), MoShiroJsonContext.Default.UpdateOfferParams, ct);
|
||||
}
|
||||
@@ -44,7 +44,7 @@ public sealed class AgentEndToEndTests
|
||||
{
|
||||
// 默认模板的端口（7443/7444）不在这个 Agent 实际监听的端口里：任务被拒绝，原因是 policy_denied
|
||||
await using AgentHarness agent = await AgentHarness.StartAsync(_backend);
|
||||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == 2);
|
||||
await BackendFixture.EventuallyAsync(async () => (await AgentViewAsync(agent))["rejected"]!.AsArray().Count == Defaults.TaskCount);
|
||||
JsonNode rejected = (await AgentViewAsync(agent))["rejected"]!;
|
||||
Assert.All(rejected.AsArray(), item => Assert.Equal("policy_denied", item!["reason"]!.GetValue<string>()));
|
||||
|
||||
|
||||
@@ -88,16 +88,19 @@ public sealed class AgentHarness : IAsyncDisposable
|
||||
public async Task OverrideLineTasksAsync(int? rateHz = null)
|
||||
{
|
||||
JsonNode tasks = await Backend.GetJsonAsync(Backend.Admin, "/api/admin/tasks");
|
||||
foreach (JsonNode? task in tasks["items"]!.AsArray())
|
||||
// 先处理 connect（关掉），再改 udp_echo / tcp_echo 的速率：否则中途的带宽预算会多出 connect 那一份
|
||||
foreach (JsonNode? task in tasks["items"]!.AsArray().OrderBy(t => t!["kind"]!.GetValue<string>() == "connect" ? 0 : 1))
|
||||
{
|
||||
if (task!["scope"]!.GetValue<string>() != "line" || task["agent_id"] is not null || task["kind"]!.GetValue<string>() is not ("udp_echo" or "tcp_echo"))
|
||||
if (task!["scope"]!.GetValue<string>() != "line" || task["agent_id"] is not null || task["kind"]!.GetValue<string>() is not ("udp_echo" or "tcp_echo" or "connect"))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
string kind = task["kind"]!.GetValue<string>();
|
||||
var parameters = new JsonObject { ["port"] = kind == "udp_echo" ? App.UdpPorts[0] : App.TcpPorts[0] };
|
||||
if (rateHz is { } hz)
|
||||
// 默认的 connect 模板在这些用例里不需要（也占预算）：对这个 Agent 关掉它
|
||||
bool enabled = kind != "connect";
|
||||
if (rateHz is { } hz && kind != "connect")
|
||||
{
|
||||
parameters["rate_hz"] = hz;
|
||||
}
|
||||
@@ -109,6 +112,7 @@ public sealed class AgentHarness : IAsyncDisposable
|
||||
["agent_id"] = Id,
|
||||
["base_task_id"] = task["id"]!.GetValue<string>(),
|
||||
["params"] = parameters,
|
||||
["enabled"] = enabled,
|
||||
}));
|
||||
Assert.True(created.StatusCode == System.Net.HttpStatusCode.Created, await created.Content.ReadAsStringAsync());
|
||||
}
|
||||
|
||||
@@ -10,6 +10,12 @@ using MoShiro.TestSupport;
|
||||
|
||||
namespace MoShiro.Backend.Tests;
|
||||
|
||||
/// <summary>首次启动种入的默认线路任务模板数：udp_echo、tcp_echo、connect。</summary>
|
||||
internal static class Defaults
|
||||
{
|
||||
public const int TaskCount = 3;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 在真实的 Kestrel 端口上启动整个后端（真实的 SQLite 与段文件，只是用临时目录和较快的轮询），
|
||||
/// 并提供管理员客户端（Bearer 令牌）和假 Agent。
|
||||
|
||||
@@ -24,8 +24,8 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
|
||||
Assert.Equal("active", agent.EnrollmentStatus);
|
||||
Assert.Equal("active", agent.AuthStatus);
|
||||
TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
Assert.Equal(["tcp_echo", "udp_echo"], set.Tasks.Select(t => t.Kind).Order().ToArray());
|
||||
TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
Assert.Equal(["connect", "tcp_echo", "udp_echo"], set.Tasks.Select(t => t.Kind).Order().ToArray());
|
||||
Assert.All(set.Tasks, t =>
|
||||
{
|
||||
Assert.Equal("line", t.Scope);
|
||||
@@ -56,7 +56,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
public async Task Reconnecting_AlwaysGetsTheCurrentSnapshotAgain()
|
||||
{
|
||||
AgentSimulator first = await _backend.ConnectAgentAsync();
|
||||
TaskSet before = await first.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
TaskSet before = await first.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
await first.DisposeAsync();
|
||||
|
||||
// 同一个身份重新连接：新连接的服务端状态为空，所以会重新推送当前版本
|
||||
@@ -157,7 +157,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
|
||||
using HttpResponseMessage approve = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.NodeId.Value}/approve", null);
|
||||
Assert.Equal(HttpStatusCode.OK, approve.StatusCode);
|
||||
TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
TaskSet set = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
Assert.NotNull(set);
|
||||
}
|
||||
|
||||
@@ -165,7 +165,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
public async Task DisablingAConnectedAgent_RevokesItAndRefusesFutureConnections()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
|
||||
using HttpResponseMessage disable = await _backend.Admin.PostAsync($"/api/admin/agents/{agent.NodeId.Value}/disable", null);
|
||||
Assert.Equal(HttpStatusCode.OK, disable.StatusCode);
|
||||
@@ -239,7 +239,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
public async Task TaskOverride_ChangesTheDesiredStateAndBumpsTheVersion()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
TaskSet initial = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
TaskSet initial = await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
TaskSpec udp = initial.Tasks.Single(t => t.Kind == "udp_echo");
|
||||
|
||||
using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject
|
||||
@@ -254,7 +254,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
Assert.StartsWith("/api/admin/tasks/", created.Headers.Location!.ToString());
|
||||
|
||||
TaskSet updated = await agent.WaitForTaskSetAsync(s => s.Version > initial.Version && s.Tasks.Any(t => t.Kind == "udp_echo" && t.Params["rate_hz"].GetInt64() == 20));
|
||||
Assert.Equal(2, updated.Tasks.Count);
|
||||
Assert.Equal(Defaults.TaskCount, updated.Tasks.Count);
|
||||
Assert.Equal(7443, updated.Tasks.Single(t => t.Kind == "udp_echo").Params["port"].GetInt64());
|
||||
|
||||
// 停用覆盖 → 该 Agent 不再运行这条任务（模板对其他 Agent 不受影响）
|
||||
@@ -262,7 +262,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
using HttpResponseMessage disable = await _backend.Admin.PatchAsync("/api/admin/tasks/" + overrideId, BackendFixture.Json(new JsonObject { ["enabled"] = false }, "application/merge-patch+json"));
|
||||
Assert.Equal(HttpStatusCode.OK, disable.StatusCode);
|
||||
TaskSet disabled = await agent.WaitForTaskSetAsync(s => s.Version > updated.Version && s.Tasks.All(t => t.Kind != "udp_echo"));
|
||||
Assert.Single(disabled.Tasks);
|
||||
Assert.Equal(Defaults.TaskCount - 1, disabled.Tasks.Count);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -314,7 +314,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
public async Task TaskSet_RejectedItemsReportedByTheAgent_AreVisibleToTheAdmin()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
agent.RejectNextState = [new Fleet.Rpc.RejectedItem { Id = "tsk_x", Reason = "policy_denied" }];
|
||||
|
||||
using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/tasks", BackendFixture.Json(new JsonObject
|
||||
@@ -330,7 +330,7 @@ public sealed class EnrollmentAndTaskSetTests
|
||||
public async Task ProbeLifecycle_CreateRotateRevokeDelete_IsReflectedInEveryAgentsPeerList()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
|
||||
using HttpResponseMessage created = await _backend.Admin.PostAsync("/api/admin/probes", BackendFixture.Json(new JsonObject { ["name"] = _backend.UniqueName("home") }));
|
||||
Assert.Equal(HttpStatusCode.Created, created.StatusCode);
|
||||
|
||||
@@ -25,7 +25,7 @@ public sealed class OnDemandTestsTests
|
||||
private async Task<AgentSimulator> AgentWithProbeAsync()
|
||||
{
|
||||
AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
await agent.SendEventAsync(AgentEventKinds.ProbeConnected, _backend.DefaultProbeId);
|
||||
await BackendFixture.EventuallyAsync(async () => (await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.NodeId.Value))["probe_connected"]!.GetValue<bool>());
|
||||
return agent;
|
||||
@@ -37,7 +37,7 @@ public sealed class OnDemandTestsTests
|
||||
public async Task Start_RequiresAProbeSession_APlanThatExists_AndValidParams()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
|
||||
using HttpResponseMessage offline = await StartAsync(agent, Plan("hy2_ladder"));
|
||||
Assert.Equal(HttpStatusCode.Conflict, offline.StatusCode);
|
||||
|
||||
@@ -139,7 +139,7 @@ public sealed class OperationsAndSecurityTests
|
||||
public async Task OversizedFrames_CloseTheConnection_WithoutHarmingTheServer()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
try
|
||||
{
|
||||
await agent.SendBinaryAsync(new byte[17 * 1024 * 1024]);
|
||||
@@ -153,14 +153,14 @@ public sealed class OperationsAndSecurityTests
|
||||
|
||||
// 服务端仍然正常
|
||||
await using AgentSimulator another = await _backend.ConnectAgentAsync();
|
||||
Assert.NotNull(await another.WaitForTaskSetAsync(s => s.Tasks.Count == 2));
|
||||
Assert.NotNull(await another.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task MalformedAndUnknownMessages_AreIgnoredOrRefused_AndTheConnectionStaysUsable()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
|
||||
await agent.SendTextAsync("{not json");
|
||||
await agent.SendTextAsync("[]");
|
||||
@@ -187,7 +187,7 @@ public sealed class OperationsAndSecurityTests
|
||||
Assert.Empty(agent.TaskSets);
|
||||
|
||||
await agent.HelloAsync();
|
||||
Assert.NotNull(await agent.WaitForTaskSetAsync(s => s.Tasks.Count == 2));
|
||||
Assert.NotNull(await agent.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
@@ -205,9 +205,9 @@ public sealed class OperationsAndSecurityTests
|
||||
public async Task ReconnectingTheSameNode_ReplacesTheOldConnection()
|
||||
{
|
||||
AgentSimulator first = await _backend.ConnectAgentAsync();
|
||||
await first.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await first.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
await using AgentSimulator second = await _backend.ReconnectAsync(first);
|
||||
await second.WaitForTaskSetAsync(s => s.Tasks.Count == 2);
|
||||
await second.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount);
|
||||
|
||||
// 旧连接被替换掉，但节点仍然在线（在线状态属于新连接）
|
||||
await first.WaitUntilAsync(() => first.CloseReason is not null);
|
||||
|
||||
@@ -57,7 +57,7 @@ public sealed class PersistenceTests : IAsyncLifetime
|
||||
|
||||
// 同一个身份可以重新连接，并拿回当前版本的期望状态
|
||||
await using AgentSimulator again = await _backend.ReconnectAsync(agent);
|
||||
Assert.NotNull(await again.WaitForTaskSetAsync(s => s.Tasks.Count == 2));
|
||||
Assert.NotNull(await again.WaitForTaskSetAsync(s => s.Tasks.Count == Defaults.TaskCount));
|
||||
|
||||
// 已经落盘的块在重启后仍然可以按 ID 下载
|
||||
JsonNode chunks = await _backend.GetJsonAsync(admin, $"/api/admin/raw/chunks?agent_id={id}&limit=1");
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
using System.Net;
|
||||
using System.Text.Json.Nodes;
|
||||
using MoShiro.Backend.Application;
|
||||
using MoShiro.Protocol;
|
||||
using MoShiro.TestSupport;
|
||||
|
||||
namespace MoShiro.Backend.Tests;
|
||||
|
||||
/// <summary>定时重启：默认开、北京时间 04:30；节点连上时补发，管理端改了设置马上群发，关掉就发空。</summary>
|
||||
[Collection("probe-e2e")]
|
||||
public sealed class ScheduleTests
|
||||
{
|
||||
private readonly BackendFixture _backend;
|
||||
|
||||
public ScheduleTests(BackendFixture backend)
|
||||
{
|
||||
_backend = backend;
|
||||
}
|
||||
|
||||
private async Task PatchAsync(JsonObject settings)
|
||||
{
|
||||
using HttpResponseMessage response = await _backend.Admin.PatchAsync("/api/admin/settings", BackendFixture.Json(settings, "application/merge-patch+json"));
|
||||
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
|
||||
}
|
||||
|
||||
private static Task<bool> LastIs(AgentSimulator agent, string? restartAt) =>
|
||||
Task.FromResult(agent.Schedules.ToArray() is { Length: > 0 } all && all[^1].RestartAt == restartAt);
|
||||
|
||||
[Fact]
|
||||
public void Format_WritesTheOffsetTheNodeParses()
|
||||
{
|
||||
Assert.Equal("04:30+08:00", ScheduleService.Format(4, 30, 480));
|
||||
Assert.Equal("00:00+00:00", ScheduleService.Format(0, 0, 0));
|
||||
Assert.Equal("23:05-05:30", ScheduleService.Format(23, 5, -330));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ANewNode_GetsTheDefaultSchedule_AndLaterChangesAreBroadcast()
|
||||
{
|
||||
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
|
||||
await BackendFixture.EventuallyAsync(() => LastIs(agent, "04:30+08:00"), because: "defaults: on, 04:30 Beijing time, sent right after hello");
|
||||
|
||||
try
|
||||
{
|
||||
await PatchAsync(new JsonObject { ["fleet.restart_hour"] = 3, ["fleet.restart_minute"] = 5, ["fleet.restart_utc_offset_minutes"] = 0 });
|
||||
await BackendFixture.EventuallyAsync(() => LastIs(agent, "03:05+00:00"), because: "a changed time is pushed to online nodes");
|
||||
|
||||
await PatchAsync(new JsonObject { ["fleet.restart_enabled"] = false });
|
||||
await BackendFixture.EventuallyAsync(() => LastIs(agent, null), because: "switching it off sends an empty schedule");
|
||||
|
||||
// 关着的时候新连上的节点也拿到「关」
|
||||
await using AgentSimulator late = await _backend.ConnectAgentAsync();
|
||||
await BackendFixture.EventuallyAsync(() => LastIs(late, null), because: "a node connecting while it is off is told so");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await PatchAsync(new JsonObject { ["fleet.restart_enabled"] = true, ["fleet.restart_hour"] = 4, ["fleet.restart_minute"] = 30, ["fleet.restart_utc_offset_minutes"] = 480 });
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task BadValues_AreRefused()
|
||||
{
|
||||
using HttpResponseMessage response = await _backend.Admin.PatchAsync("/api/admin/settings",
|
||||
BackendFixture.Json(new JsonObject { ["fleet.restart_hour"] = 24, ["fleet.restart_minute"] = 60, ["fleet.restart_utc_offset_minutes"] = 900 }, "application/merge-patch+json"));
|
||||
Assert.Equal(HttpStatusCode.BadRequest, response.StatusCode);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task DefaultLineTemplates_IncludeTheConnectTimings()
|
||||
{
|
||||
using HttpResponseMessage response = await _backend.Admin.GetAsync("/api/admin/tasks");
|
||||
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
|
||||
JsonArray items = JsonNode.Parse(await response.Content.ReadAsStringAsync())!["items"]!.AsArray();
|
||||
string[] kinds = [.. items.Select(i => i!["kind"]!.GetValue<string>())];
|
||||
Assert.Contains("connect", kinds);
|
||||
Assert.Contains("udp_echo", kinds);
|
||||
Assert.Contains("tcp_echo", kinds);
|
||||
}
|
||||
}
|
||||
@@ -51,6 +51,9 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
/// <summary>面板发来的发布清单（<c>moshiro.update.offer</c>）和更新请求（<c>moshiro.update.apply</c>）。</summary>
|
||||
public ConcurrentQueue<UpdateOfferParams> UpdateOffers { get; } = new();
|
||||
|
||||
/// <summary>面板下发的定时重启设置（<c>moshiro.schedule</c>），按到达顺序。</summary>
|
||||
public ConcurrentQueue<ScheduleParams> Schedules { get; } = new();
|
||||
|
||||
private int _updateApplies;
|
||||
|
||||
public int UpdateApplies => Volatile.Read(ref _updateApplies);
|
||||
@@ -406,6 +409,9 @@ public sealed class AgentSimulator : IAsyncDisposable
|
||||
case MoShiroMethods.UpdateOffer:
|
||||
UpdateOffers.Enqueue(message.Params!.Value.Deserialize(MoShiroJsonContext.Default.UpdateOfferParams)!);
|
||||
break;
|
||||
case MoShiroMethods.Schedule:
|
||||
Schedules.Enqueue(message.Params!.Value.Deserialize(MoShiroJsonContext.Default.ScheduleParams)!);
|
||||
break;
|
||||
case MoShiroMethods.UpdateApply:
|
||||
Interlocked.Increment(ref _updateApplies);
|
||||
break;
|
||||
|
||||
Reference in new issue
Block a user