77 lines
3.1 KiB
C#
77 lines
3.1 KiB
C#
using Fleet.Core;
|
|
using Fleet.Link;
|
|
using MoShiro.Measurement;
|
|
using MoShiro.Protocol;
|
|
|
|
namespace MoShiro.Agent;
|
|
|
|
/// <summary>
|
|
/// Probe 的列块到达之后：找到对应的流 → 与 Agent 在数据面记录的列合并 → 盖章（块头、时钟锚点）→ 编码 → 放进发件箱（通道 16）。
|
|
/// 找不到对应流的列块（任务已经被撤销、测试已经结束）被丢弃并计数；这只会发生在配置变更的瞬间。
|
|
/// </summary>
|
|
public sealed class ChunkAssembler
|
|
{
|
|
private readonly StreamRegistry _streams;
|
|
private readonly MonoClock _clock;
|
|
private readonly AgentCounters _counters;
|
|
private readonly Outbox _outbox;
|
|
private readonly NodeId _agent;
|
|
private readonly TimeProvider _time;
|
|
private readonly INodeLog _log;
|
|
|
|
public ChunkAssembler(StreamRegistry streams, MonoClock clock, AgentCounters counters, Outbox outbox, NodeId agent, TimeProvider time, INodeLog log)
|
|
{
|
|
_streams = streams;
|
|
_clock = clock;
|
|
_counters = counters;
|
|
_outbox = outbox;
|
|
_agent = agent;
|
|
_time = time;
|
|
_log = log;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 处理某个 Probe 交来的一个列块（已经解码）。返回是否被接受。
|
|
/// 块头里的 <c>probe_id</c> 以认证过的会话为准（Probe 自报的值不可信），别的 Probe 的流也不会被命中。
|
|
/// </summary>
|
|
public bool Accept(RawChunk fragment, ProbeSession probe)
|
|
{
|
|
RawChunkHeader header = fragment.Header;
|
|
StreamInfo? stream = header switch
|
|
{
|
|
{ TestId: { } test, Phase: { } phase } => _streams.FindByTest(test, phase) is { } found && ReferenceEquals(found.Owner, probe) ? found : null,
|
|
{ TaskId: { } task } => _streams.FindByTask(task, probe),
|
|
_ => null,
|
|
};
|
|
header.ProbeId = new ProbeId(probe.ProbeId);
|
|
header.IpVersion = probe.IpVersion; // 地址族以认证过的控制会话为准，不接受列块自报值
|
|
KindSpec? spec = stream is null ? null : Kinds.Find(stream.Kind);
|
|
if (stream is null || spec is null || spec.Id != header.Kind)
|
|
{
|
|
_counters.FragmentDiscarded();
|
|
return false;
|
|
}
|
|
|
|
Result<IReadOnlyList<RawStream>> merged = FragmentMerger.Merge(spec, fragment, stream.Id, _streams);
|
|
if (!merged.TryGet(out IReadOnlyList<RawStream>? streams, out FleetError? error))
|
|
{
|
|
_log.Warn("fragment discarded", ("kind", spec.Id), ("err", error.Code));
|
|
_counters.FragmentDiscarded();
|
|
stream.FragmentDone();
|
|
return false;
|
|
}
|
|
|
|
RawChunk chunk = ChunkFinalizer.Finalize(spec, fragment, streams, _agent, _clock.Anchor(), stream.Params.Values, stream.Port, ChunkId.New(_time));
|
|
Emit(chunk);
|
|
_counters.FragmentAccepted();
|
|
stream.FragmentDone();
|
|
return true;
|
|
}
|
|
|
|
/// <summary>Agent 自己采集的块（目标任务）直接发出。</summary>
|
|
public void Emit(RawChunk chunk)
|
|
{
|
|
_outbox.Enqueue(MoShiroChannels.RawChunk, chunk.Header.ChunkId.Value, RawChunkCodec.Encode(chunk));
|
|
}
|
|
}
|