Joseph Doherty
e9e44a9ef0
The WP-4 runs measured the stream path only up to the bounded send channel; the probe deliberately substituted the socket hop with a controllable reader, which is exactly where the R2 SiteStreamEventBatcher lives — so batching had never been in any benchmark's hot path. The harness now also attaches per-site wire probes: a real Kestrel h2c listener hosting the production SiteStreamGrpcServer, dialed by the production SiteStreamGrpcClient with batching negotiated, batch sizes read from the production batch_size histogram via a MeterListener (--wire-probes-per-site, default 10; 0 restores the old shape). Full-scale run (10x500x75, 20-min window, results doc §8 + raw JSON): 899,807 events in 12,462 frames (mean 72.2 ev/frame, p50=p99=75), wire P99 43.4 ms vs the 250 ms budget — every batch pays the full 25 ms linger since a 75-event burst never hits the 100 cap — zero drops, zero faults, offered load 100.0% of nominal with fan-out doubled. CI smoke Fact now asserts the wire path carries traffic and observes frames.
356 lines
19 KiB
C#
356 lines
19 KiB
C#
using System.Diagnostics;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
|
|
using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
|
|
using ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios;
|
|
|
|
namespace ZB.MOM.WW.ScadaBridge.LoadHarness;
|
|
|
|
/// <summary>Everything one harness run measured, ready for serialization.</summary>
|
|
/// <param name="Config">The configuration the run executed under.</param>
|
|
/// <param name="Environment">Host/runtime description.</param>
|
|
/// <param name="StartedUtc">Run start.</param>
|
|
/// <param name="TotalSeconds">Total wall time including ramp and teardown.</param>
|
|
/// <param name="SiteRampSeconds">Wall time to build all site fixtures (before instances).</param>
|
|
/// <param name="InstanceRampSeconds">Wall time to create every Instance Actor across all sites.</param>
|
|
/// <param name="SlowestSiteInstanceRampSeconds">Slowest single site's instance ramp — the "deploy 500 instances to a site" figure.</param>
|
|
/// <param name="TagUpdateLatency">End-to-end DCL-boundary to stream-subscriber latency.</param>
|
|
/// <param name="EmittedTagUpdates">Tag updates offered during the whole run.</param>
|
|
/// <param name="SteadyStateEmittedTagUpdates">Tag updates offered during the measurement window only.</param>
|
|
/// <param name="AchievedUpdatesPerSecond">Offered load actually achieved in the measurement window.</param>
|
|
/// <param name="NominalUpdatesPerSecond">Offered load the configuration called for.</param>
|
|
/// <param name="DriverLagSeconds">Cumulative driver slice overrun (harness-bound load shortfall).</param>
|
|
/// <param name="DriverSkippedNoCallback">Emissions skipped before subscriptions existed.</param>
|
|
/// <param name="SteadyStateResources">Resource behaviour over the measurement window.</param>
|
|
/// <param name="WholeRunResources">Resource behaviour over the whole run.</param>
|
|
/// <param name="ResourceSamples">
|
|
/// Every raw resource sample. Retained because the summary alone cannot distinguish a
|
|
/// slow leak from a heap that simply sawtooths — especially when the run records zero
|
|
/// gen-2 collections, where a positive least-squares slope may be nothing more than
|
|
/// gen-2 garbage that was never collected.
|
|
/// </param>
|
|
/// <param name="HealthReportLatency">Health report collect+ingest latency.</param>
|
|
/// <param name="HealthReportsDelivered">Health reports ingested by the central aggregator.</param>
|
|
/// <param name="SitesTrackedByAggregator">Sites the central aggregator ended up tracking.</param>
|
|
/// <param name="DebugSnapshotLatency">Debug view snapshot round-trip latency under load.</param>
|
|
/// <param name="DebugSnapshotsCompleted">Debug snapshots that completed.</param>
|
|
/// <param name="DebugSnapshotTimeouts">Debug snapshots that timed out.</param>
|
|
/// <param name="StreamProbeReceived">Events delivered to live stream subscribers.</param>
|
|
/// <param name="StreamProbeDropped">Events evicted by live subscribers' bounded channels.</param>
|
|
/// <param name="StoreAndForwardDrain">Store-and-forward drain measurement (register row 50).</param>
|
|
/// <param name="SlowSubscriber">Slow-subscriber isolation measurement (register row 50).</param>
|
|
/// <param name="WireStream">
|
|
/// Wire-path stream measurement — real Kestrel + real <c>SiteStreamGrpcClient</c> with
|
|
/// R2 batching negotiated — or null when <see cref="HarnessConfig.WireProbesPerSite"/> is 0.
|
|
/// </param>
|
|
public sealed record HarnessRunResult(
|
|
HarnessConfig Config,
|
|
EnvironmentInfo Environment,
|
|
DateTimeOffset StartedUtc,
|
|
double TotalSeconds,
|
|
double SiteRampSeconds,
|
|
double InstanceRampSeconds,
|
|
double SlowestSiteInstanceRampSeconds,
|
|
LatencySnapshot TagUpdateLatency,
|
|
long EmittedTagUpdates,
|
|
long SteadyStateEmittedTagUpdates,
|
|
double AchievedUpdatesPerSecond,
|
|
double NominalUpdatesPerSecond,
|
|
double DriverLagSeconds,
|
|
long DriverSkippedNoCallback,
|
|
ResourceWindowSummary? SteadyStateResources,
|
|
ResourceWindowSummary? WholeRunResources,
|
|
IReadOnlyList<ResourceSample> ResourceSamples,
|
|
LatencySnapshot HealthReportLatency,
|
|
long HealthReportsDelivered,
|
|
int SitesTrackedByAggregator,
|
|
LatencySnapshot DebugSnapshotLatency,
|
|
long DebugSnapshotsCompleted,
|
|
long DebugSnapshotTimeouts,
|
|
long StreamProbeReceived,
|
|
long StreamProbeDropped,
|
|
StoreAndForwardDrainResult? StoreAndForwardDrain,
|
|
SlowSubscriberResult? SlowSubscriber,
|
|
WireStreamResult? WireStream);
|
|
|
|
/// <summary>
|
|
/// Steady-state-window measurement of the wire-path stream probes: the production
|
|
/// gRPC server (Kestrel h2c) and client with the R2 batcher, protobuf serialization
|
|
/// and HTTP/2 flow control INSIDE the measured path. All figures are deltas over the
|
|
/// steady-state window only.
|
|
/// </summary>
|
|
/// <param name="ProbeCount">Wire subscribers attached across the fleet.</param>
|
|
/// <param name="Latency">End-to-end DCL-boundary → post-unpack client callback latency.</param>
|
|
/// <param name="EventsReceived">Domain events delivered by the client in the window.</param>
|
|
/// <param name="ServerDroppedEvents">Events evicted by the wire servers' bounded send channels in the window.</param>
|
|
/// <param name="StreamFaults">Stream faults reported by the client (expected 0).</param>
|
|
/// <param name="GracefulCompletions">Graceful server-side stream completions (expected 0 inside the 4 h lifetime).</param>
|
|
/// <param name="Batches">The frame/batch-size distribution the R2 batcher actually produced.</param>
|
|
public sealed record WireStreamResult(
|
|
int ProbeCount,
|
|
LatencySnapshot Latency,
|
|
long EventsReceived,
|
|
long ServerDroppedEvents,
|
|
long StreamFaults,
|
|
long GracefulCompletions,
|
|
BatchSizeWindow Batches);
|
|
|
|
/// <summary>Host and runtime facts recorded alongside the numbers.</summary>
|
|
/// <param name="MachineName">Host name.</param>
|
|
/// <param name="OsDescription">Operating system description.</param>
|
|
/// <param name="ProcessorCount">Logical processors visible to the process.</param>
|
|
/// <param name="RuntimeVersion">.NET runtime version.</param>
|
|
/// <param name="ServerGc">Whether server GC is active.</param>
|
|
public sealed record EnvironmentInfo(
|
|
string MachineName,
|
|
string OsDescription,
|
|
int ProcessorCount,
|
|
string RuntimeVersion,
|
|
bool ServerGc);
|
|
|
|
/// <summary>
|
|
/// Orchestrates a full run: build sites, ramp instances, attach subscribers, drive
|
|
/// tag updates for the sustained window while sampling resources and probing
|
|
/// observability, then run the two register-row-50 scenarios on a dedicated site.
|
|
/// </summary>
|
|
public static class HarnessRun
|
|
{
|
|
/// <summary>Data connections each site spreads its tags across.</summary>
|
|
public const int ConnectionsPerSite = 5;
|
|
|
|
/// <summary>Subscribers attached in the slow-subscriber isolation probe.</summary>
|
|
public const int SlowSubscriberProbeCount = 5;
|
|
|
|
/// <summary>Concurrent enqueue tasks in the store-and-forward drain probe.</summary>
|
|
public const int StoreAndForwardConcurrency = 25;
|
|
|
|
/// <summary>Executes a run end to end.</summary>
|
|
/// <param name="config">Scale and duration configuration.</param>
|
|
/// <param name="log">Progress sink (stdout in the console app).</param>
|
|
/// <param name="cancellationToken">Cancels the run.</param>
|
|
/// <returns>The measured result.</returns>
|
|
public static async Task<HarnessRunResult> ExecuteAsync(
|
|
HarnessConfig config, Action<string> log, CancellationToken cancellationToken)
|
|
{
|
|
var startedUtc = DateTimeOffset.UtcNow;
|
|
var totalWatch = Stopwatch.StartNew();
|
|
var dataRoot = config.DataDirectory
|
|
?? Path.Combine(Path.GetTempPath(), $"scadabridge-loadharness-{Guid.NewGuid():N}");
|
|
Directory.CreateDirectory(dataRoot);
|
|
|
|
var sampler = ResourceSampler.Start(config.SampleInterval);
|
|
var sites = new List<SiteRuntimeFixture>(config.Sites);
|
|
var latency = new LatencyHistogram();
|
|
|
|
TagUpdateDriver? driver = null;
|
|
ObservabilityProbes? probes = null;
|
|
StoreAndForwardDrainResult? drainResult = null;
|
|
SlowSubscriberResult? slowResult = null;
|
|
BatchSizeAggregator? batchAggregator = null;
|
|
|
|
var aggregator = new CentralHealthAggregator(
|
|
Options.Create(new HealthMonitoringOptions
|
|
{
|
|
ReportInterval = config.HealthReportInterval,
|
|
OfflineTimeout = config.HealthReportInterval * 2,
|
|
}),
|
|
NullLogger<CentralHealthAggregator>.Instance);
|
|
|
|
try
|
|
{
|
|
log($"Building {config.Sites} sites ({config.InstancesPerSite} instances x " +
|
|
$"{config.TagsPerInstance} tags each = {config.TotalSubscriptions:N0} subscriptions)...");
|
|
|
|
var siteWatch = Stopwatch.StartNew();
|
|
for (var s = 0; s < config.Sites; s++)
|
|
sites.Add(await SiteRuntimeFixture.CreateAsync(s, config, dataRoot, ConnectionsPerSite));
|
|
siteWatch.Stop();
|
|
log($" sites built in {siteWatch.Elapsed.TotalSeconds:F1}s");
|
|
|
|
// Instance ramp — sites in parallel (they are independent actor systems,
|
|
// exactly as 10 real sites would be), each site internally staggered at the
|
|
// production StartupBatchSize/StartupBatchDelayMs pacing.
|
|
var rampWatch = Stopwatch.StartNew();
|
|
await Task.WhenAll(sites.Select(site => site.StartInstancesAsync(cancellationToken)));
|
|
rampWatch.Stop();
|
|
var slowestSiteRamp = sites.Max(s => s.InstanceRampDuration.TotalSeconds);
|
|
log($" {config.Sites * config.InstancesPerSite:N0} instance actors created in " +
|
|
$"{rampWatch.Elapsed.TotalSeconds:F1}s (slowest site {slowestSiteRamp:F1}s)");
|
|
|
|
foreach (var site in sites)
|
|
site.AttachStreamProbes(latency);
|
|
log($" {sites.Sum(s => s.Probes.Count)} live stream subscribers attached");
|
|
|
|
// Wire path (real gRPC): the aggregator must exist BEFORE the first stream
|
|
// opens or the MeterListener misses early frames.
|
|
var wireLatency = new LatencyHistogram();
|
|
if (config.WireProbesPerSite > 0)
|
|
{
|
|
batchAggregator = new BatchSizeAggregator();
|
|
foreach (var site in sites)
|
|
await site.AttachWireStreamProbesAsync(config.WireProbesPerSite, wireLatency, log);
|
|
log($" {sites.Sum(s => s.WireSubscribers.Count)} wire stream subscribers attached " +
|
|
"(real Kestrel + SiteStreamGrpcClient, batching negotiated)");
|
|
}
|
|
|
|
// Let Instance Actors complete their DCL subscribe round-trips before the
|
|
// driver starts; an emit before SubscribeBatchAsync has captured the
|
|
// callback would be silently discarded.
|
|
await Task.Delay(config.SubscribeSettleDuration, cancellationToken);
|
|
|
|
driver = TagUpdateDriver.Start(sites, config);
|
|
log($" tag driver started, nominal {config.NominalUpdatesPerSecond:N0} updates/s");
|
|
|
|
probes = ObservabilityProbes.Start(
|
|
sites, aggregator, config.HealthReportInterval, config.DebugProbeInterval);
|
|
|
|
log($"Settling for {config.SettleDuration.TotalMinutes:F1} min...");
|
|
await Task.Delay(config.SettleDuration, cancellationToken);
|
|
|
|
// Switch to a fresh latency histogram so ramp-window outliers do not
|
|
// contaminate the steady-state percentiles. Everything reported as "steady
|
|
// state" is measured strictly after this point; the subscriptions
|
|
// themselves are left untouched.
|
|
var steadyStateStartSeconds = sampler.Snapshot().LastOrDefault()?.ElapsedSeconds ?? 0;
|
|
var emittedAtWindowStart = driver.EmittedCount;
|
|
var receivedAtWindowStart = sites.Sum(s => s.Probes.Sum(p => p.ReceivedEvents));
|
|
var droppedAtWindowStart = sites.Sum(s => s.Probes.Sum(p => p.DroppedEvents));
|
|
|
|
var steadyLatency = new LatencyHistogram();
|
|
foreach (var site in sites)
|
|
{
|
|
foreach (var probe in site.Probes)
|
|
probe.RetargetLatency(steadyLatency);
|
|
}
|
|
|
|
// Wire path: same ramp/steady histogram separation, plus window-start
|
|
// baselines for the client/server counters and the batch distribution.
|
|
var steadyWireLatency = new LatencyHistogram();
|
|
var wireReceivedAtWindowStart = sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents));
|
|
var wireDroppedAtWindowStart = sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0);
|
|
var batchesAtWindowStart = batchAggregator?.Snapshot();
|
|
foreach (var site in sites)
|
|
{
|
|
foreach (var wire in site.WireSubscribers)
|
|
wire.RetargetLatency(steadyWireLatency);
|
|
}
|
|
|
|
var windowWatch = Stopwatch.StartNew();
|
|
log($"Sustained measurement window: {config.SustainDuration.TotalMinutes:F1} min...");
|
|
await Task.Delay(config.SustainDuration, cancellationToken);
|
|
windowWatch.Stop();
|
|
|
|
var steadyStateEndSeconds = sampler.Snapshot().LastOrDefault()?.ElapsedSeconds ?? 0;
|
|
var emittedInWindow = driver.EmittedCount - emittedAtWindowStart;
|
|
|
|
var steadyResources = sampler.Summarize(steadyStateStartSeconds, steadyStateEndSeconds);
|
|
log($" window complete: {emittedInWindow:N0} updates offered, " +
|
|
$"{emittedInWindow / windowWatch.Elapsed.TotalSeconds:N0}/s achieved");
|
|
|
|
var probeReceived = sites.Sum(s => s.Probes.Sum(p => p.ReceivedEvents)) - receivedAtWindowStart;
|
|
var probeDropped = sites.Sum(s => s.Probes.Sum(p => p.DroppedEvents)) - droppedAtWindowStart;
|
|
|
|
// Wire-path window deltas, captured BEFORE the row-50 scenarios below so
|
|
// their publishing cannot contaminate the wire figures.
|
|
WireStreamResult? wireResult = null;
|
|
if (batchAggregator is not null)
|
|
{
|
|
wireResult = new WireStreamResult(
|
|
ProbeCount: sites.Sum(s => s.WireSubscribers.Count),
|
|
Latency: steadyWireLatency.Snapshot(),
|
|
EventsReceived: sites.Sum(s => s.WireSubscribers.Sum(p => p.ReceivedEvents)) - wireReceivedAtWindowStart,
|
|
ServerDroppedEvents: sites.Sum(s => s.WireHost?.Server.DroppedStreamEventCount ?? 0) - wireDroppedAtWindowStart,
|
|
StreamFaults: sites.Sum(s => s.WireSubscribers.Sum(p => p.Faults)),
|
|
GracefulCompletions: sites.Sum(s => s.WireSubscribers.Sum(p => p.GracefulCompletions)),
|
|
Batches: BatchSizeWindow.From(batchesAtWindowStart!, batchAggregator.Snapshot()));
|
|
log($" wire path: {wireResult.EventsReceived:N0} events in {wireResult.Batches.Frames:N0} frames " +
|
|
$"(mean {wireResult.Batches.MeanEventsPerFrame:F1} ev/frame, max {wireResult.Batches.Max}), " +
|
|
$"P99 {wireResult.Latency.P99Ms:F2}ms");
|
|
}
|
|
|
|
// Register row 50 — measured after the sustained window so the numbers are
|
|
// not competing with the full tag load for CPU, and reported separately for
|
|
// the same reason.
|
|
log("Stopping tag driver for the register row 50 scenarios...");
|
|
var driverLagSeconds = driver.EmitLagSeconds;
|
|
var driverSkipped = driver.SkippedNoCallback;
|
|
await driver.DisposeAsync();
|
|
driver = null;
|
|
await probes.DisposeAsync();
|
|
|
|
log($"Store-and-forward drain: {config.StoreAndForwardDrainMessages:N0} messages...");
|
|
drainResult = await StoreAndForwardDrainScenario.RunAsync(
|
|
sites[0], config.StoreAndForwardDrainMessages, StoreAndForwardConcurrency, cancellationToken);
|
|
log($" buffered {drainResult.EnqueuePerSecond:N0}/s, drained {drainResult.DrainPerSecond:N0}/s");
|
|
|
|
log($"Slow-subscriber isolation: {SlowSubscriberProbeCount} subscribers, " +
|
|
$"{config.SlowSubscriberEvents:N0} events...");
|
|
slowResult = await SlowSubscriberScenario.RunAsync(
|
|
sites[1 % sites.Count], SlowSubscriberProbeCount, config.SlowSubscriberEvents, cancellationToken);
|
|
log($" healthy min delivery {slowResult.HealthyMinDeliveryRatio:P2}, " +
|
|
$"stalled {slowResult.SlowDeliveryRatio:P2}");
|
|
|
|
totalWatch.Stop();
|
|
|
|
return new HarnessRunResult(
|
|
Config: config,
|
|
Environment: CaptureEnvironment(),
|
|
StartedUtc: startedUtc,
|
|
TotalSeconds: totalWatch.Elapsed.TotalSeconds,
|
|
SiteRampSeconds: siteWatch.Elapsed.TotalSeconds,
|
|
InstanceRampSeconds: rampWatch.Elapsed.TotalSeconds,
|
|
SlowestSiteInstanceRampSeconds: slowestSiteRamp,
|
|
TagUpdateLatency: steadyLatency.Snapshot(),
|
|
EmittedTagUpdates: emittedAtWindowStart + emittedInWindow,
|
|
SteadyStateEmittedTagUpdates: emittedInWindow,
|
|
AchievedUpdatesPerSecond: emittedInWindow / windowWatch.Elapsed.TotalSeconds,
|
|
NominalUpdatesPerSecond: config.NominalUpdatesPerSecond,
|
|
DriverLagSeconds: driverLagSeconds,
|
|
DriverSkippedNoCallback: driverSkipped,
|
|
SteadyStateResources: steadyResources,
|
|
WholeRunResources: sampler.Summarize(0, double.MaxValue),
|
|
ResourceSamples: sampler.Snapshot(),
|
|
HealthReportLatency: probes.HealthReportLatency.Snapshot(),
|
|
HealthReportsDelivered: probes.HealthReportsDelivered,
|
|
SitesTrackedByAggregator: aggregator.GetAllSiteStates().Count,
|
|
DebugSnapshotLatency: probes.DebugSnapshotLatency.Snapshot(),
|
|
DebugSnapshotsCompleted: probes.DebugSnapshotsCompleted,
|
|
DebugSnapshotTimeouts: probes.DebugSnapshotTimeouts,
|
|
StreamProbeReceived: probeReceived,
|
|
StreamProbeDropped: probeDropped,
|
|
StoreAndForwardDrain: drainResult,
|
|
SlowSubscriber: slowResult,
|
|
WireStream: wireResult);
|
|
}
|
|
finally
|
|
{
|
|
if (driver != null) await driver.DisposeAsync();
|
|
if (probes != null) await probes.DisposeAsync();
|
|
batchAggregator?.Dispose();
|
|
await sampler.DisposeAsync();
|
|
|
|
log("Tearing down sites...");
|
|
foreach (var site in sites)
|
|
await site.DisposeAsync();
|
|
|
|
try
|
|
{
|
|
if (config.DataDirectory == null && Directory.Exists(dataRoot))
|
|
Directory.Delete(dataRoot, recursive: true);
|
|
}
|
|
catch (IOException)
|
|
{
|
|
// Temp cleanup only.
|
|
}
|
|
}
|
|
}
|
|
|
|
private static EnvironmentInfo CaptureEnvironment() => new(
|
|
System.Environment.MachineName,
|
|
System.Runtime.InteropServices.RuntimeInformation.OSDescription,
|
|
System.Environment.ProcessorCount,
|
|
System.Runtime.InteropServices.RuntimeInformation.FrameworkDescription,
|
|
System.Runtime.GCSettings.IsServerGC);
|
|
}
|