chore: initial project import
This commit is contained in:
justme committed
commit
26620707a0
347 files changed
+56172
No files matched your search
@@ -0,0 +1,163 @@
|
||||
using System.Net;
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
using Monitor.Measurement;
|
||||
using Monitor.Protocol;
|
||||
|
||||
namespace Monitor.Agent;
|
||||
|
||||
/// <summary>
|
||||
/// 目标任务（Agent 自己探测一个地址，没有对端）：用和线路任务同一套积木（<see cref="BlockCatalog"/> 里的拨号方，例如 icmp_echo、connect），
|
||||
/// 只是列块不交给 Agent 合并，而是自己盖章、10 秒一块直接上传。目标地址必须在本地 <c>allowed_target_cidrs</c> 内
|
||||
/// （由 <see cref="BudgetGuard"/> 在接受任务之前检查）。
|
||||
/// </summary>
|
||||
public sealed class TargetRunners
|
||||
{
|
||||
private static readonly TimeSpan s_chunkSpan = TimeSpan.FromSeconds(ProtocolLimits.ChunkSeconds);
|
||||
|
||||
private readonly Dictionary<string, Running> _running = [];
|
||||
private readonly object _gate = new();
|
||||
private readonly ChunkAssembler _chunks;
|
||||
private readonly NodeId _agent;
|
||||
private readonly MonoClock _clock;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly INodeLog _log;
|
||||
|
||||
public TargetRunners(ChunkAssembler chunks, NodeId agent, MonoClock clock, TimeProvider time, INodeLog log)
|
||||
{
|
||||
_chunks = chunks;
|
||||
_agent = agent;
|
||||
_clock = clock;
|
||||
_time = time;
|
||||
_log = log;
|
||||
}
|
||||
|
||||
/// <summary>让运行中的任务与期望集合一致：参数没有变化的保持，其余启动或停止。</summary>
|
||||
public void Reconcile(IReadOnlyList<(TaskSpec Spec, TaskDecision Decision)> desired)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
var wanted = desired.ToDictionary(d => d.Spec.Id.Value, d => d);
|
||||
foreach ((string id, Running running) in _running.ToList())
|
||||
{
|
||||
if (!wanted.TryGetValue(id, out (TaskSpec Spec, TaskDecision Decision) entry) || Signature(entry.Spec) != running.Signature)
|
||||
{
|
||||
running.Cancel.Cancel();
|
||||
_running.Remove(id);
|
||||
}
|
||||
}
|
||||
|
||||
foreach ((string id, (TaskSpec spec, TaskDecision decision)) in wanted)
|
||||
{
|
||||
if (_running.ContainsKey(id))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var cts = new CancellationTokenSource();
|
||||
_running[id] = new Running(Signature(spec), cts);
|
||||
_ = Task.Run(() => RunAsync(spec, decision, cts.Token), CancellationToken.None);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static string Signature(TaskSpec spec) =>
|
||||
spec.Kind + "|" + spec.Target + "|" + string.Join(',', spec.Params.OrderBy(p => p.Key, StringComparer.Ordinal).Select(p => p.Key + "=" + p.Value.GetRawText()));
|
||||
|
||||
private async Task RunAsync(TaskSpec spec, TaskDecision decision, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!TryResolve(spec.Target!, decision.Resolved!, spec.Kind, out IPAddress? address, out int port))
|
||||
{
|
||||
_log.Warn("target is not an address", ("task", spec.Id.Value));
|
||||
return;
|
||||
}
|
||||
|
||||
var env = new BlockEnv
|
||||
{
|
||||
Clock = _clock,
|
||||
Log = _log,
|
||||
Session = () => null,
|
||||
PeerAddress = address,
|
||||
Fragments = new LocalSink(this, spec),
|
||||
};
|
||||
IBlock block = BlockCatalog.Find(spec.Kind)!.Dialer(env, new BlockSpec(decision.Kind!, decision.Resolved!, 0, port, new FragmentScope(spec.Id, null, null)));
|
||||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
Task run = block.RunAsync(cts.Token);
|
||||
try
|
||||
{
|
||||
using var timer = new PeriodicTimer(s_chunkSpan, _time);
|
||||
while (await timer.WaitForNextTickAsync(ct))
|
||||
{
|
||||
block.Flush(FlushMode.Settled);
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
await cts.CancelAsync();
|
||||
await run;
|
||||
block.Flush(FlushMode.All);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// 任务被撤销
|
||||
}
|
||||
catch (Exception ex) when (ex is System.Net.Sockets.SocketException or InvalidOperationException or BlockFaultException)
|
||||
{
|
||||
_log.Warn("target task stopped", ("task", spec.Id.Value), ("err", ex.Message));
|
||||
}
|
||||
}
|
||||
|
||||
private static bool TryResolve(string target, ParamValues values, string kind, out IPAddress address, out int port)
|
||||
{
|
||||
port = kind == "connect" ? (int)values.GetInt("port") : 0;
|
||||
if (IPEndPoint.TryParse(target, out IPEndPoint? endpoint))
|
||||
{
|
||||
address = endpoint.Address;
|
||||
if (endpoint.Port != 0)
|
||||
{
|
||||
port = endpoint.Port;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
return IPAddress.TryParse(target, out address!);
|
||||
}
|
||||
|
||||
/// <summary>目标任务的行没有 Probe 一侧可以合并：积木交出的就是完整的行，Agent 自己盖章（目标、时钟锚点）并放进发件箱。</summary>
|
||||
private sealed class LocalSink : IFragmentSink
|
||||
{
|
||||
private readonly TargetRunners _owner;
|
||||
private readonly TaskSpec _spec;
|
||||
|
||||
public LocalSink(TargetRunners owner, TaskSpec spec)
|
||||
{
|
||||
_owner = owner;
|
||||
_spec = spec;
|
||||
}
|
||||
|
||||
public void Ship(string kind, FragmentScope scope, IReadOnlyList<RawStream> streams, IReadOnlyDictionary<string, JsonElement>? parameters, long startUtcMs, int durationMs, int port)
|
||||
{
|
||||
ClockAnchor anchor = _owner._clock.Anchor();
|
||||
var header = new RawChunkHeader
|
||||
{
|
||||
ChunkId = ChunkId.New(_owner._time),
|
||||
Kind = kind,
|
||||
TaskId = scope.Task,
|
||||
AgentId = _owner._agent,
|
||||
Target = _spec.Target,
|
||||
Params = parameters is null ? [] : new Dictionary<string, JsonElement>(parameters),
|
||||
Port = port,
|
||||
StartUtcMs = startUtcMs,
|
||||
DurationMs = Math.Clamp(durationMs, 1, ChunkFinalizer.MaxDurationMs),
|
||||
Clock = new ChunkClock { AgentMonoUs = anchor.AgentMonoUs, UtcMs = anchor.UtcMs, ProbeOffsetUs = 0 },
|
||||
};
|
||||
_owner._chunks.Emit(new RawChunk(header, streams));
|
||||
}
|
||||
}
|
||||
|
||||
private sealed record Running(string Signature, CancellationTokenSource Cancel);
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
using System.Text.Json;
|
||||
using Fleet.Core;
|
||||
using Fleet.Rpc;
|
||||
using Monitor.Measurement;
|
||||
using Monitor.Protocol;
|
||||
|
||||
namespace Monitor.Agent;
|
||||
|
||||
/// <summary>
|
||||
/// 期望任务集（<c>task_set</c>）的调和器：每个任务先经过 <see cref="BudgetGuard"/>（参数校验 + 本地硬上限）。
|
||||
/// 通过的线路任务<b>为每个 Probe 会话各登记一组流</b>（流编号和监听方积木都属于会话，互不干扰），并汇总成 <c>line_task_set</c> 下发给这个 Probe；
|
||||
/// 通过的目标任务在本地运行，与 Probe 无关。被拒绝的任务带着原因（<c>rejected[]</c>）回报给 Backend，其余任务照常运行。
|
||||
/// 每次调用都应用完整的快照，所以是幂等的。预算（码率、包速率）按单个 Probe 会话计算，会话总数由 <c>PeerListenerOptions.MaxSessions</c> 限制。
|
||||
/// </summary>
|
||||
public sealed class TaskManager
|
||||
{
|
||||
// 线路任务 = 有监听方积木、并且不是只用于按需测试的 Kind（udp_train、tcp_bulk 会占满链路）；目标任务 = 可以由 Agent 自己运行的 Kind
|
||||
private static readonly HashSet<string> s_lineKinds = [.. BlockCatalog.All.Where(p => p.Listener is not null && !p.OnDemandOnly).Select(p => p.Kind)];
|
||||
private static readonly HashSet<string> s_targetKinds = [.. BlockCatalog.All.Where(p => p.RunsAsTarget).Select(p => p.Kind)];
|
||||
|
||||
private readonly object _gate = new();
|
||||
private readonly Func<BudgetGuard> _guardFactory;
|
||||
private readonly StreamRegistry _streams;
|
||||
private readonly BlockEnv _env;
|
||||
private readonly TargetRunners _targets;
|
||||
private readonly Func<IReadOnlyList<int>> _udpPorts;
|
||||
private readonly Func<IReadOnlyList<int>> _tcpPorts;
|
||||
private readonly Dictionary<ProbeSession, Dictionary<string, LineEntry>> _sessionLines = [];
|
||||
private volatile Dictionary<string, byte[]> _peers = new(StringComparer.Ordinal);
|
||||
private Dictionary<string, (TaskSpec Task, TaskDecision Decision)> _lineSpecs = [];
|
||||
private long _version;
|
||||
private BudgetGuard? _guard;
|
||||
|
||||
public TaskManager(Func<BudgetGuard> guardFactory, StreamRegistry streams, BlockEnv env, TargetRunners targets, Func<IReadOnlyList<int>> udpPorts, Func<IReadOnlyList<int>> tcpPorts)
|
||||
{
|
||||
_guardFactory = guardFactory;
|
||||
_streams = streams;
|
||||
_env = env;
|
||||
_targets = targets;
|
||||
_udpPorts = udpPorts;
|
||||
_tcpPorts = tcpPorts;
|
||||
}
|
||||
|
||||
public BudgetGuard Guard => _guard ??= _guardFactory();
|
||||
|
||||
/// <summary>某个 Probe 的线路任务集变化（版本递增）：由 <see cref="PeerHub"/> 推送给这个 Probe。</summary>
|
||||
public Action<ProbeSession, LineTaskSet>? OnLineTasksChanged { get; set; }
|
||||
|
||||
/// <summary>允许连接的 Probe 变了（期望状态里的 peers）：已经不在名单里的会话由 <see cref="PeerHub"/> 关闭。</summary>
|
||||
public Action<IReadOnlySet<string>>? OnPeersChanged { get; set; }
|
||||
|
||||
/// <summary>点对点握手时查找 Probe 的公钥（SPKI）。</summary>
|
||||
public byte[]? LookupPeerKey(string peerId) => _peers.TryGetValue(peerId, out byte[]? key) ? key : null;
|
||||
|
||||
/// <summary>一个新的 Probe 会话开始：为它登记全部线路任务的流并启动监听方积木，返回要下发给它的线路任务集。</summary>
|
||||
public LineTaskSet OpenSession(ProbeSession probe)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
var entries = new Dictionary<string, LineEntry>();
|
||||
foreach ((string id, (TaskSpec task, TaskDecision decision)) in _lineSpecs)
|
||||
{
|
||||
LineEntry entry = BuildLineEntry(task, decision, probe);
|
||||
Start(entry);
|
||||
entries[id] = entry;
|
||||
}
|
||||
|
||||
_sessionLines[probe] = entries;
|
||||
return SetFor(entries);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Probe 会话结束：释放它的所有流和积木（记录随之丢弃，下一个会话从零开始）。</summary>
|
||||
public async Task CloseSessionAsync(ProbeSession probe)
|
||||
{
|
||||
List<LineEntry> entries;
|
||||
lock (_gate)
|
||||
{
|
||||
if (!_sessionLines.Remove(probe, out Dictionary<string, LineEntry>? lines))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
entries = [.. lines.Values];
|
||||
}
|
||||
|
||||
foreach (LineEntry entry in entries)
|
||||
{
|
||||
await Retire(entry);
|
||||
}
|
||||
}
|
||||
|
||||
public async ValueTask<StateSyncResult> ApplyAsync(long version, TaskSet set, CancellationToken ct)
|
||||
{
|
||||
_ = ct;
|
||||
var result = new StateSyncResult { AppliedVersion = version };
|
||||
|
||||
var peers = new Dictionary<string, byte[]>(StringComparer.Ordinal);
|
||||
foreach (PeerSpec peer in set.Peers)
|
||||
{
|
||||
try
|
||||
{
|
||||
peers[peer.ProbeId.Value] = Convert.FromBase64String(peer.PublicKey);
|
||||
}
|
||||
catch (FormatException)
|
||||
{
|
||||
result.Rejected.Add(new RejectedItem { Id = peer.ProbeId.Value, Reason = RejectReasons.InvalidParams });
|
||||
}
|
||||
}
|
||||
|
||||
_peers = peers;
|
||||
|
||||
BudgetGuard guard = Guard;
|
||||
double usedMbps = 0;
|
||||
long usedPps = 0;
|
||||
var line = new Dictionary<string, (TaskSpec Task, TaskDecision Decision)>();
|
||||
var targets = new List<(TaskSpec Spec, TaskDecision Decision)>();
|
||||
|
||||
foreach (TaskSpec task in set.Tasks)
|
||||
{
|
||||
if (!task.Enabled)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
bool shapeOk = task.Scope == "target" ? s_targetKinds.Contains(task.Kind) : s_lineKinds.Contains(task.Kind);
|
||||
TaskDecision decision = shapeOk ? guard.Check(task, ref usedMbps, ref usedPps) : TaskDecision.Reject(RejectReasons.Unsupported);
|
||||
if (!decision.Accepted)
|
||||
{
|
||||
result.Rejected.Add(new RejectedItem { Id = task.Id.Value, Reason = decision.Reason! });
|
||||
continue;
|
||||
}
|
||||
|
||||
if (task.Scope == "target")
|
||||
{
|
||||
targets.Add((task, decision));
|
||||
}
|
||||
else
|
||||
{
|
||||
line[task.Id.Value] = (task, decision);
|
||||
}
|
||||
}
|
||||
|
||||
_targets.Reconcile(targets);
|
||||
|
||||
// 参数没有变化的任务沿用原来的流和积木，数据面上的会话（重放窗口、记录）不被打断
|
||||
var retired = new List<LineEntry>();
|
||||
var changes = new List<(ProbeSession Probe, LineTaskSet Set)>();
|
||||
lock (_gate)
|
||||
{
|
||||
_lineSpecs = line;
|
||||
_version = version;
|
||||
foreach ((ProbeSession probe, Dictionary<string, LineEntry> current) in _sessionLines)
|
||||
{
|
||||
var next = new Dictionary<string, LineEntry>();
|
||||
foreach ((string id, (TaskSpec task, TaskDecision decision)) in line)
|
||||
{
|
||||
LineEntry candidate = BuildLineEntry(task, decision, probe);
|
||||
if (current.TryGetValue(id, out LineEntry? existing) && existing.Signature == candidate.Signature)
|
||||
{
|
||||
next[id] = existing;
|
||||
}
|
||||
else
|
||||
{
|
||||
Start(candidate);
|
||||
next[id] = candidate;
|
||||
}
|
||||
}
|
||||
|
||||
foreach ((string id, LineEntry entry) in current)
|
||||
{
|
||||
if (!next.TryGetValue(id, out LineEntry? kept) || !ReferenceEquals(kept, entry))
|
||||
{
|
||||
retired.Add(entry);
|
||||
}
|
||||
}
|
||||
|
||||
current.Clear();
|
||||
foreach ((string id, LineEntry entry) in next)
|
||||
{
|
||||
current[id] = entry;
|
||||
}
|
||||
|
||||
changes.Add((probe, SetFor(current)));
|
||||
}
|
||||
}
|
||||
|
||||
foreach (LineEntry entry in retired)
|
||||
{
|
||||
await Retire(entry);
|
||||
}
|
||||
|
||||
OnPeersChanged?.Invoke(new HashSet<string>(peers.Keys, StringComparer.Ordinal));
|
||||
foreach ((ProbeSession probe, LineTaskSet next) in changes)
|
||||
{
|
||||
OnLineTasksChanged?.Invoke(probe, next);
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
/// <summary>停止时释放所有 Probe 会话的流和所有目标任务（Backend 失联的时候任务继续运行：节点自治）。</summary>
|
||||
public async Task StopAsync()
|
||||
{
|
||||
_targets.Reconcile([]);
|
||||
List<LineEntry> entries;
|
||||
lock (_gate)
|
||||
{
|
||||
entries = [.. _sessionLines.Values.SelectMany(l => l.Values)];
|
||||
_sessionLines.Clear();
|
||||
}
|
||||
|
||||
foreach (LineEntry entry in entries)
|
||||
{
|
||||
await Retire(entry);
|
||||
}
|
||||
}
|
||||
|
||||
private LineTaskSet SetFor(Dictionary<string, LineEntry> entries) =>
|
||||
new()
|
||||
{
|
||||
Version = _version,
|
||||
Tasks = [.. entries.Values.Select(e => e.Task)],
|
||||
UdpPorts = [.. _udpPorts()],
|
||||
TcpPorts = [.. _tcpPorts()],
|
||||
};
|
||||
|
||||
private LineEntry BuildLineEntry(TaskSpec task, TaskDecision decision, ProbeSession probe)
|
||||
{
|
||||
var info = new StreamInfo(_streams.Allocate(), task.Kind, decision.Resolved!) { TaskId = task.Id, Owner = probe };
|
||||
var lineTask = new LineTask
|
||||
{
|
||||
Id = task.Id,
|
||||
Kind = task.Kind,
|
||||
Params = new Dictionary<string, JsonElement>(decision.Resolved!.Values),
|
||||
StreamId = info.Id,
|
||||
Port = info.Port,
|
||||
};
|
||||
var spec = new BlockSpec(decision.Kind!, decision.Resolved, info.Id, info.Port, new FragmentScope(task.Id, null, null));
|
||||
return new LineEntry(lineTask, info, spec, _env.WithSession(probe));
|
||||
}
|
||||
|
||||
/// <summary>在数据面登记这个流并启动它的监听方积木。</summary>
|
||||
private void Start(LineEntry entry)
|
||||
{
|
||||
IBlock block = BlockCatalog.Find(entry.Stream.Kind)!.Listener!(entry.Env, entry.Spec);
|
||||
entry.Stream.Running = RunningBlock.Start(block, entry.Signature);
|
||||
_streams.Add(entry.Stream);
|
||||
}
|
||||
|
||||
private async Task Retire(LineEntry entry)
|
||||
{
|
||||
_streams.Remove(entry.Stream.Id);
|
||||
if (entry.Stream.Running is { } running)
|
||||
{
|
||||
await running.StopAsync();
|
||||
}
|
||||
|
||||
entry.Stream.Owner?.ForgetStream(entry.Stream.Id);
|
||||
}
|
||||
|
||||
private sealed record LineEntry(LineTask Task, StreamInfo Stream, BlockSpec Spec, BlockEnv Env)
|
||||
{
|
||||
public string Signature => Spec.ConfigSignature;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user