shared: add independent IPv4 and IPv6 collection [CI] #15

Merged
justme merged 1 commits from feature/ipv6-collection into main 2026-10-06 20:48:49 -07:00
60 changed files with 699 additions and 137 deletions

No files matched your search

-1
View File
@@ -4,7 +4,6 @@
## 测量
- IPv6
- 路由跳数、路径漂移和 trace
- 握手耗时:TCP 连接耗时,TLS / QUIC 握手耗时(QUIC 目前没有真实握手积木)
- 网页长截图
@@ -15,6 +15,7 @@ public sealed class AgentApp : IDisposable
public const string Version = MoShiro.Protocol.BuildInfo.Version;
private readonly AgentConfig _config;
private readonly bool _ipv6Supported;
private readonly LoadedNodeIdentity _identity;
private readonly TimeProvider _time;
private readonly INodeLog _log;
@@ -47,6 +48,12 @@ public sealed class AgentApp : IDisposable
_uplink = new BackendUplink(time);
IPAddress bind = IPAddress.Parse(config.Listen.Bind);
if (bind.Equals(IPAddress.IPv6Any) && !System.Net.Sockets.Socket.OSSupportsIPv6)
{
bind = IPAddress.Any;
}
_ipv6Supported = bind.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6;
_admission = new AgentAdmission(_streams);
_udp = new UdpMux(_sessions, _admission, _clock, _counters.Data, log, bind, config.Listen.UdpPorts);
_tcp = new TcpAcceptor(bind, config.Listen.TcpPorts, identity.Certificate, _sessions, _counters.Data);
@@ -257,6 +264,7 @@ public sealed class AgentApp : IDisposable
{
var meta = new AgentHelloMeta
{
Ipv6Supported = _ipv6Supported,
Platform = MoShiro.Measurement.PlatformDetector.Detect(),
TlsSpki = _identity.File.TlsSpki,
ProbeTcpPort = ControlPort,
@@ -46,7 +46,7 @@ public sealed class FleetSection
public sealed class ListenSection
{
[JsonPropertyName("bind")]
public string Bind { get; set; } = "0.0.0.0";
public string Bind { get; set; } = "::";
/// <summary>Probe 控制通道(TLS,点对点会话)。</summary>
[JsonPropertyName("control_port")]
@@ -44,6 +44,7 @@ public sealed class ChunkAssembler
_ => 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)
{
@@ -56,6 +56,10 @@ public sealed class ProbeSession : DataSession
public string ProbeId => Info.PeerId;
public string IpVersion => Info.Remote is IPEndPoint endpoint && endpoint.Address.AddressFamily == AddressFamily.InterNetworkV6 && !endpoint.Address.IsIPv4MappedToIPv6 ? IpVersions.V6 : IpVersions.V4;
public string PathKey => ProbeId + "~" + IpVersion;
/// <summary>Probe 的程序版本(握手里自报,只用于展示)。</summary>
public string? Version => Info.Version;
@@ -72,7 +76,7 @@ public sealed class ProbeSession : DataSession
get
{
SocketAddress? source = Volatile.Read(ref _udpSource);
return source is null ? null : (IPEndPoint)new IPEndPoint(IPAddress.Any, 0).Create(source);
return source is null ? null : (IPEndPoint)new IPEndPoint(source.Family == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, 0).Create(source);
}
}
@@ -125,7 +129,7 @@ public sealed class ProbeSession : DataSession
public void ForgetStream(uint streamId) => _windows.TryRemove(streamId, out _);
}
/// <summary>当前连着的 Probe 会话:每个 Probe 一个;同一个 Probe 重连时新会话取代它自己的旧会话。</summary>
/// <summary>每个 Probe 的 IPv4、IPv6 会话独立;重连只替换同地址族的旧会话。</summary>
public sealed class SessionRegistry : IDataSessions
{
private readonly object _gate = new();
@@ -143,11 +147,11 @@ public sealed class SessionRegistry : IDataSessions
}
}
public ProbeSession? Get(string probeId)
public ProbeSession? Get(string probeId, string ipVersion = IpVersions.V4)
{
lock (_gate)
{
return _byProbe.GetValueOrDefault(probeId);
return _byProbe.GetValueOrDefault(probeId + "~" + ipVersion);
}
}
@@ -156,13 +160,13 @@ public sealed class SessionRegistry : IDataSessions
{
lock (_gate)
{
_byProbe.TryGetValue(session.ProbeId, out ProbeSession? old);
_byProbe.TryGetValue(session.PathKey, out ProbeSession? old);
if (old is not null)
{
_bySession.Remove(old.SessionId);
}
_byProbe[session.ProbeId] = session;
_byProbe[session.PathKey] = session;
_bySession[session.SessionId] = session;
return old;
}
@@ -172,9 +176,9 @@ public sealed class SessionRegistry : IDataSessions
{
lock (_gate)
{
if (_byProbe.TryGetValue(session.ProbeId, out ProbeSession? current) && ReferenceEquals(current, session))
if (_byProbe.TryGetValue(session.PathKey, out ProbeSession? current) && ReferenceEquals(current, session))
{
_byProbe.Remove(session.ProbeId);
_byProbe.Remove(session.PathKey);
}
if (_bySession.TryGetValue(session.SessionId, out ProbeSession? bySession) && ReferenceEquals(bySession, session))
@@ -176,7 +176,7 @@ public sealed class PeerHub
await _tasks.CloseSessionAsync(probe);
_log.Info("probe disconnected", ("probe", info.PeerId));
OnProbeLost?.Invoke(probe);
await _uplink.EmitEventAsync(AgentEventKinds.ProbeDisconnected, info.PeerId);
await _uplink.EmitEventAsync(AgentEventKinds.ProbeDisconnected, info.PeerId, Describe(probe));
}
}
@@ -208,7 +208,7 @@ public sealed class PeerHub
/// <summary>当前连着的 Probe(心跳和 Backend 重连之后的补报都用它)。</summary>
public IReadOnlyList<ConnectedProbe> ConnectedProbes() => [.. _sessions.All.Select(Describe)];
private static ConnectedProbe Describe(ProbeSession p) => new() { Id = new ProbeId(p.ProbeId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs };
private static ConnectedProbe Describe(ProbeSession p) => new() { Id = new ProbeId(p.ProbeId), Version = p.Version, Remote = p.RemoteAddress, SinceUtcMs = p.ConnectedAtUtcMs, IpVersion = p.IpVersion };
public IReadOnlyList<string> ConnectedProbeIds() => [.. _sessions.All.Select(p => p.ProbeId)];
@@ -218,13 +218,17 @@ public sealed class PeerHub
/// <summary>Backend 要求断开某个 Probe(它会按退避重连,除非已经被吊销)。没有这个会话时返回 false。</summary>
public async ValueTask<bool> DisconnectAsync(string probeId)
{
if (_sessions.Get(probeId)?.Rpc is not { } rpc)
bool disconnected = false;
foreach (ProbeSession probe in _sessions.All)
{
return false;
if (probe.ProbeId == probeId && probe.Rpc is { } rpc)
{
await rpc.CloseAsync("disconnected", CancellationToken.None);
disconnected = true;
}
}
await rpc.CloseAsync("disconnected", CancellationToken.None);
return true;
return disconnected;
}
/// <summary>名单(期望状态里的 peers)里已经没有的 Probe:立刻关闭它的会话(吊销生效)。</summary>
@@ -80,7 +80,7 @@ public sealed class TargetRunners
Log = _log,
Session = () => null,
PeerAddress = address,
Fragments = new LocalSink(this, spec),
Fragments = new LocalSink(this, spec, address.AddressFamily == System.Net.Sockets.AddressFamily.InterNetworkV6 ? IpVersions.V6 : IpVersions.V4),
};
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);
@@ -132,11 +132,13 @@ public sealed class TargetRunners
{
private readonly TargetRunners _owner;
private readonly TaskSpec _spec;
private readonly string _ipVersion;
public LocalSink(TargetRunners owner, TaskSpec spec)
public LocalSink(TargetRunners owner, TaskSpec spec, string ipVersion)
{
_owner = owner;
_spec = spec;
_ipVersion = ipVersion;
}
public void Ship(string kind, FragmentScope scope, IReadOnlyList<RawStream> streams, IReadOnlyDictionary<string, JsonElement>? parameters, long startUtcMs, int durationMs, int port)
@@ -149,6 +151,7 @@ public sealed class TargetRunners
TaskId = scope.Task,
AgentId = _owner._agent,
Target = _spec.Target,
IpVersion = _ipVersion,
Params = parameters is null ? [] : new Dictionary<string, JsonElement>(parameters),
Port = port,
StartUtcMs = startUtcMs,
@@ -46,7 +46,12 @@ public sealed class TestRunner
/// <summary>校验并接受一个测试(立即返回),实际运行在后台进行,状态通过 <c>moshiro.test.status</c> 回报。</summary>
public Result<TestRunResult> Start(TestRunParams run)
{
if (_sessions.Get(run.ProbeId.Value) is not { } probe)
if (!IpVersions.IsValid(run.IpVersion))
{
return Result<TestRunResult>.Fail(MoShiroErrors.ValidationFailed, "ip_version must be ipv4 or ipv6");
}
if (_sessions.Get(run.ProbeId.Value, run.IpVersion) is not { } probe)
{
return Result<TestRunResult>.Fail(MoShiroErrors.ProbeOffline, $"probe {run.ProbeId.Value} is not connected to this agent");
}
+7 -1
View File
@@ -57,6 +57,11 @@ public sealed class PeerListener
public async Task RunAsync(CancellationToken ct)
{
var listener = new TcpListener(_options.Listen);
if (_options.Listen.Address.Equals(IPAddress.IPv6Any))
{
listener.Server.DualMode = true;
}
listener.Start();
Port = ((IPEndPoint)listener.LocalEndpoint).Port;
Started.TrySetResult();
@@ -219,7 +224,8 @@ public static class PeerConnector
public static async Task<(PeerSessionInfo Info, TlsFrameTransport Transport)> ConnectAsync(
string host, int port, byte[] expectedSpkiSha256, NodeIdentity identity, string peerId, string? version, int maxFrameBytes, TimeProvider time, CancellationToken ct)
{
var client = new TcpClient { NoDelay = true };
TcpClient client = IPAddress.TryParse(host, out IPAddress? address) ? new TcpClient(address.AddressFamily) : new TcpClient();
client.NoDelay = true;
PeerSockets.EnableKeepAlive(client.Client);
try
{
+58
View File
@@ -61,6 +61,13 @@
"schema": {
"type": "string"
}
},
{
"name": "ip_version",
"in": "query",
"schema": {
"type": "string"
}
}
],
"responses": {
@@ -706,6 +713,15 @@
],
"summary": "Server-sent events: nodes, overview, series and test topics",
"operationId": "live_stream",
"parameters": [
{
"name": "ip_version",
"in": "query",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "OK"
@@ -4333,6 +4349,14 @@
},
"last_seen_at": {
"format": "date-time"
},
"ipv6_host": {
"type": "string",
"default": ""
},
"ipv6_supported": {
"type": "boolean",
"default": false
}
}
},
@@ -4371,6 +4395,12 @@
"null",
"boolean"
]
},
"ipv6_host": {
"type": [
"null",
"string"
]
}
}
},
@@ -4624,6 +4654,10 @@
},
"connected_since": {
"format": "date-time"
},
"ip_version": {
"type": "string",
"default": "ipv4"
}
}
},
@@ -5555,6 +5589,14 @@
"$ref": "#/components/schemas/TestSummaryDto"
}
]
},
"ip_version": {
"type": "string",
"default": "ipv4"
},
"ipv6_available": {
"type": "boolean",
"default": false
}
}
},
@@ -6131,6 +6173,10 @@
"$ref": "#/components/schemas/TestSummaryDto"
}
]
},
"ip_version": {
"type": "string",
"default": "ipv4"
}
}
},
@@ -6727,6 +6773,10 @@
},
"connected_since": {
"format": "date-time"
},
"ip_version": {
"type": "string",
"default": "ipv4"
}
}
},
@@ -7206,6 +7256,10 @@
"additionalProperties": {
"type": "string"
}
},
"ip_version": {
"type": "string",
"default": "ipv4"
}
}
},
@@ -7458,6 +7512,10 @@
"$ref": "#/components/schemas/TestResultDto"
}
]
},
"ip_version": {
"type": "string",
"default": "ipv4"
}
}
},
@@ -74,8 +74,20 @@ public sealed class AgentConnection
private async Task<string> RunOnceAsync(Backoff backoff, CancellationToken ct)
{
AddressFamily family = Endpoint.IpVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork;
using var resolveTimeout = new CancellationTokenSource(TimeSpan.FromSeconds(10), _env.Time);
using CancellationTokenSource resolveCt = CancellationTokenSource.CreateLinkedTokenSource(ct, resolveTimeout.Token);
IPAddress[] addresses = IPAddress.TryParse(Endpoint.Host, out IPAddress? literal)
? [literal]
: await Dns.GetHostAddressesAsync(Endpoint.Host, family, resolveCt.Token);
IPAddress? address = addresses.FirstOrDefault(a => a.AddressFamily == family && !a.IsIPv4MappedToIPv6);
if (address is null)
{
throw new SocketException((int)SocketError.HostNotFound);
}
(PeerSessionInfo info, TlsFrameTransport transport) = await PeerConnector.ConnectAsync(
Endpoint.Host, Endpoint.TcpPort, Convert.FromBase64String(Endpoint.Spki), _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct);
address.ToString(), Endpoint.TcpPort, Convert.FromBase64String(Endpoint.Spki), _env.Identity, _env.ProbeId, ProbeApp.Version, 8 << 20, _env.Time, ct);
var remote = (IPEndPoint)info.Remote!;
var data = new DataSession(info.Credentials.SessionIdValue, info.Credentials.Key);
using var udp = new UdpMux(new SingleSession(data), null, _env.Clock, new DataPlaneCounters(), _env.Log, remote.AddressFamily == AddressFamily.InterNetworkV6 ? IPAddress.IPv6Any : IPAddress.Any, []);
@@ -67,7 +67,19 @@ public sealed class AgentDirectory
/// <summary>应用面板下发的列表。版本不比当前新的忽略(乱序到达的旧推送)。</summary>
public void Apply(ProbeAgentList result)
{
var next = result.Agents.Select(a => new KnownAgent { AgentId = a.AgentId.Value, Host = a.Host, TcpPort = a.TcpPort, Spki = a.Spki }).ToList();
var next = new List<KnownAgent>();
foreach (AgentEndpoint agent in result.Agents)
{
if (!string.IsNullOrWhiteSpace(agent.Host))
{
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Host, TcpPort = agent.TcpPort, Spki = agent.Spki });
}
if (!string.IsNullOrWhiteSpace(agent.Ipv6Host))
{
next.Add(new KnownAgent { AgentId = agent.AgentId.Value, Host = agent.Ipv6Host, TcpPort = agent.TcpPort, Spki = agent.Spki, IpVersion = IpVersions.V6 });
}
}
lock (_gate)
{
if (result.Version < _version || (result.Version == _version && Same(next, _agents)))
@@ -96,5 +108,5 @@ public sealed class AgentDirectory
a.Count == b.Count && a.Zip(b).All(p => Equal(p.First, p.Second));
public static bool Equal(KnownAgent a, KnownAgent b) =>
a.AgentId == b.AgentId && a.Host == b.Host && a.TcpPort == b.TcpPort && a.Spki == b.Spki;
a.AgentId == b.AgentId && a.IpVersion == b.IpVersion && a.Host == b.Host && a.TcpPort == b.TcpPort && a.Spki == b.Spki;
}
+5 -5
View File
@@ -70,19 +70,19 @@ public sealed class ProbeApp : IDisposable
public AgentDirectory Directory => _env.Directory;
/// <summary>某个 Agent 的会话是否已经建立(测试和状态查看用)。</summary>
public bool IsConnected(string agentId)
public bool IsConnected(string agentId, string ipVersion = IpVersions.V4)
{
lock (_connections)
{
return _connections.TryGetValue(agentId, out Running? running) && running.Connection.IsConnected;
return _connections.TryGetValue(agentId + "~" + ipVersion, out Running? running) && running.Connection.IsConnected;
}
}
public AgentConnection? Connection(string agentId)
public AgentConnection? Connection(string agentId, string ipVersion = IpVersions.V4)
{
lock (_connections)
{
return _connections.TryGetValue(agentId, out Running? running) ? running.Connection : null;
return _connections.TryGetValue(agentId + "~" + ipVersion, out Running? running) ? running.Connection : null;
}
}
@@ -121,7 +121,7 @@ public sealed class ProbeApp : IDisposable
private async Task ReconcileAsync(IReadOnlyList<KnownAgent> agents, CancellationToken ct)
{
var wanted = agents.ToDictionary(a => a.AgentId);
var wanted = agents.ToDictionary(a => a.PathKey);
var stopped = new List<Running>();
lock (_connections)
{
@@ -1,4 +1,5 @@
using System.Text.Json.Serialization;
using MoShiro.Protocol;
namespace MoShiro.Probe;
@@ -25,6 +26,12 @@ public sealed class ProbeFile
public sealed class KnownAgent
{
[JsonPropertyName("ip_version")]
public string IpVersion { get; set; } = IpVersions.V4;
[JsonIgnore]
public string PathKey => AgentId + "~" + IpVersion;
[JsonPropertyName("agent_id")]
public string AgentId { get; set; } = "";
@@ -6,6 +6,31 @@ namespace MoShiro.Probe.Tests;
public sealed class ProbeTests
{
[Fact]
public void AgentDirectory_DualStack_PersistsBothPaths_AndRemovingIpv6KeepsIpv4()
{
string directory = NewDirectory();
try
{
var agents = new AgentDirectory(new ConfigStore(directory), directory, NullNodeLog.Instance);
ProbeAgentList list = List(1, "agt_dual");
list.Agents[0].Ipv6Host = "2001:db8::1";
agents.Apply(list);
Assert.Equal(new[] { "ipv4", "ipv6" }, agents.Agents.Select(a => a.IpVersion));
var reloaded = new AgentDirectory(new ConfigStore(directory), directory, NullNodeLog.Instance);
Assert.Equal(2, reloaded.Agents.Count);
Assert.Equal("2001:db8::1", reloaded.Agents[1].Host);
list.Version = 2;
list.Agents[0].Ipv6Host = null;
reloaded.Apply(list);
Assert.Equal("ipv4", Assert.Single(reloaded.Agents).IpVersion);
}
finally
{
Directory.Delete(directory, recursive: true);
}
}
[Fact]
public void ClockEstimator_UsesTheSampleWithTheShortestRoundTrip()
{
@@ -69,6 +69,7 @@ public static class ChunkFinalizer
Phase = source.Phase,
AgentId = agent,
ProbeId = source.ProbeId,
IpVersion = source.IpVersion,
Target = null,
Params = new Dictionary<string, JsonElement>(resolvedParams),
Port = port,
@@ -39,7 +39,7 @@ public sealed class TcpLink : IDisposable
/// </summary>
public static async Task<TcpLink> OpenAsync(BlockEnv env, int port, bool tls, uint streamId, bool sendAuth, CancellationToken ct)
{
var client = new TcpClient { NoDelay = true };
var client = new TcpClient(env.PeerAddress!.AddressFamily) { NoDelay = true };
try
{
long start = env.Clock.NowUs();
@@ -153,6 +153,11 @@ public sealed class TcpAcceptor
foreach (int port in _ports)
{
var listener = new TcpListener(_bind, port);
if (_bind.Equals(IPAddress.IPv6Any))
{
listener.Server.DualMode = true;
}
listener.Start(128);
_listeners.Add(listener);
bound.Add(((IPEndPoint)listener.LocalEndpoint).Port);
@@ -0,0 +1,13 @@
namespace MoShiro.Protocol;
/// <summary>测量路径的地址族。旧契约缺省为 IPv4;IPv6 使用独立的线路和序列。</summary>
public static class IpVersions
{
public const string V4 = "ipv4";
public const string V6 = "ipv6";
public static bool IsValid(string? value) => value is V4 or V6;
public static string LineSeries(string probeId, string ipVersion) =>
RawChunkHeader.LinePrefix + probeId + (ipVersion == V6 ? ":ipv6" : "");
}
@@ -58,6 +58,10 @@ public sealed class StreamHeader
/// <summary>块头(JSON,自描述):列定义写在块里,新增列或新增 Kind 不需要升级 container_version。</summary>
public sealed class RawChunkHeader
{
[JsonPropertyName("ip_version")]
[JsonPropertyOrder(1)]
public string IpVersion { get; set; } = IpVersions.V4;
[JsonPropertyName("chunk_id")]
public ChunkId ChunkId { get; set; }
@@ -113,10 +117,10 @@ public sealed class RawChunkHeader
/// <summary>
/// "line:&lt;probe_id&gt;" 或 "target:&lt;task_id&gt;":存储和分析用的序列标识。
/// 一条线路是一对(Probe,Agent):同一个 Agent 上每个 Probe 的数据各是一个序列,互不混合。
/// 同一个 Agent 上每个 Probe 的每种地址族各是一个序列,IPv6 追加 :ipv6,互不混合。
/// </summary>
[JsonIgnore]
public string Series => TaskId is { } id && Target is not null ? "target:" + id.Value : LinePrefix + (ProbeId?.Value ?? "");
public string Series => TaskId is { } id && Target is not null ? "target:" + id.Value : IpVersions.LineSeries(ProbeId?.Value ?? "", IpVersion);
public const string LinePrefix = "line:";
}
@@ -61,6 +61,9 @@ public sealed class TestPhase
public sealed class TestRunParams
{
[JsonPropertyName("ip_version")]
public string IpVersion { get; set; } = IpVersions.V4;
[JsonPropertyName("test_id")]
public TestId TestId { get; set; }
@@ -127,6 +130,9 @@ public sealed class AgentEventParams
public sealed class AgentEndpoint
{
[JsonPropertyName("ipv6_host")]
public string? Ipv6Host { get; set; }
[JsonPropertyName("agent_id")]
public NodeId AgentId { get; set; }
@@ -289,6 +295,10 @@ public sealed class ProbeHelloMeta
/// <summary><c>fleet.hello</c> 的附加字段。</summary>
public sealed class AgentHelloMeta
{
/// <summary>本机是否启用了 IPv6 监听。旧 Agent 缺省为 false,面板不会给它安排 IPv6 会话。</summary>
[JsonPropertyName("ipv6_supported")]
public bool Ipv6Supported { get; set; }
[JsonPropertyName("platform")]
public PlatformInfo? Platform { get; set; }
@@ -353,6 +363,9 @@ public sealed class HostSnapshot
/// <summary>Agent 眼中的一个已连接 Probe。</summary>
public sealed class ConnectedProbe
{
[JsonPropertyName("ip_version")]
public string IpVersion { get; set; } = IpVersions.V4;
[JsonPropertyName("id")]
public ProbeId Id { get; set; }
@@ -8,6 +8,21 @@ namespace MoShiro.Protocol.Tests;
public sealed class RawChunkCodecTests
{
[Fact]
public void Header_LegacyDefaultsToIpv4_AndIpv6RoundTripsToAnIndependentSeries()
{
RawChunkHeader legacy = System.Text.Json.JsonSerializer.Deserialize("{\"probe_id\":\"prb_legacy\"}", MoShiroJsonContext.Default.RawChunkHeader)!;
Assert.Equal(IpVersions.V4, legacy.IpVersion);
Assert.Equal("line:prb_legacy", legacy.Series);
RawChunk chunk = SampleChunk();
chunk.Header.ProbeId = new ProbeId("prb_dual");
chunk.Header.IpVersion = IpVersions.V6;
RawChunk decoded = RawChunkCodec.Decode(RawChunkCodec.Encode(chunk)).Value;
Assert.Equal("ipv6", decoded.Header.IpVersion);
Assert.Equal("line:prb_dual:ipv6", decoded.Header.Series);
Assert.Equal(chunk.Find("msg")!.Count, decoded.Find("msg")!.Count);
}
private static readonly NodeId s_agent = new("agt_test");
private static RawChunk SampleChunk()
@@ -87,7 +87,7 @@ public static class AgentEndpoints
private static async Task<IResult> PatchAsync(HttpContext http, AgentService agents, AgentPatchRequest request, string id, CancellationToken ct)
{
Result<AgentView> result = await agents.PatchAsync(
new NodeId(id), new AgentPatch { Name = request.Name, Labels = request.Labels, PublicHost = request.PublicHost, PublicVisible = request.PublicVisible }, Concurrency.ExpectedRowVersion(http), ct);
new NodeId(id), new AgentPatch { Name = request.Name, Labels = request.Labels, PublicHost = request.PublicHost, Ipv6Host = request.Ipv6Host, PublicVisible = request.PublicVisible }, Concurrency.ExpectedRowVersion(http), ct);
if (!result.TryGet(out AgentView? view, out FleetError? error))
{
return Problems.From(http, error);
@@ -109,7 +109,7 @@ public static class TaskEndpoints
return Problems.From(http, lineError);
}
Result<TestRecord> result = await tests.StartAsync(line.Key.Agent, line.Key.Probe, new StartTestRequest(request.Plan ?? "", request.Params), "api", ct);
Result<TestRecord> result = await tests.StartAsync(line.Key.Agent, line.Key.Probe, new StartTestRequest(request.Plan ?? "", request.Params), "api", line.IpVersion, ct);
if (!result.TryGet(out TestRecord? test, out FleetError? error))
{
return Problems.From(http, error);
@@ -124,10 +124,12 @@ public static class TaskEndpoints
// 只给 Agent id(并且有多个 Probe)时列出这个 Agent 上所有 Probe 的测试
NodeId agent;
ProbeId? probe = null;
string? ipVersion = null;
if (LineKey.TryParse(line_id, out LineKey key))
{
agent = key.Agent;
probe = key.Probe;
ipVersion = key.IpVersion;
}
else
{
@@ -135,7 +137,7 @@ public static class TaskEndpoints
}
return TypedResults.Json(
new PageDto<TestDto>([.. (await tests.ListAsync(agent, probe, Math.Clamp(limit ?? 50, 1, 500), ct)).Select(Mappers.ToDto)], null),
new PageDto<TestDto>([.. (await tests.ListAsync(agent, probe, Math.Clamp(limit ?? 50, 1, 500), ipVersion, ct)).Select(Mappers.ToDto)], null),
ApiJsonContext.Default.PageDtoTestDto);
}
@@ -31,34 +31,34 @@ public static class Mappers
public static OverviewLineDto ToDto(OverviewLine line) =>
new(
line.Line.Key.Id, line.Line.Agent.Agent.Id.Value, line.Line.Probe.Id, line.Line.Probe.Name, line.Line.Agent.Agent.Name, line.Line.Agent.Agent.Labels,
line.Line.Agent.Online, line.Line.Connected, line.Line.Agent.EndpointChanges, ToDto(line.Summary), Summary(line.LastTest));
line.Line.Agent.Online, line.Line.Connected, line.Line.Agent.EndpointChanges, ToDto(line.Summary), Summary(line.LastTest), line.Line.IpVersion);
public static HostDto ToDto(HostSnapshot h) =>
new(h.CpuPercent, h.MemUsed, h.MemTotal, h.DiskUsed, h.DiskTotal, h.NetRxTotal, h.NetTxTotal, h.NetRxBps, h.NetTxBps, h.NetLinkMbps, h.UptimeSeconds, Time(h.AtUtcMs));
public static ConnectedProbeDto ToDto(ConnectedProbeInfo p) => new(p.Id, p.Version, p.Remote, Time(p.SinceMs));
public static ConnectedProbeDto ToDto(ConnectedProbeInfo p) => new(p.Id, p.Version, p.Remote, Time(p.SinceMs), p.IpVersion);
public static ProbeDto ToDto(ProbeView v) =>
new(
v.Probe.Id, v.Probe.Name, Time(v.Probe.CreatedAtMs), Time(v.Probe.RevokedAtMs), v.Revoked, v.Online,
[.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs)))]);
[.. v.Connections.Select(c => new ProbeConnectionDto(c.AgentId.Value, c.AgentName, c.AgentOnline, c.Version, c.Remote, Time(c.SinceMs), c.IpVersion))]);
public static SeriesDto ToDto(SeriesResult r, bool live) =>
new(
r.Line.Id, r.Line.Agent.Value, r.Line.Probe.Value, Time(r.Range.FromMs), Time(r.Range.ToMs), Steps.Format(r.Range.StepMs), r.Plan.Name, live && r.Range.IsRelative, r.Plan.Degraded,
r.T, r.Values, ToDto(r.Summary), new EdgePartialDto(r.Range.FirstPartial, r.Range.LastPartial),
[.. r.Gaps.Select(g => new GapDto(Time(g.FromMs), Time(g.ToMs), g.Reason))], [.. r.Events.Select(ToDto)], r.Versions);
[.. r.Gaps.Select(g => new GapDto(Time(g.FromMs), Time(g.ToMs), g.Reason))], [.. r.Events.Select(ToDto)], r.Versions, r.Line.IpVersion);
public static TestDto ToDto(TestDetail detail)
{
TestRecord t = detail.Test;
return new TestDto(
t.Id.Value, t.AgentId.Value, t.ProbeId.Value, t.Plan, detail.PhaseKinds, t.Status, t.Phase, t.ErrorCode, t.Trigger,
Time(t.RequestedAtMs), Time(t.StartedAtMs), Time(t.FinishedAtMs), detail.Result is null ? null : ToDto(detail.Result));
Time(t.RequestedAtMs), Time(t.StartedAtMs), Time(t.FinishedAtMs), detail.Result is null ? null : ToDto(detail.Result), t.IpVersion);
}
public static TestDto ToDto(TestRecord t) =>
new(t.Id.Value, t.AgentId.Value, t.ProbeId.Value, t.Plan, [.. (Plans.Find(t.Plan)?.Phases ?? []).Select(p => p.Kind)], t.Status, t.Phase, t.ErrorCode, t.Trigger, Time(t.RequestedAtMs), Time(t.StartedAtMs), Time(t.FinishedAtMs), null);
new(t.Id.Value, t.AgentId.Value, t.ProbeId.Value, t.Plan, [.. (Plans.Find(t.Plan)?.Phases ?? []).Select(p => p.Kind)], t.Status, t.Phase, t.ErrorCode, t.Trigger, Time(t.RequestedAtMs), Time(t.StartedAtMs), Time(t.FinishedAtMs), null, t.IpVersion);
public static TestResultDto ToDto(TestResultView r) =>
new(
@@ -103,7 +103,7 @@ public static class Mappers
: null;
return new AgentDto(
a.Id.Value, a.Name, StatusName(a.Status), view.Online, view.ProbeConnected, [.. view.Probes.Select(ToDto)], view.Host is null ? null : ToDto(view.Host), a.PublicHost, a.ProbeTcpPort, a.ProbeUdpPorts, a.Labels, a.PublicVisible,
a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, Time(a.CreatedAtMs), Time(a.LastSeenAtMs));
a.Kinds, a.AgentVersion, limits, a.TaskVersion, a.AppliedTaskVersion, rejected, heartbeat, view.EndpointChanges, Time(a.CreatedAtMs), Time(a.LastSeenAtMs), a.Ipv6Host, a.Ipv6Supported);
}
public static TaskDto ToDto(TaskRecord t) =>
@@ -37,10 +37,12 @@ public sealed record AgentDto(
HeartbeatDto? Heartbeat,
int EndpointChanges,
DateTimeOffset CreatedAt,
DateTimeOffset? LastSeenAt);
DateTimeOffset? LastSeenAt,
string Ipv6Host = "",
bool Ipv6Supported = false);
/// <summary>merge-patch:缺省或为 null 的字段不修改。</summary>
public sealed record AgentPatchRequest(string? Name, IReadOnlyList<string>? Labels, string? PublicHost, bool? PublicVisible);
public sealed record AgentPatchRequest(string? Name, IReadOnlyList<string>? Labels, string? PublicHost, bool? PublicVisible, string? Ipv6Host = null);
public sealed record JoinTokenDto(string Id, string? Note, IReadOnlyList<string> Labels, bool AutoApprove, int MaxUses, int Uses, DateTimeOffset ExpiresAt, DateTimeOffset? RevokedAt, DateTimeOffset CreatedAt);
@@ -52,10 +54,10 @@ public sealed record InstallCommandsDto(string Linux, string Macos, string Windo
public sealed record CreatedJoinTokenDto(JoinTokenDto Token, string Plaintext, InstallCommandsDto InstallCommands);
/// <summary>Agent 眼中连着的一个 Probe(Agent 上报)。</summary>
public sealed record ConnectedProbeDto(string ProbeId, string? Version, string? Remote, DateTimeOffset ConnectedSince);
public sealed record ConnectedProbeDto(string ProbeId, string? Version, string? Remote, DateTimeOffset ConnectedSince, string IpVersion = "ipv4");
/// <summary>某个 Probe 现在连着的一个 Agent。</summary>
public sealed record ProbeConnectionDto(string AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, DateTimeOffset ConnectedSince);
public sealed record ProbeConnectionDto(string AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, DateTimeOffset ConnectedSince, string IpVersion = "ipv4");
public sealed record ProbeDto(string Id, string Name, DateTimeOffset CreatedAt, DateTimeOffset? RevokedAt, bool Revoked, bool Online, IReadOnlyList<ProbeConnectionDto> Connections);
@@ -22,7 +22,8 @@ public sealed record OverviewLineDto(
bool ProbeConnected,
int EndpointChanges,
SummaryDto Summary,
TestSummaryDto? LastTest);
TestSummaryDto? LastTest,
string IpVersion = "ipv4");
public sealed record OverviewDto(string Range, DateTimeOffset From, DateTimeOffset To, string Step, string Source, bool Degraded, IReadOnlyList<OverviewLineDto> Lines);
@@ -43,7 +44,8 @@ public sealed record SeriesDto(
EdgePartialDto EdgePartial,
IReadOnlyList<GapDto> Gaps,
IReadOnlyList<EventDto> Events,
IReadOnlyDictionary<string, string> Versions);
IReadOnlyDictionary<string, string> Versions,
string IpVersion = "ipv4");
public sealed record CompareDto(
string Metric,
@@ -81,7 +83,9 @@ public sealed record LineDto(
bool ProbeConnected,
DateTimeOffset? DataSince,
IReadOnlyList<string> Kinds,
TestSummaryDto? LastTest);
TestSummaryDto? LastTest,
string IpVersion = "ipv4",
bool Ipv6Available = false);
public sealed record DimensionDto(string Id, string Label, double Weight, double? Value, double? Score);
@@ -109,7 +113,8 @@ public sealed record TestDto(
DateTimeOffset RequestedAt,
DateTimeOffset? StartedAt,
DateTimeOffset? FinishedAt,
TestResultDto? Result);
TestResultDto? Result,
string IpVersion = "ipv4");
// ── 目录(由代码注册表生成) ──────────────────────────────────────────
@@ -33,8 +33,14 @@ public static class LiveEndpoints
private static async Task<IResult> StreamAsync(
HttpContext http, AgentService agents, LineService lineService, TestService tests, QueryService query, LiveHub hub, SseLimiter limiter,
BackendOptions options, BackendCounters counters, TimeProvider time, IHostApplicationLifetime lifetime)
BackendOptions options, BackendCounters counters, TimeProvider time, IHostApplicationLifetime lifetime, string? ip_version)
{
string ipVersion = ip_version ?? IpVersions.V4;
if (!IpVersions.IsValid(ipVersion))
{
return Problems.From(http, ValidationError.Single("ip_version", "must be ipv4 or ipv6"));
}
StringValues raw = http.Request.Query["topics"];
Result<IReadOnlyList<LiveTopic>> parsed = LiveTopic.ParseAll(raw.Where(v => v is not null)!, options.MaxSseTopics);
if (!parsed.TryGet(out IReadOnlyList<LiveTopic>? topics, out FleetError? error))
@@ -78,7 +84,7 @@ public static class LiveEndpoints
http.Response.Headers["X-Accel-Buffering"] = "no";
await http.Response.Body.FlushAsync(http.RequestAborted);
var session = new LiveSession(http, agents, lineService, tests, query, hub, options, time, topics, lineKeys);
var session = new LiveSession(http, agents, lineService, tests, query, hub, options, time, topics, lineKeys, ipVersion);
using var linked = CancellationTokenSource.CreateLinkedTokenSource(http.RequestAborted, lifetime.ApplicationStopping);
await session.RunAsync(linked.Token);
}
@@ -101,6 +107,7 @@ public static class LiveEndpoints
private readonly AgentService _agents;
private readonly LineService _lines;
private readonly IReadOnlyDictionary<string, LineKey> _lineKeys;
private readonly string _ipVersion;
private readonly TestService _tests;
private readonly QueryService _query;
private readonly LiveHub _hub;
@@ -117,12 +124,13 @@ public static class LiveEndpoints
private long _lastWriteMs;
private long _lastPeriodicMs;
public LiveSession(HttpContext http, AgentService agents, LineService lines, TestService tests, QueryService query, LiveHub hub, BackendOptions options, TimeProvider time, IReadOnlyList<LiveTopic> topics, IReadOnlyDictionary<string, LineKey> lineKeys)
public LiveSession(HttpContext http, AgentService agents, LineService lines, TestService tests, QueryService query, LiveHub hub, BackendOptions options, TimeProvider time, IReadOnlyList<LiveTopic> topics, IReadOnlyDictionary<string, LineKey> lineKeys, string ipVersion)
{
_http = http;
_agents = agents;
_lines = lines;
_lineKeys = lineKeys;
_ipVersion = ipVersion;
_tests = tests;
_query = query;
_hub = hub;
@@ -297,7 +305,7 @@ public static class LiveEndpoints
{
MetricSelection selection = MetricSelection.Resolve(null, topic.Preset, _options.MaxMetricsPerQuery).Value;
IReadOnlyList<LineView> visible = await _lines.ListAsync(a => a.Agent.Status == Fleet.Server.NodeStatus.Active && IsVisible(a), ct);
Result<OverviewResult> result = await _query.OverviewAsync(visible, new RangeRequest(topic.Range, null, null, null, null), selection, ct);
Result<OverviewResult> result = await _query.OverviewAsync([.. visible.Where(l => l.IpVersion == _ipVersion)], new RangeRequest(topic.Range, null, null, null, null), selection, ct);
if (result.TryGet(out OverviewResult? overview, out _))
{
var dto = new OverviewDto(
@@ -43,9 +43,15 @@ public static class QueryEndpoints
private static Task<IResult> OverviewAsync(
HttpContext http, LineService lineService, QueryService query, BackendOptions options, TimeProvider time,
string? range, DateTimeOffset? from, DateTimeOffset? to, string? preset, string? metrics, string? sort, string? order) =>
string? range, DateTimeOffset? from, DateTimeOffset? to, string? preset, string? metrics, string? sort, string? order, string? ip_version) =>
QueryHelpers.GuardedAsync(http, async ct =>
{
string ipVersion = ip_version ?? IpVersions.V4;
if (!IpVersions.IsValid(ipVersion))
{
return Problems.From(http, ValidationError.Single("ip_version", "must be ipv4 or ipv6"));
}
Result<MetricSelection> selection = MetricSelection.Resolve(metrics, preset ?? (metrics is null ? DefaultOverviewPreset : null), options.MaxMetricsPerQuery);
if (!selection.TryGet(out MetricSelection? sel, out FleetError? error))
{
@@ -53,7 +59,7 @@ public static class QueryEndpoints
}
IReadOnlyList<LineView> visible = await lineService.ListAsync(a => a.Agent.Status == Fleet.Server.NodeStatus.Active && (a.Agent.PublicVisible || PublicGuard.CanSeeHidden(http)), ct);
Result<OverviewResult> result = await query.OverviewAsync(visible, new RangeRequest(range, from, to, null, null), sel, ct);
Result<OverviewResult> result = await query.OverviewAsync([.. visible.Where(l => l.IpVersion == ipVersion)], new RangeRequest(range, from, to, null, null), sel, ct);
if (!result.TryGet(out OverviewResult? overview, out error))
{
return Problems.From(http, error);
@@ -81,12 +87,12 @@ public static class QueryEndpoints
}
AgentView view = line.Agent;
IReadOnlyList<TestRecord> recent = await tests.ListAsync(view.Agent.Id, new ProbeId(line.Probe.Id), 1, ct);
IReadOnlyList<TestRecord> recent = await tests.ListAsync(view.Agent.Id, new ProbeId(line.Probe.Id), 1, line.IpVersion, ct);
TestRecord? last = recent.Count > 0 ? recent[0] : null;
return TypedResults.Json(
new LineDto(
line.Key.Id, view.Agent.Id.Value, line.Probe.Id, line.Probe.Name, view.Agent.Name, view.Agent.Labels, view.Online, line.Connected,
Mappers.Time(await query.EarliestAsync(line.Key, ct)), view.Agent.Kinds, Mappers.Summary(last)),
Mappers.Time(await query.EarliestAsync(line.Key, ct)), view.Agent.Kinds, Mappers.Summary(last), line.IpVersion, view.Agent.Ipv6Host.Length > 0),
ApiJsonContext.Default.LineDto);
}
@@ -34,11 +34,11 @@ public sealed class AgentEventService
switch (e.Kind)
{
case AgentEventKinds.ProbeConnected when !string.IsNullOrEmpty(e.Detail):
state.AddProbe(e.Probe is { } p ? new ConnectedProbeInfo(e.Detail, p.Version, p.Remote, p.SinceUtcMs > 0 ? p.SinceUtcMs : at) : new ConnectedProbeInfo(e.Detail, null, null, at));
state.AddProbe(e.Probe is { } p ? new ConnectedProbeInfo(e.Detail, p.Version, p.Remote, p.SinceUtcMs > 0 ? p.SinceUtcMs : at, p.IpVersion) : new ConnectedProbeInfo(e.Detail, null, null, at));
_nodes.ObserveProbeInBackground(e.Detail, e.Probe?.Remote);
break;
case AgentEventKinds.ProbeDisconnected when !string.IsNullOrEmpty(e.Detail):
state.RemoveProbe(e.Detail);
state.RemoveProbe(e.Detail, e.Probe?.IpVersion ?? IpVersions.V4);
break;
case AgentEventKinds.EndpointChanged:
state.CountEndpointChange();
@@ -5,7 +5,7 @@ using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>Agent 眼中的一个已连接 Probe(Agent 在事件和心跳里上报)。</summary>
public sealed record ConnectedProbeInfo(string Id, string? Version, string? Remote, long SinceMs);
public sealed record ConnectedProbeInfo(string Id, string? Version, string? Remote, long SinceMs, string IpVersion = IpVersions.V4);
/// <summary>Agent 的运行时状态(只在内存中):连着哪些 Probe、最近的进程内计数、端点变化次数。</summary>
public sealed class AgentRuntimeState
@@ -19,22 +19,22 @@ public sealed class AgentRuntimeState
public bool ProbeConnected => Probes.Count > 0;
public bool IsConnected(string probeId) => Probes.Any(p => p.Id == probeId);
public bool IsConnected(string probeId, string ipVersion = IpVersions.V4) => Probes.Any(p => p.Id == probeId && p.IpVersion == ipVersion);
/// <summary>事件 <c>probe_connected</c>:事件里带着 Probe 的版本和地址;心跳里的完整快照随后会再校正一次。</summary>
public void AddProbe(ConnectedProbeInfo info)
{
lock (_gate)
{
Volatile.Write(ref _probes, [.. _probes.Where(p => p.Id != info.Id), info]);
Volatile.Write(ref _probes, [.. _probes.Where(p => p.Id != info.Id || p.IpVersion != info.IpVersion), info]);
}
}
public void RemoveProbe(string id)
public void RemoveProbe(string id, string ipVersion = IpVersions.V4)
{
lock (_gate)
{
Volatile.Write(ref _probes, [.. _probes.Where(p => p.Id != id)]);
Volatile.Write(ref _probes, [.. _probes.Where(p => p.Id != id || p.IpVersion != ipVersion)]);
}
}
@@ -1,3 +1,5 @@
using System.Net;
using System.Net.Sockets;
using System.Text.RegularExpressions;
using Fleet.Core;
using Fleet.Server;
@@ -49,6 +51,16 @@ public sealed partial class AgentService
return new AgentView(agent, _gateway.IsOnline(agent.Id), state?.Probes ?? [], state?.LastHeartbeat, state?.EndpointChanges ?? 0, state?.LastHeartbeatAtMs, state?.Host);
}
private static bool ValidHost(string host, string ipVersion)
{
if (IPAddress.TryParse(host, out IPAddress? address))
{
return !address.IsIPv4MappedToIPv6 && address.AddressFamily == (ipVersion == IpVersions.V6 ? AddressFamily.InterNetworkV6 : AddressFamily.InterNetwork);
}
return host.Length <= 253 && Uri.CheckHostName(host) == UriHostNameType.Dns;
}
public async ValueTask<Result<AgentView>> PatchAsync(NodeId id, AgentPatch patch, long? expectedRowVersion, CancellationToken ct)
{
var errors = new Dictionary<string, string>();
@@ -57,11 +69,16 @@ public sealed partial class AgentService
errors["name"] = "must be 1-64 characters: letters (any language), digits, '.', '_' or '-'";
}
if (patch.PublicHost is not null && !HostPattern().IsMatch(patch.PublicHost))
if (patch.PublicHost is not null && !ValidHost(patch.PublicHost, IpVersions.V4))
{
errors["public_host"] = "must be a host name or IP address";
}
if (patch.Ipv6Host is not null && patch.Ipv6Host.Length > 0 && !ValidHost(patch.Ipv6Host, IpVersions.V6))
{
errors["ipv6_host"] = "must be an IPv6 address or host name; empty disables IPv6 collection";
}
if (patch.Labels is not null && (patch.Labels.Count > 16 || patch.Labels.Any(l => l.Length is 0 or > 32)))
{
errors["labels"] = "at most 16 labels of 1-32 characters";
@@ -78,7 +95,7 @@ public sealed partial class AgentService
return Result<AgentView>.Fail(error);
}
if (patch.Name is not null || patch.PublicHost is not null)
if (patch.Name is not null || patch.PublicHost is not null || patch.Ipv6Host is not null)
{
await AfterAgentSetChangedAsync(ct);
}
@@ -139,6 +156,7 @@ public sealed partial class AgentService
AgentId = agent.Id,
Name = agent.Name,
Host = agent.PublicHost,
Ipv6Host = agent.Ipv6Supported && agent.Ipv6Host.Length > 0 ? agent.Ipv6Host : null,
TcpPort = agent.ProbeTcpPort,
UdpPorts = [.. agent.ProbeUdpPorts],
Spki = Convert.ToBase64String(agent.TlsSpki),
@@ -170,7 +188,4 @@ public sealed partial class AgentService
[GeneratedRegex(@"^[\p{L}\p{M}\p{N}._-]{1,64}$")]
private static partial Regex NamePattern();
[GeneratedRegex(@"^[A-Za-z0-9.:\-\[\]]{1,253}$")]
private static partial Regex HostPattern();
}
@@ -99,7 +99,7 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
_releases.RecordStatus(id, beat?.Update);
if (beat is not null)
{
state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs)));
state.ReplaceProbes(beat.Probes.Select(p => new ConnectedProbeInfo(p.Id.Value, p.Version, p.Remote, p.SinceUtcMs, p.IpVersion)));
foreach (ConnectedProbe p in beat.Probes)
{
_nodes.ObserveProbeInBackground(p.Id.Value, p.Remote);
@@ -196,14 +196,14 @@ public sealed class MoShiroNodeEventSink : INodeEventSink
AgentRecord? before = await _agents.GetAsync(id, ct);
string? limitsJson = meta.LocalLimits is null ? null : JsonSerializer.Serialize(meta.LocalLimits, LocalLimitsJson.Default.LocalLimits);
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts), ct);
await _agents.ApplyHelloAsync(id, new AgentHello(kinds, hello.NodeVersion, limitsJson, spki is { Length: 32 } ? spki : null, meta.ProbeTcpPort, meta.ProbeUdpPorts, meta.Ipv6Supported), ct);
// 端口或证书指纹变了,Probe 需要新的列表
AgentRecord? after = await _agents.GetAsync(id, ct);
_nodes.ObserveAgentInBackground(id.Value, after?.PublicHost, meta.Platform);
await _releases.OfferToNodeAsync(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.TlsSpki.AsSpan().SequenceEqual(after.TlsSpki) || before.ProbeTcpPort != after.ProbeTcpPort || !before.ProbeUdpPorts.SequenceEqual(after.ProbeUdpPorts) || before.Ipv6Supported != after.Ipv6Supported)
&& after.Status == NodeStatus.Active)
{
await _service.AfterAgentSetChangedAsync(ct);
@@ -13,7 +13,7 @@ public sealed record ProbeLimits(double MaxTestMbps, long MaxTotalPps);
public sealed record ProbeIdentityFile(ProbeId ProbeId, string PrivateKey, string Server, ProbeLimits Limits);
/// <summary>某个 Probe 现在连着的一个 Agent(Agent 上报)。</summary>
public sealed record ProbeConnection(NodeId AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, long SinceMs);
public sealed record ProbeConnection(NodeId AgentId, string AgentName, bool AgentOnline, string? Version, string? Remote, long SinceMs, string IpVersion = IpVersions.V4);
public sealed record ProbeView(ProbeRecord Probe, bool Online, IReadOnlyList<ProbeConnection> Connections)
{
@@ -178,7 +178,7 @@ public sealed partial class ProbeService
AgentRecord? agent = agents.FirstOrDefault(a => a.Id == agentId);
if (agent is not null)
{
connections.Add(new ProbeConnection(agentId, agent.Name, _gateway.IsOnline(agentId), info.Version, info.Remote, info.SinceMs));
connections.Add(new ProbeConnection(agentId, agent.Name, _gateway.IsOnline(agentId), info.Version, info.Remote, info.SinceMs, info.IpVersion));
}
}
@@ -69,7 +69,8 @@ public sealed partial class IngestService
RawChunkHeader header = chunk.Header;
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
string? problem =
header.AgentId != agent ? "agent_id does not match the authenticated node"
!IpVersions.IsValid(header.IpVersion) ? "unknown ip_version"
: header.AgentId != agent ? "agent_id does not match the authenticated node"
: header.ChunkId.Value != msgId ? "chunk_id does not match the message id"
: Kinds.Find(header.Kind) is null ? $"unknown kind {header.Kind}"
: header.DurationMs is <= 0 or > MaxDurationMs ? "duration out of range"
@@ -22,7 +22,9 @@ public sealed record AgentRecord(
string? RejectedJson,
long CreatedAtMs,
long? LastSeenAtMs,
long RowVersion);
long RowVersion,
string Ipv6Host = "",
bool Ipv6Supported = false);
/// <summary>管理员可以修改的字段(merge-patch 语义:为 null 的字段不改)。</summary>
public sealed class AgentPatch
@@ -33,6 +35,8 @@ public sealed class AgentPatch
public string? PublicHost { get; init; }
public string? Ipv6Host { get; init; }
public bool? PublicVisible { get; init; }
}
@@ -43,7 +47,8 @@ public sealed record AgentHello(
string? LocalLimitsJson,
byte[]? TlsSpki,
int? ProbeTcpPort,
IReadOnlyList<int>? ProbeUdpPorts);
IReadOnlyList<int>? ProbeUdpPorts,
bool Ipv6Supported = false);
/// <summary>注册之后 MoShiro 的补充:公开地址和端口。</summary>
public sealed record AgentEnrollment(string PublicHost, int ProbeTcpPort, IReadOnlyList<int> ProbeUdpPorts, IReadOnlyList<string> Kinds);
@@ -30,7 +30,8 @@ public sealed record TestRecord(
int PhaseCount,
long RequestedAtMs,
long? StartedAtMs,
long? FinishedAtMs);
long? FinishedAtMs,
string IpVersion = IpVersions.V4);
/// <summary>测试结果:指标与曲线都以 JSON 保存。</summary>
public sealed record TestResultRecord(TestId TestId, byte[] MetricsJson, byte[] SeriesJson);
@@ -108,6 +108,8 @@ public interface ITestStore
ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, CancellationToken ct);
ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, string? ipVersion, CancellationToken ct);
ValueTask<bool> UpdateStatusAsync(TestId id, string status, int? phase, string? errorCode, long nowMs, CancellationToken ct);
ValueTask SetPhaseCountAsync(TestId id, int phaseCount, CancellationToken ct);
@@ -3,35 +3,35 @@ using MoShiro.Protocol;
namespace MoShiro.Backend.Application;
/// <summary>一条线路 = 一对(Agent,Probe)。存储里它的序列是 <c>line:&lt;probe_id&gt;</c>,对外的 id 是 <c>&lt;agent_id&gt;~&lt;probe_id&gt;</c>。</summary>
public readonly record struct LineKey(NodeId Agent, ProbeId Probe)
/// <summary>线路按(Agent,Probe,地址族)隔离。IPv4 保留 <c>agent~probe</c> 和 <c>line:probe</c>;IPv6 分别追加 <c>~ipv6</c>、<c>:ipv6</c>。</summary>
public readonly record struct LineKey(NodeId Agent, ProbeId Probe, string IpVersion = IpVersions.V4)
{
public string Series => RawChunkHeader.LinePrefix + Probe.Value;
public string Series => IpVersions.LineSeries(Probe.Value, IpVersion);
public string Id => Agent.Value + "~" + Probe.Value;
public string Id => Agent.Value + "~" + Probe.Value + (IpVersion == IpVersions.V6 ? "~ipv6" : "");
public static bool TryParse(string text, out LineKey key)
{
int split = text.IndexOf('~', StringComparison.Ordinal);
if (split <= 0 || split == text.Length - 1)
string[] parts = text.Split('~');
if (parts.Length is < 2 or > 3 || parts[0].Length == 0 || parts[1].Length == 0 || (parts.Length == 3 && parts[2] != IpVersions.V6))
{
key = default;
return false;
}
key = new LineKey(new NodeId(text[..split]), new ProbeId(text[(split + 1)..]));
key = new LineKey(new NodeId(parts[0]), new ProbeId(parts[1]), parts.Length == 3 ? IpVersions.V6 : IpVersions.V4);
return true;
}
}
/// <summary>一条线路的展示视图:Agent 的状态、Probe 记录,以及这一对现在是否连着。</summary>
public sealed record LineView(AgentView Agent, ProbeRecord Probe)
public sealed record LineView(AgentView Agent, ProbeRecord Probe, string IpVersion = IpVersions.V4)
{
public LineKey Key => new(Agent.Agent.Id, new ProbeId(Probe.Id));
public LineKey Key => new(Agent.Agent.Id, new ProbeId(Probe.Id), IpVersion);
public bool Connected => Agent.Probes.Any(p => p.Id == Probe.Id);
public bool Connected => Agent.Probes.Any(p => p.Id == Probe.Id && p.IpVersion == IpVersion);
public ConnectedProbeInfo? Connection => Agent.Probes.FirstOrDefault(p => p.Id == Probe.Id);
public ConnectedProbeInfo? Connection => Agent.Probes.FirstOrDefault(p => p.Id == Probe.Id && p.IpVersion == IpVersion);
}
/// <summary>线路目录:把 id 解析成(Agent,Probe),列出所有线路。</summary>
@@ -54,7 +54,20 @@ public sealed class LineService
IReadOnlyList<AgentView> agents = await _agents.ListAsync(ct);
IReadOnlyList<ProbeRecord> probes = [.. (await _probes.ListAsync(ct)).Where(p => p.RevokedAtMs is null)];
LinkSnapshot links = await _links.SnapshotAsync(ct);
return [.. agents.Where(include).SelectMany(a => probes.Where(p => links.IsLinked(p.Id, a.Agent.Id.Value)).Select(p => new LineView(a, p)))];
var result = new List<LineView>();
foreach (AgentView agent in agents.Where(include))
{
foreach (ProbeRecord probe in probes.Where(p => links.IsLinked(p.Id, agent.Agent.Id.Value)))
{
result.Add(new LineView(agent, probe));
if (agent.Agent.Ipv6Host.Length > 0)
{
result.Add(new LineView(agent, probe, IpVersions.V6));
}
}
}
return result;
}
/// <summary>
@@ -64,6 +77,7 @@ public sealed class LineService
{
string agentPart = id;
string? probePart = null;
string ipVersion = IpVersions.V4;
if (id.Contains('~', StringComparison.Ordinal))
{
if (!LineKey.TryParse(id, out LineKey key))
@@ -73,6 +87,7 @@ public sealed class LineService
agentPart = key.Agent.Value;
probePart = key.Probe.Value;
ipVersion = key.IpVersion;
}
AgentView? agent = await _agents.GetAsync(new NodeId(agentPart), ct);
@@ -102,6 +117,6 @@ public sealed class LineService
probe = active[0];
}
return Result<LineView>.Ok(new LineView(agent, probe));
return Result<LineView>.Ok(new LineView(agent, probe, ipVersion));
}
}
@@ -167,7 +167,7 @@ public sealed class QueryService
}
IReadOnlyDictionary<string, double> overlay = selection.WantsScores ? await TestOverlayAsync(line.Key, ct) : new Dictionary<string, double>();
IReadOnlyList<TestRecord> recent = await _tests.ListAsync(line.Key.Agent, line.Key.Probe, 1, ct);
IReadOnlyList<TestRecord> recent = await _tests.ListAsync(line.Key.Agent, line.Key.Probe, 1, line.Key.IpVersion, ct);
result.Add(new OverviewLine(line, Summarize(merged, count, p.Plan, p.Range, selection, overlay), recent.Count > 0 ? recent[0] : null));
}
@@ -372,7 +372,7 @@ public sealed class QueryService
long cutoff = _time.GetUtcNow().ToUnixTimeMilliseconds() - (7 * Steps.Day);
var overlay = new Dictionary<string, double>(StringComparer.Ordinal);
var seenPlans = new HashSet<string>();
foreach (TestRecord test in await _tests.ListAsync(line.Agent, line.Probe, 20, ct))
foreach (TestRecord test in await _tests.ListAsync(line.Agent, line.Probe, 20, line.IpVersion, ct))
{
if (test.Status != TestStatuses.Succeeded || test.FinishedAtMs is not { } finished || finished < cutoff || !seenPlans.Add(test.Plan))
{
@@ -139,7 +139,8 @@ public sealed class RowLoader
IReadOnlyList<MetricRow> all = await _metrics.ReadAgentsAsync(ToLevel(plan.Source), agents, bySeries.Key, range.FromMs, range.ToMs, ct);
foreach (IGrouping<NodeId, MetricRow> group in all.GroupBy(r => r.AgentId))
{
result[new LineKey(group.Key, new ProbeId(bySeries.Key[RawChunkHeader.LinePrefix.Length..]))] = Decode([.. group]);
LineKey key = bySeries.First(l => l.Agent == group.Key);
result[key] = Decode([.. group]);
}
}
@@ -28,8 +28,16 @@ public sealed class TestService
_time = time;
}
public async ValueTask<Result<TestRecord>> StartAsync(NodeId agentId, ProbeId probeId, StartTestRequest request, string trigger, CancellationToken ct)
public ValueTask<Result<TestRecord>> StartAsync(NodeId agentId, ProbeId probeId, StartTestRequest request, string trigger, CancellationToken ct) =>
StartAsync(agentId, probeId, request, trigger, IpVersions.V4, ct);
public async ValueTask<Result<TestRecord>> StartAsync(NodeId agentId, ProbeId probeId, StartTestRequest request, string trigger, string ipVersion, CancellationToken ct)
{
if (!IpVersions.IsValid(ipVersion))
{
return ValidationError.Fail<TestRecord>("ip_version", "must be ipv4 or ipv6");
}
PlanSpec? plan = Plans.Find(request.Plan);
if (plan is null)
{
@@ -47,6 +55,11 @@ public sealed class TestService
return Result<TestRecord>.Fail(MoShiroErrors.AgentNotActive, "agent is not active");
}
if (ipVersion == IpVersions.V6 && !agent.Ipv6Supported)
{
return Result<TestRecord>.Fail(FleetErrors.Unsupported, "this agent has not enabled IPv6 collection");
}
// Agent 在 hello 里声明了它支持的采集方法:方案里有它不支持的就不下发(否则要等到阶段开始才失败)
string? unsupported = plan.Phases.Select(p => p.Kind).FirstOrDefault(kind => !agent.Kinds.Contains(kind));
if (unsupported is not null)
@@ -65,7 +78,7 @@ public sealed class TestService
return Result<TestRecord>.Fail(MoShiroErrors.AgentOffline, "agent is offline");
}
if (!(_runtime.TryGet(agentId, out AgentRuntimeState? state) && state.IsConnected(probeId.Value)))
if (!(_runtime.TryGet(agentId, out AgentRuntimeState? state) && state.IsConnected(probeId.Value, ipVersion)))
{
return Result<TestRecord>.Fail(MoShiroErrors.ProbeOffline, $"probe {probeId.Value} has no session with this agent");
}
@@ -78,10 +91,10 @@ public sealed class TestService
var testId = new TestId(Ulid.NewPrefixed(TestId.Prefix, _time));
long now = _time.GetUtcNow().ToUnixTimeMilliseconds();
string paramsJson = request.Params is null ? "{}" : JsonSerializer.Serialize(new Dictionary<string, JsonElement>(request.Params), TestJson.Default.ParamMap);
var record = new TestRecord(testId, agentId, probeId, plan.Id, paramsJson, trigger, TestStatuses.Queued, null, null, phaseList.Count, now, null, null);
var record = new TestRecord(testId, agentId, probeId, plan.Id, paramsJson, trigger, TestStatuses.Queued, null, null, phaseList.Count, now, null, null, ipVersion);
await _tests.CreateAsync(record, ct);
Result<bool> sent = await _gateway.RunTestAsync(agentId, new TestRunParams { TestId = testId, ProbeId = probeId, Plan = plan.Id, Phases = phaseList }, ct);
Result<bool> sent = await _gateway.RunTestAsync(agentId, new TestRunParams { TestId = testId, ProbeId = probeId, IpVersion = ipVersion, Plan = plan.Id, Phases = phaseList }, ct);
if (!sent.TryGet(out _, out FleetError? sendError))
{
await _tests.UpdateStatusAsync(testId, TestStatuses.Failed, null, sendError.Code.Value, now, ct);
@@ -149,6 +162,8 @@ public sealed class TestService
public ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, CancellationToken ct) => _tests.ListAsync(agentId, probeId, limit, ct);
public ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, string? ipVersion, CancellationToken ct) => _tests.ListAsync(agentId, probeId, limit, ipVersion, ct);
public static TestResultView ToView(TestResultRecord stored)
{
StoredTestMetrics metrics = JsonSerializer.Deserialize(stored.MetricsJson, TestJson.Default.StoredTestMetrics) ?? new StoredTestMetrics();
@@ -5,7 +5,7 @@ namespace MoShiro.Backend.Storage.Sqlite;
/// <summary>表结构。版本记录在 <c>PRAGMA user_version</c>;以后的迁移是按版本号追加的内置 SQL。</summary>
internal static class Schema
{
public const int CurrentVersion = 5;
public const int CurrentVersion = 6;
public static void Migrate(SqliteConnection connection)
{
@@ -90,6 +90,17 @@ internal static class Schema
connection.Exec("PRAGMA user_version = 5");
transaction.Commit();
}
if (version < 6)
{
using SqliteTransaction transaction = connection.BeginTransaction();
connection.Exec("ALTER TABLE agents ADD COLUMN ipv6_host TEXT NOT NULL DEFAULT ''");
connection.Exec("ALTER TABLE agents ADD COLUMN ipv6_supported INTEGER NOT NULL DEFAULT 0");
connection.Exec("ALTER TABLE tests ADD COLUMN ip_version TEXT NOT NULL DEFAULT 'ipv4'");
connection.Exec("CREATE INDEX ix_tests_path ON tests (agent_id, probe_id, ip_version, requested_at)");
connection.Exec("PRAGMA user_version = 6");
transaction.Commit();
}
}
/// <summary>
@@ -14,7 +14,7 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
{
private const string AgentColumns =
"id, name, status, pubkey, tls_spki, public_host, probe_tcp_port, probe_udp_ports, labels, public_visible, kinds, agent_version, local_limits, " +
"task_version, applied_task_version, rejected, created_at, last_seen_at, row_version";
"task_version, applied_task_version, rejected, created_at, last_seen_at, row_version, ipv6_host, ipv6_supported";
private readonly SqliteDatabase _db;
@@ -61,10 +61,10 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
}
c.Exec(
"UPDATE agents SET name = @name, labels = @labels, public_host = @host, public_visible = @visible, row_version = row_version + 1 WHERE id = @id",
"UPDATE agents SET name = @name, labels = @labels, public_host = @host, ipv6_host = @v6, public_visible = @visible, row_version = row_version + 1 WHERE id = @id",
("@id", id.Value), ("@name", patch.Name ?? current.Name),
("@labels", StorageJson.ToJson(patch.Labels ?? current.Labels)),
("@host", patch.PublicHost ?? current.PublicHost), ("@visible", (patch.PublicVisible ?? current.PublicVisible) ? 1 : 0));
("@host", patch.PublicHost ?? current.PublicHost), ("@v6", patch.Ipv6Host ?? current.Ipv6Host), ("@visible", (patch.PublicVisible ?? current.PublicVisible) ? 1 : 0));
return Result<AgentRecord>.Ok(GetInternal(c, id)!);
}, ct);
@@ -117,9 +117,9 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
{
c.Exec(
"UPDATE agents SET kinds = @kinds, agent_version = @version, local_limits = COALESCE(@limits, local_limits), tls_spki = COALESCE(@spki, tls_spki), " +
"probe_tcp_port = COALESCE(@tcp, probe_tcp_port), probe_udp_ports = COALESCE(@udp, probe_udp_ports), row_version = row_version + 1 WHERE id = @id",
"probe_tcp_port = COALESCE(@tcp, probe_tcp_port), probe_udp_ports = COALESCE(@udp, probe_udp_ports), ipv6_supported = @v6, row_version = row_version + 1 WHERE id = @id",
("@id", id.Value), ("@kinds", StorageJson.ToJson(hello.Kinds)), ("@version", hello.AgentVersion), ("@limits", hello.LocalLimitsJson),
("@spki", hello.TlsSpki), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)));
("@spki", hello.TlsSpki), ("@tcp", hello.ProbeTcpPort), ("@udp", hello.ProbeUdpPorts is null ? null : StorageJson.ToJson(hello.ProbeUdpPorts)), ("@v6", hello.Ipv6Supported ? 1 : 0));
}, ct);
public ValueTask SetTaskVersionAsync(NodeId id, long version, string taskHash, CancellationToken ct) =>
@@ -217,7 +217,9 @@ public sealed class SqliteAgentStore : IAgentStore, IFleetStore
r.NullableString(15),
r.GetInt64(16),
r.NullableLong(17),
r.GetInt64(18));
r.GetInt64(18),
r.GetString(19),
r.GetInt64(20) != 0);
private static NodeRecord ToNode(AgentRecord a, string? metaJson) =>
new(a.Id, a.Name, a.Status, a.PublicKey, a.TlsSpki, a.Labels, metaJson, a.AgentVersion, a.CreatedAtMs, a.LastSeenAtMs);
@@ -7,7 +7,7 @@ namespace MoShiro.Backend.Storage.Sqlite;
public sealed class SqliteTestStore : ITestStore
{
private const string Columns = "id, agent_id, plan, params, trigger, status, error_code, phase, phase_count, requested_at, started_at, finished_at, probe_id";
private const string Columns = "id, agent_id, plan, params, trigger, status, error_code, phase, phase_count, requested_at, started_at, finished_at, probe_id, ip_version";
private readonly SqliteDatabase _db;
@@ -19,19 +19,22 @@ public sealed class SqliteTestStore : ITestStore
public ValueTask CreateAsync(TestRecord test, CancellationToken ct) =>
_db.ExecuteAsync(
c => c.Exec(
"INSERT INTO tests (id, agent_id, probe_id, plan, params, trigger, status, error_code, phase, phase_count, requested_at) VALUES (@id, @agent, @probe, @plan, @params, @trigger, @status, @err, @phase, @count, @at)",
"INSERT INTO tests (id, agent_id, probe_id, plan, params, trigger, status, error_code, phase, phase_count, requested_at, ip_version) VALUES (@id, @agent, @probe, @plan, @params, @trigger, @status, @err, @phase, @count, @at, @ip)",
("@id", test.Id.Value), ("@agent", test.AgentId.Value), ("@probe", test.ProbeId.Value), ("@plan", test.Plan), ("@params", test.ParamsJson), ("@trigger", test.Trigger),
("@status", test.Status), ("@err", test.ErrorCode), ("@phase", test.Phase), ("@count", test.PhaseCount), ("@at", test.RequestedAtMs)),
("@status", test.Status), ("@err", test.ErrorCode), ("@phase", test.Phase), ("@count", test.PhaseCount), ("@at", test.RequestedAtMs), ("@ip", test.IpVersion)),
ct);
public ValueTask<TestRecord?> GetAsync(TestId id, CancellationToken ct) =>
_db.ReadAsync(c => c.QueryOne($"SELECT {Columns} FROM tests WHERE id = @id", Read, ("@id", id.Value)), ct);
public ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, CancellationToken ct) =>
ListAsync(agentId, probeId, limit, null, ct);
public ValueTask<IReadOnlyList<TestRecord>> ListAsync(NodeId? agentId, ProbeId? probeId, int limit, string? ipVersion, CancellationToken ct) =>
_db.ReadAsync<IReadOnlyList<TestRecord>>(
c => c.Query(
$"SELECT {Columns} FROM tests WHERE (@agent IS NULL OR agent_id = @agent) AND (@probe IS NULL OR probe_id = @probe) ORDER BY requested_at DESC, id DESC LIMIT @limit",
Read, ("@agent", agentId?.Value), ("@probe", probeId?.Value), ("@limit", limit)),
$"SELECT {Columns} FROM tests WHERE (@agent IS NULL OR agent_id = @agent) AND (@probe IS NULL OR probe_id = @probe) AND (@ip IS NULL OR ip_version = @ip) ORDER BY requested_at DESC, id DESC LIMIT @limit",
Read, ("@agent", agentId?.Value), ("@probe", probeId?.Value), ("@limit", limit), ("@ip", ipVersion)),
ct);
public ValueTask<bool> UpdateStatusAsync(TestId id, string status, int? phase, string? errorCode, long nowMs, CancellationToken ct) =>
@@ -88,5 +91,6 @@ public sealed class SqliteTestStore : ITestStore
r.GetInt32(8),
r.GetInt64(9),
r.NullableLong(10),
r.NullableLong(11));
r.NullableLong(11),
r.GetString(13));
}
@@ -30,7 +30,7 @@ public sealed class HostSamplerTests
var time = new Microsoft.Extensions.Time.Testing.FakeTimeProvider();
string stat = Stat1;
string dev = Dev1;
var sampler = new HostSampler(time, path => path switch { "/proc/stat" => stat, "/proc/net/dev" => dev, "/proc/uptime" => "100.0 1.0", _ => null }, () => (1000, 250), procRoot: "/proc", sysRoot: "/sys");
var sampler = new HostSampler(time, path => path.Replace('\\', '/') switch { "/proc/stat" => stat, "/proc/net/dev" => dev, "/proc/uptime" => "100.0 1.0", _ => null }, () => (1000, 250), procRoot: "/proc", sysRoot: "/sys");
HostSnapshot first = sampler.Sample();
Assert.Null(first.CpuPercent); // 第一次没有基线
Assert.Null(first.NetRxBps);
@@ -53,6 +53,7 @@ public sealed class HostSamplerTests
var reads = new List<string>();
var sampler = new HostSampler(TimeProvider.System, path =>
{
path = path.Replace('\\', '/'); // 测试里的 /proc 格式统一用斜杠,兼容 Windows 的 Path.Combine
reads.Add(path);
return path switch
{
@@ -0,0 +1,165 @@
using System.Net;
using System.Net.Sockets;
using System.Text.Json.Nodes;
using Microsoft.Extensions.DependencyInjection;
using MoShiro.Backend.Application;
using MoShiro.Protocol;
using MoShiro.TestSupport;
namespace MoShiro.Backend.Tests;
/// <summary>真实双栈回环:同一身份的两种会话、原始块、查询与测试结果必须各自独立。</summary>
[Collection("probe-e2e")]
public sealed class Ipv6Tests
{
private readonly BackendFixture _backend;
public Ipv6Tests(BackendFixture backend)
{
_backend = backend;
}
[Fact]
public async Task Queries_DifferentFamilyLoss_RemainsIndependentInHotPersistedAndCompareViews()
{
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
long start = ((TimeProvider.System.GetUtcNow().ToUnixTimeMilliseconds() / 60_000) - 4) * 60_000;
for (int k = 0; k < 6; k++)
{
foreach (string version in new[] { IpVersions.V4, IpVersions.V6 })
{
RawChunk chunk = ChunkFactory.Echo(agent.NodeId, "udp_echo", start + k * 10_000, 100, 100,
i => new EchoPacket(false, version == IpVersions.V6 && i % 2 == 0, 20_000, 20_000));
chunk.Header.ProbeId = new ProbeId(_backend.DefaultProbeId);
chunk.Header.IpVersion = version;
Assert.True(await agent.SendChunkAsync(chunk));
}
}
string range = "from=" + DateTimeOffset.FromUnixTimeMilliseconds(start).UtcDateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffZ")
+ "&to=" + DateTimeOffset.FromUnixTimeMilliseconds(start + 60_000).UtcDateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffZ");
string v4 = _backend.Line(agent.NodeId);
string v6 = v4 + "~ipv6";
foreach (string step in new[] { "1s", "10s", "1m" })
{
foreach ((string line, double loss) in new[] { (v4, 0.0), (v6, 50.0) })
{
JsonNode series = null!;
await BackendFixture.EventuallyAsync(async () =>
{
series = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{line}/series?{range}&step={step}&preset=udp");
return series["summary"]!["samples"]!["udp_echo"]?.GetValue<long>() == 600;
});
Assert.Equal(loss, series["summary"]!["metrics"]!["loss.total"]!.GetValue<double>(), 6);
}
}
JsonNode compare = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/compare?lines={v4},{v6}&metric=loss.total&step=1m&{range}");
Assert.Equal(0, compare["summary"]![v4]!.GetValue<double>());
Assert.Equal(50, compare["summary"]![v6]!.GetValue<double>());
await using SseClient live = await SseClient.ConnectAsync(_backend.Admin, $"/api/public/live?topics=series:{v6}:10s:udp");
SseEvent snapshot = await live.WaitForAsync(e => e.Name == "series" && e.Data?["line_id"]?.GetValue<string>() == v6);
Assert.Equal("ipv6", snapshot.Data!["ip_version"]!.GetValue<string>());
}
[Fact]
public async Task Ipv6Configuration_RejectsWrongFamily_AndDoesNotArrangeIpv6ForALegacyAgent()
{
await using AgentSimulator agent = await _backend.ConnectAgentAsync();
using HttpResponseMessage invalid = await _backend.Admin.PatchAsync("/api/admin/agents/" + agent.NodeId.Value,
BackendFixture.Json(new JsonObject { ["ipv6_host"] = "127.0.0.1" }));
Assert.Equal(HttpStatusCode.BadRequest, invalid.StatusCode);
Assert.Contains("ipv6_host", await invalid.Content.ReadAsStringAsync(), StringComparison.Ordinal);
using HttpResponseMessage valid = await _backend.Admin.PatchAsync("/api/admin/agents/" + agent.NodeId.Value,
BackendFixture.Json(new JsonObject { ["ipv6_host"] = "::1" }));
Assert.True(valid.IsSuccessStatusCode);
ProbeAgentList endpoints = await _backend.Services.GetRequiredService<AgentService>().ProbeAgentsAsync(_backend.DefaultProbeId, default);
Assert.Null(Assert.Single(endpoints.Agents, a => a.AgentId == agent.NodeId).Ipv6Host);
JsonNode identity = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.NodeId.Value);
Assert.Equal("::1", identity["ipv6_host"]!.GetValue<string>());
Assert.False(identity["ipv6_supported"]!.GetValue<bool>());
}
[Fact]
public async Task DualStack_CollectsBothFamilies_DisablingIpv6KeepsIpv4Running()
{
Assert.SkipUnless(Socket.OSSupportsIPv6, "IPv6 sockets are unavailable on this host");
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, c => c.Listen.Bind = "::");
await agent.OverrideLineTasksAsync(rateHz: 30);
using HttpResponseMessage configured = await _backend.Admin.PatchAsync("/api/admin/agents/" + agent.Id,
BackendFixture.Json(new JsonObject { ["public_host"] = "127.0.0.1", ["ipv6_host"] = "::1" }));
Assert.True(configured.IsSuccessStatusCode, await configured.Content.ReadAsStringAsync());
await using ProbeHarness probe = await ProbeHarness.StartAsync(_backend);
await BackendFixture.EventuallyAsync(() => Task.FromResult(probe.App.IsConnected(agent.Id) && probe.App.IsConnected(agent.Id, IpVersions.V6)));
await BackendFixture.EventuallyAsync(async () =>
{
JsonNode raw = await _backend.GetJsonAsync(_backend.Admin, "/api/admin/raw/chunks?agent_id=" + agent.Id);
return new[] { "line:" + probe.ProbeId, "line:" + probe.ProbeId + ":ipv6" }.All(series =>
new[] { "udp_echo", "tcp_echo" }.All(kind => raw["items"]!.AsArray().Any(c => c!["series"]!.GetValue<string>() == series && c["kind"]!.GetValue<string>() == kind)));
}, TimeSpan.FromSeconds(45));
string v4 = agent.Id + "~" + probe.ProbeId;
string v6 = v4 + "~ipv6";
foreach (string step in new[] { "1s", "10s" })
{
foreach ((string line, string family) in new[] { (v4, IpVersions.V4), (v6, IpVersions.V6) })
{
JsonNode series = await _backend.GetJsonAsync(_backend.Admin, $"/api/public/lines/{line}/series?range=1m&step={step}&preset=udp,tcp");
Assert.Equal(family, series["ip_version"]!.GetValue<string>());
Assert.True(series["summary"]!["samples"]!["udp_echo"]!.GetValue<long>() > 0, series.ToJsonString());
Assert.True(series["summary"]!["samples"]!["tcp_echo"]!.GetValue<long>() > 0, series.ToJsonString());
}
}
JsonNode overview = await _backend.GetJsonAsync(_backend.Admin, "/api/public/overview?range=1m&ip_version=ipv6");
Assert.Contains(overview["lines"]!.AsArray(), l => l!["id"]!.GetValue<string>() == v6);
Assert.All(overview["lines"]!.AsArray(), l => Assert.Equal("ipv6", l!["ip_version"]!.GetValue<string>()));
var ipv4Connection = probe.App.Connection(agent.Id);
long previous = agent.App.Counters.Fragments;
using HttpResponseMessage disabled = await _backend.Admin.PatchAsync("/api/admin/agents/" + agent.Id, BackendFixture.Json(new JsonObject { ["ipv6_host"] = "" }));
Assert.True(disabled.IsSuccessStatusCode);
await BackendFixture.EventuallyAsync(() => Task.FromResult(probe.App.Connection(agent.Id, IpVersions.V6) is null));
Assert.Same(ipv4Connection, probe.App.Connection(agent.Id));
Assert.True(probe.App.IsConnected(agent.Id));
await BackendFixture.EventuallyAsync(() => Task.FromResult(agent.App.Counters.Fragments > previous), TimeSpan.FromSeconds(20));
Assert.Equal(0, agent.App.Counters.FragmentsDiscarded);
}
[Theory]
[InlineData("hy2_ladder")]
[InlineData("reality_session")]
public async Task Ipv6Test_UsesIpv6Session_AndDoesNotAppearInIpv4History(string plan)
{
Assert.SkipUnless(Socket.OSSupportsIPv6, "IPv6 sockets are unavailable on this host");
await using AgentHarness agent = await AgentHarness.StartAsync(_backend, c => c.Listen.Bind = "::");
await agent.OverrideLineTasksAsync();
using HttpResponseMessage configured = await _backend.Admin.PatchAsync("/api/admin/agents/" + agent.Id,
BackendFixture.Json(new JsonObject { ["ipv6_host"] = "::1" }));
Assert.True(configured.IsSuccessStatusCode);
await using ProbeHarness probe = await ProbeHarness.StartAsync(_backend);
string v4 = agent.Id + "~" + probe.ProbeId;
string v6 = v4 + "~ipv6";
await BackendFixture.EventuallyAsync(async () =>
(await _backend.GetJsonAsync(_backend.Admin, "/api/admin/agents/" + agent.Id))["probes"]!.AsArray()
.Any(p => p!["probe_id"]!.GetValue<string>() == probe.ProbeId && p["ip_version"]!.GetValue<string>() == "ipv6"));
var parameters = plan == "hy2_ladder" ? new JsonObject { ["mbps"] = 1, ["duration_s"] = 1 } : new JsonObject { ["duration_s"] = 1 };
using HttpResponseMessage accepted = await _backend.Admin.PostAsync($"/api/admin/lines/{v6}/tests", BackendFixture.Json(new JsonObject { ["plan"] = plan, ["params"] = parameters }));
Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode);
string id = JsonNode.Parse(await accepted.Content.ReadAsStringAsync())!["id"]!.GetValue<string>();
JsonNode result = null!;
await BackendFixture.EventuallyAsync(async () =>
{
result = await _backend.GetJsonAsync(_backend.Admin, "/api/public/tests/" + id);
return result["status"]!.GetValue<string>() is "failed" or "aborted" || result["result"] is not null;
}, TimeSpan.FromSeconds(100));
Assert.Equal("succeeded", result["status"]!.GetValue<string>());
Assert.Equal("ipv6", result["ip_version"]!.GetValue<string>());
Assert.NotNull(result["result"]);
JsonNode history = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/lines/{v4}/tests");
Assert.Empty(history["items"]!.AsArray());
JsonNode v6History = await _backend.GetJsonAsync(_backend.Admin, $"/api/admin/lines/{v6}/tests");
Assert.Contains(v6History["items"]!.AsArray(), t => t!["id"]!.GetValue<string>() == id);
}
}
+8
View File
@@ -4,6 +4,14 @@
## 1. 分层
### IPv4 / IPv6
- `PATCH /api/admin/agents/{id}` 新增可选 `ipv6_host`:IPv6 地址或域名;空字符串关闭 IPv6,缺省或 null 不修改。原 `public_host` 用于 IPv4。
- `GET /api/public/overview` 新增可选 `ip_version=ipv4|ipv6`,缺省 IPv4。SSE `/api/public/live` 的同名参数选择 overview 主题的地址族;series 主题仍按线路 id 定位。
- IPv4 线路 id 保持 `agent~probe`,IPv6 为 `agent~probe~ipv6`。series、distribution、heatmap、compare 和管理端测试端点共用这个标识。测试记录与评分叠加独立。
- 线路、时序、测试和连接 DTO 新增 `ip_version`;管理端 Agent DTO 新增 `ipv6_host`、`ipv6_supported`;线路详情新增 `ipv6_available`(是否配置了 IPv6 地址)。公开 DTO 不包含任一种 IP 地址。
- API v1 采用新增字段和可选参数,现有 IPv4 请求及历史链接继续有效。前端类型从 OpenAPI 重新生成。
```text
/api/public/* 只读。站点设为公开时匿名可访问,设为私有时需要登录或只读令牌;
public_visible = false 的 Agent 对匿名访客隐藏
+8
View File
@@ -4,6 +4,14 @@
## 1. 构建与打包
### 启用 IPv6 测量
升级 Backend、Agent、Probe 后,在管理端的 Agent 编辑页保留 IPv4 地址,并填写 IPv6 地址或带 AAAA 记录的域名。未填写 IPv6 地址的 VPS 只采集 IPv4;Probe 没有 IPv6 路由或 IPv6 连接失败时,IPv4 会话继续工作。
Agent 的新默认监听地址为 `::`(双栈);现有 `agent.json` 如果显式配置了 `0.0.0.0`,需要在宿主机改为 `::` 并重启节点。系统不支持 IPv6 套接字时,新默认值退回 IPv4 监听。防火墙需允许 IPv6 的 `7443/tcp`、`7443/udp`、`7444/tcp`(自定义端口时用实际端口);容器推荐宿主网络。管理端的「IPv6 监听」显示节点实际是否启用。
原来的 SQLite 数据库自动迁移,旧原始块、指标和测试结果继续属于 IPv4。先升级 Backend,再升级节点,最后填写 IPv6 地址;旧 Agent 不会上报 IPv6 支持,Backend 不会给它安排第二条会话。
```text
scripts/publish.sh <agent|probe|backend|fleet-enroll> <rid> → backend/artifacts/publish/<project>/<rid>/
scripts/package.sh <agent|probe> <rid> → backend/artifacts/packages/moshiro-<role>-<version>-<rid>.tar.gz(附 .sha256)
+2
View File
@@ -8,6 +8,8 @@
## 2. 任务模型
同一个 Probe 与 Agent 的 IPv4、IPv6 路径并行执行相同的持续线路任务模板,各自记录、上传、分析。IPv6 只在管理端配置了地址且 Agent 上报支持时建立;两端的控制和数据通道固定使用同一地址族。按需测试通过所选线路 id 指定地址族,重负载并发限制仍按 Agent 和 Probe 的本机策略执行。两个地址族分别使用有界发件箱(Probe 每路径 8 MB,Agent 的共同上传队列 64 MB)。
```text
Task = { id, scope, kind, target, params, schedule, enabled }
+11
View File
@@ -94,6 +94,17 @@ Probe 只记录自己一侧的列,Agent 在数据面记录另一侧,由 Agen
## 4. 版本演进
### IPv4 / IPv6 路径
- 一个 Probe 与 Agent 可以同时建立 IPv4、IPv6 两个会话,共用身份公钥,各自使用独立的会话密钥、流、时钟估计与发件箱。重连只替换同地址族的会话。
- `probe_agents.agents[]` 保留 `host`(IPv4 地址或 A 记录域名),新增可选 `ipv6_host`(IPv6 地址或 AAAA 记录域名)。未配置时不建立 IPv6 会话;Probe 按地址族解析,不跨地址族回退。
- Agent 在 hello 的元数据里上报 `ipv6_supported`,只在本机监听 IPv6 时为 true。Backend 只向支持它的 Agent 安排 IPv6 路径,旧 Agent 缺省为 false。
- `ConnectedProbe`、`TestRunParams` 和 RawChunk 块头新增 `ip_version`,取值 `ipv4` / `ipv6`,缺省 `ipv4`。Agent 合并列块时以认证过的控制会话地址族为准,覆盖 Probe 上报的值。
- IPv4 的线路 id `agent~probe`、序列 `line:probe` 保持不变;IPv6 使用 `agent~probe~ipv6`、`line:probe:ipv6`。指标、重算、SSE 和测试结果按路径隔离。
- 本次是 JSON 字段扩展,二进制包、列编码和容器布局没有变化,协议与 `container_version` 保持原版本。双栈采集需要升级 Agent 和 Probe;旧节点继续提供 IPv4。
### 通用规则
- Fleet 层的兼容规则见 [../backend/fleet/README.md](../backend/fleet/README.md)。
- `moshiro.*` 方法:新增方法或可选字段不需要升级版本,接收方忽略未知字段。
- RawChunk:块头自描述,新增列或新增 Kind 不需要升级 `container_version`;只有容器格式变化时才升级。
+1
View File
@@ -22,6 +22,7 @@ npm run lint:css # 颜色只能写在 shared/theme/tokens.css,
连别的后端(本机跑前端、连线上的):登录页左边选后端(这个页面自己所在的、保存过的,或者「添加后端」),右边都是用户名和密码;添加后端填地址、用户名、密码,登录成功才保存。跨域用不了 Cookie,所以登录时后端会返回会话令牌(`return_session`),前端存在浏览器里、之后每个请求带上,12 小时过期、退出登录即作废;用户看不到也不需要管它。后端需要放行这个页面的来源(`MoShiro__CorsOrigins`)。展示端和管理端侧栏的插头图标是同一个切换器。
约定:
- 总览、线路详情、对比和管理端按需测试均可切换 IPv4 / IPv6;查询和测试结果独立。管理端 Agent 编辑页分别配置 IPv4、IPv6 地址,IPv6 留空只采 IPv4。展示端只显示地址族,不显示 IP。
- 展示端能拿到的主机数据只有汇总的数字(CPU、内存、磁盘占用,网卡累计流量与瞬时网速,运行时长),**没有 CPU 型号、主机名、网卡名,也没有任何 IP**——后端的公开接口不返回它们,Agent 也根本不采集。展示端不显示 Probe 的版本和来源地址(这些只在管理端)。
- 线路详情页:先选**哪一天**(前一天 / 后一天 / 日期选择,URL 的 `day=YYYY-MM-DD`),下面是这一天的丢包缩略图,上面有可拖动的选区(拖空白处新建、拖两边调长短、拖中间平移;URL 的 `at=起_止`,单位是从选中那天 0 点算起的分钟,可以是负数或超过 1440)。缩略图的**刻度**缺省是 48 小时(连续两天,选区可以跨零点,比如晚高峰 20:00 到凌晨 02:00),也可以切成 24 小时(只看一天;URL 的 `span=24`,选区跨零点时自动用 48 小时)。刻度下方是**选区长度**:30 分钟、1 小时、3 小时、6 小时、12 小时、整天,点一下把选区设成这么长——今天且选区到现在为止的,继续看「最近一段」(实时刷新),否则保持选区的起点往后取。刻度上每小时一根短线,48 小时刻度每 3 小时标一个钟点(24 小时刻度每 2 小时),零点标日期;鼠标移到缩略图上会标出那个位置的时间;点一下空白处或者刻度下面的钟点:离那里近的端点(左端或右端)贴过去,另一个端点不动。鼠标停在轴上或刻度上时,会动的那个端点变色,它到目标位置之间染色:扩大是橙色,收缩是灰色。往日默认是整天;「整天」是本地时区的自然日。时间的文字不写「次日」,跨零点两头都写日期(2026年10月5日 21:06 – 2026年10月6日 02:31)。
- 图表上方有 **EWMA 平滑**开关(关、轻 α=0.3、中 α=0.1、重 α=0.03;URL 的 `ewma=low|mid|high`):开了之后每条线画成指数加权移动平均,原始数据以淡线垫在后面,缺口处断开;只在浏览器里算,不影响汇总的指标数字。
+16 -6
View File
@@ -13,6 +13,7 @@ function EditDialog({ agent, onClose, onSaved }: { agent: AgentDto; onClose: ()
const [name, setName] = useState(agent.name)
const [labels, setLabels] = useState(agent.labels.join(', '))
const [host, setHost] = useState(agent.public_host)
const [ipv6Host, setIpv6Host] = useState(agent.ipv6_host ?? '')
const [visible, setVisible] = useState(agent.public_visible)
return (
<Dialog
@@ -29,7 +30,7 @@ function EditDialog({ agent, onClose, onSaved }: { agent: AgentDto; onClose: ()
disabled={busy}
onClick={async () => {
const ok = await run(
() => api.patch(`/api/admin/agents/${agent.id}`, { name, labels: labels.split(',').map((x) => x.trim()).filter(Boolean), public_host: host, public_visible: visible }),
() => api.patch(`/api/admin/agents/${agent.id}`, { name, labels: labels.split(',').map((x) => x.trim()).filter(Boolean), public_host: host.trim(), ipv6_host: ipv6Host.trim(), public_visible: visible }),
'已保存',
)
if (ok) {
@@ -50,10 +51,15 @@ function EditDialog({ agent, onClose, onSaved }: { agent: AgentDto; onClose: ()
</Field>
</div>
<div className="wide">
<Field label="公开地址" hint="Probe 用它连接这个 Agent:域名或 IP">
<Field label="IPv4 地址" hint="Probe 用它测 IPv4:IPv4 地址或域名(A 记录)">
<input className="input" value={host} onChange={(e) => setHost(e.target.value)} />
</Field>
</div>
<div className="wide">
<Field label="IPv6 地址" hint="IPv6 地址或域名(AAAA 记录);留空只采集 IPv4">
<input className="input" value={ipv6Host} onChange={(e) => setIpv6Host(e.target.value)} placeholder="例如 2001:db8::1 或 agent.example.com" />
</Field>
</div>
<div className="wide">
<Field label="标签" hint="逗号分隔">
<input className="input" value={labels} onChange={(e) => setLabels(e.target.value)} />
@@ -167,9 +173,9 @@ export function AgentDetail() {
</thead>
<tbody>
{a.probes.map((p) => (
<tr key={p.probe_id}>
<tr key={`${p.probe_id}~${p.ip_version}`}>
<td>
<div style={{ fontWeight: 600 }}>{probeName(p.probe_id)}</div>
<div style={{ fontWeight: 600 }}>{probeName(p.probe_id)} <Chip>{p.ip_version === 'ipv6' ? 'IPv6' : 'IPv4'}</Chip></div>
<div className="mono faint">{p.probe_id}</div>
</td>
<td>{p.version ?? '—'}</td>
@@ -179,7 +185,7 @@ export function AgentDetail() {
</td>
<td className="r">
<span className="row" style={{ justifyContent: 'flex-end' }}>
<Link className="btn small" to={`/tests?agent=${a.id}&probe=${p.probe_id}`}>
<Link className="btn small" to={`/tests?agent=${a.id}&probe=${p.probe_id}&ip_version=${p.ip_version}`}>
<FlaskConical size={14} />
测试
</Link>
@@ -209,8 +215,12 @@ export function AgentDetail() {
<dl className="kv">
<dt>ID</dt>
<dd className="mono">{a.id}</dd>
<dt>公开地址</dt>
<dt>IPv4 地址</dt>
<dd className="mono">{a.public_host || '—'}</dd>
<dt>IPv6 地址</dt>
<dd className="mono">{a.ipv6_host || '未配置'}</dd>
<dt>IPv6 监听</dt>
<dd>{a.ipv6_supported ? '已启用' : '未启用,请升级 Agent 并配置双栈监听'}</dd>
<dt>控制端口</dt>
<dd className="mono">{a.probe_tcp_port}/tcp</dd>
<dt>数据端口</dt>
+3 -2
View File
@@ -209,7 +209,7 @@ export function Probes() {
) : (
<>
<Chip tone={p.online ? 'ok' : ''}>{p.online ? '已连面板' : '未连面板'}</Chip>
<Chip tone={p.connections.length > 0 ? 'ok' : ''}>{p.connections.length > 0 ? `连着 ${p.connections.length} 个 Agent` : '没有连接'}</Chip>
<Chip tone={p.connections.length > 0 ? 'ok' : ''}>{p.connections.length > 0 ? `连着 ${new Set(p.connections.map((c) => c.agent_id)).size} 个 Agent` : '没有连接'}</Chip>
</>
)}
<span className="mono faint">{p.id}</span>
@@ -252,10 +252,11 @@ export function Probes() {
</thead>
<tbody>
{p.connections.map((c) => (
<tr key={c.agent_id}>
<tr key={`${c.agent_id}~${c.ip_version}`}>
<td>
<Link to={`/agents/${c.agent_id}`} style={{ fontWeight: 600 }}>
{c.agent_name}
<Chip>{c.ip_version === 'ipv6' ? 'IPv6' : 'IPv4'}</Chip>
</Link>
</td>
<td>{c.version ?? '—'}</td>
+9 -4
View File
@@ -1,5 +1,6 @@
import { FlaskConical, Square } from 'lucide-react'
import { useEffect, useState } from 'react'
import { IpVersionPicker, ipVersionOf } from '@shared/ipVersions'
import { useSearchParams } from 'react-router-dom'
import { api } from '@shared/api/client'
import type { AgentDto, Page, PlanDto, ProbeDto, TestDto } from '@shared/api/types'
@@ -84,6 +85,7 @@ function Result({ test }: { test: TestDto }) {
export function Tests() {
const [params, setParams] = useSearchParams()
const ipVersion = ipVersionOf(params.get('ip_version'))
const agents = useQuery(() => api.get<Page<AgentDto>>('/api/admin/agents'), [], 10_000)
const probes = useQuery(() => api.get<Page<ProbeDto>>('/api/admin/probes'), [])
const plans = useQuery(() => api.get<Page<PlanDto>>('/api/public/catalog/plans'), [])
@@ -92,16 +94,18 @@ export function Tests() {
const active = (agents.data?.items ?? []).filter((a) => a.status === 'active')
const agentId = params.get('agent') ?? active[0]?.id ?? ''
const agent = active.find((a) => a.id === agentId)
const connected = agent?.probes ?? []
const connected = (agent?.probes ?? []).filter((p) => ipVersionOf(p.ip_version) === ipVersion)
const probeId = params.get('probe') ?? connected[0]?.probe_id ?? ''
const [planId, setPlanId] = useState('')
const [selected, setSelected] = useState<string>()
const line = agentId && probeId ? `${agentId}~${probeId}` : ''
const line = agentId && probeId ? `${agentId}~${probeId}${ipVersion === 'ipv6' ? '~ipv6' : ''}` : ''
useEffect(() => {
if (!planId && plans.data?.items[0]) setPlanId(plans.data.items[0].id)
}, [plans.data, planId])
useEffect(() => { setSelected(undefined) }, [line])
const tests = useQuery(() => (line ? api.get<Page<TestDto>>(`/api/admin/lines/${encodeURIComponent(line)}/tests`, { limit: 20 }) : Promise.resolve(undefined)), [line], 3000)
const detail = useQuery(() => (selected ? api.get<TestDto>(`/api/public/tests/${selected}`) : Promise.resolve(undefined)), [selected, tests.data?.items[0]?.status], 3000)
const probeName = (id: string) => probes.data?.items.find((p) => p.id === id)?.name ?? id
@@ -118,6 +122,7 @@ export function Tests() {
<>
<header>
<h1>按需测试</h1>
<IpVersionPicker value={ipVersion} onChange={(v) => set({ ip_version: v })} />
<span className="right">{agents.loading && !agents.data ? <Spinner /> : null}</span>
</header>
<ErrorNotice error={agents.error ?? plans.error ?? tests.error} />
@@ -190,7 +195,7 @@ export function Tests() {
) : (
<table className="table hover">
<tbody>
{tests.data?.items.map((t) => (
{tests.data?.items.filter((t) => t.agent_id === agentId && t.probe_id === probeId && ipVersionOf(t.ip_version) === ipVersion).map((t) => (
<tr key={t.id} onClick={() => setSelected(t.id)} style={selected === t.id ? { background: 'var(--app-card-hover)' } : undefined}>
<td>
<Chip tone={STATUS[t.status]?.tone ?? ''}>{STATUS[t.status]?.label ?? t.status}</Chip>
@@ -221,7 +226,7 @@ export function Tests() {
</div>
<div className="card">
<div className="card-title">结果</div>
{detail.data ? (
{detail.data && detail.data.id === selected && ipVersionOf(detail.data.ip_version) === ipVersion ? (
<>
<p className="faint" style={{ marginBottom: 12 }}>
{detail.data.plan} · {STATUS[detail.data.status]?.label ?? detail.data.status}
+12 -4
View File
@@ -1,20 +1,28 @@
import { BuildFooter } from './BuildFooter'
import { NavLink, Navigate, Route, Routes } from 'react-router-dom'
import { NavLink, Navigate, Route, Routes, useLocation, useParams } from 'react-router-dom'
import { SourceButton, ThemeToggle } from '@shared/ui'
import { Compare } from './pages/Compare'
import { Line } from './pages/Line'
import { Overview } from './pages/Overview'
function LineRoute() {
const { id } = useParams()
return <Line key={id} />
}
export function App() {
const location = useLocation()
const ipv6 = location.pathname.endsWith('~ipv6') || new URLSearchParams(location.search).get('ip_version') === 'ipv6'
const familyQuery = ipv6 ? '?ip_version=ipv6' : ''
return (
<>
<header className="topbar">
<div className="inner">
<nav className="topnav">
<NavLink to="/" end className={({ isActive }) => (isActive ? 'active' : '')}>
<NavLink to={`/${familyQuery}`} end className={({ isActive }) => (isActive ? 'active' : '')}>
总览
</NavLink>
<NavLink to="/compare" className={({ isActive }) => (isActive ? 'active' : '')}>
<NavLink to={`/compare${familyQuery}`} className={({ isActive }) => (isActive ? 'active' : '')}>
对比
</NavLink>
<a href="/admin/">管理</a>
@@ -25,7 +33,7 @@ export function App() {
</header>
<Routes>
<Route path="/" element={<Overview />} />
<Route path="/line/:id" element={<Line />} />
<Route path="/line/:id" element={<LineRoute />} />
<Route path="/compare" element={<Compare />} />
<Route path="*" element={<Navigate to="/" replace />} />
</Routes>
+7 -4
View File
@@ -1,5 +1,6 @@
import { Star } from 'lucide-react'
import { useMemo } from 'react'
import { IpVersionPicker, ipVersionOf } from '@shared/ipVersions'
import { useSearchParams } from 'react-router-dom'
import { api } from '@shared/api/client'
import type { CompareDto, OverviewDto, OverviewLineDto } from '@shared/api/types'
@@ -54,15 +55,16 @@ interface Cell {
export function Compare() {
const [params, setParams] = useSearchParams()
const ipVersion = ipVersionOf(params.get('ip_version'))
const range: PresetValue = isPreset(params.get('range')) ? (params.get('range') as PresetValue) : '1d'
const metric = METRICS.find((m) => m.id === params.get('metric')) ?? METRICS[0]
const picked = useMemo(() => (params.get('lines') ?? '').split(',').filter(Boolean), [params])
const picked = useMemo(() => (params.get('lines') ?? '').split(',').filter((id) => id && id.endsWith('~ipv6') === (ipVersion === 'ipv6')), [params, ipVersion])
const overview = useQuery(() => api.get<OverviewDto>('/api/public/overview', { range, metrics: metric.id }), [range, metric.id], 15_000)
const overview = useQuery(() => api.get<OverviewDto>('/api/public/overview', { range, metrics: metric.id, ip_version: ipVersion }), [range, metric.id, ipVersion], 15_000)
// 矩阵:行是 Agent,列是 Probe;每一列内部排名(同一个 Probe 看出去,哪个 Agent 的线路好)
const matrix = useMemo(() => {
const lines = overview.data?.lines ?? []
const lines = (overview.data?.lines ?? []).filter((l) => ipVersionOf(l.ip_version) === ipVersion)
const probes = [...new Map(lines.map((l) => [l.probe_id, l.probe_name])).entries()].sort((a, b) => a[1].localeCompare(b[1]))
const agents = [...new Map(lines.map((l) => [l.agent_id, l.name])).entries()].sort((a, b) => a[1].localeCompare(b[1]))
const byProbe = new Map<string, Cell[]>()
@@ -79,7 +81,7 @@ export function Compare() {
byProbe.set(probeId, cells)
}
return { probes, agents, byProbe }
}, [overview.data, metric])
}, [overview.data, metric, ipVersion])
// 默认选每个 Probe 里最好的那条线路(最多 6 条)
const defaults = useMemo(() => {
@@ -123,6 +125,7 @@ export function Compare() {
<h1>线路对比</h1>
<p className="sub">同一个指标,在每个 Probe 看出去,哪个 Agent 的线路更好。</p>
<div className="toolbar">
<IpVersionPicker value={ipVersion} onChange={(v) => update({ ip_version: v, lines: null })} />
<Segmented options={PRESETS.map((p) => ({ value: p.value, label: p.label }))} value={range} onChange={(v) => update({ range: v })} />
<select className="select" style={{ width: 'auto' }} value={metric.id} onChange={(e) => update({ metric: e.target.value })}>
{METRICS.map((x) => (
+14 -1
View File
@@ -1,5 +1,6 @@
import { ChevronLeft } from 'lucide-react'
import { useMemo } from 'react'
import { IP_VERSIONS, lineIdFor } from '@shared/ipVersions'
import { Link, useParams, useSearchParams } from 'react-router-dom'
import { api } from '@shared/api/client'
import type { LineDto, TestDto } from '@shared/api/types'
@@ -245,7 +246,7 @@ export function Line() {
return (
<main className="page">
<div className="crumbs">
<Link to="/" className="row">
<Link to={id.endsWith('~ipv6') ? '/?ip_version=ipv6' : '/'} className="row">
<ChevronLeft size={16} />
总览
</Link>
@@ -256,6 +257,18 @@ export function Line() {
{line.data?.name ?? '线路'}
{line.data ? <span className="muted" style={{ fontWeight: 400 }}> → {line.data.probe_name}</span> : null}
</h1>
<div className="row" style={{ marginTop: 12 }}>
{IP_VERSIONS.map((v) => {
const query = new URLSearchParams(params)
query.set('ip_version', v.value)
return (
<Link key={v.value} className={`btn small${(id.endsWith('~ipv6') ? 'ipv6' : 'ipv4') === v.value ? ' primary' : ''}`} to={`/line/${encodeURIComponent(lineIdFor(id, v.value))}?${query.toString()}`}>
{v.label}
</Link>
)
})}
{line.data && !line.data.ipv6_available && id.endsWith('~ipv6') ? <span className="muted">此 Agent 未配置 IPv6 地址</span> : null}
</div>
{line.data ? (
<div className="row wrap" style={{ marginTop: 12 }}>
<Dot tone={line.data.online && line.data.probe_connected ? 'ok' : line.data.online ? 'warn' : 'bad'} />
+7 -3
View File
@@ -9,6 +9,7 @@ import { fmtBytes, fmtSpeed, fmtUptime, pct, useHosts, type HostItem } from '../
import { Flag, IspIcon, useNodeInfos } from '@shared/nodes'
import { OsIcon } from '@shared/osIcon'
import { PageError } from '../lib'
import { IpVersionPicker, ipVersionOf } from '@shared/ipVersions'
import { LineStrips, OVERVIEW_RANGES, type OverviewRange } from '../LineStrips'
const METRICS = ['rtt.p50', 'loss.total', 'loss.up', 'loss.down', 'tcp.rtt.p50', 'tcp.retrans'].join(',')
@@ -153,6 +154,7 @@ export function Overview() {
const infos = useNodeInfos()
// 近多久的汇总和走势:3 / 6 / 12 小时,记在链接里(?range=6h)
const [params, setParams] = useSearchParams()
const ipVersion = ipVersionOf(params.get('ip_version'))
const range: OverviewRange = OVERVIEW_RANGES.find((r) => r.value === params.get('range'))?.value ?? '3h'
const setRange = (v: OverviewRange) =>
setParams(
@@ -164,12 +166,13 @@ export function Overview() {
},
{ replace: true },
)
const { data, error, loading } = useQuery(() => api.get<OverviewDto>('/api/public/overview', { range, metrics: METRICS }), [range], 10_000)
const { data, error, loading } = useQuery(() => api.get<OverviewDto>('/api/public/overview', { range, metrics: METRICS, ip_version: ipVersion }), [range, ipVersion], 10_000)
const probes = useMemo(() => [...new Set((data?.lines ?? []).map((l) => l.probe_name))].sort(), [data])
const cards = useMemo(() => {
const byAgent = new Map<string, AgentCard>()
for (const l of data?.lines ?? []) {
if (ipVersionOf(l.ip_version) !== ipVersion) continue
if (probe && l.probe_name !== probe) continue
const card = byAgent.get(l.agent_id) ?? { id: l.agent_id, name: l.name, labels: l.labels, online: l.online, lines: [] }
card.lines.push(l)
@@ -190,7 +193,7 @@ export function Overview() {
if (y === null) return -1
return x - y
})
}, [data, probe, sort])
}, [data, probe, sort, ipVersion])
const online = cards.filter((c) => c.online).length
@@ -203,6 +206,7 @@ export function Overview() {
<span className={`live-dot ${live ? 'on' : ''}`} title={live ? '实时更新中' : '实时连接中断,正在重连'} />
</p>
<div className="toolbar">
<IpVersionPicker value={ipVersion} onChange={(v) => { const next = new URLSearchParams(params); next.set('ip_version', v); setProbe(''); setParams(next, { replace: true }) }} />
{probes.length > 1 ? <Segmented options={[{ value: '', label: '全部 Probe' }, ...probes.map((p) => ({ value: p, label: p }))]} value={probe} onChange={setProbe} /> : null}
<span className="right row">
<Segmented<OverviewRange> options={OVERVIEW_RANGES.map((r) => ({ value: r.value, label: `近 ${r.label}` }))} value={range} onChange={setRange} />
@@ -225,7 +229,7 @@ export function Overview() {
{data ? (
cards.length === 0 ? (
<div className="card" style={{ marginTop: 24 }}>
<Empty>还没有线路</Empty>
<Empty>{ipVersion === 'ipv6' ? '还没有 IPv6 线路,请在管理端配置 Agent 的 IPv6 地址' : '还没有线路'}</Empty>
</div>
) : (
<div className="cards">
+23 -1
View File
@@ -1007,12 +1007,17 @@ export interface components {
created_at: unknown;
/** Format: date-time */
last_seen_at: unknown;
/** @default */
ipv6_host: string;
/** @default false */
ipv6_supported: boolean;
};
AgentPatchRequest: {
name: null | string;
labels: null | string[];
public_host: null | string;
public_visible: null | boolean;
ipv6_host?: null | string;
};
ApiTokenDto: {
id: string;
@@ -1075,6 +1080,8 @@ export interface components {
remote: null | string;
/** Format: date-time */
connected_since: unknown;
/** @default ipv4 */
ip_version: string;
};
CreateApiTokenRequest: {
name: string;
@@ -1285,6 +1292,10 @@ export interface components {
data_since: unknown;
kinds: string[];
last_test: null | components["schemas"]["TestSummaryDto"];
/** @default ipv4 */
ip_version: string;
/** @default false */
ipv6_available: boolean;
};
LinkDto: {
probe_id: string;
@@ -1408,6 +1419,8 @@ export interface components {
endpoint_changes: number | string;
summary: components["schemas"]["SummaryDto"];
last_test: null | components["schemas"]["TestSummaryDto"];
/** @default ipv4 */
ip_version: string;
};
PageDtoOfAgentDto: {
items: components["schemas"]["AgentDto"][];
@@ -1527,6 +1540,8 @@ export interface components {
remote: null | string;
/** Format: date-time */
connected_since: unknown;
/** @default ipv4 */
ip_version: string;
};
ProbeDto: {
id: string;
@@ -1642,6 +1657,8 @@ export interface components {
versions: {
[key: string]: string;
};
/** @default ipv4 */
ip_version: string;
};
SetLinkRequest: {
enabled: null | boolean;
@@ -1702,6 +1719,8 @@ export interface components {
/** Format: date-time */
finished_at: unknown;
result: null | components["schemas"]["TestResultDto"];
/** @default ipv4 */
ip_version: string;
};
TestResultDto: {
metrics: {
@@ -1762,6 +1781,7 @@ export interface operations {
metrics?: string;
sort?: string;
order?: string;
ip_version?: string;
};
header?: never;
path?: never;
@@ -2191,7 +2211,9 @@ export interface operations {
};
live_stream: {
parameters: {
query?: never;
query?: {
ip_version?: string;
};
header?: never;
path?: never;
cookie?: never;
+10
View File
@@ -0,0 +1,10 @@
import { Segmented } from './ui'
export type IpVersion = 'ipv4' | 'ipv6'
export const ipVersionOf = (value: string | null | undefined): IpVersion => value === 'ipv6' ? 'ipv6' : 'ipv4'
export const IP_VERSIONS: { value: IpVersion; label: string }[] = [{ value: 'ipv4', label: 'IPv4' }, { value: 'ipv6', label: 'IPv6' }]
export const lineIdFor = (id: string, version: IpVersion): string => id.replace(/~ipv6$/, '') + (version === 'ipv6' ? '~ipv6' : '')
export function IpVersionPicker({ value, onChange }: { value: IpVersion; onChange: (value: IpVersion) => void }) {
return <Segmented options={IP_VERSIONS} value={value} onChange={onChange} />
}