test(loadharness)+docs: wire-path stream probes — R2 batching benchmark-proven at target scale
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.
This commit is contained in:
Joseph Doherty
@@ -68,6 +68,19 @@ public sealed record HarnessConfig
|
||||
/// </summary>
|
||||
public int StreamProbesPerSite { get; init; } = 10;
|
||||
|
||||
/// <summary>
|
||||
/// Instances per site additionally watched over a REAL gRPC hop: a per-site
|
||||
/// Kestrel h2c listener hosts the production <c>SiteStreamGrpcServer</c> and the
|
||||
/// production <c>SiteStreamGrpcClient</c> dials in with batching negotiated, so
|
||||
/// the R2 <c>SiteStreamEventBatcher</c>, protobuf serialization and HTTP/2 flow
|
||||
/// control are inside the measured path (they are NOT for
|
||||
/// <see cref="StreamProbesPerSite"/>, whose final hop is deliberately an
|
||||
/// in-process reader). 0 disables the wire path entirely. Like the in-process
|
||||
/// probes, each wire subscriber's stream graph sees the full site event flow and
|
||||
/// filters it, so this adds to the fan-out multiplier.
|
||||
/// </summary>
|
||||
public int WireProbesPerSite { get; init; } = 10;
|
||||
|
||||
/// <summary>Store-and-forward messages enqueued for the drain-rate measurement (register row 50).</summary>
|
||||
public int StoreAndForwardDrainMessages { get; init; } = 20_000;
|
||||
|
||||
@@ -131,6 +144,7 @@ public sealed record HarnessConfig
|
||||
"health-interval-seconds" => config with { HealthReportInterval = TimeSpan.FromSeconds(double.Parse(value)) },
|
||||
"debug-probe-interval-seconds" => config with { DebugProbeInterval = TimeSpan.FromSeconds(double.Parse(value)) },
|
||||
"stream-probes-per-site" => config with { StreamProbesPerSite = int.Parse(value) },
|
||||
"wire-probes-per-site" => config with { WireProbesPerSite = int.Parse(value) },
|
||||
"sf-drain-messages" => config with { StoreAndForwardDrainMessages = int.Parse(value) },
|
||||
"slow-subscriber-events" => config with { SlowSubscriberEvents = int.Parse(value) },
|
||||
"data-dir" => config with { DataDirectory = value },
|
||||
|
||||
@@ -40,6 +40,10 @@ namespace ZB.MOM.WW.ScadaBridge.LoadHarness;
|
||||
/// <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,
|
||||
@@ -67,7 +71,30 @@ public sealed record HarnessRunResult(
|
||||
long StreamProbeReceived,
|
||||
long StreamProbeDropped,
|
||||
StoreAndForwardDrainResult? StoreAndForwardDrain,
|
||||
SlowSubscriberResult? SlowSubscriber);
|
||||
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>
|
||||
@@ -120,6 +147,7 @@ public static class HarnessRun
|
||||
ObservabilityProbes? probes = null;
|
||||
StoreAndForwardDrainResult? drainResult = null;
|
||||
SlowSubscriberResult? slowResult = null;
|
||||
BatchSizeAggregator? batchAggregator = null;
|
||||
|
||||
var aggregator = new CentralHealthAggregator(
|
||||
Options.Create(new HealthMonitoringOptions
|
||||
@@ -154,6 +182,18 @@ public static class HarnessRun
|
||||
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.
|
||||
@@ -184,6 +224,18 @@ public static class HarnessRun
|
||||
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);
|
||||
@@ -199,6 +251,24 @@ public static class HarnessRun
|
||||
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.
|
||||
@@ -250,12 +320,14 @@ public static class HarnessRun
|
||||
StreamProbeReceived: probeReceived,
|
||||
StreamProbeDropped: probeDropped,
|
||||
StoreAndForwardDrain: drainResult,
|
||||
SlowSubscriber: slowResult);
|
||||
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...");
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
using System.Diagnostics.Metrics;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
|
||||
|
||||
/// <summary>
|
||||
/// Captures the production <c>scadabridge.site.stream.batch_size</c> histogram via a
|
||||
/// <see cref="MeterListener"/>, so a wire-path run can report the actual frame/batch
|
||||
/// distribution the R2 batcher produced — not a client-side reconstruction. The
|
||||
/// instrument is recorded once per emitted gRPC frame (negotiated streams only), with
|
||||
/// the frame's event count as the value, which makes <c>count = frames</c> and
|
||||
/// <c>sum = events</c> exactly.
|
||||
///
|
||||
/// <para>
|
||||
/// The harness hosts every site's <see cref="ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer"/>
|
||||
/// in-process, so this listener sees every wire frame the run emits and nothing else —
|
||||
/// no other component in the harness records this instrument.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class BatchSizeAggregator : IDisposable
|
||||
{
|
||||
/// <summary>
|
||||
/// Sizes above this are clamped into the last bucket. The production cap is
|
||||
/// <c>GrpcStreamBatchMaxEvents</c> = 100, so 512 leaves generous headroom for a
|
||||
/// configured-up run without an unbounded array.
|
||||
/// </summary>
|
||||
public const int MaxTrackedBatchSize = 512;
|
||||
|
||||
private readonly MeterListener _listener;
|
||||
private readonly long[] _framesBySize = new long[MaxTrackedBatchSize + 1];
|
||||
|
||||
/// <summary>Starts listening. Must be created BEFORE the first stream opens.</summary>
|
||||
public BatchSizeAggregator()
|
||||
{
|
||||
_listener = new MeterListener
|
||||
{
|
||||
InstrumentPublished = (instrument, listener) =>
|
||||
{
|
||||
if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName &&
|
||||
instrument.Name == "scadabridge.site.stream.batch_size")
|
||||
{
|
||||
listener.EnableMeasurementEvents(instrument);
|
||||
}
|
||||
},
|
||||
};
|
||||
_listener.SetMeasurementEventCallback<int>((_, value, _, _) =>
|
||||
{
|
||||
var size = Math.Clamp(value, 1, MaxTrackedBatchSize);
|
||||
Interlocked.Increment(ref _framesBySize[size]);
|
||||
});
|
||||
_listener.Start();
|
||||
}
|
||||
|
||||
/// <summary>Copies the cumulative per-size frame counts.</summary>
|
||||
/// <returns>Index = events-per-frame (clamped), value = frames emitted at that size.</returns>
|
||||
public long[] Snapshot()
|
||||
{
|
||||
var copy = new long[_framesBySize.Length];
|
||||
for (var i = 0; i < copy.Length; i++)
|
||||
copy[i] = Interlocked.Read(ref _framesBySize[i]);
|
||||
return copy;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose() => _listener.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Batch-size distribution over one measurement window, computed from two
|
||||
/// <see cref="BatchSizeAggregator.Snapshot"/>s. Percentiles are exact (integer sizes,
|
||||
/// full counts — nothing is bucketed away).
|
||||
/// </summary>
|
||||
/// <param name="Frames">gRPC frames written in the window.</param>
|
||||
/// <param name="Events">Events those frames carried.</param>
|
||||
/// <param name="BatchedFrames">Frames that carried more than one event.</param>
|
||||
/// <param name="MeanEventsPerFrame">Events / Frames.</param>
|
||||
/// <param name="P50">Median events-per-frame.</param>
|
||||
/// <param name="P95">95th percentile events-per-frame.</param>
|
||||
/// <param name="P99">99th percentile events-per-frame.</param>
|
||||
/// <param name="Max">Largest frame observed in the window.</param>
|
||||
public sealed record BatchSizeWindow(
|
||||
long Frames,
|
||||
long Events,
|
||||
long BatchedFrames,
|
||||
double MeanEventsPerFrame,
|
||||
int P50,
|
||||
int P95,
|
||||
int P99,
|
||||
int Max)
|
||||
{
|
||||
/// <summary>Computes the window distribution from cumulative start/end snapshots.</summary>
|
||||
/// <param name="start">Snapshot at window start.</param>
|
||||
/// <param name="end">Snapshot at window end.</param>
|
||||
/// <returns>The distribution of frames emitted between the two snapshots.</returns>
|
||||
public static BatchSizeWindow From(long[] start, long[] end)
|
||||
{
|
||||
long frames = 0, events = 0, batched = 0;
|
||||
var max = 0;
|
||||
var delta = new long[end.Length];
|
||||
for (var size = 1; size < end.Length; size++)
|
||||
{
|
||||
var n = end[size] - start[size];
|
||||
if (n <= 0) continue;
|
||||
delta[size] = n;
|
||||
frames += n;
|
||||
events += n * size;
|
||||
if (size > 1) batched += n;
|
||||
max = size;
|
||||
}
|
||||
|
||||
return new BatchSizeWindow(
|
||||
frames, events, batched,
|
||||
frames == 0 ? 0 : events / (double)frames,
|
||||
Percentile(delta, frames, 50),
|
||||
Percentile(delta, frames, 95),
|
||||
Percentile(delta, frames, 99),
|
||||
max);
|
||||
}
|
||||
|
||||
private static int Percentile(long[] delta, long frames, double percentile)
|
||||
{
|
||||
if (frames == 0) return 0;
|
||||
var target = (long)Math.Ceiling(frames * percentile / 100.0);
|
||||
if (target < 1) target = 1;
|
||||
|
||||
long cumulative = 0;
|
||||
for (var size = 1; size < delta.Length; size++)
|
||||
{
|
||||
cumulative += delta[size];
|
||||
if (cumulative >= target)
|
||||
return size;
|
||||
}
|
||||
|
||||
return delta.Length - 1;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.AspNetCore.Hosting;
|
||||
using Microsoft.AspNetCore.Hosting.Server;
|
||||
using Microsoft.AspNetCore.Hosting.Server.Features;
|
||||
using Microsoft.AspNetCore.Server.Kestrel.Core;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;
|
||||
|
||||
/// <summary>
|
||||
/// The site half of the wire-path stream probe: a real Kestrel HTTP/2 (h2c) listener on
|
||||
/// a loopback ephemeral port, serving the PRODUCTION <see cref="SiteStreamGrpcServer"/>
|
||||
/// over the site's real <see cref="ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming.SiteStreamManager"/>.
|
||||
///
|
||||
/// <para>
|
||||
/// This is the hop <see cref="StreamSubscriberProbe"/> deliberately substitutes with a
|
||||
/// controllable in-process reader (which the slow-subscriber scenario needs), and
|
||||
/// therefore the hop the original WP-4 runs never exercised: the
|
||||
/// <c>SiteStreamEventBatcher</c> pump (R2), protobuf serialization,
|
||||
/// <c>IServerStreamWriter.WriteAsync</c>, and HTTP/2 flow control. The server is built
|
||||
/// through its production DI constructor with default <see cref="CommunicationOptions"/>,
|
||||
/// so the batch negotiation (100 events / 25 ms window), channel capacities and the
|
||||
/// concurrency cap are exactly what ships.
|
||||
/// </para>
|
||||
///
|
||||
/// <para>
|
||||
/// Deliberately NOT in the path: <c>ControlPlaneAuthInterceptor</c> (it lives in the
|
||||
/// Host project and authenticates once per CALL — one header compare per stream
|
||||
/// lifetime, nothing on the per-event path) and TLS (production is h2c too).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class SiteWireStreamHost : IAsyncDisposable
|
||||
{
|
||||
private readonly WebApplication _app;
|
||||
|
||||
/// <summary>The hosted production gRPC server (drop counters, active stream count).</summary>
|
||||
public SiteStreamGrpcServer Server { get; }
|
||||
|
||||
/// <summary>The dialable endpoint, e.g. <c>http://127.0.0.1:54321</c>.</summary>
|
||||
public string Endpoint { get; }
|
||||
|
||||
private SiteWireStreamHost(WebApplication app, SiteStreamGrpcServer server, string endpoint)
|
||||
{
|
||||
_app = app;
|
||||
Server = server;
|
||||
Endpoint = endpoint;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Builds, starts and readies the listener for one site.
|
||||
/// </summary>
|
||||
/// <param name="site">The site whose stream manager and actor system back the server.</param>
|
||||
/// <returns>The started host.</returns>
|
||||
public static async Task<SiteWireStreamHost> StartAsync(SiteRuntimeFixture site)
|
||||
{
|
||||
var server = new SiteStreamGrpcServer(
|
||||
site.StreamManager,
|
||||
Microsoft.Extensions.Logging.Abstractions.NullLogger<SiteStreamGrpcServer>.Instance,
|
||||
Options.Create(new CommunicationOptions()));
|
||||
|
||||
var builder = WebApplication.CreateSlimBuilder();
|
||||
builder.Logging.ClearProviders();
|
||||
builder.WebHost.ConfigureKestrel(options =>
|
||||
options.Listen(System.Net.IPAddress.Loopback, 0, listenOptions =>
|
||||
listenOptions.Protocols = HttpProtocols.Http2));
|
||||
builder.Services.AddGrpc();
|
||||
builder.Services.AddSingleton(server);
|
||||
|
||||
var app = builder.Build();
|
||||
app.MapGrpcService<SiteStreamGrpcServer>();
|
||||
await app.StartAsync();
|
||||
|
||||
// Same post-start readiness handoff the production host performs.
|
||||
server.SetReady(site.System);
|
||||
|
||||
var address = app.Services.GetRequiredService<IServer>()
|
||||
.Features.Get<IServerAddressesFeature>()?.Addresses.FirstOrDefault()
|
||||
?? throw new InvalidOperationException("Kestrel bound no address.");
|
||||
|
||||
return new SiteWireStreamHost(app, server, address);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
// Mirrors the production shutdown ordering: cancel active streams first so
|
||||
// their pumps observe cancellation, then stop the listener.
|
||||
Server.CancelAllStreams();
|
||||
await _app.StopAsync();
|
||||
await _app.DisposeAsync();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,108 @@
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes;
|
||||
|
||||
/// <summary>
|
||||
/// The central half of the wire-path stream probe: one instance subscription opened
|
||||
/// through the PRODUCTION <see cref="SiteStreamGrpcClient"/> — the exact
|
||||
/// <c>SubscribeAsync</c> path central's <c>DebugStreamBridgeActor</c> uses, including
|
||||
/// the R2 batch negotiation (<c>BatchingSupported = true</c>) and the client-side
|
||||
/// batch unpack + proto→domain conversion.
|
||||
///
|
||||
/// <para>
|
||||
/// Latency is recorded in <c>onEvent</c>, i.e. AFTER wire deserialization, batch
|
||||
/// unpacking and domain mapping — the same "DCL boundary → subscriber" measurement as
|
||||
/// <see cref="StreamSubscriberProbe"/>, extended across the real socket. The emit
|
||||
/// instant travels verbatim on the proto <c>Timestamp</c>, so the subtraction stays a
|
||||
/// genuine end-to-end figure (same process, same clock).
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class WireStreamSubscriber
|
||||
{
|
||||
private Task _streamTask = Task.CompletedTask;
|
||||
private volatile LatencyHistogram? _latency;
|
||||
private long _received;
|
||||
private long _faults;
|
||||
private long _completions;
|
||||
|
||||
/// <summary>Probe name (also the stream's correlation id).</summary>
|
||||
public string Name { get; }
|
||||
|
||||
/// <summary>Domain events delivered by the client callback.</summary>
|
||||
public long ReceivedEvents => Interlocked.Read(ref _received);
|
||||
|
||||
/// <summary>Stream faults reported via <c>onError</c>. Expected 0 for a healthy run.</summary>
|
||||
public long Faults => Interlocked.Read(ref _faults);
|
||||
|
||||
/// <summary>Graceful server-side completions via <c>onCompleted</c>. Expected 0 inside a run shorter than the 4 h stream lifetime.</summary>
|
||||
public long GracefulCompletions => Interlocked.Read(ref _completions);
|
||||
|
||||
private WireStreamSubscriber(string name, LatencyHistogram? latency)
|
||||
{
|
||||
Name = name;
|
||||
_latency = latency;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Opens one instance stream on the shared per-site client. The subscription task
|
||||
/// runs until <paramref name="ct"/> cancels — the harness deliberately does NOT
|
||||
/// reconnect on fault (a fault inside a 25-minute loopback run is a finding, not
|
||||
/// an operational event to paper over), it counts it.
|
||||
/// </summary>
|
||||
/// <param name="client">The site's shared production gRPC client (one channel per site, streams multiplexed).</param>
|
||||
/// <param name="instanceUniqueName">Instance whose events this probe receives.</param>
|
||||
/// <param name="name">Probe/correlation name (must be a valid Akka path element).</param>
|
||||
/// <param name="latency">Histogram fed with end-to-end event latency.</param>
|
||||
/// <param name="log">Progress sink for fault reporting.</param>
|
||||
/// <param name="ct">Stops the subscription.</param>
|
||||
/// <returns>The attached subscriber.</returns>
|
||||
public static WireStreamSubscriber Attach(
|
||||
SiteStreamGrpcClient client,
|
||||
string instanceUniqueName,
|
||||
string name,
|
||||
LatencyHistogram? latency,
|
||||
Action<string> log,
|
||||
CancellationToken ct)
|
||||
{
|
||||
var self = new WireStreamSubscriber(name, latency);
|
||||
|
||||
self._streamTask = client.SubscribeAsync(
|
||||
correlationId: name,
|
||||
instanceUniqueName: instanceUniqueName,
|
||||
onEvent: evt =>
|
||||
{
|
||||
Interlocked.Increment(ref self._received);
|
||||
if (self._latency is { } histogram && evt is AttributeValueChanged changed)
|
||||
histogram.Record(DateTimeOffset.UtcNow - changed.Timestamp);
|
||||
},
|
||||
onError: ex =>
|
||||
{
|
||||
Interlocked.Increment(ref self._faults);
|
||||
log($" wire probe {name} stream fault: {ex.GetType().Name}: {ex.Message}");
|
||||
},
|
||||
onCompleted: () => Interlocked.Increment(ref self._completions),
|
||||
ct);
|
||||
|
||||
return self;
|
||||
}
|
||||
|
||||
/// <summary>Repoints the latency histogram (steady-state window separation) — mirrors <see cref="StreamSubscriberProbe.RetargetLatency"/>.</summary>
|
||||
/// <param name="latency">The histogram to record into from now on, or null to stop recording.</param>
|
||||
public void RetargetLatency(LatencyHistogram? latency) => _latency = latency;
|
||||
|
||||
/// <summary>Awaits the subscription task after its token has been cancelled.</summary>
|
||||
/// <returns>A task that completes when the stream loop has exited.</returns>
|
||||
public async Task WaitForShutdownAsync()
|
||||
{
|
||||
try
|
||||
{
|
||||
await _streamTask;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Normal teardown.
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,7 @@ Console.WriteLine($" sites {config.Sites}");
|
||||
Console.WriteLine($" instances/site {config.InstancesPerSite}");
|
||||
Console.WriteLine($" tags/instance {config.TagsPerInstance}");
|
||||
Console.WriteLine($" total subscriptions {config.TotalSubscriptions:N0}");
|
||||
Console.WriteLine($" stream probes/site {config.StreamProbesPerSite} in-process + {config.WireProbesPerSite} wire (real gRPC)");
|
||||
Console.WriteLine($" nominal update rate {config.NominalUpdatesPerSecond:N0}/s");
|
||||
Console.WriteLine($" settle / sustain {config.SettleDuration.TotalMinutes:F1} / {config.SustainDuration.TotalMinutes:F1} min");
|
||||
Console.WriteLine();
|
||||
|
||||
@@ -47,6 +47,18 @@ public static class ResultsFormatter
|
||||
builder.AppendLine(culture, $"events dropped {result.StreamProbeDropped:N0}");
|
||||
builder.AppendLine();
|
||||
|
||||
if (result.WireStream is { } wire)
|
||||
{
|
||||
builder.AppendLine("-- wire stream subscribers (real gRPC, R2 batching in path) --");
|
||||
builder.AppendLine(culture, $"probes {wire.ProbeCount} (Kestrel h2c + SiteStreamGrpcClient, batching negotiated)");
|
||||
AppendLatency(builder, culture, wire.Latency);
|
||||
builder.AppendLine(culture, $"events delivered {wire.EventsReceived:N0} in {wire.Batches.Frames:N0} frames");
|
||||
builder.AppendLine(culture, $"batching mean {wire.Batches.MeanEventsPerFrame:F2} ev/frame p50 {wire.Batches.P50} p95 {wire.Batches.P95} p99 {wire.Batches.P99} max {wire.Batches.Max} ({wire.Batches.BatchedFrames:N0} frames >1 event)");
|
||||
builder.AppendLine(culture, $"server channel drops {wire.ServerDroppedEvents:N0}");
|
||||
builder.AppendLine(culture, $"faults / completions {wire.StreamFaults:N0} / {wire.GracefulCompletions:N0}");
|
||||
builder.AppendLine();
|
||||
}
|
||||
|
||||
builder.AppendLine("-- health report delivery --");
|
||||
AppendLatency(builder, culture, result.HealthReportLatency);
|
||||
builder.AppendLine(culture, $"reports ingested {result.HealthReportsDelivered:N0}");
|
||||
|
||||
@@ -6,6 +6,7 @@ using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using ZB.MOM.WW.LocalDb;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Messages.DataConnection;
|
||||
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Flattening;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer;
|
||||
using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors;
|
||||
@@ -43,7 +44,11 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
private readonly ILocalDb _localDb;
|
||||
private readonly List<IActorRef> _instanceActors = new();
|
||||
private readonly List<StreamSubscriberProbe> _probes = new();
|
||||
private readonly List<WireStreamSubscriber> _wireSubscribers = new();
|
||||
private readonly string _dataDirectory;
|
||||
private SiteWireStreamHost? _wireHost;
|
||||
private SiteStreamGrpcClient? _wireClient;
|
||||
private CancellationTokenSource? _wireCts;
|
||||
|
||||
/// <summary>Site identifier, e.g. <c>site-01</c>.</summary>
|
||||
public string SiteId { get; }
|
||||
@@ -72,6 +77,12 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
/// <summary>The live stream subscribers attached to this site.</summary>
|
||||
public IReadOnlyList<StreamSubscriberProbe> Probes => _probes;
|
||||
|
||||
/// <summary>The wire-path (real gRPC) subscribers attached to this site, if any.</summary>
|
||||
public IReadOnlyList<WireStreamSubscriber> WireSubscribers => _wireSubscribers;
|
||||
|
||||
/// <summary>The site's real gRPC listener, when wire probes are attached.</summary>
|
||||
public SiteWireStreamHost? WireHost => _wireHost;
|
||||
|
||||
/// <summary>Instance Actors created on this site.</summary>
|
||||
public IReadOnlyList<IActorRef> InstanceActors => _instanceActors;
|
||||
|
||||
@@ -344,12 +355,63 @@ public sealed class SiteRuntimeFixture : IAsyncDisposable
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the site's real Kestrel/gRPC listener and attaches
|
||||
/// <paramref name="count"/> subscribers through the production
|
||||
/// <see cref="SiteStreamGrpcClient"/> — one channel per site with the streams
|
||||
/// multiplexed on it, exactly the production shape (central keys one client per
|
||||
/// <c>(siteId, endpoint)</c>). Instance selection uses the same stride as
|
||||
/// <see cref="AttachStreamProbes"/>, so the wire probes watch the SAME instances
|
||||
/// as the in-process probes and the two latency distributions are directly
|
||||
/// comparable within one run.
|
||||
/// </summary>
|
||||
/// <param name="count">Wire subscribers to attach.</param>
|
||||
/// <param name="latency">Histogram that receives end-to-end (over-the-wire) latencies.</param>
|
||||
/// <param name="log">Progress sink for fault reporting.</param>
|
||||
/// <returns>
|
||||
/// A task that completes when the listener is up and every stream has been
|
||||
/// initiated. Stream ESTABLISHMENT finishes asynchronously — comfortably inside
|
||||
/// the subscribe-settle + settle windows that separate attach from measurement.
|
||||
/// </returns>
|
||||
public async Task AttachWireStreamProbesAsync(int count, LatencyHistogram latency, Action<string> log)
|
||||
{
|
||||
_wireHost = await SiteWireStreamHost.StartAsync(this);
|
||||
_wireClient = new SiteStreamGrpcClient(_wireHost.Endpoint, NullLogger.Instance);
|
||||
_wireCts = new CancellationTokenSource();
|
||||
|
||||
var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, count));
|
||||
for (var p = 0; p < count; p++)
|
||||
{
|
||||
var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1);
|
||||
var subscriber = WireStreamSubscriber.Attach(
|
||||
_wireClient, InstanceName(instanceIndex), $"wire-{SiteId}-p{p:D2}",
|
||||
latency, log, _wireCts.Token);
|
||||
_wireSubscribers.Add(subscriber);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Total tag paths registered across this site's connections.</summary>
|
||||
public int TotalTagPaths => TagPathsByConnection.Sum(list => list.Count);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
// Wire path first: cancel the client streams, wait for their loops to exit,
|
||||
// then drop the channel and the listener — the same order production teardown
|
||||
// takes (client goes away, server cancels, actor system last).
|
||||
if (_wireCts is not null)
|
||||
{
|
||||
await _wireCts.CancelAsync();
|
||||
foreach (var subscriber in _wireSubscribers)
|
||||
await subscriber.WaitForShutdownAsync();
|
||||
_wireCts.Dispose();
|
||||
}
|
||||
|
||||
if (_wireClient is not null)
|
||||
await _wireClient.DisposeAsync();
|
||||
if (_wireHost is not null)
|
||||
await _wireHost.DisposeAsync();
|
||||
|
||||
foreach (var probe in _probes)
|
||||
await probe.DisposeAsync();
|
||||
|
||||
|
||||
@@ -27,6 +27,14 @@
|
||||
<ConcurrentGarbageCollection>true</ConcurrentGarbageCollection>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- Wire-path probes: each site hosts the production SiteStreamGrpcServer on a
|
||||
real Kestrel h2c listener so the R2 batcher + serialization + HTTP/2 flow
|
||||
control sit inside the measured path (see Probes/SiteWireStreamHost.cs). -->
|
||||
<FrameworkReference Include="Microsoft.AspNetCore.App" />
|
||||
<PackageReference Include="Grpc.AspNetCore" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.Commons/ZB.MOM.WW.ScadaBridge.Commons.csproj" />
|
||||
<ProjectReference Include="../../src/ZB.MOM.WW.ScadaBridge.SiteRuntime/ZB.MOM.WW.ScadaBridge.SiteRuntime.csproj" />
|
||||
|
||||
+13
@@ -49,6 +49,7 @@ public class TargetScaleHarnessSmokeTests
|
||||
DebugProbeInterval = TimeSpan.FromSeconds(2),
|
||||
SubscribeSettleDuration = TimeSpan.FromSeconds(5),
|
||||
StreamProbesPerSite = 3,
|
||||
WireProbesPerSite = 2,
|
||||
StoreAndForwardDrainMessages = 200,
|
||||
SlowSubscriberEvents = 2_000,
|
||||
ResultsPath = Path.Combine(Path.GetTempPath(), $"loadharness-smoke-{Guid.NewGuid():N}.json"),
|
||||
@@ -64,6 +65,18 @@ public class TargetScaleHarnessSmokeTests
|
||||
Assert.True(result.StreamProbeReceived > 0, "Live stream subscribers received nothing.");
|
||||
Assert.Equal(0, result.DriverSkippedNoCallback);
|
||||
|
||||
// The wire path (real Kestrel + real SiteStreamGrpcClient, R2 batching
|
||||
// negotiated) carried traffic end to end: events crossed a real HTTP/2
|
||||
// socket, the batch-size histogram observed real frames, and nothing faulted.
|
||||
Assert.NotNull(result.WireStream);
|
||||
Assert.True(result.WireStream!.EventsReceived > 0,
|
||||
"Wire subscribers received nothing — the Kestrel/gRPC hop did not carry traffic.");
|
||||
Assert.True(result.WireStream.Latency.Count > 0,
|
||||
"No wire latency samples — the over-the-wire measurement path is broken.");
|
||||
Assert.True(result.WireStream.Batches.Frames > 0,
|
||||
"The batch-size MeterListener observed no frames — the R2 histogram path is broken.");
|
||||
Assert.Equal(0, result.WireStream.StreamFaults);
|
||||
|
||||
// Observability held up.
|
||||
Assert.Equal(config.Sites, result.SitesTrackedByAggregator);
|
||||
Assert.True(result.DebugSnapshotsCompleted > 0, "No debug snapshot completed.");
|
||||
|
||||
Reference in New Issue
Block a user