372 lines
13 KiB
C#
372 lines
13 KiB
C#
using System.Net.NetworkInformation;
|
||
using System.Net.Sockets;
|
||
using System.Security.Authentication;
|
||
using Fleet.Core;
|
||
using MoShiro.Protocol;
|
||
|
||
namespace MoShiro.Measurement;
|
||
|
||
/// <summary>什么也不做的监听方：connect 的另一端就是 TCP 监听器本身（接受连接、认证失败就关闭），不需要积木，但需要占一个流编号。</summary>
|
||
public sealed class IdleListener : BlockBase
|
||
{
|
||
public IdleListener(BlockEnv env, BlockSpec spec)
|
||
: base(env, spec)
|
||
{
|
||
}
|
||
|
||
public override Task RunAsync(CancellationToken ct) => WaitUntilCancelledAsync(ct);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 反复建立连接（TCP，可选 TLS）并记录每个阶段的耗时和结果。结果编码与 Backend 的 <c>ConnectOutcome</c> 一致：
|
||
/// 0 成功、1 超时、2 被重置或拒绝、3 TLS 告警、4 其他。拨号方是 Probe 时对端是 Agent（按指纹固定证书，建好连接即关闭）；
|
||
/// 作为目标任务在 Agent 上运行时对端是任意地址（不校验证书，只测握手耗时）。
|
||
/// </summary>
|
||
public sealed class ConnectDialer : BlockBase
|
||
{
|
||
private readonly DriverLog _log;
|
||
private readonly int _attempt;
|
||
private readonly int _ts;
|
||
private readonly int _tcp;
|
||
private readonly int _tls;
|
||
private readonly int _outcome;
|
||
|
||
public ConnectDialer(BlockEnv env, BlockSpec spec)
|
||
: base(env, spec)
|
||
{
|
||
_log = new DriverLog(spec.Kind.FindStream("conn")!, Side.Probe, 1024);
|
||
_attempt = _log.Store.ColumnIndex("attempt");
|
||
_ts = _log.Store.ColumnIndex("ts");
|
||
_tcp = _log.Store.ColumnIndex("tcp_us");
|
||
_tls = _log.Store.ColumnIndex("tls_us");
|
||
_outcome = _log.Store.ColumnIndex("outcome");
|
||
}
|
||
|
||
public override long Produced => _log.Count;
|
||
|
||
public override async Task RunAsync(CancellationToken ct)
|
||
{
|
||
TimeSpan interval = TimeSpan.FromSeconds(60.0 / Spec.Int("per_minute"));
|
||
TimeSpan timeout = TimeSpan.FromMilliseconds(Spec.Int("timeout_ms"));
|
||
bool tls = Spec.Bool("tls");
|
||
try
|
||
{
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
long start = Env.Clock.NowUs();
|
||
await AttemptAsync(tls, timeout, start, ct);
|
||
TimeSpan wait = interval - TimeSpan.FromMicroseconds(Env.Clock.NowUs() - start);
|
||
if (wait > TimeSpan.Zero)
|
||
{
|
||
await Task.Delay(wait, Env.Time, ct);
|
||
}
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// 积木被停止
|
||
}
|
||
}
|
||
|
||
public override int Flush(FlushMode mode, long peerProduced = 0)
|
||
{
|
||
RawStream? stream = _log.Take(null, all: true);
|
||
if (stream is null)
|
||
{
|
||
return 0;
|
||
}
|
||
|
||
Ship(stream, "ts");
|
||
return 1;
|
||
}
|
||
|
||
private async Task AttemptAsync(bool tls, TimeSpan timeout, long startUs, CancellationToken ct)
|
||
{
|
||
using var timeoutCts = new CancellationTokenSource(timeout, Env.Time);
|
||
using CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource(ct, timeoutCts.Token);
|
||
long tcpUs = 0;
|
||
long? tlsUs = null;
|
||
int outcome;
|
||
try
|
||
{
|
||
// 对端不需要认证首帧：只测量建立连接
|
||
using TcpLink link = await TcpLink.OpenAsync(Env, Spec.Port, tls, Spec.StreamId, sendAuth: false, linked.Token);
|
||
tcpUs = link.TcpUs;
|
||
tlsUs = link.TlsUs;
|
||
outcome = 0;
|
||
}
|
||
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
||
{
|
||
outcome = 1;
|
||
}
|
||
catch (SocketException ex) when (ex.SocketErrorCode is SocketError.ConnectionRefused or SocketError.ConnectionReset)
|
||
{
|
||
outcome = 2;
|
||
}
|
||
catch (Exception ex) when (ex is AuthenticationException || ex.InnerException is AuthenticationException)
|
||
{
|
||
outcome = 3;
|
||
}
|
||
catch (Exception ex) when (ex is SocketException or IOException)
|
||
{
|
||
outcome = ex is IOException { InnerException: SocketException { SocketErrorCode: SocketError.ConnectionReset } } ? 2 : 4;
|
||
}
|
||
|
||
// 失败时 tcp_us/tls_us 缺失；成功时没有 TLS 则 tls_us 缺失
|
||
long key = _log.Store.Append();
|
||
_log.Store.Set(key, _attempt, key);
|
||
_log.Store.Set(key, _ts, startUs);
|
||
_log.Store.Set(key, _outcome, outcome);
|
||
if (outcome == 0)
|
||
{
|
||
_log.Store.Set(key, _tcp, tcpUs);
|
||
if (tlsUs is { } t)
|
||
{
|
||
_log.Store.Set(key, _tls, t);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 保持，拨号方：一条长时间保持的 TLS 连接，每隔 <c>heartbeat_s</c> 发一个心跳并等回复。记录打开、心跳、重置（连接意外中断）和关闭，
|
||
/// 每一行带着连接已经存活的时间。连接中断之后按退避重连。监听方是不记录的 <see cref="TcpEchoListener"/>。
|
||
/// </summary>
|
||
public sealed class HoldDialer : BlockBase
|
||
{
|
||
private const int Open = 0;
|
||
private const int Heartbeat = 1;
|
||
private const int Reset = 2;
|
||
private const int Close = 3;
|
||
|
||
private readonly DriverLog _log;
|
||
private readonly int _ts;
|
||
private readonly int _event;
|
||
private readonly int _lifetime;
|
||
|
||
public HoldDialer(BlockEnv env, BlockSpec spec)
|
||
: base(env, spec)
|
||
{
|
||
_log = new DriverLog(spec.Kind.FindStream("hold")!, Side.Probe, 1024);
|
||
_ts = _log.Store.ColumnIndex("ts");
|
||
_event = _log.Store.ColumnIndex("event");
|
||
_lifetime = _log.Store.ColumnIndex("lifetime_ms");
|
||
}
|
||
|
||
public override long Produced => _log.Count;
|
||
|
||
public override async Task RunAsync(CancellationToken ct)
|
||
{
|
||
var backoff = new Fleet.Core.Backoff(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(30));
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
long openedUs = 0;
|
||
bool opened = false;
|
||
try
|
||
{
|
||
using TcpLink link = await TcpLink.OpenAsync(Env, Spec.Port, tls: true, Spec.StreamId, sendAuth: true, ct);
|
||
openedUs = Env.Clock.NowUs();
|
||
opened = true;
|
||
backoff.Reset();
|
||
Record(Open, openedUs, openedUs);
|
||
await HoldAsync(link.Stream, openedUs, ct);
|
||
}
|
||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||
{
|
||
if (opened)
|
||
{
|
||
Record(Close, Env.Clock.NowUs(), openedUs);
|
||
}
|
||
|
||
return;
|
||
}
|
||
catch (Exception ex) when (ex is SocketException or IOException or AuthenticationException or OperationCanceledException or InvalidDataException or InvalidOperationException)
|
||
{
|
||
if (opened)
|
||
{
|
||
Record(Reset, Env.Clock.NowUs(), openedUs);
|
||
}
|
||
}
|
||
|
||
try
|
||
{
|
||
await Task.Delay(backoff.Next(), Env.Time, ct);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
return;
|
||
}
|
||
}
|
||
}
|
||
|
||
public override int Flush(FlushMode mode, long peerProduced = 0)
|
||
{
|
||
RawStream? stream = _log.Take(null, all: true);
|
||
if (stream is null)
|
||
{
|
||
return 0;
|
||
}
|
||
|
||
Ship(stream, "ts");
|
||
return 1;
|
||
}
|
||
|
||
private async Task HoldAsync(Stream stream, long openedUs, CancellationToken ct)
|
||
{
|
||
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||
Task reader = ReadAsync(stream, openedUs, cts.Token);
|
||
Task writer = WriteAsync(stream, cts.Token);
|
||
Task finished = await Task.WhenAny(reader, writer);
|
||
await cts.CancelAsync();
|
||
try
|
||
{
|
||
await finished; // 连接中断时这里抛出，由调用方记录 Reset
|
||
if (!ct.IsCancellationRequested)
|
||
{
|
||
throw new IOException("the peer closed the connection");
|
||
}
|
||
}
|
||
finally
|
||
{
|
||
foreach (Task task in new[] { reader, writer })
|
||
{
|
||
try
|
||
{
|
||
await task;
|
||
}
|
||
catch (Exception ex) when (ex is OperationCanceledException or IOException or SocketException or ObjectDisposedException)
|
||
{
|
||
// 已经处理过
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
private async Task WriteAsync(Stream stream, CancellationToken ct)
|
||
{
|
||
byte[] body = new byte[TcpWire.MinEchoBody];
|
||
uint seq = 0;
|
||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(Spec.Int("heartbeat_s")), Env.Time);
|
||
while (await timer.WaitForNextTickAsync(ct))
|
||
{
|
||
TcpWire.WriteEchoBody(body, seq++, Env.Clock.NowUs());
|
||
await TcpWire.WriteMessageAsync(stream, body, ct);
|
||
await stream.FlushAsync(ct);
|
||
}
|
||
}
|
||
|
||
private async Task ReadAsync(Stream stream, long openedUs, CancellationToken ct)
|
||
{
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
byte[]? reply = await TcpWire.ReadMessageAsync(stream, ct);
|
||
if (reply is null)
|
||
{
|
||
return;
|
||
}
|
||
|
||
Record(Heartbeat, Env.Clock.NowUs(), openedUs);
|
||
}
|
||
}
|
||
|
||
private void Record(int eventCode, long nowUs, long openedUs)
|
||
{
|
||
long key = _log.Store.Append();
|
||
_log.Store.Set(key, _ts, nowUs);
|
||
_log.Store.Set(key, _event, eventCode);
|
||
_log.Store.Set(key, _lifetime, Math.Max(0, (nowUs - openedUs) / 1000));
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// ICMP 回显（只有一侧：Agent 对目标地址）：按速率发 ICMP 并记录发出和收到的时刻（本地时钟）；超时没有回复的行 t4 缺失。
|
||
/// 没有对端，所以没有监听方；行交给 <see cref="BlockEnv.Fragments"/>（Agent 自己盖章上传）。
|
||
/// </summary>
|
||
public sealed class IcmpEchoSolo : BlockBase
|
||
{
|
||
private readonly DriverLog _log;
|
||
private readonly int _seq;
|
||
private readonly int _size;
|
||
private readonly int _t1;
|
||
private readonly int _t4;
|
||
|
||
public IcmpEchoSolo(BlockEnv env, BlockSpec spec)
|
||
: base(env, spec)
|
||
{
|
||
_log = new DriverLog(spec.Kind.FindStream("msg")!, Side.Probe, 1024);
|
||
_seq = _log.Store.ColumnIndex("seq");
|
||
_size = _log.Store.ColumnIndex("size");
|
||
_t1 = _log.Store.ColumnIndex("t1");
|
||
_t4 = _log.Store.ColumnIndex("t4");
|
||
}
|
||
|
||
public override long Produced => _log.Count;
|
||
|
||
public override async Task RunAsync(CancellationToken ct)
|
||
{
|
||
TimeSpan interval = TimeSpan.FromSeconds(1.0 / Spec.Int("rate_hz"));
|
||
int size = (int)Spec.Int("size");
|
||
byte[] payload = new byte[Math.Max(0, size - 28)];
|
||
using LinuxIcmpEcho? linuxPing = OperatingSystem.IsLinux() ? LinuxIcmpEcho.TryCreate(Env.PeerAddress!, payload.Length) : null;
|
||
using Ping? ping = linuxPing is null ? new Ping() : null;
|
||
try
|
||
{
|
||
while (!ct.IsCancellationRequested)
|
||
{
|
||
long t1 = Env.Clock.NowUs();
|
||
long? t4 = null;
|
||
try
|
||
{
|
||
bool receivedReply = linuxPing is not null
|
||
? await linuxPing.SendAsync(TimeSpan.FromSeconds(2), ct)
|
||
: (await ping!.SendPingAsync(Env.PeerAddress!, TimeSpan.FromSeconds(2), payload, null, ct)).Status == IPStatus.Success;
|
||
if (receivedReply)
|
||
{
|
||
t4 = Env.Clock.NowUs();
|
||
}
|
||
}
|
||
catch (PingException)
|
||
{
|
||
// 发送失败（例如本机没有路由）：记为丢失
|
||
}
|
||
catch (PlatformNotSupportedException) when (OperatingSystem.IsLinux())
|
||
{
|
||
Env.Log.Warn("ICMP permission unavailable", ("hint", "allow the mapped process GID in net.ipv4.ping_group_range, or grant effective NET_RAW"));
|
||
throw;
|
||
}
|
||
|
||
long key = _log.Store.Append();
|
||
_log.Store.Set(key, _seq, key);
|
||
_log.Store.Set(key, _size, size);
|
||
_log.Store.Set(key, _t1, t1);
|
||
if (t4 is { } received)
|
||
{
|
||
_log.Store.Set(key, _t4, received);
|
||
}
|
||
|
||
TimeSpan wait = interval - TimeSpan.FromMicroseconds(Env.Clock.NowUs() - t1);
|
||
if (wait > TimeSpan.Zero)
|
||
{
|
||
await Task.Delay(wait, Env.Time, ct);
|
||
}
|
||
}
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
// 积木被停止
|
||
}
|
||
}
|
||
|
||
public override int Flush(FlushMode mode, long peerProduced = 0)
|
||
{
|
||
RawStream? stream = _log.Take(null, all: true);
|
||
if (stream is null)
|
||
{
|
||
return 0;
|
||
}
|
||
|
||
Ship(stream, "t1");
|
||
return 1;
|
||
}
|
||
}
|