diff --git a/Proxd-man/Config/Outbound/ShadowSocksOutboundSetting.cs b/Proxd-man/Config/Outbound/ShadowSocksOutboundSetting.cs index 6e89f36..b937a31 100644 --- a/Proxd-man/Config/Outbound/ShadowSocksOutboundSetting.cs +++ b/Proxd-man/Config/Outbound/ShadowSocksOutboundSetting.cs @@ -12,7 +12,6 @@ public class ShadowSocksOutboundSetting : BaseOutboundSetting public string? UserEmail { get; init; } = null; public required string Method { get; init; } - public new required string Address { get; init; } public int Port { get; init; } public required string Password { get; init; } public bool EnableUdpOverTcp { get; init; } = false; diff --git a/Proxd-man/Config/ProxdConfigs.cs b/Proxd-man/Config/ProxdConfigs.cs index 142b7e5..dfc9e94 100644 --- a/Proxd-man/Config/ProxdConfigs.cs +++ b/Proxd-man/Config/ProxdConfigs.cs @@ -16,6 +16,23 @@ public class InboundConfig { public HttpInboundSetting? Http { get; init; } public SocksInboundSetting? Socks { get; init; } + + public List ToInboundSettings() + { + var inboundSettings = new List(); + if (Http != null) + { + inboundSettings.Add(Http); + } + + if (Socks != null) + { + inboundSettings.Add(Socks); + } + + return inboundSettings; + } + } public interface IInboundSetting @@ -142,4 +159,5 @@ public class BasicInboundAuthSetting(string username, string password) public class SubscribeSetting { public List Subscribes { get; } = []; + public string SubscribeSaveFile { get; } = "./subscribes.json"; } \ No newline at end of file diff --git a/Proxd-man/Monitor/OutboundMonitor.cs b/Proxd-man/Monitor/OutboundMonitor.cs index f913128..f1fc067 100644 --- a/Proxd-man/Monitor/OutboundMonitor.cs +++ b/Proxd-man/Monitor/OutboundMonitor.cs @@ -1,4 +1,5 @@ -using System.Diagnostics; +using System.Collections.Concurrent; +using System.Diagnostics; using System.Net; using System.Net.NetworkInformation; using Microsoft.Extensions.Logging; @@ -20,7 +21,7 @@ public class OutboundMonitor private readonly SubscribeManager _subscribeManager; private readonly XrayCoreManager _xrayCoreManager; - private ProxdConfig _config => _configProvider.CurrentValue; + private ProxdConfig Config => _configProvider.CurrentValue; public OutboundMonitor(ILogger logger, IOptionsMonitor configProvider, SubscribeManager subscribeManager, XrayCoreManager xrayCoreManager) { @@ -32,86 +33,230 @@ public class OutboundMonitor { await ReselectOutbound(subscribeOutbounds); }; + + _xrayCoreManager.MainXrayCoreProcessStarted += StartMonitor; } - private async Task ReselectOutbound(IDictionary> subscribeOutbounds) + public async Task ReselectOutbound(IDictionary> subscribeOutbounds) { - - _logger.LogInformation("开始重新选择代理..."); - foreach (var (source, outboundSettings) in subscribeOutbounds) + try { - _logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name); - foreach (var outboundSetting in outboundSettings) + _logger.LogInformation("开始重新选择代理..."); + var outboundLatencies = new ConcurrentDictionary(); + foreach (var (source, outboundSettings) in subscribeOutbounds) { - if (outboundSetting.Address == null) + _logger.LogDebug("开始测试订阅 {SubscribeName} 的代理...", source.Name); + List tasks = []; + var semaphore = new SemaphoreSlim(4); + foreach (var outboundSetting in outboundSettings) { - _logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空", - outboundSetting.Name, outboundSetting.GetTag()); - continue; - } - - using var ping = new Ping(); - var pingReply = ping.Send(outboundSetting.Address, 5000); - - if (pingReply.Status != IPStatus.Success) - { - _logger.LogWarning("无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, pingReply.Status); - continue; - } - - // 如果 Ping 成功, 则开始测试代理的延迟和带宽 - _logger.LogDebug("开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); - - // 随机选择一个端口进行测试 - var port = (ushort) new Random().Next(50000, 60000); - var outboundTestProcess = _xrayCoreManager.CreateOutboundTestProcess(port, outboundSetting); - outboundTestProcess.Start(); + var task = Task.Run(async () => + { + if (outboundSetting.Address == null) + { + _logger.LogDebug("跳过代理 {OutboundSettingName} ({OutboundSettingTag}), 因为 Address 为空", + outboundSetting.Name, outboundSetting.GetTag()); + return; + } - // 等 Xray-Core 就绪. - await Task.Delay(500); - - // 先测试代理的延迟. - var (latencies, bandwidths) = await MeasureOutbound(port); - if (latencies.All(l => l == -1)) - { - _logger.LogWarning("代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); - continue; + await semaphore.WaitAsync(); + + try + { + using var ping = new Ping(); + var pingReply = ping.Send(outboundSetting.Address, 5000); + + if (pingReply.Status != IPStatus.Success) + { + _logger.LogWarning( + "无法 Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}, 错误: {PingReplyStatus}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, + pingReply.Status); + return; + } + + _logger.LogDebug( + "Ping 到代理 {OutboundSettingName} ({OutboundSettingTag}) 成功, 地址: {OutboundSettingAddress}, 延迟: {PingReply.RoundtripTime}ms", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address, + pingReply.RoundtripTime); + + // 如果 Ping 成功, 则开始测试代理的延迟和带宽 + _logger.LogDebug( + "开始测试代理 {OutboundSettingName} ({OutboundSettingTag}), 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); + + // 随机选择一个端口进行测试 + var port = (ushort)new Random().Next(50000, 60000); + var outboundTestProcess = + await _xrayCoreManager.CreateOutboundTestProcess(port, outboundSetting); + + if (!outboundTestProcess.Start()) + { + _logger.LogError("无法启动 Xray-Core 进程, 地址: {OutboundSettingAddress}", + outboundSetting.Address); + } + else + { + _logger.LogDebug("Xray-Core 进程已启动, 地址: {OutboundSettingAddress}", + outboundSetting.Address); + _xrayCoreManager.RedirectXrayCoreProcessOutput(outboundTestProcess); + } + + try + { + // 等 Xray-Core 就绪. + await Task.Delay(200); + _logger.LogDebug("Xray-Core 进程已就绪, 开始测试延迟..."); + + // 先测试代理的延迟. + var (latencies, _) = await MeasureOutbound(port); + if (latencies.Count(l => l == -1) > 4) + { + _logger.LogWarning( + "代理 {OutboundSettingName} ({OutboundSettingTag}) 的延迟测试失败, 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), outboundSetting.Address); + return; + } + + // 计算平均延迟. + var avgLatency = latencies.Where(l => l != -1).Average(); + outboundLatencies[outboundSetting] = avgLatency; + _logger.LogDebug( + "代理 {OutboundSettingName} ({OutboundSettingTag}) 的平均延迟为 {AvgLatency}ms, 地址: {OutboundSettingAddress}", + outboundSetting.Name, outboundSetting.GetTag(), avgLatency, + outboundSetting.Address); + } + finally + { + outboundTestProcess.Kill(); + } + } + finally + { + semaphore.Release(); + } + }); + + tasks.Add(task); } - - // 计算平均延迟. - var avgLatency = latencies.Where(l => l != -1).Average(); - _logger.LogDebug("代理 {OutboundSettingName} ({OutboundSettingTag}) 的平均延迟为 {AvgLatency}ms, 地址: {OutboundSettingAddress}", - outboundSetting.Name, outboundSetting.GetTag(), avgLatency, outboundSetting.Address); + + Task.WaitAll(tasks); + _logger.LogDebug("订阅 {SubscribeName} 的代理测试完成.", source.Name); + } + + var sortedOutboundSettings = outboundLatencies + .OrderBy(kv => kv.Value) + .ToList(); + + // 展示延迟前10的节点名称和延迟 + var index = 1; + var latenciesRanking = "---------------- 延迟排名 ----------------\n"; + sortedOutboundSettings + .Take(10) + .ToList() + .ForEach(kv => latenciesRanking += $"{index++}. {kv.Key.Name} ({kv.Key.GetTag()}) [{kv.Key.Address}]: {kv.Value}ms\n"); + _logger.LogInformation(latenciesRanking); + + var bestOutbound = sortedOutboundSettings + .FirstOrDefault().Key; + + if (bestOutbound != null) + { + _logger.LogInformation( + "选择代理 {OutboundSettingName} ({OutboundSettingTag}) 作为新的出站代理, 地址: {OutboundSettingAddress}", + bestOutbound.Name, bestOutbound.GetTag(), bestOutbound.Address); + await _xrayCoreManager.StartMainXrayCoreProcess(bestOutbound); + } + else + { + _logger.LogWarning("没有可用的代理."); } } - } - - public void StartMonitor() - { - Task.Run(async () => + catch (Exception e) { - - }); + _logger.LogError(e, "重新选择代理时发生错误."); + } + } + + private WebProxy GetWebProxyFromConfig() + { + var httpInboundSetting = Config.Inbounds.Http; + if (httpInboundSetting != null) + { + return new WebProxy(new Uri($"http://localhost:{httpInboundSetting.Port}")); + } + + var socksInboundSetting = Config.Inbounds.Socks; + if (socksInboundSetting != null) + { + return new WebProxy(new Uri($"socks5://localhost:{socksInboundSetting.Port}")); + } + + throw new Exception("无法从配置中获取代理信息."); + } + + private CancellationTokenSource? _monitorTaskCts; + + public async void StartMonitor() + { + if (_monitorTaskCts != null) + { + await StopMonitor(); + } + + _monitorTaskCts = new CancellationTokenSource(); + _ = Task.Run(async () => + { + while (!_monitorTaskCts.Token.IsCancellationRequested) + { + var mainXrayCoreProcess = _xrayCoreManager.MainXrayCoreProcess; + if (mainXrayCoreProcess == null || mainXrayCoreProcess.HasExited) + { + _logger.LogDebug("因 MainXrayCore 退出, OutboundMonitor 任务结束."); + break; + } + + var webProxyFromConfig = GetWebProxyFromConfig(); + var httpLatency = await MeasureHttpLatency(webProxyFromConfig); + + // 检查单次失败率是否在 50% 以上 + var failedCount = httpLatency.Count(l => l == -1); + if (failedCount >= 6) + { + _ = ReselectOutbound(_subscribeManager.GetSubscribeOutbounds()); + _ = StopMonitor(); + + break; + } + } + + return Task.CompletedTask; + }, _monitorTaskCts.Token); + } + + public async Task StopMonitor() + { + if (_monitorTaskCts != null) + { + await _monitorTaskCts.CancelAsync(); + _monitorTaskCts = null; + } } private static async Task<(long[] latencies, long[] bandwidths)> MeasureOutbound(ushort port, bool withBandwidth = false, CancellationToken cancellationToken = default) { - var latencies = await MeasureHttpLatency(port, cancellationToken); - var bandwidths = withBandwidth ? await MeasureHttpBandwidth(port, cancellationToken) : []; + var proxy = new WebProxy(new Uri($"http://localhost:{port}")); + + var latencies = await MeasureHttpLatency(proxy, cancellationToken); + var bandwidths = withBandwidth ? await MeasureHttpBandwidth(proxy, cancellationToken) : []; return (latencies, bandwidths); } - private static async Task MeasureHttpLatency(ushort port, CancellationToken cancellationToken = default) + private static async Task MeasureHttpLatency(IWebProxy proxy, CancellationToken cancellationToken = default) { var httpClient = new HttpClient(new HttpClientHandler { - Proxy = new WebProxy - { - Address = new Uri($"http://127.0.0.1:{port}") - }, + Proxy = proxy, UseProxy = true, }) { @@ -149,14 +294,11 @@ public class OutboundMonitor return latencies; } - private static async Task MeasureHttpBandwidth(ushort port, CancellationToken cancellationToken = default) + private static async Task MeasureHttpBandwidth(IWebProxy proxy, CancellationToken cancellationToken = default) { var httpClient = new HttpClient(new HttpClientHandler { - Proxy = new WebProxy - { - Address = new Uri($"http://127.0.0.1:{port}") - }, + Proxy = proxy, UseProxy = true, }) { diff --git a/Proxd-man/Program.cs b/Proxd-man/Program.cs index ca7f864..a69cd65 100644 --- a/Proxd-man/Program.cs +++ b/Proxd-man/Program.cs @@ -30,6 +30,7 @@ internal class Program { configure.AddConfiguration(configuration); configure.AddConsole(); + configure.SetMinimumLevel(LogLevel.Debug); }) .Configure(configuration.GetSection("Proxd")) .AddSingleton() @@ -44,7 +45,18 @@ internal class Program var subscribeManager = serviceProvider.GetRequiredService(); await subscribeManager.CheckAndUpdateSubscribes(); - + var outboundMonitor = serviceProvider.GetRequiredService(); + if (subscribeManager.GetSubscribeOutbounds().Count > 0) + { + logger.LogInformation("已通过订阅获取到出站代理, 正在尝试选取最优出站代理..."); + _ = outboundMonitor.ReselectOutbound(subscribeManager.GetSubscribeOutbounds()); + } + else + { + logger.LogWarning("没有获取到任何出站代理, 等待下一次订阅更新..."); + } + subscribeManager.StartUpdateSubscribes(); + // 监听 SIGTERM 和 SIGINT 信号, 以优雅退出 var cts = new CancellationTokenSource(); AppDomain.CurrentDomain.ProcessExit += (_, _) => cts.Cancel(); diff --git a/Proxd-man/Subscribe/SubscribeManager.cs b/Proxd-man/Subscribe/SubscribeManager.cs index 5b950df..becea4f 100644 --- a/Proxd-man/Subscribe/SubscribeManager.cs +++ b/Proxd-man/Subscribe/SubscribeManager.cs @@ -1,6 +1,7 @@ using System.Buffers.Text; using System.Collections.Immutable; using System.Text; +using System.Text.Json; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Proxd_man.Config; @@ -73,7 +74,10 @@ public class SubscribeManager _logger.LogWarning(e, "更新订阅失败: {subscribeName}", subscribe.Name); } - newSubscribeOutbounds[subscribe] = outbounds; + if (outbounds.Count > 0) + { + newSubscribeOutbounds[subscribe] = outbounds; + } _logger.LogInformation("订阅 {subscribeName} 更新完成. (共 {outboundCount} 个节点)", subscribe.Name, outbounds.Count); } @@ -94,14 +98,30 @@ public class SubscribeManager public void StartUpdateSubscribes() { + if (_cts != null) + { + StopUpdateSubscribes(); + } + var cts = new CancellationTokenSource(); Task.Run(async () => { - while (!cts.IsCancellationRequested) + while (!cts.Token.IsCancellationRequested) { + if (_subscribeOutbounds.Count > 0) + { + _logger.LogDebug("订阅更新完成, 将在 12 小时后再次更新."); + await Task.Delay(TimeSpan.FromHours(12), cts.Token); + } + else + { + _logger.LogDebug("没有可用的订阅, 将在 5 分钟后再次更新."); + await Task.Delay(TimeSpan.FromMinutes(5), cts.Token); + } + + _logger.LogInformation("开始更新订阅..."); await CheckAndUpdateSubscribes(cts.Token); - await Task.Delay(TimeSpan.FromHours(12), cts.Token); } }, cts.Token); _cts = cts; @@ -111,6 +131,7 @@ public class SubscribeManager public void StopUpdateSubscribes() { _cts?.Cancel(); + _cts = null; } } \ No newline at end of file diff --git a/Proxd-man/XrayCoreManager.cs b/Proxd-man/XrayCoreManager.cs index 769e840..a708e61 100644 --- a/Proxd-man/XrayCoreManager.cs +++ b/Proxd-man/XrayCoreManager.cs @@ -15,6 +15,7 @@ namespace Proxd_man; public class XrayCoreManager( ILogger logger, + ILoggerFactory loggerFactory, IOptionsMonitor config) { private const string XrayLatestReleaseApiUrl = "https://api.github.com/repos/XTLS/Xray-core/releases/latest"; @@ -25,8 +26,6 @@ public class XrayCoreManager( private string CoreProgramDir => config.CurrentValue.XrayCoreProgramDirectory; private string WorkingDirectory => config.CurrentValue.XrayCoreWorkingDirectory; - // TODO: 看看注册订阅更新事件后, 需要进行哪些操作. - public string GetXrayCoreProgramPath() { if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows)) @@ -107,7 +106,7 @@ public class XrayCoreManager( return $"{XrayLatestReleaseAssetDownloadUrlPrefix}Xray-{os}-{arch}.zip"; } - private Process CreateXrayCoreProcess(string args, bool redirectOutput = true) + private Process CreateXrayCoreProcess(List args, bool redirectOutput = true, bool redirectError = true) { var programPath = GetXrayCoreProgramPath(); if (!File.Exists(programPath)) @@ -122,11 +121,10 @@ public class XrayCoreManager( return new Process { - StartInfo = new ProcessStartInfo + StartInfo = new ProcessStartInfo(programPath, args) { - FileName = programPath, - Arguments = args, RedirectStandardOutput = redirectOutput, + RedirectStandardError = redirectError, UseShellExecute = false, CreateNoWindow = true, WorkingDirectory = WorkingDirectory @@ -147,7 +145,7 @@ public class XrayCoreManager( return null; } - var process = CreateXrayCoreProcess("--version"); + var process = CreateXrayCoreProcess(["--version"]); process.Start(); await process.WaitForExitAsync(); var output = (await process.StandardOutput.ReadToEndAsync()).Trim(); @@ -284,9 +282,11 @@ public class XrayCoreManager( public async Task VerifyXrayCoreConfigFile(string configFilePath) { - var process = CreateXrayCoreProcess($"run -test -config {configFilePath}"); - + var process = CreateXrayCoreProcess(["run", "-test", "-c", configFilePath]); process.Start(); + + RedirectXrayCoreProcessOutput(process); + await process.WaitForExitAsync(); return process.ExitCode == 0; } @@ -427,7 +427,7 @@ public class XrayCoreManager( // ------------ Outbound 配置 ------------ // ------------ Routing 配置 ------------- - + var routingSettings = new JsonObject { { "domainStrategy", "AsIs" } @@ -472,7 +472,7 @@ public class XrayCoreManager( return proxyConfig; } - public Process CreateOutboundTestProcess(ushort inboundHttpPort, BaseOutboundSetting outboundSetting) + public async Task CreateOutboundTestProcess(ushort inboundHttpPort, BaseOutboundSetting outboundSetting) { var configJson = GenerateXrayCoreConfigJson([ new HttpInboundSetting @@ -481,12 +481,105 @@ public class XrayCoreManager( } ], outboundSetting, true); - var configFilePath = Path.Combine(Path.Combine(WorkingDirectory, "test-config"), Path.GetRandomFileName() + ".json"); - using var configFileStream = File.CreateText(configFilePath); - configFileStream.Write(JsonSerializer.Serialize(configJson)); - return CreateXrayCoreProcess($"run -config {configFilePath}"); + var testingConfigDir = Path.Combine(WorkingDirectory, "test-config"); + if (!Directory.Exists(testingConfigDir)) + { + Directory.CreateDirectory(testingConfigDir); + } + var configFilePath = Path.GetFullPath(Path.Combine(testingConfigDir, Path.GetRandomFileName() + ".json")); + await using (var configFileStream = File.CreateText(configFilePath)) + { + await configFileStream.WriteAsync(JsonSerializer.Serialize(configJson)); + } + + if (!await VerifyXrayCoreConfigFile(configFilePath)) + { + throw new Exception($"测试配置验证失败: {configFilePath}"); + } + + logger.LogDebug("Xray-Core 测试配置文件: {configFilePath}", configFilePath); + + var xrayCoreProcess = CreateXrayCoreProcess(["run", "-c", configFilePath], true, true); + xrayCoreProcess.Exited += (_, _) => File.Delete(configFilePath); + return xrayCoreProcess; + } + + public Process? MainXrayCoreProcess { get; private set; } + public BaseOutboundSetting? CurrentOutboundSetting { get; private set; } + + public delegate void MainXrayCoreProcessExitedHandler(); + public event MainXrayCoreProcessExitedHandler? MainXrayCoreProcessExited; + public delegate void MainXrayCoreProcessStartedHandler(); + + public event MainXrayCoreProcessStartedHandler? MainXrayCoreProcessStarted; + + public async Task StartMainXrayCoreProcess(BaseOutboundSetting outboundSetting, bool noStart = false) + { + if (MainXrayCoreProcess != null) + { + await StopMainXrayCoreProcess(); + } + + logger.LogInformation("正在启动主要 Xray-Core 进程..."); + var configJson = GenerateXrayCoreConfigJson(config.CurrentValue.Inbounds.ToInboundSettings(), outboundSetting); + var configFilePath = Path.GetFullPath(Path.Combine(WorkingDirectory, "main-config.json")); + await File.WriteAllTextAsync(configFilePath, JsonSerializer.Serialize(configJson)); + + logger.LogDebug("配置文件已写入: {configPath}", configFilePath); + + var process = CreateXrayCoreProcess(["run", "-c", configFilePath]); + process.Exited += async (_, _) => + { + MainXrayCoreProcessExited?.Invoke(); + await StopMainXrayCoreProcess(); + }; + process.Start(); + RedirectXrayCoreProcessOutput(process); + logger.LogInformation("主要 Xray-Core 进程已启动."); + MainXrayCoreProcessStarted?.Invoke(); + + MainXrayCoreProcess = process; + CurrentOutboundSetting = outboundSetting; + } + + public async Task StopMainXrayCoreProcess() + { + if (MainXrayCoreProcess == null) + { + return; + } + + if (!MainXrayCoreProcess.HasExited) + { + MainXrayCoreProcess.Kill(); + await MainXrayCoreProcess.WaitForExitAsync(); + } + MainXrayCoreProcess = null; + CurrentOutboundSetting = null; + } + + public void RedirectXrayCoreProcessOutput(Process process) + { + var processLogger = loggerFactory.CreateLogger($"Xray-Core.Process[{process.Id}]"); + process.BeginOutputReadLine(); + process.BeginErrorReadLine(); + + process.OutputDataReceived += (sender, e) => + { + if (!string.IsNullOrEmpty(e.Data)) + { + processLogger.LogInformation("{content}", e.Data); + } + }; + + process.ErrorDataReceived += (sender, e) => + { + if (!string.IsNullOrEmpty(e.Data)) + { + processLogger.LogError("{content}", e.Data); + } + }; } - } \ No newline at end of file