diff --git a/perf-hunt/.gitignore b/perf-hunt/.gitignore new file mode 100644 index 0000000000..cd42ee34e8 --- /dev/null +++ b/perf-hunt/.gitignore @@ -0,0 +1,2 @@ +bin/ +obj/ diff --git a/perf-hunt/Directory.Build.props b/perf-hunt/Directory.Build.props new file mode 100644 index 0000000000..01314b6c67 --- /dev/null +++ b/perf-hunt/Directory.Build.props @@ -0,0 +1,10 @@ + + + diff --git a/perf-hunt/Directory.Build.targets b/perf-hunt/Directory.Build.targets new file mode 100644 index 0000000000..0371595aa7 --- /dev/null +++ b/perf-hunt/Directory.Build.targets @@ -0,0 +1,3 @@ + + + diff --git a/perf-hunt/E2EBench/E2EBench.csproj b/perf-hunt/E2EBench/E2EBench.csproj new file mode 100644 index 0000000000..086303fd05 --- /dev/null +++ b/perf-hunt/E2EBench/E2EBench.csproj @@ -0,0 +1,25 @@ + + + + Exe + net9.0 + enable + disable + latest + true + + true + true + true + true + true + E2EBench + E2EBench + + + + + + + + diff --git a/perf-hunt/E2EBench/Program.cs b/perf-hunt/E2EBench/Program.cs new file mode 100644 index 0000000000..3cdba50309 --- /dev/null +++ b/perf-hunt/E2EBench/Program.cs @@ -0,0 +1,349 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// +// E2EBench: realistic end-to-end throughput/latency harness for the full MapReverseProxy pipeline. +// +// One process performs ONE measurement of a (target, http) combination so working-set and allocation +// snapshots stay clean. It hosts a real Kestrel backend, optionally a real YARP proxy (MapReverseProxy +// + LoadFromMemory, 1 route/1 cluster/1 destination -> backend), and an in-process async load client. +// +// --target direct : client -> backend (framework/Kestrel baseline, no YARP) +// --target proxy : client -> YARP proxy -> backend (framework + YARP) +// (proxy - direct) isolates YARP's end-to-end cost. +// +// HTTP: +// --http 1 : HTTP/1.1 everywhere +// --http 2 : HTTP/2 (h2c prior knowledge) client->front and, for proxy, front->backend +// +// Metrics over the measured window: throughput (req/s), latency percentiles, process-wide +// allocated-bytes/request (before/after differencing cancels the identical client+backend cost so the +// delta reflects the proxy change), and working set. Everything else in the process is byte-identical +// across a before/after comparison, so a before-vs-after allocation delta attributes to the proxy. + +using System; +using System.Diagnostics; +using System.Globalization; +using System.Linq; +using System.Net; +using System.Net.Http; +using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Server.Kestrel.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Yarp.ReverseProxy.Configuration; +using Yarp.ReverseProxy.Forwarder; + +namespace E2EBench; + +internal static class Program +{ + private static async Task Main(string[] args) + { + var target = GetStr(args, "--target", "proxy"); // proxy | direct + var http = GetInt(args, "--http", 1); // 1 | 2 + var connections = GetInt(args, "--connections", 64); + var warmupSec = GetInt(args, "--warmup", 5); + var measureSec = GetInt(args, "--duration", 10); + var bodyBytes = GetInt(args, "--body", 100); + var label = GetStr(args, "--label", "run"); + + var proto = http == 2 ? HttpProtocols.Http2 : HttpProtocols.Http1; + var reqVersion = http == 2 ? HttpVersion.Version20 : HttpVersion.Version11; + + var backendPort = FreePort(); + var backend = BuildBackend(backendPort, proto, bodyBytes); + await backend.StartAsync(); + var backendUrl = $"http://127.0.0.1:{backendPort}/"; + + IHost? proxy = null; + string targetUrl; + if (target == "proxy") + { + var proxyPort = FreePort(); + proxy = BuildProxy(proxyPort, proto, backendUrl, reqVersion); + await proxy.StartAsync(); + targetUrl = $"http://127.0.0.1:{proxyPort}/"; + } + else + { + targetUrl = backendUrl; + } + + var handler = new SocketsHttpHandler + { + EnableMultipleHttp2Connections = true, + MaxConnectionsPerServer = connections, + PooledConnectionLifetime = TimeSpan.FromMinutes(10), + PooledConnectionIdleTimeout = TimeSpan.FromMinutes(10), + AutomaticDecompression = DecompressionMethods.None, + UseProxy = false, + UseCookies = false, + }; + var client = new HttpMessageInvoker(handler); + var uri = new Uri(targetUrl); + + // Phase 1: warmup (also opens the connection pool). + await RunLoad(client, uri, reqVersion, connections, TimeSpan.FromSeconds(warmupSec), measure: false); + + // Phase 2: measured window. + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + var proc = Process.GetCurrentProcess(); + proc.Refresh(); + var wsBefore = proc.WorkingSet64; + var allocBefore = GC.GetTotalAllocatedBytes(precise: true); + var gc0 = GC.CollectionCount(0); + var gc1 = GC.CollectionCount(1); + var gc2 = GC.CollectionCount(2); + + var sw = Stopwatch.StartNew(); + var result = await RunLoad(client, uri, reqVersion, connections, TimeSpan.FromSeconds(measureSec), measure: true); + sw.Stop(); + + var allocAfter = GC.GetTotalAllocatedBytes(precise: true); + proc.Refresh(); + var wsAfter = proc.WorkingSet64; + + var elapsed = sw.Elapsed.TotalSeconds; + var rps = result.Count / elapsed; + var allocPerReq = result.Count > 0 ? (double)(allocAfter - allocBefore) / result.Count : 0; + var hist = result.Histogram; + var total = result.Count; + + Console.WriteLine(); + Console.WriteLine($"RESULT label={label} target={target} http={http} conns={connections} dur={elapsed:F1}s " + + $"requests={total} errors={result.Errors} rps={rps:F0} " + + $"p50us={Percentile(hist, total, 0.50)} p90us={Percentile(hist, total, 0.90)} " + + $"p99us={Percentile(hist, total, 0.99)} p999us={Percentile(hist, total, 0.999)} maxus={result.MaxMicros} " + + $"alloc_per_req_B={allocPerReq:F1} ws_MB={wsAfter / (1024.0 * 1024.0):F1} " + + $"gc0={GC.CollectionCount(0) - gc0} gc1={GC.CollectionCount(1) - gc1} gc2={GC.CollectionCount(2) - gc2}"); + + client.Dispose(); + if (proxy is not null) + { + await proxy.StopAsync(); + proxy.Dispose(); + } + await backend.StopAsync(); + backend.Dispose(); + return 0; + } + + private sealed class LoadResult + { + public long Count; + public long Errors; + public long MaxMicros; + public long[] Histogram = Array.Empty(); + } + + private const int HistBuckets = 200_000; // 1us resolution up to 200ms; above clamps to last bucket. + + private static async Task RunLoad( + HttpMessageInvoker client, Uri uri, Version version, int workers, TimeSpan duration, bool measure) + { + var deadline = Stopwatch.GetTimestamp() + (long)(duration.TotalSeconds * Stopwatch.Frequency); + var perWorker = new (long count, long errors, long max, long[] hist)[workers]; + var tasks = new Task[workers]; + + for (var w = 0; w < workers; w++) + { + var idx = w; + var hist = measure ? new long[HistBuckets] : Array.Empty(); + tasks[w] = Task.Run(async () => + { + long count = 0, errors = 0, max = 0; + var buffer = new byte[8192]; + while (Stopwatch.GetTimestamp() < deadline) + { + var start = Stopwatch.GetTimestamp(); + try + { + using var req = new HttpRequestMessage(HttpMethod.Get, uri) + { + Version = version, + VersionPolicy = HttpVersionPolicy.RequestVersionExact, + }; + using var resp = await client.SendAsync(req, CancellationToken.None); + var body = resp.Content; + using var s = await body.ReadAsStreamAsync(); + while (await s.ReadAsync(buffer) > 0) { } + if (!resp.IsSuccessStatusCode) + { + errors++; + continue; + } + } + catch + { + errors++; + continue; + } + + if (measure) + { + var us = (Stopwatch.GetTimestamp() - start) * 1_000_000L / Stopwatch.Frequency; + if (us > max) + { + max = us; + } + var bucket = us >= HistBuckets ? HistBuckets - 1 : (int)us; + hist[bucket]++; + count++; + } + } + perWorker[idx] = (count, errors, max, hist); + }); + } + + await Task.WhenAll(tasks); + + var result = new LoadResult { Histogram = measure ? new long[HistBuckets] : Array.Empty() }; + foreach (var (count, errors, max, hist) in perWorker) + { + result.Count += count; + result.Errors += errors; + if (max > result.MaxMicros) + { + result.MaxMicros = max; + } + if (measure && hist.Length > 0) + { + for (var i = 0; i < HistBuckets; i++) + { + result.Histogram[i] += hist[i]; + } + } + } + return result; + } + + private static long Percentile(long[] hist, long total, double p) + { + if (total == 0 || hist.Length == 0) + { + return -1; + } + var target = (long)Math.Ceiling(p * total); + long cum = 0; + for (var i = 0; i < hist.Length; i++) + { + cum += hist[i]; + if (cum >= target) + { + return i; + } + } + return hist.Length - 1; + } + + private static IHost BuildBackend(int port, HttpProtocols proto, int bodyBytes) + { + var payload = new byte[bodyBytes]; + for (var i = 0; i < payload.Length; i++) + { + payload[i] = (byte)('a' + (i % 26)); + } + + return Host.CreateDefaultBuilder() + .ConfigureLogging(b => b.ClearProviders().SetMinimumLevel(LogLevel.Warning)) + .ConfigureWebHostDefaults(web => + { + web.UseKestrel(k => k.Listen(IPAddress.Loopback, port, lo => lo.Protocols = proto)); + web.Configure(app => app.Run(async ctx => + { + ctx.Response.StatusCode = 200; + ctx.Response.ContentType = "application/octet-stream"; + ctx.Response.ContentLength = payload.Length; + await ctx.Response.Body.WriteAsync(payload); + })); + }) + .Build(); + } + + private static IHost BuildProxy(int port, HttpProtocols proto, string backendUrl, Version forwardVersion) + { + var routes = new[] + { + new RouteConfig + { + RouteId = "route0", + ClusterId = "cluster0", + Match = new RouteMatch { Path = "/{**catchall}" }, + }, + }; + var clusters = new[] + { + new ClusterConfig + { + ClusterId = "cluster0", + Destinations = new System.Collections.Generic.Dictionary + { + ["dest0"] = new DestinationConfig { Address = backendUrl }, + }, + HttpRequest = new ForwarderRequestConfig + { + Version = forwardVersion, + VersionPolicy = HttpVersionPolicy.RequestVersionExact, + }, + }, + }; + + return Host.CreateDefaultBuilder() + .ConfigureLogging(b => b.ClearProviders().SetMinimumLevel(LogLevel.Warning)) + .ConfigureWebHostDefaults(web => + { + web.UseKestrel(k => k.Listen(IPAddress.Loopback, port, lo => lo.Protocols = proto)); + web.ConfigureServices(services => + { + services.AddReverseProxy().LoadFromMemory(routes, clusters); + }); + web.Configure(app => + { + app.UseRouting(); + app.UseEndpoints(e => e.MapReverseProxy()); + }); + }) + .Build(); + } + + private static int FreePort() + { + var l = new TcpListener(IPAddress.Loopback, 0); + l.Start(); + var port = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + return port; + } + + private static int GetInt(string[] args, string name, int def) + { + for (var i = 0; i < args.Length - 1; i++) + { + if (args[i] == name && int.TryParse(args[i + 1], NumberStyles.Integer, CultureInfo.InvariantCulture, out var v)) + { + return v; + } + } + return def; + } + + private static string GetStr(string[] args, string name, string def) + { + for (var i = 0; i < args.Length - 1; i++) + { + if (args[i] == name) + { + return args[i + 1]; + } + } + return def; + } +} diff --git a/perf-hunt/PipelineBench/PipelineBench.csproj b/perf-hunt/PipelineBench/PipelineBench.csproj new file mode 100644 index 0000000000..46e343ee23 --- /dev/null +++ b/perf-hunt/PipelineBench/PipelineBench.csproj @@ -0,0 +1,30 @@ + + + + Exe + net9.0 + enable + disable + latest + true + true + + false + false + true + false + true + PipelineBench + PipelineBench + $(NoWarn);CS1591 + + + + + + + + + + diff --git a/perf-hunt/PipelineBench/Program.cs b/perf-hunt/PipelineBench/Program.cs new file mode 100644 index 0000000000..3daa5a506d --- /dev/null +++ b/perf-hunt/PipelineBench/Program.cs @@ -0,0 +1,579 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// +// PipelineBench: in-process micro-benchmark of the full MapReverseProxy request pipeline. +// +// Goal: isolate YARP per-request pipeline cost (routing endpoint metadata lookup, pipeline +// initializer, session affinity, load balancing, passive health, limits, forwarder middleware) +// from Kestrel and from the network/backend. It does this by: +// * Building the REAL public pipeline: UseRouting() -> UseEndpoints(e => e.MapReverseProxy()). +// * Replacing IHttpForwarder with a stub so we measure the pipeline AROUND the forwarder, +// not the forwarder internals (those were Phase 1 / direct-IHttpForwarder scope). +// * Driving a fresh DefaultHttpContext through the built RequestDelegate on a single thread. +// * Measuring exact allocations/request via GC.GetTotalAllocatedBytes(precise:true) (process-wide, +// so thread-pool continuation allocations in async mode are also counted) and CPU/request via +// Stopwatch, over N iterations x T trials (min + mean +/- sample stddev). +// +// Two modes per scenario: +// SYNC : stub forwarder completes synchronously. Cleanest CPU signal; captures all non-async +// allocations (feature object, feature store, lookups). +// ASYNC: stub forwarder awaits Task.Yield() once, forcing every genuinely-async middleware frame +// to suspend and heap-allocate its state machine + ExecutionContext capture, as happens in +// production (real network I/O suspends). CPU here is dominated by the thread-pool hop and +// is NOT a CPU signal -- only its allocation delta is meaningful. +// +// YARP-attributable cost = (scenario) - (routing-only baseline with the same route count). The +// baseline maps the identical route patterns to a trivial terminal endpoint, so DefaultHttpContext +// creation + ASP.NET routing cost cancels out. + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Globalization; +using System.IO.Hashing; +using System.Linq; +using System.Net.Http; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Routing; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Logging; +using Yarp.ReverseProxy.Configuration; +using Yarp.ReverseProxy.Forwarder; +using Yarp.ReverseProxy.Health; +using Yarp.ReverseProxy.LoadBalancing; +using Yarp.ReverseProxy.SessionAffinity; + +namespace PipelineBench; + +internal static class Program +{ + private static volatile int _sink; + + private static async Task Main(string[] args) + { + var iters = GetArg(args, "--iters", 50_000); + var trials = GetArg(args, "--trials", 12); + var warmup = GetArg(args, "--warmup", 50_000); + var label = GetArgString(args, "--label", "run"); + var only = GetArgString(args, "--only", null); + + Console.WriteLine($"# PipelineBench label={label}"); + Console.WriteLine($"runtime={System.Runtime.InteropServices.RuntimeInformation.FrameworkDescription}"); + Console.WriteLine($"os={System.Runtime.InteropServices.RuntimeInformation.OSDescription} arch={System.Runtime.InteropServices.RuntimeInformation.OSArchitecture}"); + Console.WriteLine($"gcServer={System.Runtime.GCSettings.IsServerGC} processors={Environment.ProcessorCount}"); + Console.WriteLine($"iters={iters} trials={trials} warmup={warmup}"); + Console.WriteLine(); + + var scenarios = BuildScenarios(); + if (only is not null) + { + scenarios = scenarios.Where(s => s.Name.Contains(only, StringComparison.OrdinalIgnoreCase)).ToList(); + } + + // Route-count -> routing-only baseline results (sync/async), built lazily. + var results = new List(); + + foreach (var scenario in scenarios) + { + Console.Error.WriteLine($"[running] {scenario.Name} ..."); + var (routes, clusters, reqPath, cookie) = BuildConfig(scenario); + + foreach (var forceAsync in new[] { false, true }) + { + RequestDelegate pipeline; + IServiceProvider provider; + try + { + (pipeline, provider) = BuildProxyPipeline(routes, clusters, forceAsync, scenario.Shape); + } + catch (Exception ex) + { + Console.WriteLine($"FAILED to build pipeline for {scenario.Name} (async={forceAsync}):"); + Console.WriteLine(ex); + return 2; + } + + Func makeCtx = () => MakeContext(provider, reqPath, cookie); + + // Correctness validation on first (sync) build. + if (!forceAsync) + { + var ok = await ValidateAsync(scenario.Name, pipeline, makeCtx, scenario); + if (!ok) + { + return 3; + } + } + + var r = await MeasureAsync($"{scenario.Name}", forceAsync, makeCtx, pipeline, iters, trials, warmup); + r.RouteCount = scenario.RouteCount; + results.Add(r); + } + } + + // Routing-only baselines for each distinct route count present. + var routeCounts = scenarios.Select(s => s.RouteCount).Distinct().OrderBy(x => x).ToList(); + var baselines = new Dictionary<(int, bool), Result>(); + foreach (var rc in routeCounts) + { + Console.Error.WriteLine($"[running] baseline routes={rc} ..."); + var (routes, _, reqPath, _) = BuildConfig(new Scenario($"baseline-routes{rc}", rc, 1, null, false, false, false)); + foreach (var forceAsync in new[] { false, true }) + { + var (pipeline, provider) = BuildRoutingOnlyPipeline(routes, forceAsync); + Func makeCtx = () => MakeContext(provider, reqPath, null); + var r = await MeasureAsync($"baseline-routes{rc}", forceAsync, makeCtx, pipeline, iters, trials, warmup); + r.RouteCount = rc; + r.IsBaseline = true; + baselines[(rc, forceAsync)] = r; + results.Add(r); + } + } + + PrintTable(results, baselines); + return 0; + } + + // ---- Scenario matrix ------------------------------------------------------------------- + + private static List BuildScenarios() => new() + { + // Name routes dest lb aff affHit passive + new Scenario("min-1r-1c-1d", 1, 1, null, false, false, false), + // Pipeline decomposition: same minimal config, different middleware chains, to attribute + // the async-frame cost. Default = SA+LB+PassiveHealth; NoPH = SA+LB; Minimal = none. + new Scenario("min-no-passivehealth", 1, 1, null, false, false, false, PipelineShape.NoPassiveHealth), + new Scenario("min-minimal-pipeline", 1, 1, null, false, false, false, PipelineShape.Minimal), + new Scenario("routes-100", 100, 1, null, false, false, false), + new Scenario("routes-1000", 1000, 1, null, false, false, false), + new Scenario("dest-8-roundrobin", 1, 8, LoadBalancingPolicies.RoundRobin, false, false, false), + new Scenario("dest-64-roundrobin", 1, 64, LoadBalancingPolicies.RoundRobin, false, false, false), + new Scenario("dest-8-p2c", 1, 8, LoadBalancingPolicies.PowerOfTwoChoices,false, false, false), + new Scenario("dest-64-p2c", 1, 64, LoadBalancingPolicies.PowerOfTwoChoices,false, false, false), + new Scenario("dest-5-healthyfilter", 1, 5, LoadBalancingPolicies.PowerOfTwoChoices,false, false, false), + new Scenario("affinity-8-miss", 1, 8, LoadBalancingPolicies.PowerOfTwoChoices,true, false, false), + new Scenario("affinity-8-hit", 1, 8, LoadBalancingPolicies.PowerOfTwoChoices,true, true, false), + new Scenario("passivehealth-1d", 1, 1, null, false, false, true), + }; + + // ---- Config construction --------------------------------------------------------------- + + private static (RouteConfig[] routes, ClusterConfig[] clusters, string reqPath, string? cookie) BuildConfig(Scenario s) + { + var routes = new RouteConfig[s.RouteCount]; + var clusters = new ClusterConfig[s.RouteCount]; + + for (var i = 0; i < s.RouteCount; i++) + { + var clusterId = "cluster" + i.ToString(CultureInfo.InvariantCulture); + var path = s.RouteCount == 1 ? "/{**catchall}" : $"/svc{i}/{{**catchall}}"; + + var dests = new Dictionary(StringComparer.OrdinalIgnoreCase); + for (var j = 0; j < s.DestCount; j++) + { + dests["dest" + j.ToString(CultureInfo.InvariantCulture)] = + new DestinationConfig { Address = $"http://127.0.0.1:5000/c{i}_d{j}" }; + } + + clusters[i] = new ClusterConfig + { + ClusterId = clusterId, + LoadBalancingPolicy = s.LbPolicy, + Destinations = dests, + SessionAffinity = s.AffinityEnabled + ? new SessionAffinityConfig + { + Enabled = true, + Policy = SessionAffinityConstants.Policies.HashCookie, + FailurePolicy = SessionAffinityConstants.FailurePolicies.Redistribute, + AffinityKeyName = "yarp.affinity", + } + : null, + HealthCheck = s.PassiveHealth + ? new HealthCheckConfig + { + Passive = new PassiveHealthCheckConfig + { + Enabled = true, + Policy = HealthCheckConstants.PassivePolicy.TransportFailureRate, + ReactivationPeriod = TimeSpan.FromSeconds(60), + }, + } + : null, + }; + + routes[i] = new RouteConfig + { + RouteId = "route" + i.ToString(CultureInfo.InvariantCulture), + ClusterId = clusterId, + Match = new RouteMatch { Path = path }, + }; + } + + var reqPath = s.RouteCount == 1 ? "/" : $"/svc{s.RouteCount - 1}/x"; + + // Affinity hit: reproduce HashCookieSessionAffinityPolicy.GetDestinationHash("dest0") using + // only public inputs/APIs (XxHash64 of the upper-invariant destination id, hex, lowercased). + string? cookie = null; + if (s.AffinityEnabled && s.AffinityHit) + { + var hash = HashDestination("dest0"); + cookie = "yarp.affinity=" + hash; + } + + return (routes, clusters, reqPath, cookie); + } + + private static string HashDestination(string destinationId) + { + var bytes = Encoding.Unicode.GetBytes(destinationId.ToUpperInvariant()); + var hash = XxHash64.Hash(bytes); + return Convert.ToHexString(hash).ToLowerInvariant(); + } + + // ---- Pipeline construction ------------------------------------------------------------- + + private static (RequestDelegate, IServiceProvider) BuildProxyPipeline( + RouteConfig[] routes, ClusterConfig[] clusters, bool forceAsync, PipelineShape shape) + { + var services = new ServiceCollection(); + AddCommonHostServices(services); + services.AddReverseProxy().LoadFromMemory(routes, clusters); + + // Replace the real forwarder with a stub so we measure the pipeline, not forwarder internals. + services.RemoveAll(); + services.AddSingleton(new StubForwarder(forceAsync)); + + var provider = services.BuildServiceProvider(); + + var app = new ApplicationBuilder(provider); + app.UseRouting(); + app.UseEndpoints(endpoints => + { + switch (shape) + { + case PipelineShape.Default: + // Exactly what the parameterless MapReverseProxy() installs. + endpoints.MapReverseProxy(); + break; + case PipelineShape.NoPassiveHealth: + endpoints.MapReverseProxy(proxy => + { + proxy.UseSessionAffinity(); + proxy.UseLoadBalancing(); + }); + break; + case PipelineShape.Minimal: + endpoints.MapReverseProxy(proxy => { }); + break; + } + }); + return (app.Build(), provider); + } + + private static (RequestDelegate, IServiceProvider) BuildRoutingOnlyPipeline(RouteConfig[] routes, bool forceAsync) + { + var services = new ServiceCollection(); + AddCommonHostServices(services); + services.AddRouting(); + var provider = services.BuildServiceProvider(); + + RequestDelegate terminal = forceAsync + ? static async ctx => { await Task.Yield(); ctx.Response.StatusCode = 200; } + : static ctx => { ctx.Response.StatusCode = 200; return Task.CompletedTask; }; + + var app = new ApplicationBuilder(provider); + app.UseRouting(); + app.UseEndpoints(endpoints => + { + foreach (var route in routes) + { + var pattern = string.IsNullOrEmpty(route.Match.Path) ? "/{**catchall}" : route.Match.Path!; + endpoints.Map(pattern, terminal); + } + }); + return (app.Build(), provider); + } + + private static void AddCommonHostServices(IServiceCollection services) + { + services.AddLogging(b => b.SetMinimumLevel(LogLevel.Warning)); + services.AddMetrics(); + // EndpointRoutingMiddleware requires a DiagnosticListener (normally provided by the host). + var listener = new DiagnosticListener("PipelineBench"); + services.AddSingleton(listener); + services.AddSingleton(listener); + } + + // ---- Request construction -------------------------------------------------------------- + + private static HttpContext MakeContext(IServiceProvider provider, string path, string? cookie) + { + var ctx = new DefaultHttpContext(); + ctx.RequestServices = provider; + var req = ctx.Request; + req.Method = "GET"; + req.Scheme = "http"; + req.Host = new HostString("localhost"); + req.Path = path; + req.Protocol = "HTTP/1.1"; + if (cookie is not null) + { + req.Headers.Cookie = cookie; + } + return ctx; + } + + // ---- Measurement ----------------------------------------------------------------------- + + private static async Task MeasureAsync( + string name, bool forceAsync, Func makeCtx, RequestDelegate pipeline, + int iters, int trials, int warmup) + { + for (var i = 0; i < warmup; i++) + { + var ctx = makeCtx(); + await pipeline(ctx); + _sink += ctx.Response.StatusCode; + } + + var ns = new double[trials]; + var bytes = new double[trials]; + + for (var t = 0; t < trials; t++) + { + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + var alloc0 = GC.GetTotalAllocatedBytes(precise: true); + var sw = Stopwatch.StartNew(); + for (var i = 0; i < iters; i++) + { + var ctx = makeCtx(); + await pipeline(ctx); + _sink += ctx.Response.StatusCode; + } + sw.Stop(); + var alloc1 = GC.GetTotalAllocatedBytes(precise: true); + + ns[t] = sw.Elapsed.TotalNanoseconds / iters; + bytes[t] = (double)(alloc1 - alloc0) / iters; + } + + return new Result + { + Name = name, + Async = forceAsync, + NsPerReqMin = ns.Min(), + NsPerReqMean = Mean(ns), + NsPerReqStd = Std(ns), + BytesPerReqMin = bytes.Min(), + BytesPerReqMean = Mean(bytes), + BytesPerReqStd = Std(bytes), + }; + } + + private static async Task ValidateAsync(string name, RequestDelegate pipeline, Func makeCtx, Scenario s) + { + try + { + var ctx = makeCtx(); + await pipeline(ctx); + var status = ctx.Response.StatusCode; + var endpoint = ctx.GetEndpoint(); + if (endpoint is null) + { + Console.WriteLine($"VALIDATION FAILED [{name}]: no endpoint matched (routing did not select a route)."); + return false; + } + if (status != 200) + { + Console.WriteLine($"VALIDATION FAILED [{name}]: expected status 200 from stub forwarder, got {status}. Endpoint='{endpoint.DisplayName}'."); + return false; + } + Console.Error.WriteLine($"[validate] {name}: status={status} endpoint='{endpoint.DisplayName}' OK"); + return true; + } + catch (Exception ex) + { + Console.WriteLine($"VALIDATION EXCEPTION [{name}]:"); + Console.WriteLine(ex); + return false; + } + } + + // ---- Output ---------------------------------------------------------------------------- + + private static void PrintTable(List results, Dictionary<(int, bool), Result> baselines) + { + Console.WriteLine(); + Console.WriteLine("## Raw results (min of trials; mean +/- sample stddev)"); + Console.WriteLine(); + Console.WriteLine("| scenario | mode | ns/req min | ns/req mean±sd | B/req min | B/req mean±sd |"); + Console.WriteLine("|---|---|--:|--:|--:|--:|"); + foreach (var r in results) + { + var mode = r.Async ? "async" : "sync"; + Console.WriteLine( + $"| {r.Name} | {mode} | {r.NsPerReqMin,8:F1} | {r.NsPerReqMean,8:F1}±{r.NsPerReqStd,-5:F1} | " + + $"{r.BytesPerReqMin,7:F1} | {r.BytesPerReqMean,7:F1}±{r.BytesPerReqStd,-4:F1} |"); + } + + Console.WriteLine(); + Console.WriteLine("## YARP-attributable delta vs routing-only baseline (same route count)"); + Console.WriteLine("(delta = scenario - baseline; isolates the proxy middleware chain from ASP.NET routing + DefaultHttpContext.)"); + Console.WriteLine(); + Console.WriteLine("| scenario | mode | Δ ns/req | Δ B/req |"); + Console.WriteLine("|---|---|--:|--:|"); + foreach (var r in results.Where(r => !r.IsBaseline)) + { + if (!baselines.TryGetValue((r.RouteCount, r.Async), out var b)) + { + continue; + } + var dns = r.NsPerReqMin - b.NsPerReqMin; + var db = r.BytesPerReqMin - b.BytesPerReqMin; + var mode = r.Async ? "async" : "sync"; + Console.WriteLine($"| {r.Name} | {mode} | {dns,8:F1} | {db,7:F1} |"); + } + + // Highlight the async-only extra (async delta minus sync delta) for the ubiquitous min path: + // this quantifies the state-machine/ExecutionContext overhead the proxy chain adds over a + // single-frame endpoint. + Console.WriteLine(); + Console.WriteLine("## Async-frame overhead (proxy chain extra async cost over baseline)"); + Console.WriteLine("(= (scenarioAsync - scenarioSync) - (baselineAsync - baselineSync); ~ heap state machines + EC captures the proxy chain adds)"); + Console.WriteLine(); + Console.WriteLine("| scenario | Δ async-frame B/req |"); + Console.WriteLine("|---|--:|"); + var byName = results.Where(r => !r.IsBaseline).GroupBy(r => r.Name); + foreach (var g in byName) + { + var sync = g.FirstOrDefault(r => !r.Async); + var asyncR = g.FirstOrDefault(r => r.Async); + if (sync is null || asyncR is null) + { + continue; + } + if (!baselines.TryGetValue((sync.RouteCount, false), out var bs) || + !baselines.TryGetValue((sync.RouteCount, true), out var ba)) + { + continue; + } + var extra = (asyncR.BytesPerReqMin - sync.BytesPerReqMin) - (ba.BytesPerReqMin - bs.BytesPerReqMin); + Console.WriteLine($"| {g.Key} | {extra,7:F1} |"); + } + + Console.WriteLine(); + Console.WriteLine($"(sink={_sink})"); + } + + // ---- helpers --------------------------------------------------------------------------- + + private static double Mean(double[] xs) => xs.Average(); + + private static double Std(double[] xs) + { + if (xs.Length < 2) + { + return 0; + } + var m = xs.Average(); + var s = xs.Sum(x => (x - m) * (x - m)); + return Math.Sqrt(s / (xs.Length - 1)); + } + + private static int GetArg(string[] args, string name, int def) + { + for (var i = 0; i < args.Length - 1; i++) + { + if (args[i] == name && int.TryParse(args[i + 1], out var v)) + { + return v; + } + } + return def; + } + + private static string? GetArgString(string[] args, string name, string? def) + { + for (var i = 0; i < args.Length - 1; i++) + { + if (args[i] == name) + { + return args[i + 1]; + } + } + return def; + } + + private sealed class Scenario + { + public Scenario(string name, int routeCount, int destCount, string? lbPolicy, bool affinityEnabled, bool affinityHit, bool passiveHealth, PipelineShape shape = PipelineShape.Default) + { + Name = name; + RouteCount = routeCount; + DestCount = destCount; + LbPolicy = lbPolicy; + AffinityEnabled = affinityEnabled; + AffinityHit = affinityHit; + PassiveHealth = passiveHealth; + Shape = shape; + } + + public string Name { get; } + public int RouteCount { get; } + public int DestCount { get; } + public string? LbPolicy { get; } + public bool AffinityEnabled { get; } + public bool AffinityHit { get; } + public bool PassiveHealth { get; } + public PipelineShape Shape { get; } + } + + private enum PipelineShape + { + Default, + NoPassiveHealth, + Minimal, + } + + private sealed class Result + { + public string Name { get; set; } = ""; + public bool Async { get; set; } + public bool IsBaseline { get; set; } + public int RouteCount { get; set; } + public double NsPerReqMin { get; set; } + public double NsPerReqMean { get; set; } + public double NsPerReqStd { get; set; } + public double BytesPerReqMin { get; set; } + public double BytesPerReqMean { get; set; } + public double BytesPerReqStd { get; set; } + } + + private sealed class StubForwarder : IHttpForwarder + { + private readonly bool _forceAsync; + + public StubForwarder(bool forceAsync) => _forceAsync = forceAsync; + + public async ValueTask SendAsync( + HttpContext context, string destinationPrefix, HttpMessageInvoker httpClient, + ForwarderRequestConfig requestConfig, HttpTransformer transformer) + { + if (_forceAsync) + { + await Task.Yield(); + } + context.Response.StatusCode = 200; + return ForwarderError.None; + } + } +} diff --git a/perf-hunt/README.md b/perf-hunt/README.md new file mode 100644 index 0000000000..a348c53390 --- /dev/null +++ b/perf-hunt/README.md @@ -0,0 +1,68 @@ +# perf-hunt — YARP `MapReverseProxy` pipeline perf harness + +Durable micro/E2E benchmarks used to find and guard a per-request allocation win in the full +`MapReverseProxy` request pipeline (routing → pipeline initializer → session affinity → load +balancing → passive health → limits → forwarder). + +These projects are **intentionally isolated** from the product build: + +* Not referenced by `YARP.slnx`, so `./build.sh` / `dotnet build YARP.slnx` never build them. +* A local `Directory.Build.props`/`Directory.Build.targets` stops the Arcade walk-up, so the harness + builds as a plain SDK project. Each project `ProjectReference`s `src/ReverseProxy` (built in its own + Arcade context). + +Build/run with the repo-pinned SDK: `./.dotnet/dotnet run -c Release --project perf-hunt/ -- `. + +## PipelineBench (allocation + CPU A/B, in-process) + +Drives a fresh `DefaultHttpContext` through the **real** public pipeline +(`UseRouting()` → `UseEndpoints(e => e.MapReverseProxy())`) with `IHttpForwarder` replaced by a stub, +so it measures the YARP pipeline **around** the forwarder (not forwarder internals). It reports exact +allocations/request via `GC.GetTotalAllocatedBytes(precise:true)` and ns/request via `Stopwatch` +(min + mean ± sample stddev over T trials), in two modes: + +* **sync** — stub completes synchronously (clean CPU signal; non-async allocations). +* **async** — stub `await Task.Yield()` once, forcing the genuinely-async frames to suspend and + heap-allocate their state machines, as happens in production (network I/O suspends). + +YARP-attributable cost = scenario − routing-only baseline (same route count), which cancels +`DefaultHttpContext` creation + ASP.NET routing. Scenarios cover: 1 / 100 / 1000 routes; 8 / 64 +destinations; round-robin and power-of-two; session affinity hit/miss; healthy-destination filtering; +passive health enabled; plus pipeline-decomposition shapes (`min-no-passivehealth`, `min-minimal`). + +```bash +./.dotnet/dotnet run -c Release --project perf-hunt/PipelineBench -- --iters 50000 --trials 12 --warmup 50000 +``` + +## E2EBench (throughput + latency, real Kestrel) + +One process per measurement: a real Kestrel backend, optionally a real YARP proxy +(`MapReverseProxy` + `LoadFromMemory`), and an in-process async load client. `--target direct` isolates +framework/Kestrel cost; `--target proxy` adds YARP. Supports HTTP/1.1 and HTTP/2 (h2c). Reports req/s, +latency percentiles, process-wide allocations/request, working set, and GC counts. Because the client +and backend are byte-identical across a before/after comparison, the before−after allocation delta +attributes to the proxy. + +```bash +./.dotnet/dotnet run -c Release --project perf-hunt/E2EBench -- --target proxy --http 1 --duration 10 --warmup 4 --connections 32 +``` + +## Finding this harness produced + +`PassiveHealthCheckMiddleware.Invoke` was an `async` method that always awaited `_next`, forcing a +**~112 B/request async state-machine allocation on every proxied request through the default pipeline**, +even though passive health checks are **disabled by default**. The parameterless `MapReverseProxy()` +pipeline now records passive-health outcomes in the terminal forwarder's existing async frame instead. +The public `UsePassiveHealthChecks()` middleware is unchanged, preserving custom-pipeline ordering and +`ReassignProxyRequest` semantics. + +Measured on Apple M-series (arm64), .NET 9.0.2: + +| Measurement | Before | After | Δ | +| --- | ---: | ---: | ---: | +| PipelineBench `min-1r-1c-1d` async B/req | 1856 | 1744 | **−112** | +| …all default-pipeline shapes (routes/dests/LB/affinity) | — | — | **−112** each | +| PipelineBench `passivehealth-1d` (enabled) | 1856 | 1744 | **−112** | +| E2E proxy HTTP/1.1 alloc/req | ~4372 | ~4259 | **−112** | +| E2E proxy HTTP/2 alloc/req | ~5900 | ~5795 | **−106** | +| E2E proxy HTTP/1.1 & HTTP/2 throughput/latency | — | — | flat (no regression) | diff --git a/src/ReverseProxy/Forwarder/ForwarderMiddleware.cs b/src/ReverseProxy/Forwarder/ForwarderMiddleware.cs index e48a5aba4c..71abab0a4b 100644 --- a/src/ReverseProxy/Forwarder/ForwarderMiddleware.cs +++ b/src/ReverseProxy/Forwarder/ForwarderMiddleware.cs @@ -2,10 +2,13 @@ // The .NET Foundation licenses this file to you under the MIT license. using System; +using System.Collections.Frozen; +using System.Collections.Generic; using System.Diagnostics; using System.Threading.Tasks; using Microsoft.AspNetCore.Http; using Microsoft.Extensions.Logging; +using Yarp.ReverseProxy.Health; using Yarp.ReverseProxy.Model; using Yarp.ReverseProxy.Utilities; @@ -20,17 +23,31 @@ internal sealed class ForwarderMiddleware private readonly RequestDelegate _next; // Unused, this middleware is always terminal private readonly ILogger _logger; private readonly IHttpForwarder _forwarder; - - public ForwarderMiddleware(RequestDelegate next, ILogger logger, IHttpForwarder forwarder, IRandomFactory randomFactory) + private readonly FrozenDictionary _passiveHealthCheckPolicies; + private readonly bool _recordPassiveHealthChecks; + + public ForwarderMiddleware( + RequestDelegate next, + ILogger logger, + IHttpForwarder forwarder, + IRandomFactory randomFactory, + IEnumerable passiveHealthCheckPolicies, + bool recordPassiveHealthChecks = false) { ArgumentNullException.ThrowIfNull(next); ArgumentNullException.ThrowIfNull(logger); ArgumentNullException.ThrowIfNull(forwarder); ArgumentNullException.ThrowIfNull(randomFactory); + ArgumentNullException.ThrowIfNull(passiveHealthCheckPolicies); + _next = next; _logger = logger; _forwarder = forwarder; _randomFactory = randomFactory; + _recordPassiveHealthChecks = recordPassiveHealthChecks; + _passiveHealthCheckPolicies = recordPassiveHealthChecks + ? passiveHealthCheckPolicies.ToDictionaryByUniqueId(p => p.Name) + : FrozenDictionary.Empty; } /// @@ -93,6 +110,11 @@ public async Task Invoke(HttpContext context) destination.ConcurrencyCounter.Decrement(); cluster.ConcurrencyCounter.Decrement(); } + + if (_recordPassiveHealthChecks) + { + PassiveHealthCheckMiddleware.RecordOutcome(context, _passiveHealthCheckPolicies); + } } private static class Log diff --git a/src/ReverseProxy/Health/PassiveHealthCheckMiddleware.cs b/src/ReverseProxy/Health/PassiveHealthCheckMiddleware.cs index 7cbcc83147..d5a2074a3c 100644 --- a/src/ReverseProxy/Health/PassiveHealthCheckMiddleware.cs +++ b/src/ReverseProxy/Health/PassiveHealthCheckMiddleware.cs @@ -26,7 +26,13 @@ public PassiveHealthCheckMiddleware(RequestDelegate next, IEnumerable policies) + { var proxyFeature = context.GetReverseProxyFeature(); var options = proxyFeature.Cluster.Config.HealthCheck?.Passive; @@ -36,7 +42,7 @@ public async Task Invoke(HttpContext context) return; } - var policy = _policies.GetRequiredServiceById(options.Policy, HealthCheckConstants.PassivePolicy.TransportFailureRate); + var policy = policies.GetRequiredServiceById(options.Policy, HealthCheckConstants.PassivePolicy.TransportFailureRate); var cluster = context.GetRouteModel().Cluster!; policy.RequestProxied(context, cluster, proxyFeature.ProxiedDestination); } diff --git a/src/ReverseProxy/Routing/ReverseProxyIEndpointRouteBuilderExtensions.cs b/src/ReverseProxy/Routing/ReverseProxyIEndpointRouteBuilderExtensions.cs index b35cc06aa3..b1c0a032df 100644 --- a/src/ReverseProxy/Routing/ReverseProxyIEndpointRouteBuilderExtensions.cs +++ b/src/ReverseProxy/Routing/ReverseProxyIEndpointRouteBuilderExtensions.cs @@ -24,12 +24,11 @@ public static class ReverseProxyIEndpointRouteBuilderExtensions /// public static ReverseProxyConventionBuilder MapReverseProxy(this IEndpointRouteBuilder endpoints) { - return endpoints.MapReverseProxy(app => + return MapReverseProxy(endpoints, app => { app.UseSessionAffinity(); app.UseLoadBalancing(); - app.UsePassiveHealthChecks(); - }); + }, recordPassiveHealthChecks: true); } /// @@ -37,6 +36,12 @@ public static ReverseProxyConventionBuilder MapReverseProxy(this IEndpointRouteB /// by default the initialization step and the final proxy step, but not LoadBalancingMiddleware or other intermediate components. /// public static ReverseProxyConventionBuilder MapReverseProxy(this IEndpointRouteBuilder endpoints, Action configureApp) + => MapReverseProxy(endpoints, configureApp, recordPassiveHealthChecks: false); + + private static ReverseProxyConventionBuilder MapReverseProxy( + IEndpointRouteBuilder endpoints, + Action configureApp, + bool recordPassiveHealthChecks) { ArgumentNullException.ThrowIfNull(endpoints); ArgumentNullException.ThrowIfNull(configureApp); @@ -45,7 +50,7 @@ public static ReverseProxyConventionBuilder MapReverseProxy(this IEndpointRouteB proxyAppBuilder.UseMiddleware(); configureApp(proxyAppBuilder); proxyAppBuilder.UseMiddleware(); - proxyAppBuilder.UseMiddleware(); + proxyAppBuilder.UseMiddleware(recordPassiveHealthChecks); var app = proxyAppBuilder.Build(); var proxyEndpointFactory = endpoints.ServiceProvider.GetRequiredService(); diff --git a/test/ReverseProxy.Tests/Forwarder/ForwarderMiddlewareTests.cs b/test/ReverseProxy.Tests/Forwarder/ForwarderMiddlewareTests.cs index e39b7bc873..cfd9778227 100644 --- a/test/ReverseProxy.Tests/Forwarder/ForwarderMiddlewareTests.cs +++ b/test/ReverseProxy.Tests/Forwarder/ForwarderMiddlewareTests.cs @@ -7,11 +7,14 @@ using System.Net.Http; using System.Threading.Tasks; using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.Logging; using Moq; using Xunit; using Yarp.Tests.Common; using Yarp.ReverseProxy.Configuration; +using Yarp.ReverseProxy.Health; using Yarp.ReverseProxy.Model; +using Yarp.ReverseProxy.Utilities; namespace Yarp.ReverseProxy.Forwarder.Tests; @@ -163,4 +166,235 @@ public async Task NoDestinations_503() Assert.Equal(ForwarderError.NoAvailableDestinations, errorFeature?.Error); Assert.Null(errorFeature.Exception); } + + [Fact] + public async Task Invoke_RecordPassiveHealthChecks_RecordsAfterForwardingCompletes() + { + var (context, cluster, destination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: true); + var order = new List(); + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns("policy1"); + policy.Setup(p => p.RequestProxied(context, cluster, destination)) + .Callback(() => + { + Assert.Equal(0, cluster.ConcurrencyCounter.Value); + Assert.Equal(0, destination.ConcurrentRequestCount); + order.Add("recorded"); + }); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + destination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(() => + { + order.Add("forwarded"); + return ForwarderError.None; + }); + + var sut = CreateMiddleware(forwarder.Object, new[] { policy.Object }, recordPassiveHealthChecks: true); + + await sut.Invoke(context); + + Assert.Equal(new[] { "forwarded", "recorded" }, order); + policy.Verify(p => p.RequestProxied(context, cluster, destination), Times.Once); + } + + [Fact] + public async Task Invoke_ForwarderReportsError_RecordsPassiveHealthOutcome() + { + var (context, cluster, destination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: true); + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns("policy1"); + policy.Setup(p => p.RequestProxied(context, cluster, destination)) + .Callback(() => + { + var error = context.Features.Get(); + Assert.Equal(ForwarderError.RequestTimedOut, error?.Error); + }); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + destination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(() => + { + context.Features.Set( + new ForwarderErrorFeature(ForwarderError.RequestTimedOut, new TimeoutException())); + return ForwarderError.RequestTimedOut; + }); + + var sut = CreateMiddleware(forwarder.Object, new[] { policy.Object }, recordPassiveHealthChecks: true); + + await sut.Invoke(context); + + policy.Verify(p => p.RequestProxied(context, cluster, destination), Times.Once); + } + + [Fact] + public async Task Invoke_ForwarderThrows_DoesNotRecordPassiveHealthOutcome() + { + var (context, cluster, destination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: true); + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns("policy1"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + destination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ThrowsAsync(new InvalidOperationException("Forwarder failure")); + + var sut = CreateMiddleware(forwarder.Object, new[] { policy.Object }, recordPassiveHealthChecks: true); + + await Assert.ThrowsAsync(() => sut.Invoke(context)); + + policy.VerifyGet(p => p.Name, Times.Once); + policy.VerifyNoOtherCalls(); + Assert.Equal(0, cluster.ConcurrencyCounter.Value); + Assert.Equal(0, destination.ConcurrentRequestCount); + } + + [Fact] + public async Task Invoke_RecordPassiveHealthChecksDisabled_DoesNotRecord() + { + var (context, _, destination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: false); + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns("policy1"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + destination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(ForwarderError.None); + + var sut = CreateMiddleware(forwarder.Object, new[] { policy.Object }, recordPassiveHealthChecks: true); + + await sut.Invoke(context); + + policy.VerifyGet(p => p.Name, Times.Once); + policy.VerifyNoOtherCalls(); + } + + [Fact] + public async Task Invoke_PassiveHealthRecordingNotIntegrated_DoesNotRecord() + { + var (context, _, destination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: true); + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns("policy1"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + destination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(ForwarderError.None); + + var sut = CreateMiddleware(forwarder.Object, new[] { policy.Object }, recordPassiveHealthChecks: false); + + await sut.Invoke(context); + + policy.VerifyNoOtherCalls(); + } + + [Fact] + public async Task Invoke_ReassignedDuringForwarding_RecordsAgainstFinalClusterAndDestination() + { + var (context, initialCluster, initialDestination) = CreateContext("cluster1", "policy1", passiveHealthEnabled: true); + var (_, finalCluster, finalDestination) = CreateContext("cluster2", "policy2", passiveHealthEnabled: true); + var initialPolicy = new Mock(); + initialPolicy.SetupGet(p => p.Name).Returns("policy1"); + var finalPolicy = new Mock(); + finalPolicy.SetupGet(p => p.Name).Returns("policy2"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + context, + initialDestination.Model.Config.Address, + context.GetReverseProxyFeature().Cluster.HttpClient, + It.IsAny(), + It.IsAny())) + .ReturnsAsync(() => + { + var finalRoute = new RouteModel(new RouteConfig(), finalCluster, HttpTransformer.Default); + context.ReassignProxyRequest(finalRoute, finalCluster); + context.GetReverseProxyFeature().ProxiedDestination = finalDestination; + return ForwarderError.None; + }); + + var sut = CreateMiddleware( + forwarder.Object, + new[] { initialPolicy.Object, finalPolicy.Object }, + recordPassiveHealthChecks: true); + + await sut.Invoke(context); + + initialPolicy.VerifyGet(p => p.Name, Times.Once); + initialPolicy.VerifyNoOtherCalls(); + finalPolicy.Verify(p => p.RequestProxied(context, finalCluster, finalDestination), Times.Once); + Assert.Equal(0, initialCluster.ConcurrencyCounter.Value); + Assert.Equal(0, initialDestination.ConcurrentRequestCount); + } + + private ForwarderMiddleware CreateMiddleware( + IHttpForwarder forwarder, + IEnumerable policies, + bool recordPassiveHealthChecks) + { + return new ForwarderMiddleware( + _ => Task.CompletedTask, + Mock>().Object, + forwarder, + Mock().Object, + policies, + recordPassiveHealthChecks); + } + + private static (DefaultHttpContext Context, ClusterState Cluster, DestinationState Destination) CreateContext( + string clusterId, + string policy, + bool passiveHealthEnabled) + { + var context = new DefaultHttpContext(); + var httpClient = new HttpMessageInvoker(new Mock().Object); + var cluster = new ClusterState(clusterId); + var clusterModel = new ClusterModel( + new ClusterConfig + { + ClusterId = clusterId, + HealthCheck = new HealthCheckConfig + { + Passive = new PassiveHealthCheckConfig + { + Enabled = passiveHealthEnabled, + Policy = policy, + } + } + }, + httpClient); + cluster.Model = clusterModel; + var destination = cluster.Destinations.GetOrAdd( + "destination1", + id => new DestinationState(id) + { + Model = new DestinationModel(new DestinationConfig { Address = "https://localhost:123/" }) + }); + var route = new RouteModel(new RouteConfig { RouteId = "route1" }, cluster, HttpTransformer.Default); + context.Features.Set( + new ReverseProxyFeature + { + AvailableDestinations = new List { destination }.AsReadOnly(), + Cluster = clusterModel, + Route = route, + }); + + return (context, cluster, destination); + } } diff --git a/test/ReverseProxy.Tests/Health/PassiveHealthCheckMiddlewareTests.cs b/test/ReverseProxy.Tests/Health/PassiveHealthCheckMiddlewareTests.cs index 7cda8f9f53..b8124ecdbc 100644 --- a/test/ReverseProxy.Tests/Health/PassiveHealthCheckMiddlewareTests.cs +++ b/test/ReverseProxy.Tests/Health/PassiveHealthCheckMiddlewareTests.cs @@ -1,6 +1,7 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System.Collections.Generic; using System.Linq; using System.Net.Http; using System.Threading.Tasks; @@ -95,6 +96,126 @@ public async Task Invoke_PassiveHealthCheckIsEnabledButNoDestinationSelected_DoN policies[1].VerifyNoOtherCalls(); } + [Fact] + public async Task Invoke_NextIsInvokedBeforePolicyRecording() + { + // Ordering: post-processing (health recording) must happen AFTER _next completes. + var policies = new[] { GetPolicy("policy0") }; + var cluster0 = GetClusterInfo("cluster0", "policy0"); + var order = new List(); + var middleware = new PassiveHealthCheckMiddleware(_ => + { + order.Add("next"); + return Task.CompletedTask; + }, policies.Select(p => p.Object)); + + var context0 = GetContext(cluster0, selectedDestination: 1, error: null); + policies[0].Setup(p => p.RequestProxied(It.IsAny(), It.IsAny(), It.IsAny())) + .Callback(() => order.Add("recorded")); + + await middleware.Invoke(context0); + + Assert.Equal(new[] { "next", "recorded" }, order); + } + + [Fact] + public async Task Invoke_EnabledAtEntry_ReassignedToAnotherEnabledClusterDuringNext_RecordsAgainstFinalCluster() + { + // Preserves ReassignProxyRequest semantics: when the request is reassigned to a different, + // also-enabled cluster during downstream processing, the outcome is recorded against the + // FINAL cluster/destination (read after _next) using the final cluster's policy. + var policies = new[] { GetPolicy("policy0"), GetPolicy("policy1") }; + var cluster0 = GetClusterInfo("cluster0", "policy0"); + var cluster1 = GetClusterInfo("cluster1", "policy1"); + + var context = GetContext(cluster0, selectedDestination: 0, error: null); + + var middleware = new PassiveHealthCheckMiddleware(ctx => + { + ReassignProxyRequest(ctx, cluster1, selectedDestination: 1); + return Task.CompletedTask; + }, policies.Select(p => p.Object)); + + await middleware.Invoke(context); + + policies[1].Verify(p => p.RequestProxied(context, cluster1, cluster1.DestinationsState.AllDestinations[1]), Times.Once); + policies[1].VerifyGet(p => p.Name, Times.Once); + policies[1].VerifyNoOtherCalls(); + // The entry cluster's policy must NOT be used for recording. + policies[0].VerifyGet(p => p.Name, Times.Once); + policies[0].VerifyNoOtherCalls(); + } + + [Fact] + public async Task Invoke_EnabledAtEntry_ReassignedToDisabledClusterDuringNext_DoesNotRecord() + { + // Preserves semantics: if the final (reassigned) cluster has passive health disabled, nothing + // is recorded even though the entry cluster had it enabled. + var policies = new[] { GetPolicy("policy0"), GetPolicy("policy1") }; + var cluster0 = GetClusterInfo("cluster0", "policy0"); + var cluster1 = GetClusterInfo("cluster1", "policy1", enabled: false); + + var context = GetContext(cluster0, selectedDestination: 0, error: null); + + var middleware = new PassiveHealthCheckMiddleware(ctx => + { + ReassignProxyRequest(ctx, cluster1, selectedDestination: 1); + return Task.CompletedTask; + }, policies.Select(p => p.Object)); + + await middleware.Invoke(context); + + policies[0].VerifyGet(p => p.Name, Times.Once); + policies[0].VerifyNoOtherCalls(); + policies[1].VerifyGet(p => p.Name, Times.Once); + policies[1].VerifyNoOtherCalls(); + } + + [Fact] + public async Task Invoke_DisabledAtEntry_ReassignedToEnabledClusterDuringNext_RecordsAgainstFinalCluster() + { + var policies = new[] { GetPolicy("policy0"), GetPolicy("policy1") }; + var cluster0 = GetClusterInfo("cluster0", "policy0", enabled: false); + var cluster1 = GetClusterInfo("cluster1", "policy1"); + + var context = GetContext(cluster0, selectedDestination: 0, error: null); + var middleware = new PassiveHealthCheckMiddleware(ctx => + { + ReassignProxyRequest(ctx, cluster1, selectedDestination: 1); + return Task.CompletedTask; + }, policies.Select(p => p.Object)); + + await middleware.Invoke(context); + + policies[1].Verify(p => p.RequestProxied(context, cluster1, cluster1.DestinationsState.AllDestinations[1]), Times.Once); + policies[0].VerifyGet(p => p.Name, Times.Once); + policies[0].VerifyNoOtherCalls(); + policies[1].VerifyGet(p => p.Name, Times.Once); + policies[1].VerifyNoOtherCalls(); + } + + [Fact] + public async Task Invoke_DisabledAtEntry_InvokesNextAndDoesNotRecord() + { + var policies = new[] { GetPolicy("policy0"), GetPolicy("policy1") }; + var cluster0 = GetClusterInfo("cluster0", "policy0", enabled: false); + var nextInvoked = false; + var middleware = new PassiveHealthCheckMiddleware(_ => + { + nextInvoked = true; + return Task.CompletedTask; + }, policies.Select(p => p.Object)); + + var context0 = GetContext(cluster0, selectedDestination: 0, error: null); + await middleware.Invoke(context0); + + Assert.True(nextInvoked); + policies[0].VerifyGet(p => p.Name, Times.Once); + policies[0].VerifyNoOtherCalls(); + policies[1].VerifyGet(p => p.Name, Times.Once); + policies[1].VerifyNoOtherCalls(); + } + private HttpContext GetContext(ClusterState cluster, int selectedDestination, IForwarderErrorFeature error) { var context = new DefaultHttpContext(); @@ -103,6 +224,13 @@ private HttpContext GetContext(ClusterState cluster, int selectedDestination, IF return context; } + private static void ReassignProxyRequest(HttpContext context, ClusterState cluster, int selectedDestination) + { + var route = new RouteModel(new RouteConfig(), cluster, HttpTransformer.Default); + context.ReassignProxyRequest(route, cluster); + context.GetReverseProxyFeature().ProxiedDestination = cluster.DestinationsState.AllDestinations[selectedDestination]; + } + private Mock GetPolicy(string name) { var policy = new Mock(); diff --git a/test/ReverseProxy.Tests/Routing/ReverseProxyIEndpointRouteBuilderExtensionsTests.cs b/test/ReverseProxy.Tests/Routing/ReverseProxyIEndpointRouteBuilderExtensionsTests.cs new file mode 100644 index 0000000000..85be814958 --- /dev/null +++ b/test/ReverseProxy.Tests/Routing/ReverseProxyIEndpointRouteBuilderExtensionsTests.cs @@ -0,0 +1,330 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System; +using System.Collections.Generic; +using System.IO; +using System.Net; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Moq; +using Xunit; +using Yarp.ReverseProxy.Configuration; +using Yarp.ReverseProxy.Forwarder; +using Yarp.ReverseProxy.Health; +using Yarp.ReverseProxy.Model; + +namespace Yarp.ReverseProxy.Routing.Tests; + +public class ReverseProxyIEndpointRouteBuilderExtensionsTests +{ + private const string PassivePolicyName = "TestPassivePolicy"; + private static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(10); + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MapReverseProxy_Success_RecordsPassiveHealthAfterForwardingExactlyOnce(bool useDefaultPipeline) + { + var order = new List(); + var forwardingStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var completeForwarding = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(async () => + { + forwardingStarted.SetResult(true); + await completeForwarding.Task; + order.Add("forwarded"); + return ForwarderError.None; + }); + var policy = CreatePassivePolicy(); + policy.Setup(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => order.Add("recorded")); + + using var host = await CreateHostAsync(useDefaultPipeline, forwarder.Object, policy.Object); + var requestTask = host.GetTestClient().GetAsync("/"); + + await forwardingStarted.Task.WaitAsync(TestTimeout); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + + completeForwarding.SetResult(true); + using var response = await requestTask.WaitAsync(TestTimeout); + + response.EnsureSuccessStatusCode(); + Assert.Equal(new[] { "forwarded", "recorded" }, order); + forwarder.Verify(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Once); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Once); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MapReverseProxy_ForwarderThrows_PropagatesExceptionWithoutRecording(bool useDefaultPipeline) + { + var expectedException = new InvalidOperationException("Forwarder failure"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns(async () => + { + await Task.Yield(); + throw expectedException; + }); + var policy = CreatePassivePolicy(); + + using var host = await CreateHostAsync(useDefaultPipeline, forwarder.Object, policy.Object); + + var exception = await Assert.ThrowsAsync( + () => host.GetTestClient().GetAsync("/").WaitAsync(TestTimeout)); + + Assert.Same(expectedException, exception); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MapReverseProxy_ClientCancellation_RemainsCanceledWithoutRecording(bool useDefaultPipeline) + { + var forwardingStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var forwardingCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns((HttpContext context, string _, HttpMessageInvoker _, ForwarderRequestConfig _, HttpTransformer _) => + WaitForCancellationAsync(context.RequestAborted, forwardingStarted, forwardingCanceled)); + var policy = CreatePassivePolicy(); + + using var host = await CreateHostAsync(useDefaultPipeline, forwarder.Object, policy.Object); + using var cancellationSource = new CancellationTokenSource(); + var requestTask = host.GetTestClient().GetAsync("/", cancellationSource.Token); + + await forwardingStarted.Task.WaitAsync(TestTimeout); + cancellationSource.Cancel(); + + await forwardingCanceled.Task.WaitAsync(TestTimeout); + await Assert.ThrowsAnyAsync(() => requestTask.WaitAsync(TestTimeout)); + Assert.True(requestTask.IsCanceled); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MapReverseProxy_ForwarderError_RecordsPassiveHealthAfterForwardingExactlyOnce(bool useDefaultPipeline) + { + var order = new List(); + var forwarderException = new IOException("Destination failure"); + var forwarder = new Mock(); + forwarder.Setup(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Returns((HttpContext context, string _, HttpMessageInvoker _, ForwarderRequestConfig _, HttpTransformer _) => + { + return CompleteWithErrorAsync(context, order, forwarderException); + }); + var policy = CreatePassivePolicy(); + policy.Setup(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback((context, _, _) => + { + var error = Assert.IsAssignableFrom(context.Features.Get()); + Assert.Equal(ForwarderError.Request, error.Error); + Assert.Same(forwarderException, error.Exception); + order.Add("recorded"); + }); + + using var host = await CreateHostAsync(useDefaultPipeline, forwarder.Object, policy.Object); + using var response = await host.GetTestClient().GetAsync("/").WaitAsync(TestTimeout); + + Assert.Equal(HttpStatusCode.BadGateway, response.StatusCode); + Assert.Equal(new[] { "forwarded", "recorded" }, order); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Once); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task MapReverseProxy_NoDestinations_ReturnsServiceUnavailableWithoutRecording(bool useDefaultPipeline) + { + var forwarder = new Mock(); + var policy = CreatePassivePolicy(); + + using var host = await CreateHostAsync( + useDefaultPipeline, + forwarder.Object, + policy.Object, + includeDestination: false); + using var response = await host.GetTestClient().GetAsync("/").WaitAsync(TestTimeout); + + Assert.Equal(HttpStatusCode.ServiceUnavailable, response.StatusCode); + forwarder.Verify(f => f.SendAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + policy.Verify(p => p.RequestProxied( + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Never); + } + + private static Mock CreatePassivePolicy() + { + var policy = new Mock(); + policy.SetupGet(p => p.Name).Returns(PassivePolicyName); + return policy; + } + + private static async ValueTask WaitForCancellationAsync( + CancellationToken cancellationToken, + TaskCompletionSource forwardingStarted, + TaskCompletionSource forwardingCanceled) + { + forwardingStarted.SetResult(true); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + return ForwarderError.None; + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + forwardingCanceled.SetResult(true); + throw; + } + } + + private static async ValueTask CompleteWithErrorAsync( + HttpContext context, + List order, + Exception exception) + { + await Task.Yield(); + context.Response.StatusCode = StatusCodes.Status502BadGateway; + context.Features.Set( + new ForwarderErrorFeature(ForwarderError.Request, exception)); + order.Add("forwarded"); + return ForwarderError.Request; + } + + private static Task CreateHostAsync( + bool useDefaultPipeline, + IHttpForwarder forwarder, + IPassiveHealthCheckPolicy policy, + bool includeDestination = true) + { + var routes = new[] + { + new RouteConfig + { + RouteId = "route1", + ClusterId = "cluster1", + Match = new RouteMatch { Path = "/{**catchall}" }, + } + }; + var destinations = new Dictionary(StringComparer.OrdinalIgnoreCase); + if (includeDestination) + { + destinations.Add("destination1", new DestinationConfig { Address = "http://localhost/" }); + } + + var clusters = new[] + { + new ClusterConfig + { + ClusterId = "cluster1", + Destinations = destinations, + HealthCheck = new HealthCheckConfig + { + Passive = new PassiveHealthCheckConfig + { + Enabled = true, + Policy = PassivePolicyName, + } + } + } + }; + + return new HostBuilder() + .ConfigureWebHost(webHost => + { + webHost.UseTestServer(); + webHost.ConfigureServices(services => + { + services.AddReverseProxy().LoadFromMemory(routes, clusters); + services.AddSingleton(forwarder); + services.AddSingleton(policy); + }); + webHost.Configure(app => + { + app.UseRouting(); + app.UseEndpoints(endpoints => + { + if (useDefaultPipeline) + { + endpoints.MapReverseProxy(); + } + else + { + endpoints.MapReverseProxy(proxyApp => + { + proxyApp.UseLoadBalancing(); + proxyApp.UsePassiveHealthChecks(); + }); + } + }); + }); + }) + .StartAsync(); + } +}