363 lines
14 KiB
C#
363 lines
14 KiB
C#
using System.Globalization;
|
||
using System.Runtime.InteropServices;
|
||
using System.Security.Cryptography.X509Certificates;
|
||
using System.Text;
|
||
using System.Text.Json;
|
||
using System.Text.Json.Serialization;
|
||
using System.Text.Json.Serialization.Metadata;
|
||
using Fleet.Core;
|
||
using Fleet.Identity;
|
||
|
||
namespace Fleet.Hosting;
|
||
|
||
/// <summary>logfmt 日志：<c>ts=… level=info msg="…" key=value</c>。令牌、会话密钥和 MAC 不能出现在日志里。</summary>
|
||
public sealed class ConsoleNodeLog : INodeLog
|
||
{
|
||
private readonly TimeProvider _time;
|
||
private readonly TextWriter _writer;
|
||
private readonly object _gate = new();
|
||
|
||
public ConsoleNodeLog(TimeProvider time, TextWriter? writer = null)
|
||
{
|
||
_time = time;
|
||
_writer = writer ?? Console.Error;
|
||
}
|
||
|
||
public void Write(NodeLogLevel level, string message, params (string Key, object? Value)[] fields)
|
||
{
|
||
var line = new StringBuilder();
|
||
line.Append("ts=").Append(_time.GetUtcNow().ToString("yyyy-MM-dd'T'HH:mm:ss.fff'Z'", CultureInfo.InvariantCulture));
|
||
line.Append(" level=").Append(level.ToString().ToLowerInvariant());
|
||
line.Append(" msg=").Append(Quote(message));
|
||
foreach ((string key, object? value) in fields)
|
||
{
|
||
line.Append(' ').Append(key).Append('=').Append(Quote(Convert.ToString(value, CultureInfo.InvariantCulture) ?? ""));
|
||
}
|
||
|
||
lock (_gate)
|
||
{
|
||
_writer.WriteLine(line.ToString());
|
||
}
|
||
}
|
||
|
||
private static string Quote(string text) =>
|
||
text.Length > 0 && text.All(c => char.IsLetterOrDigit(c) || c is '_' or '-' or '.' or ':' or '/') ? text : "\"" + text.Replace("\\", "\\\\", StringComparison.Ordinal).Replace("\"", "\\\"", StringComparison.Ordinal) + "\"";
|
||
}
|
||
|
||
/// <summary>可以独立运行的功能模块（连接、数据面、切块……）。</summary>
|
||
public interface IModule
|
||
{
|
||
string Name { get; }
|
||
|
||
Task RunAsync(CancellationToken ct);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 模块宿主：同时运行所有模块，隔离故障（一个模块崩溃只重启它自己，指数退避），收到取消之后等所有模块退出。
|
||
/// 模块之间通过构造函数传入的接口或 <c>Channel<T></c> 通信，不使用静态全局状态。
|
||
/// </summary>
|
||
public sealed class ModuleHost
|
||
{
|
||
private readonly List<IModule> _modules = [];
|
||
private readonly INodeLog _log;
|
||
private readonly TimeProvider _time;
|
||
|
||
public ModuleHost(INodeLog log, TimeProvider time)
|
||
{
|
||
_log = log;
|
||
_time = time;
|
||
}
|
||
|
||
public ModuleHost Add(IModule module)
|
||
{
|
||
_modules.Add(module);
|
||
return this;
|
||
}
|
||
|
||
public Task RunAsync(CancellationToken ct) => Task.WhenAll(_modules.Select(m => RunModuleAsync(m, ct)));
|
||
|
||
private async Task RunModuleAsync(IModule module, CancellationToken ct)
|
||
{
|
||
var backoff = new Backoff(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
try
|
||
{
|
||
await module.RunAsync(ct);
|
||
return; // 模块正常结束
|
||
}
|
||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||
{
|
||
return;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
// 模块边界：一个模块的故障不能带倒整个进程
|
||
TimeSpan delay = backoff.Next();
|
||
_log.Error("module crashed", ("module", module.Name), ("err", ex.GetType().Name + ": " + ex.Message), ("retry_ms", (long)delay.TotalMilliseconds));
|
||
await Task.Delay(delay, _time, ct);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 启动自检：Linux 上 .NET 的加密和 TLS 依赖系统的 OpenSSL（<c>libssl.so.3</c>）。OpenWrt 默认用 mbedTLS/wolfSSL，需要另外安装 <c>libopenssl3</c>。
|
||
/// 缺少时在启动的第一步明确报错，而不是在第一次握手时抛出一个难以理解的异常。
|
||
/// </summary>
|
||
public static class PlatformCheck
|
||
{
|
||
/// <summary>返回 null 表示正常，否则是给人看的错误说明。</summary>
|
||
public static string? VerifyCrypto()
|
||
{
|
||
try
|
||
{
|
||
using var key = System.Security.Cryptography.ECDsa.Create(System.Security.Cryptography.ECCurve.NamedCurves.nistP256);
|
||
_ = key.SignData("check"u8, System.Security.Cryptography.HashAlgorithmName.SHA256);
|
||
return null;
|
||
}
|
||
catch (Exception ex) when (ex is DllNotFoundException or TypeInitializationException or System.Security.Cryptography.CryptographicException or PlatformNotSupportedException or InvalidOperationException)
|
||
{
|
||
return $"cannot use the system cryptography library ({ex.GetType().Name}: {ex.Message}). Install OpenSSL 3: Debian/Ubuntu `apt install libssl3`, Alpine `apk add libssl3`, OpenWrt `opkg install libopenssl3`";
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>处理 Ctrl+C 和 SIGTERM：取消令牌，让各个模块有序退出。</summary>
|
||
public static class ServiceShell
|
||
{
|
||
/// <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 =>
|
||
{
|
||
context.Cancel = true;
|
||
log.Info("stopping", ("result", "SIGTERM"));
|
||
cts.Cancel();
|
||
});
|
||
using PosixSignalRegistration interrupt = PosixSignalRegistration.Create(PosixSignal.SIGINT, context =>
|
||
{
|
||
context.Cancel = true;
|
||
log.Info("stopping", ("result", "SIGINT"));
|
||
cts.Cancel();
|
||
});
|
||
// Windows 上 Ctrl+Break 对应 SIGQUIT（没有控制台窗口的后台运行方式用它来请求有序退出）
|
||
using PosixSignalRegistration quit = PosixSignalRegistration.Create(PosixSignal.SIGQUIT, context =>
|
||
{
|
||
context.Cancel = true;
|
||
log.Info("stopping", ("result", "SIGQUIT"));
|
||
cts.Cancel();
|
||
});
|
||
// Windows 上关闭控制台窗口 / taskkill（不带 /F）对应 SIGHUP
|
||
using PosixSignalRegistration hangup = PosixSignalRegistration.Create(PosixSignal.SIGHUP, context =>
|
||
{
|
||
context.Cancel = true;
|
||
log.Info("stopping", ("result", "SIGHUP"));
|
||
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);
|
||
}
|
||
catch (OperationCanceledException) when (cts.IsCancellationRequested)
|
||
{
|
||
// 正常退出
|
||
}
|
||
finally
|
||
{
|
||
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;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 节点进程的托管堆只有几百 KB，但工作站 GC 的第 0 代预算很大，按默认设置要攒很多垃圾才回收一次，
|
||
/// 在 1 GB 内存的小机器上表现为常驻内存缓慢爬升（不是泄漏，强制回收后曲线是平的）。
|
||
/// 每 30 秒回收一次：堆这么小，一次完整回收不到 1 ms。
|
||
/// </summary>
|
||
private static async Task TrimHeapAsync(CancellationToken ct)
|
||
{
|
||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
|
||
try
|
||
{
|
||
while (await timer.WaitForNextTickAsync(ct))
|
||
{
|
||
GC.Collect(2, GCCollectionMode.Forced, blocking: true, compacting: false);
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// 退出
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>读取 JSON 配置（源生成，不使用反射）。解析失败时给出明确的错误，而不是带着默认值悄悄运行。</summary>
|
||
public static class NodeConfig
|
||
{
|
||
public static T Load<T>(string path, JsonTypeInfo<T> typeInfo)
|
||
where T : class
|
||
{
|
||
try
|
||
{
|
||
if (!File.Exists(path))
|
||
{
|
||
throw new InvalidDataException($"config file {path} does not exist");
|
||
}
|
||
|
||
byte[] bytes = File.ReadAllBytes(path);
|
||
ReadOnlySpan<byte> json = bytes;
|
||
if (json.StartsWith(""u8)) // Windows 编辑器常带 UTF-8 BOM，System.Text.Json 不接受
|
||
{
|
||
json = json[3..];
|
||
}
|
||
|
||
return JsonSerializer.Deserialize(json, typeInfo) ?? throw new InvalidDataException($"{path} is empty");
|
||
}
|
||
catch (JsonException ex)
|
||
{
|
||
throw new InvalidDataException($"{path} is not valid: {ex.Message}", ex);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 常驻业务进程里唯一允许写文件的地方，而且只能写配置/身份目录（原子替换，类 Unix 上权限 0600）。
|
||
/// 例如 Probe 把最新的 Agent 列表写回 bootstrap。测量数据本来就不落盘。
|
||
/// </summary>
|
||
public sealed class ConfigStore
|
||
{
|
||
private readonly string _directory;
|
||
|
||
public ConfigStore(string directory)
|
||
{
|
||
_directory = Path.GetFullPath(directory);
|
||
}
|
||
|
||
public void Write(string fileName, byte[] content)
|
||
{
|
||
string path = Path.GetFullPath(Path.Combine(_directory, fileName));
|
||
if (!path.StartsWith(_directory, StringComparison.Ordinal))
|
||
{
|
||
throw new InvalidOperationException("config files must stay inside the config directory");
|
||
}
|
||
|
||
// 相对路径里的子目录（例如 update/）也要建：新建的命名卷里只有镜像自带的目录
|
||
Directory.CreateDirectory(Path.GetDirectoryName(path)!);
|
||
string temp = path + ".tmp";
|
||
#pragma warning disable RS0030 // 纯采集边界的唯一豁免点：ConfigStore 只写配置目录，先写临时文件再原子替换
|
||
File.WriteAllBytes(temp, content);
|
||
#pragma warning restore RS0030
|
||
if (!OperatingSystem.IsWindows())
|
||
{
|
||
File.SetUnixFileMode(temp, UnixFileMode.UserRead | UnixFileMode.UserWrite);
|
||
}
|
||
|
||
File.Move(temp, path, overwrite: true);
|
||
}
|
||
}
|
||
|
||
/// <summary><c>node.json</c>：<c>fleet-enroll</c> 写入，节点只读。</summary>
|
||
public sealed class NodeFile
|
||
{
|
||
[JsonPropertyName("node_id")]
|
||
public string NodeId { get; set; } = "";
|
||
|
||
[JsonPropertyName("server")]
|
||
public string Server { get; set; } = "";
|
||
|
||
/// <summary>本节点 TLS 证书的 SPKI 指纹（base64），注册时上报。</summary>
|
||
[JsonPropertyName("tls_spki")]
|
||
public string TlsSpki { get; set; } = "";
|
||
}
|
||
|
||
[JsonSerializable(typeof(NodeFile))]
|
||
public sealed partial class NodeFileContext : JsonSerializerContext;
|
||
|
||
/// <summary>身份目录里的文件（<c>fleet-enroll</c> 生成）：<c>node.json</c>、<c>node.key</c>（PKCS#8）、<c>tls.pfx</c>。</summary>
|
||
public sealed class LoadedNodeIdentity : IDisposable
|
||
{
|
||
public const string NodeFileName = "node.json";
|
||
public const string KeyFileName = "node.key";
|
||
public const string TlsFileName = "tls.pfx";
|
||
|
||
public LoadedNodeIdentity(NodeFile file, NodeIdentity identity, X509Certificate2 certificate)
|
||
{
|
||
File = file;
|
||
Identity = identity;
|
||
Certificate = certificate;
|
||
}
|
||
|
||
public NodeFile File { get; }
|
||
|
||
public NodeId NodeId => new(File.NodeId);
|
||
|
||
public NodeIdentity Identity { get; }
|
||
|
||
public X509Certificate2 Certificate { get; }
|
||
|
||
/// <summary>
|
||
/// 对等面的签名身份：TLS 证书的那把密钥，和连服务端用的 <see cref="Identity"/> 是两把独立的密钥，
|
||
/// 各自只用于自己的用途（一把泄露不影响另一把）。
|
||
/// </summary>
|
||
public NodeIdentity PeerIdentity => _peerIdentity ??= NodeIdentity.FromPkcs8(
|
||
(Certificate.GetECDsaPrivateKey() ?? throw new InvalidDataException("the node certificate has no ECDSA private key")).ExportPkcs8PrivateKey());
|
||
|
||
private NodeIdentity? _peerIdentity;
|
||
|
||
public static LoadedNodeIdentity Load(string directory)
|
||
{
|
||
if (!System.IO.File.Exists(Path.Combine(directory, KeyFileName)) || !System.IO.File.Exists(Path.Combine(directory, TlsFileName)))
|
||
{
|
||
throw new InvalidDataException($"no node identity in {directory}: run fleet-enroll first");
|
||
}
|
||
|
||
NodeFile file = NodeConfig.Load(Path.Combine(directory, NodeFileName), NodeFileContext.Default.NodeFile);
|
||
NodeIdentity identity = NodeIdentity.FromPkcs8(System.IO.File.ReadAllBytes(Path.Combine(directory, KeyFileName)));
|
||
X509Certificate2 certificate = NodeCertificate.LoadPfx(System.IO.File.ReadAllBytes(Path.Combine(directory, TlsFileName)));
|
||
return new LoadedNodeIdentity(file, identity, certificate);
|
||
}
|
||
|
||
public void Dispose()
|
||
{
|
||
Identity.Dispose();
|
||
_peerIdentity?.Dispose();
|
||
Certificate.Dispose();
|
||
}
|
||
}
|