diff --git a/ZB.MOM.WW.ScadaBridge.slnx b/ZB.MOM.WW.ScadaBridge.slnx index 21748f21..192bc1a4 100644 --- a/ZB.MOM.WW.ScadaBridge.slnx +++ b/ZB.MOM.WW.ScadaBridge.slnx @@ -53,6 +53,7 @@ + diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs new file mode 100644 index 00000000..12c2b27b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessConfig.cs @@ -0,0 +1,144 @@ +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Scale and duration knobs for a load-harness run. Defaults are the Phase-8 WP-4 +/// target scale (10 sites x 500 instances x 75 tags = 375,000 subscriptions). +/// Every value is overridable from the command line so the same binary serves both +/// the full-scale protocol run and the scaled-down CI smoke. +/// +public sealed record HarnessConfig +{ + /// Number of simulated sites (WP-4 acceptance criterion [2.5-1]). + public int Sites { get; init; } = 10; + + /// Instance Actors per site (WP-4 [2.5-2]). + public int InstancesPerSite { get; init; } = 500; + + /// Data-sourced attributes ("live tags") per instance (WP-4 [2.5-3]). + public int TagsPerInstance { get; init; } = 75; + + /// + /// Nominal per-tag update period. The driver emits each tag once per period, so + /// the site-wide event rate is InstancesPerSite * TagsPerInstance / period. + /// The 10 s default puts the target-scale fleet at 37,500 tag updates/second. + /// + public TimeSpan TagUpdatePeriod { get; init; } = TimeSpan.FromSeconds(10); + + /// + /// Length of the steady-state measurement window, measured AFTER the ramp + /// completes and after the post-ramp settle. Memory-growth and CPU figures are + /// computed over exactly this window. + /// + public TimeSpan SustainDuration { get; init; } = TimeSpan.FromMinutes(20); + + /// + /// Quiet period between the last instance starting and the start of the + /// measurement window — lets startup allocations settle so the memory-growth + /// slope reflects steady state, not the ramp. + /// + public TimeSpan SettleDuration { get; init; } = TimeSpan.FromMinutes(2); + + /// Resource-sampling cadence (working set, GC heap, CPU, thread count). + public TimeSpan SampleInterval { get; init; } = TimeSpan.FromSeconds(10); + + /// + /// Health report cadence. Defaults to the production + /// HealthMonitoringOptions.ReportInterval (30 s) so the timing measured is + /// the one that ships; only the CI smoke shortens it, because a 20-second smoke + /// window would otherwise never see a single tick. + /// + public TimeSpan HealthReportInterval { get; init; } = TimeSpan.FromSeconds(30); + + /// How often to take a debug view snapshot of a random live instance. + public TimeSpan DebugProbeInterval { get; init; } = TimeSpan.FromSeconds(5); + + /// + /// Pause between the last Instance Actor starting and the tag driver starting, to + /// let every instance complete its DCL subscribe round-trip. The adapter callback + /// is captured at subscribe time, so an emit before that lands is silently + /// discarded — this window is what keeps SkippedNoCallback at zero. + /// + public TimeSpan SubscribeSettleDuration { get; init; } = TimeSpan.FromSeconds(30); + + /// + /// Instances per site that carry a live stream subscriber (a real + /// StreamRelayActor + bounded DropOldest channel, i.e. the production + /// Debug View / central shape). Every subscriber's stream graph sees the FULL + /// site event flow and filters it, so this is the fan-out multiplier. + /// + public int StreamProbesPerSite { get; init; } = 10; + + /// Store-and-forward messages enqueued for the drain-rate measurement (register row 50). + public int StoreAndForwardDrainMessages { get; init; } = 20_000; + + /// Events published at the slow-subscriber isolation probe (register row 50). + public int SlowSubscriberEvents { get; init; } = 200_000; + + /// Directory for the site SQLite files. A temp directory is used when null. + public string? DataDirectory { get; init; } + + /// Path the JSON metrics document is written to. + public string ResultsPath { get; init; } = "loadharness-results.json"; + + /// Total live tag subscriptions across the fleet. + public int TotalSubscriptions => Sites * InstancesPerSite * TagsPerInstance; + + /// Nominal fleet-wide tag updates per second implied by the scale and update period. + public double NominalUpdatesPerSecond => TotalSubscriptions / TagUpdatePeriod.TotalSeconds; + + /// + /// Parses --key value / --key=value arguments over the defaults. + /// Unknown keys throw so a typo in a 20-minute run's command line fails fast + /// rather than silently measuring the wrong scale. + /// + /// Raw command-line arguments. + /// The parsed configuration. + public static HarnessConfig Parse(string[] args) + { + var config = new HarnessConfig(); + for (var i = 0; i < args.Length; i++) + { + var arg = args[i]; + if (!arg.StartsWith("--", StringComparison.Ordinal)) + throw new ArgumentException($"Unexpected argument '{arg}' (expected --key value)."); + + string key; + string value; + var eq = arg.IndexOf('=', StringComparison.Ordinal); + if (eq >= 0) + { + key = arg[2..eq]; + value = arg[(eq + 1)..]; + } + else + { + key = arg[2..]; + if (i + 1 >= args.Length) + throw new ArgumentException($"Option '--{key}' requires a value."); + value = args[++i]; + } + + config = key switch + { + "sites" => config with { Sites = int.Parse(value) }, + "instances-per-site" => config with { InstancesPerSite = int.Parse(value) }, + "tags-per-instance" => config with { TagsPerInstance = int.Parse(value) }, + "tag-update-period-seconds" => config with { TagUpdatePeriod = TimeSpan.FromSeconds(double.Parse(value)) }, + "sustain-minutes" => config with { SustainDuration = TimeSpan.FromMinutes(double.Parse(value)) }, + "settle-minutes" => config with { SettleDuration = TimeSpan.FromMinutes(double.Parse(value)) }, + "subscribe-settle-seconds" => config with { SubscribeSettleDuration = TimeSpan.FromSeconds(double.Parse(value)) }, + "sample-seconds" => config with { SampleInterval = TimeSpan.FromSeconds(double.Parse(value)) }, + "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) }, + "sf-drain-messages" => config with { StoreAndForwardDrainMessages = int.Parse(value) }, + "slow-subscriber-events" => config with { SlowSubscriberEvents = int.Parse(value) }, + "data-dir" => config with { DataDirectory = value }, + "results" => config with { ResultsPath = value }, + _ => throw new ArgumentException($"Unknown option '--{key}'."), + }; + } + + return config; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs new file mode 100644 index 00000000..26a3d521 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/HarnessRun.cs @@ -0,0 +1,275 @@ +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; + +/// Everything one harness run measured, ready for serialization. +/// The configuration the run executed under. +/// Host/runtime description. +/// Run start. +/// Total wall time including ramp and teardown. +/// Wall time to build all site fixtures (before instances). +/// Wall time to create every Instance Actor across all sites. +/// Slowest single site's instance ramp — the "deploy 500 instances to a site" figure. +/// End-to-end DCL-boundary to stream-subscriber latency. +/// Tag updates offered during the whole run. +/// Tag updates offered during the measurement window only. +/// Offered load actually achieved in the measurement window. +/// Offered load the configuration called for. +/// Cumulative driver slice overrun (harness-bound load shortfall). +/// Emissions skipped before subscriptions existed. +/// Resource behaviour over the measurement window. +/// Resource behaviour over the whole run. +/// Health report collect+ingest latency. +/// Health reports ingested by the central aggregator. +/// Sites the central aggregator ended up tracking. +/// Debug view snapshot round-trip latency under load. +/// Debug snapshots that completed. +/// Debug snapshots that timed out. +/// Events delivered to live stream subscribers. +/// Events evicted by live subscribers' bounded channels. +/// Store-and-forward drain measurement (register row 50). +/// Slow-subscriber isolation measurement (register row 50). +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, + LatencySnapshot HealthReportLatency, + long HealthReportsDelivered, + int SitesTrackedByAggregator, + LatencySnapshot DebugSnapshotLatency, + long DebugSnapshotsCompleted, + long DebugSnapshotTimeouts, + long StreamProbeReceived, + long StreamProbeDropped, + StoreAndForwardDrainResult? StoreAndForwardDrain, + SlowSubscriberResult? SlowSubscriber); + +/// Host and runtime facts recorded alongside the numbers. +/// Host name. +/// Operating system description. +/// Logical processors visible to the process. +/// .NET runtime version. +/// Whether server GC is active. +public sealed record EnvironmentInfo( + string MachineName, + string OsDescription, + int ProcessorCount, + string RuntimeVersion, + bool ServerGc); + +/// +/// 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. +/// +public static class HarnessRun +{ + /// Data connections each site spreads its tags across. + public const int ConnectionsPerSite = 5; + + /// Subscribers attached in the slow-subscriber isolation probe. + public const int SlowSubscriberProbeCount = 5; + + /// Concurrent enqueue tasks in the store-and-forward drain probe. + public const int StoreAndForwardConcurrency = 25; + + /// Executes a run end to end. + /// Scale and duration configuration. + /// Progress sink (stdout in the console app). + /// Cancels the run. + /// The measured result. + public static async Task ExecuteAsync( + HarnessConfig config, Action 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(config.Sites); + var latency = new LatencyHistogram(); + + TagUpdateDriver? driver = null; + ObservabilityProbes? probes = null; + StoreAndForwardDrainResult? drainResult = null; + SlowSubscriberResult? slowResult = null; + + var aggregator = new CentralHealthAggregator( + Options.Create(new HealthMonitoringOptions + { + ReportInterval = config.HealthReportInterval, + OfflineTimeout = config.HealthReportInterval * 2, + }), + NullLogger.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"); + + // 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); + } + + 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; + + // 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), + 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); + } + finally + { + if (driver != null) await driver.DisposeAsync(); + if (probes != null) await probes.DisposeAsync(); + 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); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs new file mode 100644 index 00000000..2c19d71b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/LatencyHistogram.cs @@ -0,0 +1,136 @@ +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +/// +/// Lock-free logarithmic latency histogram sized for tens of thousands of samples +/// per second across many threads. +/// +/// +/// Buckets are 16-per-octave over microseconds, i.e. bucket i covers +/// [2^(i/16), 2^((i+1)/16)) µs. That bounds relative bucket width at +/// 2^(1/16) - 1 ≈ 4.4%, so a reported percentile is within ~4.4% of the true +/// value — ample for the millisecond-scale thresholds this harness asserts, and far +/// cheaper than retaining 45 million raw samples. +/// +/// +/// Recording is a plus one +/// ; there is no allocation on the hot path. +/// +/// +public sealed class LatencyHistogram +{ + private const int SubBucketsPerOctave = 16; + private const int BucketCount = 64 * SubBucketsPerOctave; + + private readonly long[] _buckets = new long[BucketCount]; + private long _count; + private long _totalMicroseconds; + private long _maxMicroseconds; + + /// Number of samples recorded. + public long Count => Interlocked.Read(ref _count); + + /// Largest sample seen, in microseconds (exact — not bucketed). + public double MaxMs => Interlocked.Read(ref _maxMicroseconds) / 1000.0; + + /// Arithmetic mean in milliseconds (exact — accumulated, not bucketed). + public double MeanMs + { + get + { + var count = Interlocked.Read(ref _count); + return count == 0 ? 0 : Interlocked.Read(ref _totalMicroseconds) / 1000.0 / count; + } + } + + /// + /// Records one sample. Negative durations (clock skew across the emit/receive + /// boundary) are clamped to zero rather than discarded, so the sample count stays + /// an honest denominator. + /// + /// The measured latency. + public void Record(TimeSpan elapsed) + { + var micros = (long)(elapsed.TotalMilliseconds * 1000.0); + if (micros < 0) micros = 0; + + Interlocked.Increment(ref _count); + Interlocked.Add(ref _totalMicroseconds, micros); + + long observedMax; + while (micros > (observedMax = Interlocked.Read(ref _maxMicroseconds))) + { + if (Interlocked.CompareExchange(ref _maxMicroseconds, micros, observedMax) == observedMax) + break; + } + + Interlocked.Increment(ref _buckets[BucketIndex(micros)]); + } + + private static int BucketIndex(long micros) + { + if (micros <= 0) return 0; + var index = (int)(Math.Log2(micros) * SubBucketsPerOctave); + if (index < 0) return 0; + return index >= BucketCount ? BucketCount - 1 : index; + } + + /// Bucket midpoint in milliseconds, used when reconstructing a percentile. + private static double BucketMidpointMs(int index) + { + var low = Math.Pow(2, (double)index / SubBucketsPerOctave); + var high = Math.Pow(2, (double)(index + 1) / SubBucketsPerOctave); + return (low + high) / 2.0 / 1000.0; + } + + /// + /// Returns the requested percentile in milliseconds, or 0 when no samples were recorded. + /// + /// Percentile in the range 0..100 (e.g. 99 for P99). + /// The percentile value in milliseconds. + public double PercentileMs(double percentile) + { + var total = Interlocked.Read(ref _count); + if (total == 0) return 0; + + var target = (long)Math.Ceiling(total * percentile / 100.0); + if (target < 1) target = 1; + + long cumulative = 0; + for (var i = 0; i < BucketCount; i++) + { + cumulative += Interlocked.Read(ref _buckets[i]); + if (cumulative >= target) + return BucketMidpointMs(i); + } + + return MaxMs; + } + + /// Materializes the standard percentile set plus mean/max/count for reporting. + /// A snapshot record of this histogram. + public LatencySnapshot Snapshot() => new( + Count, + MeanMs, + PercentileMs(50), + PercentileMs(95), + PercentileMs(99), + PercentileMs(99.9), + MaxMs); +} + +/// Point-in-time summary of a . All times in milliseconds. +/// Samples recorded. +/// Arithmetic mean. +/// Median. +/// 95th percentile. +/// 99th percentile. +/// 99.9th percentile. +/// Largest observed sample. +public sealed record LatencySnapshot( + long Count, + double MeanMs, + double P50Ms, + double P95Ms, + double P99Ms, + double P999Ms, + double MaxMs); diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs new file mode 100644 index 00000000..a6fdd7fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Metrics/ResourceSampler.cs @@ -0,0 +1,202 @@ +using System.Diagnostics; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +/// +/// One periodic observation of process resource usage. +/// +/// Seconds since sampling started. +/// Process working set (RSS). +/// without forcing a collection. +/// Mean CPU utilization since the previous sample, as a percentage of ONE core. +/// OS threads in the process. +/// Cumulative gen-2 collections. +public sealed record ResourceSample( + double ElapsedSeconds, + double WorkingSetMb, + double ManagedHeapMb, + double CpuPercent, + int ThreadCount, + int Gen2Collections); + +/// +/// Samples working set, managed heap, CPU and thread count on a fixed cadence for the +/// life of a run. CPU is differential (processor time delta / wall delta) so a sample +/// reflects the interval it covers rather than the whole process lifetime. +/// +public sealed class ResourceSampler : IAsyncDisposable +{ + private readonly List _samples = new(); + private readonly object _lock = new(); + private readonly CancellationTokenSource _cts = new(); + private readonly Task _loop; + private readonly Stopwatch _wall = Stopwatch.StartNew(); + + private TimeSpan _lastCpu; + private double _lastElapsedSeconds; + + private ResourceSampler(TimeSpan interval) + { + _lastCpu = Process.GetCurrentProcess().TotalProcessorTime; + _loop = Task.Run(() => SampleLoopAsync(interval, _cts.Token)); + } + + /// Starts sampling at the given cadence. + /// Sampling interval. + /// The running sampler. + public static ResourceSampler Start(TimeSpan interval) => new(interval); + + /// All samples collected so far, oldest first. + /// A snapshot copy of the sample list. + public IReadOnlyList Snapshot() + { + lock (_lock) return _samples.ToList(); + } + + private async Task SampleLoopAsync(TimeSpan interval, CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + Capture(); + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private void Capture() + { + using var process = Process.GetCurrentProcess(); + process.Refresh(); + + var elapsedSeconds = _wall.Elapsed.TotalSeconds; + var cpu = process.TotalProcessorTime; + var wallDelta = elapsedSeconds - _lastElapsedSeconds; + var cpuPercent = wallDelta > 0 + ? (cpu - _lastCpu).TotalSeconds / wallDelta * 100.0 + : 0.0; + + _lastCpu = cpu; + _lastElapsedSeconds = elapsedSeconds; + + var sample = new ResourceSample( + elapsedSeconds, + process.WorkingSet64 / 1024.0 / 1024.0, + GC.GetTotalMemory(forceFullCollection: false) / 1024.0 / 1024.0, + cpuPercent, + process.Threads.Count, + GC.CollectionCount(2)); + + lock (_lock) _samples.Add(sample); + } + + /// + /// Summarizes the samples falling inside a window, expressed as seconds since + /// sampling started. Memory growth is reported both as an absolute delta and as a + /// least-squares slope, because a run that sawtooths around a stable mean and a + /// run that climbs monotonically can share the same endpoint delta. + /// + /// Window start (inclusive), seconds since start. + /// Window end (inclusive), seconds since start. + /// The window summary, or null when fewer than two samples fall inside it. + public ResourceWindowSummary? Summarize(double fromSeconds, double toSeconds) + { + var window = Snapshot() + .Where(s => s.ElapsedSeconds >= fromSeconds && s.ElapsedSeconds <= toSeconds) + .ToList(); + + if (window.Count < 2) + return null; + + var first = window[0]; + var last = window[^1]; + + return new ResourceWindowSummary( + SampleCount: window.Count, + DurationSeconds: last.ElapsedSeconds - first.ElapsedSeconds, + WorkingSetStartMb: first.WorkingSetMb, + WorkingSetEndMb: last.WorkingSetMb, + WorkingSetPeakMb: window.Max(s => s.WorkingSetMb), + WorkingSetSlopeMbPerMinute: Slope(window, s => s.WorkingSetMb) * 60.0, + ManagedHeapStartMb: first.ManagedHeapMb, + ManagedHeapEndMb: last.ManagedHeapMb, + ManagedHeapPeakMb: window.Max(s => s.ManagedHeapMb), + ManagedHeapSlopeMbPerMinute: Slope(window, s => s.ManagedHeapMb) * 60.0, + MeanCpuPercentOfOneCore: window.Average(s => s.CpuPercent), + PeakCpuPercentOfOneCore: window.Max(s => s.CpuPercent), + MeanThreadCount: window.Average(s => s.ThreadCount), + Gen2Collections: last.Gen2Collections - first.Gen2Collections); + } + + private static double Slope(IReadOnlyList samples, Func selector) + { + var n = samples.Count; + var meanX = samples.Average(s => s.ElapsedSeconds); + var meanY = samples.Average(selector); + + double numerator = 0, denominator = 0; + for (var i = 0; i < n; i++) + { + var dx = samples[i].ElapsedSeconds - meanX; + numerator += dx * (selector(samples[i]) - meanY); + denominator += dx * dx; + } + + return denominator == 0 ? 0 : numerator / denominator; + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await _loop; + } + catch (OperationCanceledException) + { + // Expected. + } + + _cts.Dispose(); + } +} + +/// Aggregate resource behaviour over a measurement window. +/// Samples in the window. +/// Window length. +/// Working set at window start. +/// Working set at window end. +/// Peak working set in the window. +/// Least-squares working-set growth rate. +/// Managed heap at window start. +/// Managed heap at window end. +/// Peak managed heap in the window. +/// Least-squares managed-heap growth rate. +/// Mean CPU as a percentage of one core (1400% = 14 cores saturated). +/// Peak single-sample CPU as a percentage of one core. +/// Mean OS thread count. +/// Gen-2 collections during the window. +public sealed record ResourceWindowSummary( + int SampleCount, + double DurationSeconds, + double WorkingSetStartMb, + double WorkingSetEndMb, + double WorkingSetPeakMb, + double WorkingSetSlopeMbPerMinute, + double ManagedHeapStartMb, + double ManagedHeapEndMb, + double ManagedHeapPeakMb, + double ManagedHeapSlopeMbPerMinute, + double MeanCpuPercentOfOneCore, + double PeakCpuPercentOfOneCore, + double MeanThreadCount, + int Gen2Collections); diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs new file mode 100644 index 00000000..8c3079f1 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Probes/StreamSubscriberProbe.cs @@ -0,0 +1,205 @@ +using System.Threading.Channels; +using Akka.Actor; +using ZB.MOM.WW.ScadaBridge.Communication.Actors; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; + +/// +/// A live site-stream subscriber assembled from the SAME parts +/// SiteStreamGrpcServer.RunSubscriptionStreamAsync uses: +/// +/// +/// SiteStreamManager.Subscribe — materializes the +/// per-subscriber graph (Where instance filter → Buffer(StreamBufferSize, +/// DropHead) → Sink.ForEach(Tell)). +/// A real , which converts the Akka +/// record to the protobuf SiteStreamEvent and TryWrites it. +/// A bounded DropOldest of the +/// production capacity (GrpcInstanceStreamChannelCapacity = 1000) with the +/// eviction counter wired to . +/// +/// +/// +/// The single substitution is the final hop: instead of +/// responseStream.WriteAsync pushing onto a socket, a reader task drains the +/// channel. That is deliberate — it is precisely the hop whose slowness register row +/// 50 asks about, and a controllable reader is the only way to hold it still. +/// +/// +public sealed class StreamSubscriberProbe : IAsyncDisposable +{ + /// Production per-instance stream channel capacity (GrpcInstanceStreamChannelCapacity). + public const int ProductionChannelCapacity = 1000; + + private readonly SiteStreamManager _manager; + private readonly string _subscriptionId; + private readonly IActorRef _relayActor; + private readonly ActorSystem _system; + private readonly Channel _channel; + private readonly CancellationTokenSource _cts = new(); + private readonly Task _readerTask; + private volatile LatencyHistogram? _latency; + + private readonly DropCounter _dropCounter; + private long _received; + private long _readerDelayMicroseconds; + private int _disposed; + + /// Human-readable probe name (also the relay actor's name suffix). + public string Name { get; } + + /// Events evicted by the bounded channel's DropOldest policy. + public long DroppedEvents => _dropCounter.Value; + + /// Events successfully drained by the reader (i.e. "sent to the client"). + public long ReceivedEvents => Interlocked.Read(ref _received); + + /// + /// Repoints the latency histogram this probe records into, without tearing the + /// subscription down. Used to separate ramp-window samples from steady-state ones: + /// re-attaching probes instead would open a zero-subscriber gap (during which + /// PublishAttributeValueChanged short-circuits) and risk reusing an actor + /// name whose previous incarnation has not finished terminating. + /// + /// The histogram to record into from now on, or null to stop recording. + public void RetargetLatency(LatencyHistogram? latency) => _latency = latency; + + /// + /// Artificial per-event reader delay, in microseconds. Zero is a healthy + /// subscriber; a large value models a stalled WAN link or a wedged client. + /// + public long ReaderDelayMicroseconds + { + get => Interlocked.Read(ref _readerDelayMicroseconds); + set => Interlocked.Exchange(ref _readerDelayMicroseconds, value); + } + + private StreamSubscriberProbe( + ActorSystem system, + SiteStreamManager manager, + string name, + Channel channel, + IActorRef relayActor, + string subscriptionId, + LatencyHistogram? latency, + DropCounter dropCounter) + { + _system = system; + _manager = manager; + Name = name; + _channel = channel; + _relayActor = relayActor; + _subscriptionId = subscriptionId; + _latency = latency; + _dropCounter = dropCounter; + _readerTask = Task.Run(() => ReadLoopAsync(_cts.Token)); + } + + /// + /// Builds and attaches a probe subscribed to one instance's events. + /// + /// The site actor system. + /// The site stream manager to subscribe against. + /// Instance whose events this probe receives. + /// Probe name, used for the relay actor's path. + /// Optional histogram fed with end-to-end event latency. + /// The attached probe. + public static StreamSubscriberProbe Attach( + ActorSystem system, + SiteStreamManager manager, + string instanceUniqueName, + string name, + LatencyHistogram? latency) + { + var dropCounter = new DropCounter(); + var channel = Channel.CreateBounded( + new BoundedChannelOptions(ProductionChannelCapacity) + { + FullMode = BoundedChannelFullMode.DropOldest, + }, + _ => dropCounter.Increment()); + + var relayActor = system.ActorOf( + Props.Create(typeof(StreamRelayActor), name, channel.Writer), + $"stream-relay-{name}"); + + var subscriptionId = manager.Subscribe(instanceUniqueName, relayActor); + + return new StreamSubscriberProbe( + system, manager, name, channel, relayActor, subscriptionId, latency, dropCounter); + } + + private async Task ReadLoopAsync(CancellationToken cancellationToken) + { + try + { + await foreach (var evt in _channel.Reader.ReadAllAsync(cancellationToken)) + { + Interlocked.Increment(ref _received); + + var latency = _latency; + if (latency != null && evt.AttributeChanged != null) + { + // The emit instant travels verbatim: the driver stamps it on + // TagValueUpdate.Timestamp, DataConnectionActor forwards it, + // InstanceActor copies it onto AttributeValueChanged.Timestamp, and + // StreamRelayActor maps it onto the proto Timestamp. So this is a + // true end-to-end DCL-boundary → subscriber measurement. + var emitted = evt.AttributeChanged.Timestamp.ToDateTimeOffset(); + latency.Record(DateTimeOffset.UtcNow - emitted); + } + + var delay = Interlocked.Read(ref _readerDelayMicroseconds); + if (delay > 0) + await Task.Delay(TimeSpan.FromMicroseconds(delay), cancellationToken); + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + /// Detaches the subscription and stops the relay actor and reader. + /// A task that completes when the probe is torn down. + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + _manager.Unsubscribe(_subscriptionId); + _channel.Writer.TryComplete(); + await _cts.CancelAsync(); + + try + { + await _readerTask; + } + catch (OperationCanceledException) + { + // Expected. + } + + _system.Stop(_relayActor); + _cts.Dispose(); + } +} + +/// +/// Thread-safe counter for a bounded channel's itemDropped callback. A tiny +/// class rather than a captured local so the probe and the channel share exactly one +/// counter instance without a second closure. +/// +public sealed class DropCounter +{ + private long _value; + + /// Current count. + public long Value => Interlocked.Read(ref _value); + + /// Increments the counter. + public void Increment() => Interlocked.Increment(ref _value); +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs new file mode 100644 index 00000000..30fc5e11 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Program.cs @@ -0,0 +1,53 @@ +using System.Text.Json; +using ZB.MOM.WW.ScadaBridge.LoadHarness; + +// Target-scale load harness (Phase-8 WP-4 / deferred-work register #25 + row 50). +// +// Full-scale protocol run (the WP-4 numbers): +// dotnet run -c Release --project tests/ZB.MOM.WW.ScadaBridge.LoadHarness -- \ +// --sustain-minutes 20 --results loadharness-results.json +// +// Full 1-hour version, unchanged in every other respect: +// ... -- --sustain-minutes 60 --results loadharness-results-1h.json +// +// Scaled-down smoke (what the CI [Fact] runs): +// ... -- --sites 2 --instances-per-site 10 --tags-per-instance 5 \ +// --settle-minutes 0.1 --sustain-minutes 0.2 --sample-seconds 2 \ +// --sf-drain-messages 200 --slow-subscriber-events 2000 + +var config = HarnessConfig.Parse(args); + +Console.WriteLine("ScadaBridge target-scale load harness"); +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($" nominal update rate {config.NominalUpdatesPerSecond:N0}/s"); +Console.WriteLine($" settle / sustain {config.SettleDuration.TotalMinutes:F1} / {config.SustainDuration.TotalMinutes:F1} min"); +Console.WriteLine(); + +using var cancellation = new CancellationTokenSource(); +Console.CancelKeyPress += (_, e) => +{ + e.Cancel = true; + cancellation.Cancel(); +}; + +try +{ + var result = await HarnessRun.ExecuteAsync(config, Console.WriteLine, cancellation.Token); + + var json = JsonSerializer.Serialize(result, new JsonSerializerOptions { WriteIndented = true }); + await File.WriteAllTextAsync(config.ResultsPath, json, cancellation.Token); + + Console.WriteLine(); + Console.WriteLine(ResultsFormatter.Format(result)); + Console.WriteLine(); + Console.WriteLine($"JSON metrics written to {Path.GetFullPath(config.ResultsPath)}"); + return 0; +} +catch (OperationCanceledException) +{ + Console.Error.WriteLine("Run cancelled."); + return 130; +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs new file mode 100644 index 00000000..5c9e40c3 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ResultsFormatter.cs @@ -0,0 +1,110 @@ +using System.Globalization; +using System.Text; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Renders a as the markdown table that goes into the +/// results document, so the published numbers and the JSON come from one source. +/// +public static class ResultsFormatter +{ + /// Formats a run result for console output and the results doc. + /// The run to format. + /// A markdown-ish plain-text report. + public static string Format(HarnessRunResult result) + { + var culture = CultureInfo.InvariantCulture; + var builder = new StringBuilder(); + + builder.AppendLine("=== MEASURED ==="); + builder.AppendLine(culture, $"host {result.Environment.MachineName} / {result.Environment.OsDescription}"); + builder.AppendLine(culture, $"cpus / runtime / gc {result.Environment.ProcessorCount} / {result.Environment.RuntimeVersion} / serverGC={result.Environment.ServerGc}"); + builder.AppendLine(culture, $"scale {result.Config.Sites} sites x {result.Config.InstancesPerSite} instances x {result.Config.TagsPerInstance} tags = {result.Config.TotalSubscriptions:N0} subscriptions"); + builder.AppendLine(culture, $"total run {result.TotalSeconds:F1}s"); + builder.AppendLine(); + + builder.AppendLine("-- deployment / ramp --"); + builder.AppendLine(culture, $"site fixtures built {result.SiteRampSeconds:F1}s"); + builder.AppendLine(culture, $"all instance actors {result.InstanceRampSeconds:F1}s"); + builder.AppendLine(culture, $"slowest single site {result.SlowestSiteInstanceRampSeconds:F1}s ({result.Config.InstancesPerSite} instances)"); + builder.AppendLine(); + + builder.AppendLine("-- offered load (steady-state window) --"); + builder.AppendLine(culture, $"nominal {result.NominalUpdatesPerSecond:N0} updates/s"); + builder.AppendLine(culture, $"achieved {result.AchievedUpdatesPerSecond:N0} updates/s ({result.AchievedUpdatesPerSecond / result.NominalUpdatesPerSecond:P1} of nominal)"); + builder.AppendLine(culture, $"updates offered {result.SteadyStateEmittedTagUpdates:N0}"); + builder.AppendLine(culture, $"driver slice overrun {result.DriverLagSeconds:F1}s cumulative"); + builder.AppendLine(culture, $"skipped (no callback) {result.DriverSkippedNoCallback:N0}"); + builder.AppendLine(); + + builder.AppendLine("-- tag update latency (DCL boundary -> stream subscriber) --"); + AppendLatency(builder, culture, result.TagUpdateLatency); + builder.AppendLine(); + + builder.AppendLine("-- live stream subscribers (steady-state window) --"); + builder.AppendLine(culture, $"events delivered {result.StreamProbeReceived:N0}"); + builder.AppendLine(culture, $"events dropped {result.StreamProbeDropped:N0}"); + builder.AppendLine(); + + builder.AppendLine("-- health report delivery --"); + AppendLatency(builder, culture, result.HealthReportLatency); + builder.AppendLine(culture, $"reports ingested {result.HealthReportsDelivered:N0}"); + builder.AppendLine(culture, $"sites tracked centrally {result.SitesTrackedByAggregator}"); + builder.AppendLine(); + + builder.AppendLine("-- debug view snapshot (under load) --"); + AppendLatency(builder, culture, result.DebugSnapshotLatency); + builder.AppendLine(culture, $"completed / timed out {result.DebugSnapshotsCompleted:N0} / {result.DebugSnapshotTimeouts:N0}"); + builder.AppendLine(); + + if (result.SteadyStateResources is { } steady) + { + builder.AppendLine("-- resources (steady-state window) --"); + builder.AppendLine(culture, $"window {steady.DurationSeconds:F0}s over {steady.SampleCount} samples"); + builder.AppendLine(culture, $"working set {steady.WorkingSetStartMb:F0} -> {steady.WorkingSetEndMb:F0} MB (peak {steady.WorkingSetPeakMb:F0} MB)"); + builder.AppendLine(culture, $"working set slope {steady.WorkingSetSlopeMbPerMinute:F2} MB/min"); + builder.AppendLine(culture, $"managed heap {steady.ManagedHeapStartMb:F0} -> {steady.ManagedHeapEndMb:F0} MB (peak {steady.ManagedHeapPeakMb:F0} MB)"); + builder.AppendLine(culture, $"managed heap slope {steady.ManagedHeapSlopeMbPerMinute:F2} MB/min"); + builder.AppendLine(culture, $"cpu mean / peak {steady.MeanCpuPercentOfOneCore:F0}% / {steady.PeakCpuPercentOfOneCore:F0}% of one core ({steady.MeanCpuPercentOfOneCore / result.Environment.ProcessorCount:F1}% of the box)"); + builder.AppendLine(culture, $"threads / gen2 GCs {steady.MeanThreadCount:F0} / {steady.Gen2Collections}"); + builder.AppendLine(); + } + + if (result.StoreAndForwardDrain is { } drain) + { + builder.AppendLine("-- store-and-forward (register row 50) --"); + builder.AppendLine(culture, $"messages {drain.MessageCount:N0}"); + builder.AppendLine(culture, $"concurrent buffering {drain.EnqueuePerSecond:N0} msg/s ({drain.EnqueueSeconds:F1}s)"); + builder.AppendLine(culture, $"retry wait before drain {drain.TimeToFirstDeliverySeconds:F1}s (DefaultRetryInterval)"); + builder.AppendLine(culture, $"drain throughput {drain.DrainPerSecond:N0} msg/s (active drain {drain.DrainSeconds - drain.TimeToFirstDeliverySeconds:F2}s)"); + builder.AppendLine(culture, $"residual depth {drain.ResidualDepth}"); + builder.AppendLine(); + } + + if (result.SlowSubscriber is { } slow) + { + builder.AppendLine("-- slow-subscriber isolation (register row 50) --"); + builder.AppendLine(culture, $"published {slow.PublishedEvents:N0} at {slow.PublishPerSecond:N0}/s"); + builder.AppendLine(culture, $"healthy min delivery {slow.HealthyMinDeliveryRatio:P2}"); + builder.AppendLine(culture, $"stalled delivery {slow.SlowDeliveryRatio:P2}"); + foreach (var outcome in slow.Outcomes) + { + builder.AppendLine(culture, + $" {outcome.Name,-28} {(outcome.IsSlow ? "STALLED" : "healthy"),-8} " + + $"recv {outcome.Received,8:N0} dropped {outcome.Dropped,8:N0} ratio {outcome.DeliveryRatio:P2}"); + } + + builder.AppendLine(); + } + + return builder.ToString(); + } + + private static void AppendLatency(StringBuilder builder, CultureInfo culture, Metrics.LatencySnapshot snapshot) + { + builder.AppendLine(culture, + $"samples {snapshot.Count:N0} mean {snapshot.MeanMs:F2}ms p50 {snapshot.P50Ms:F2}ms " + + $"p95 {snapshot.P95Ms:F2}ms p99 {snapshot.P99Ms:F2}ms p99.9 {snapshot.P999Ms:F2}ms max {snapshot.MaxMs:F2}ms"); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs new file mode 100644 index 00000000..16af25fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/ObservabilityProbes.cs @@ -0,0 +1,160 @@ +using System.Diagnostics; +using Akka.Actor; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// +/// The two "does observability still work at scale?" probes WP-4's test protocol +/// names — health report delivery timing and debug view latency — run continuously +/// alongside the sustained load rather than after it, so both are measured against a +/// site that is actually busy. +/// +public sealed class ObservabilityProbes : IAsyncDisposable +{ + private readonly CancellationTokenSource _cts = new(); + private readonly List _tasks = new(); + + /// + /// End-to-end health report latency: SiteHealthCollector.CollectReport plus + /// the transport hop plus CentralHealthAggregator.ProcessReport. The + /// interesting term at scale is CollectReport, which materializes the + /// per-connection dictionaries for a site carrying 37,500 subscriptions. + /// + public LatencyHistogram HealthReportLatency { get; } = new(); + + /// + /// Debug view snapshot round-trip: an Ask of DebugSnapshotRequest to + /// a live Instance Actor. This lands the request in the mailbox of an actor that + /// is concurrently ingesting tag updates, so the measured time includes real + /// queueing behind production traffic — which is the whole point of measuring it + /// under load. + /// + public LatencyHistogram DebugSnapshotLatency { get; } = new(); + + /// Health reports successfully ingested by the central aggregator. + public long HealthReportsDelivered => Interlocked.Read(ref _healthReports); + + /// Debug snapshots that completed within the ask timeout. + public long DebugSnapshotsCompleted => Interlocked.Read(ref _debugSnapshots); + + /// Debug snapshot asks that timed out. + public long DebugSnapshotTimeouts => Interlocked.Read(ref _debugTimeouts); + + private long _healthReports; + private long _debugSnapshots; + private long _debugTimeouts; + + /// + /// Starts both probes. + /// + /// Sites to probe. + /// The real central aggregator receiving the reports. + /// Health report cadence (production default 30 s). + /// How often to take a debug snapshot. + /// The running probes. + public static ObservabilityProbes Start( + IReadOnlyList sites, + CentralHealthAggregator aggregator, + TimeSpan reportInterval, + TimeSpan debugProbeInterval) + { + var probes = new ObservabilityProbes(); + probes._tasks.Add(Task.Run(() => probes.HealthLoopAsync(sites, aggregator, reportInterval, probes._cts.Token))); + probes._tasks.Add(Task.Run(() => probes.DebugLoopAsync(sites, debugProbeInterval, probes._cts.Token))); + return probes; + } + + private async Task HealthLoopAsync( + IReadOnlyList sites, + CentralHealthAggregator aggregator, + TimeSpan interval, + CancellationToken cancellationToken) + { + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + foreach (var site in sites) + { + var watch = Stopwatch.StartNew(); + var report = site.HealthCollector.CollectReport(site.SiteId); + aggregator.ProcessReport(report); + watch.Stop(); + + HealthReportLatency.Record(watch.Elapsed); + Interlocked.Increment(ref _healthReports); + } + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private async Task DebugLoopAsync( + IReadOnlyList sites, + TimeSpan interval, + CancellationToken cancellationToken) + { + var random = new Random(20260815); + using var timer = new PeriodicTimer(interval); + try + { + while (await timer.WaitForNextTickAsync(cancellationToken)) + { + var site = sites[random.Next(sites.Count)]; + if (site.InstanceActors.Count == 0) + continue; + + var index = random.Next(site.InstanceActors.Count); + var actor = site.InstanceActors[index]; + var request = new DebugSnapshotRequest(site.InstanceName(index), Guid.NewGuid().ToString("N")); + + var watch = Stopwatch.StartNew(); + try + { + await actor.Ask(request, TimeSpan.FromSeconds(10), cancellationToken); + watch.Stop(); + DebugSnapshotLatency.Record(watch.Elapsed); + Interlocked.Increment(ref _debugSnapshots); + } + catch (AskTimeoutException) + { + Interlocked.Increment(ref _debugTimeouts); + } + } + } + catch (OperationCanceledException) + { + // Normal teardown. + } + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + // Idempotent: the orchestrator stops the probes early (so the register row 50 + // scenarios do not compete with them) and again in its finally block. + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await Task.WhenAll(_tasks); + } + catch (OperationCanceledException) + { + // Expected. + } + + _cts.Dispose(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs new file mode 100644 index 00000000..b302ae7c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/SlowSubscriberScenario.cs @@ -0,0 +1,153 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// Per-subscriber outcome from the slow-subscriber isolation probe. +/// Probe name. +/// Whether this probe's reader was deliberately stalled. +/// Events the reader drained. +/// Events evicted by this probe's bounded DropOldest channel. +/// Received / published, before accounting for the site-stream buffer. +public sealed record SubscriberOutcome( + string Name, + bool IsSlow, + long Received, + long Dropped, + double DeliveryRatio); + +/// Result of the slow-subscriber isolation measurement. +/// Events published to the site stream during the probe. +/// Wall time the publisher took. +/// Publish throughput observed by the producer. +/// Per-subscriber outcomes. +/// Worst delivery ratio among the healthy subscribers. +/// Delivery ratio of the stalled subscriber. +public sealed record SlowSubscriberResult( + int PublishedEvents, + double PublishSeconds, + double PublishPerSecond, + IReadOnlyList Outcomes, + double HealthyMinDeliveryRatio, + double SlowDeliveryRatio); + +/// +/// Register row 50, second half: what does a slow or stalled gRPC subscriber do to +/// per-subscriber buffering when several subscribers are attached? +/// +/// +/// Several probes are attached to the SAME instance so every one of them is offered +/// exactly the same event sequence — otherwise a difference in delivery could be a +/// difference in offered load rather than a backpressure effect. One probe's reader +/// is then stalled (a large per-event delay, standing in for a wedged client or a +/// dead WAN link) while the rest read as fast as they can. Events are published +/// through the real SiteStreamManager. +/// +/// +/// The question the numbers answer: does the stalled subscriber's backlog propagate +/// upstream — evicting events for the healthy subscribers or slowing the publisher — +/// or is it confined to its own Buffer(DropHead) stage and its own bounded +/// DropOldest channel? The design intends the latter; this measures it. +/// +/// +public static class SlowSubscriberScenario +{ + /// Per-event reader delay applied to the stalled subscriber. + public const int SlowReaderDelayMicroseconds = 50_000; + + /// + /// Publish rate for the probe. Deliberately paced rather than a tight burst: the + /// publish source is a single Source.ActorRef(StreamBufferSize, DropHead) + /// SHARED by every attribute subscriber, so an unpaced burst saturates that shared + /// stage and every subscriber loses events for a reason that has nothing to do + /// with the slow one. Pacing below the shared stage's capacity is what isolates + /// the variable under test. + /// + public const int PublishRatePerSecond = 2_000; + + /// Runs the isolation probe on a dedicated site. + /// Site whose stream manager is used. + /// Total subscribers to attach (one of them is stalled). + /// Events to publish. + /// Cancels the measurement. + /// The measured result. + public static async Task RunAsync( + SiteRuntimeFixture site, + int subscriberCount, + int eventCount, + CancellationToken cancellationToken) + { + var instanceName = site.InstanceName(0); + var probes = new List(subscriberCount); + + try + { + for (var i = 0; i < subscriberCount; i++) + { + var probe = StreamSubscriberProbe.Attach( + site.System, site.StreamManager, instanceName, + $"{site.SiteId}-slowprobe-{i:D2}", latency: null); + + // Probe 0 is the pathological one. + if (i == 0) + probe.ReaderDelayMicroseconds = SlowReaderDelayMicroseconds; + + probes.Add(probe); + } + + // Let every subscription's stream graph finish materializing before the burst. + await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); + + var watch = Stopwatch.StartNew(); + const int sliceMilliseconds = 50; + var perSlice = Math.Max(1, PublishRatePerSecond * sliceMilliseconds / 1000); + var published = 0; + while (published < eventCount && !cancellationToken.IsCancellationRequested) + { + var sliceStart = Stopwatch.GetTimestamp(); + var end = Math.Min(published + perSlice, eventCount); + for (var i = published; i < end; i++) + { + site.StreamManager.PublishAttributeValueChanged(new AttributeValueChanged( + instanceName, "Tag000", "Tag000", i, "Good", DateTimeOffset.UtcNow)); + } + + published = end; + + var elapsedMs = (Stopwatch.GetTimestamp() - sliceStart) * 1000.0 / Stopwatch.Frequency; + if (elapsedMs < sliceMilliseconds) + await Task.Delay(TimeSpan.FromMilliseconds(sliceMilliseconds - elapsedMs), cancellationToken); + } + + watch.Stop(); + + // Give the healthy readers time to finish; the stalled one will not. + await Task.Delay(TimeSpan.FromSeconds(20), cancellationToken); + + var outcomes = probes + .Select((p, i) => new SubscriberOutcome( + p.Name, + IsSlow: i == 0, + p.ReceivedEvents, + p.DroppedEvents, + p.ReceivedEvents / (double)eventCount)) + .ToList(); + + var healthy = outcomes.Where(o => !o.IsSlow).ToList(); + + return new SlowSubscriberResult( + PublishedEvents: eventCount, + PublishSeconds: watch.Elapsed.TotalSeconds, + PublishPerSecond: eventCount / Math.Max(0.001, watch.Elapsed.TotalSeconds), + Outcomes: outcomes, + HealthyMinDeliveryRatio: healthy.Count == 0 ? 0 : healthy.Min(o => o.DeliveryRatio), + SlowDeliveryRatio: outcomes[0].DeliveryRatio); + } + finally + { + foreach (var probe in probes) + await probe.DisposeAsync(); + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs new file mode 100644 index 00000000..5370f281 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/Scenarios/StoreAndForwardDrainScenario.cs @@ -0,0 +1,154 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Scenarios; + +/// Result of one store-and-forward drain measurement. +/// Site the buffer belonged to. +/// Messages buffered before the drain began. +/// Wall time to buffer them (concurrent, many origin instances). +/// Buffering throughput. +/// +/// Wall time from the first sweep to the first successful delivery. With +/// attemptImmediateDelivery: false the engine stamps LastAttemptAt, so the +/// row is not due until one DefaultRetryInterval (30 s) has passed — this is the +/// configured retry latency, not drain slowness, and is reported separately for that reason. +/// +/// Wall time from the first sweep to an empty buffer (includes the retry wait). +/// +/// Drain throughput measured from the FIRST delivery to an empty buffer — the engine's +/// actual capacity, and the headline number for register row 50. +/// +/// Buffer depth left when the measurement stopped (0 = fully drained). +/// Delivered-count samples during the drain, so a steady rate can be told from a stall-then-burst. +public sealed record StoreAndForwardDrainResult( + string SiteId, + int MessageCount, + double EnqueueSeconds, + double EnqueuePerSecond, + double TimeToFirstDeliverySeconds, + double DrainSeconds, + double DrainPerSecond, + int ResidualDepth, + IReadOnlyList Progress); + +/// One observation during the drain. +/// Seconds since the drain began. +/// Cumulative successful deliveries. +/// Remaining buffer depth. +public sealed record DrainProgressSample(double ElapsedSeconds, long Delivered, int Depth); + +/// +/// Measures store-and-forward buffering and drain throughput (deferred-work register +/// row 50, first half) using the real StoreAndForwardService, the real +/// StoreAndForwardStorage and the real SQLite file — only the delivery target +/// is a counting stub, because what is being measured is the site-local buffer's +/// throughput, not a remote endpoint's. +/// +/// +/// Phase 1 buffers messages with +/// attemptImmediateDelivery: false, spread across many origin instance names +/// and issued from many concurrent tasks — the "concurrent buffering from multiple +/// instances" WP-4 asks about ([xc-7]). Phase 2 registers a delivery handler +/// that always succeeds and drives sweeps to completion, timing the drain. +/// +/// +/// The sweep is driven explicitly rather than waiting on the 10 s +/// RetryTimerInterval so the number reported is the engine's drain capacity, +/// not its polling cadence. The per-sweep batch is SweepBatchLimit (500) with +/// SweepTargetParallelism (4) lanes, both at their production defaults. +/// +/// +public static class StoreAndForwardDrainScenario +{ + /// Runs the drain measurement against one site's real S&F engine. + /// The site whose store-and-forward engine is exercised. + /// Messages to buffer. + /// Concurrent enqueue tasks (distinct origin instances). + /// Cancels the measurement. + /// The measured result. + public static async Task RunAsync( + SiteRuntimeFixture site, + int messageCount, + int concurrency, + CancellationToken cancellationToken) + { + var service = site.StoreAndForward; + var payload = $"{{\"site\":\"{site.SiteId}\",\"body\":\"{new string('x', 256)}\"}}"; + + // Phase 1 — concurrent buffering from many instances, no delivery attempted. + var enqueueWatch = Stopwatch.StartNew(); + var perTask = messageCount / concurrency; + var enqueueTasks = new List(concurrency); + for (var t = 0; t < concurrency; t++) + { + var taskIndex = t; + enqueueTasks.Add(Task.Run(async () => + { + for (var i = 0; i < perTask; i++) + { + await service.EnqueueAsync( + StoreAndForwardCategory.ExternalSystem, + target: $"load-target-{taskIndex % 4}", + payloadJson: payload, + originInstanceName: site.InstanceName(taskIndex), + attemptImmediateDelivery: false); + } + }, cancellationToken)); + } + + await Task.WhenAll(enqueueTasks); + enqueueWatch.Stop(); + var buffered = perTask * concurrency; + + // Phase 2 — a delivery target that always succeeds; time the drain to empty. + var delivered = 0L; + service.RegisterDeliveryHandler(StoreAndForwardCategory.ExternalSystem, _ => + { + Interlocked.Increment(ref delivered); + return Task.FromResult(true); + }); + + // Sweeps are driven explicitly rather than waiting on the 10 s RetryTimerInterval: + // the number wanted is the engine's drain CAPACITY, not its polling cadence. One + // sweep moves at most SweepBatchLimit (500) messages, so a large backlog needs + // many, and the progress series below is what distinguishes a genuinely slow + // drain from an artefact of this polling loop. + var drainWatch = Stopwatch.StartNew(); + var deadline = DateTimeOffset.UtcNow.AddMinutes(10); + var progress = new List(); + int depth; + while (true) + { + service.TriggerSweep(); + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + + var depths = await service.GetBufferDepthAsync(); + depth = depths.Values.Sum(); + progress.Add(new DrainProgressSample( + drainWatch.Elapsed.TotalSeconds, Interlocked.Read(ref delivered), depth)); + + if (depth == 0 || DateTimeOffset.UtcNow > deadline) + break; + } + + drainWatch.Stop(); + + // Split the retry wait from the drain: the first sample with a non-zero delivered + // count marks the moment the backlog actually became due. + var firstDelivery = progress.FirstOrDefault(s => s.Delivered > 0); + var timeToFirstDelivery = firstDelivery?.ElapsedSeconds ?? drainWatch.Elapsed.TotalSeconds; + var activeDrainSeconds = Math.Max(0.001, drainWatch.Elapsed.TotalSeconds - timeToFirstDelivery); + + return new StoreAndForwardDrainResult( + SiteId: site.SiteId, + MessageCount: buffered, + EnqueueSeconds: enqueueWatch.Elapsed.TotalSeconds, + EnqueuePerSecond: buffered / Math.Max(0.001, enqueueWatch.Elapsed.TotalSeconds), + TimeToFirstDeliverySeconds: timeToFirstDelivery, + DrainSeconds: drainWatch.Elapsed.TotalSeconds, + DrainPerSecond: Interlocked.Read(ref delivered) / activeDrainSeconds, + ResidualDepth: depth, + Progress: progress); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs new file mode 100644 index 00000000..d08bdfeb --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SimulatedDataConnection.cs @@ -0,0 +1,153 @@ +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// In-process stand-in for an OPC UA / MxGateway server, registered on the REAL +/// DataConnectionFactory under the protocol so the +/// whole Data Connection Layer above it — DataConnectionManagerActor, +/// DataConnectionActor, its _instancesByTag fan-out and the +/// TagValueUpdate hand-off to Instance Actors — runs unmodified. +/// +/// +/// It implements because the batch path is +/// the one a real site takes at this scale; the per-tag fallback would make +/// subscribe setup, not steady-state throughput, the thing being measured. +/// +/// +/// Why faking here does not invalidate the measurement. Everything this class +/// replaces is on the far side of the process boundary: socket I/O, the OPC UA SDK's +/// own session/subscription machinery, and the device. The system under test — the +/// actor hierarchy, the site stream, store-and-forward, health, and the streaming +/// relay — begins at the invocation, which is +/// exactly where the real adapter hands off. Driving 375,000 genuine monitored items +/// would measure the OPC UA stack, not ScadaBridge. +/// +/// +public sealed class SimulatedDataConnection : IDataConnection, IBatchSubscribableConnection +{ + /// Protocol discriminator this adapter registers under on the factory. + public const string ProtocolName = "LoadSim"; + + private SubscriptionCallback? _callback; + private int _subscriptionCounter; + + /// + /// The data connection name this adapter was created for, taken from the connection + /// details. The DCL factory creates adapters as its manager actor processes the + /// CreateConnectionCommands, so creation order is not connection order — the driver + /// resolves an adapter by name rather than by index. + /// + public string ConnectionName { get; private set; } = string.Empty; + + /// Key under which the connection name travels in the connection details. + public const string ConnectionNameKey = "connectionName"; + + /// + public ConnectionHealth Status { get; private set; } = ConnectionHealth.Disconnected; + + /// + public event Action? Disconnected; + + /// + /// The callback captured at subscribe time. The tag driver invokes this to inject + /// a value change, mirroring what the OPC UA SDK's notification thread does. + /// Null until the site's Instance Actors have subscribed. + /// + public SubscriptionCallback? ValueCallback => _callback; + + /// Number of tag paths this connection has accepted subscriptions for. + public int SubscribedTagCount => _subscriptionCounter; + + /// + public Task ConnectAsync(IDictionary connectionDetails, CancellationToken cancellationToken = default) + { + if (connectionDetails.TryGetValue(ConnectionNameKey, out var name)) + ConnectionName = name; + + Status = ConnectionHealth.Connected; + return Task.CompletedTask; + } + + /// + public Task DisconnectAsync(CancellationToken cancellationToken = default) + { + Status = ConnectionHealth.Disconnected; + Disconnected?.Invoke(); + return Task.CompletedTask; + } + + /// + public Task SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default) + { + _callback = callback; + return Task.FromResult($"sub-{Interlocked.Increment(ref _subscriptionCounter)}"); + } + + /// + public Task> SubscribeBatchAsync( + IReadOnlyList tagPaths, + SubscriptionCallback callback, + CancellationToken cancellationToken = default) + { + _callback = callback; + var results = new List(tagPaths.Count); + foreach (var path in tagPaths) + { + results.Add(new TagSubscribeResult( + path, true, $"sub-{Interlocked.Increment(ref _subscriptionCounter)}", null)); + } + + return Task.FromResult>(results); + } + + /// + public Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task UnsubscribeBatchAsync(IReadOnlyList subscriptionIds, CancellationToken cancellationToken = default) + => Task.CompletedTask; + + /// + public Task ReadAsync(string tagPath, CancellationToken cancellationToken = default) + => Task.FromResult(new ReadResult(true, new TagValue(0d, QualityCode.Good, DateTimeOffset.UtcNow), null)); + + /// + public Task> ReadBatchAsync( + IEnumerable tagPaths, CancellationToken cancellationToken = default) + { + var now = DateTimeOffset.UtcNow; + var results = new Dictionary(); + foreach (var path in tagPaths) + results[path] = new ReadResult(true, new TagValue(0d, QualityCode.Good, now), null); + + return Task.FromResult>(results); + } + + /// + public Task WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default) + => Task.FromResult(new WriteResult(true, null)); + + /// + public Task> WriteBatchAsync( + IDictionary values, CancellationToken cancellationToken = default) + { + var results = new Dictionary(); + foreach (var key in values.Keys) + results[key] = new WriteResult(true, null); + + return Task.FromResult>(results); + } + + /// + public Task WriteBatchAndWaitAsync( + IDictionary values, string flagPath, object? flagValue, + string responsePath, object? responseValue, TimeSpan timeout, + CancellationToken cancellationToken = default) + => Task.FromResult(true); + + /// + public ValueTask DisposeAsync() => ValueTask.CompletedTask; +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs new file mode 100644 index 00000000..649c0c9c --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/SiteRuntimeFixture.cs @@ -0,0 +1,371 @@ +using System.Diagnostics; +using System.Text.Json; +using Akka.Actor; +using Microsoft.Extensions.Configuration; +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.Commons.Types.Flattening; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Actors; +using ZB.MOM.WW.ScadaBridge.HealthMonitoring; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; +using ZB.MOM.WW.ScadaBridge.LoadHarness.Probes; +using ZB.MOM.WW.ScadaBridge.SiteRuntime; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Actors; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts; +using ZB.MOM.WW.ScadaBridge.SiteRuntime.Streaming; +using ZB.MOM.WW.ScadaBridge.StoreAndForward; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// One simulated site: its own , its own LocalDb SQLite +/// file, a real Data Connection Layer over +/// adapters, real Instance Actors, a real , real +/// store-and-forward, and a real . +/// +/// +/// The sites are separate, non-clustered ActorSystems rather than 10 real two-node +/// Akka clusters. Cluster membership, singleton placement and failover timing are +/// already measured on a real two-node rig by +/// PerformanceTests/Failover/FailoverTimingTests.cs; what WP-4 asks about is +/// the load-bearing hierarchy under each singleton, which is what this builds. +/// +/// +public sealed class SiteRuntimeFixture : IAsyncDisposable +{ + private readonly HarnessConfig _config; + private readonly List _connections; + private readonly ServiceProvider _localDbProvider; + private readonly ILocalDb _localDb; + private readonly List _instanceActors = new(); + private readonly List _probes = new(); + private readonly string _dataDirectory; + + /// Site identifier, e.g. site-01. + public string SiteId { get; } + + /// This site's actor system. + public ActorSystem System { get; } + + /// The real site-wide broadcast stream. + public SiteStreamManager StreamManager { get; } + + /// The real site health collector feeding the 30 s report. + public SiteHealthCollector HealthCollector { get; } + + /// The real store-and-forward engine for this site. + public StoreAndForwardService StoreAndForward { get; } + + /// The real DCL manager actor. + public IActorRef DataConnectionManager { get; } + + /// The simulated adapters, one per data connection, in creation order. + public IReadOnlyList Connections + { + get { lock (_connections) return _connections.ToList(); } + } + + /// The live stream subscribers attached to this site. + public IReadOnlyList Probes => _probes; + + /// Instance Actors created on this site. + public IReadOnlyList InstanceActors => _instanceActors; + + /// Tag paths per connection index, in the order they were assigned. + public IReadOnlyList> TagPathsByConnection { get; } + + /// Wall-clock time the instance ramp took, measured by . + public TimeSpan InstanceRampDuration { get; private set; } + + private SiteRuntimeFixture( + string siteId, + HarnessConfig config, + string dataDirectory, + ServiceProvider localDbProvider, + ILocalDb localDb, + ActorSystem system, + SiteStreamManager streamManager, + SiteHealthCollector healthCollector, + StoreAndForwardService storeAndForward, + IActorRef dataConnectionManager, + SiteStorageService storage, + ScriptCompilationService compilationService, + SharedScriptLibrary sharedScriptLibrary, + SiteRuntimeOptions siteOptions, + List> tagPathsByConnection, + List connections) + { + _connections = connections; + SiteId = siteId; + _config = config; + _dataDirectory = dataDirectory; + _localDbProvider = localDbProvider; + _localDb = localDb; + System = system; + StreamManager = streamManager; + HealthCollector = healthCollector; + StoreAndForward = storeAndForward; + DataConnectionManager = dataConnectionManager; + Storage = storage; + CompilationService = compilationService; + SharedScriptLibrary = sharedScriptLibrary; + SiteOptions = siteOptions; + TagPathsByConnection = tagPathsByConnection; + } + + private SiteStorageService Storage { get; } + private ScriptCompilationService CompilationService { get; } + private SharedScriptLibrary SharedScriptLibrary { get; } + private SiteRuntimeOptions SiteOptions { get; } + + /// + /// Builds a site: LocalDb file, actor system, DCL with its simulated connections, + /// stream manager, health collector and store-and-forward engine. Instance Actors + /// are created separately by so the deployment + /// ramp can be timed on its own. + /// + /// Zero-based site index. + /// Harness configuration. + /// Directory under which this site's SQLite files live. + /// Number of data connections to spread the site's tags across. + /// The started fixture. + public static async Task CreateAsync( + int siteIndex, HarnessConfig config, string rootDataDirectory, int connectionsPerSite) + { + var siteId = $"site-{siteIndex + 1:D2}"; + var dataDirectory = Path.Combine(rootDataDirectory, siteId); + Directory.CreateDirectory(dataDirectory); + + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["LocalDb:Path"] = Path.Combine(dataDirectory, "site-localdb.db"), + }) + .Build(); + + var localDbProvider = new ServiceCollection() + .AddZbLocalDb(configuration) + .BuildServiceProvider(); + var localDb = localDbProvider.GetRequiredService(); + + var storage = new SiteStorageService(localDb, NullLogger.Instance); + await storage.InitializeAsync(); + + var compilationService = new ScriptCompilationService(NullLogger.Instance); + var sharedScriptLibrary = new SharedScriptLibrary(compilationService, NullLogger.Instance); + + // Production defaults throughout — the point of the run is to measure the + // shipped configuration, so nothing here is tuned for the harness. + var siteOptions = new SiteRuntimeOptions(); + + // WARNING-level logging. At 37,500 updates/s per site, Akka's INFO output would + // itself become a measured load; and the InstanceActorInitialized dead letters are + // a harness artifact (see StartInstancesAsync) rather than a real condition. + var system = ActorSystem.Create($"loadharness-{siteId}", Akka.Configuration.ConfigurationFactory.ParseString( + "akka.loglevel = WARNING\nakka.stdout-loglevel = WARNING\nakka.log-dead-letters = 0\nakka.log-dead-letters-during-shutdown = off")); + + var streamManager = new SiteStreamManager(siteOptions, NullLogger.Instance); + streamManager.Initialize(system); + + var healthCollector = new SiteHealthCollector(); + healthCollector.SetActiveNode(true); + healthCollector.SetNodeHostname($"{siteId}-node-a"); + + var sfStorage = new StoreAndForwardStorage(localDb, NullLogger.Instance); + var storeAndForward = new StoreAndForwardService( + sfStorage, + new StoreAndForwardOptions(), + NullLogger.Instance, + siteId: siteId); + await storeAndForward.StartAsync(); + + // Real DCL, with the simulated adapter registered on the real factory via its + // documented RegisterAdapter extension point. + var loggerFactory = NullLoggerFactory.Instance; + var factory = new DataConnectionFactory(loggerFactory); + var connections = new List(); + factory.RegisterAdapter(SimulatedDataConnection.ProtocolName, _ => + { + var connection = new SimulatedDataConnection(); + lock (connections) connections.Add(connection); + return connection; + }); + + var dclManager = system.ActorOf( + Props.Create(() => new DataConnectionManagerActor( + factory, new DataConnectionOptions(), healthCollector, null, null)), + "data-connection-manager"); + + for (var c = 0; c < connectionsPerSite; c++) + { + dclManager.Tell(new CreateConnectionCommand( + ConnectionName: ConnectionName(c), + ProtocolType: SimulatedDataConnection.ProtocolName, + PrimaryConnectionDetails: new Dictionary + { + ["endpoint"] = $"sim://{siteId}/{c}", + [SimulatedDataConnection.ConnectionNameKey] = ConnectionName(c), + })); + } + + var tagPathsByConnection = new List>(); + for (var c = 0; c < connectionsPerSite; c++) + tagPathsByConnection.Add(new List()); + + var fixture = new SiteRuntimeFixture( + siteId, config, dataDirectory, localDbProvider, localDb, system, streamManager, + healthCollector, storeAndForward, dclManager, storage, compilationService, + sharedScriptLibrary, siteOptions, tagPathsByConnection, connections); + + return fixture; + } + + /// Deterministic connection name for a connection index. + /// Zero-based connection index. + /// The connection name used in configs and DCL commands. + public static string ConnectionName(int connectionIndex) => $"sim-conn-{connectionIndex:D2}"; + + /// + /// Creates this site's Instance Actors in production-shaped staggered batches + /// ( / + /// ) and records how long the + /// ramp took. This is the harness's stand-in for "deployment of 500 instances to + /// a site" — it exercises the same per-instance construction, config + /// deserialization, override load and DCL subscribe that a real deploy triggers. + /// + /// Cancels the ramp. + /// A task that completes when every instance actor exists. + public async Task StartInstancesAsync(CancellationToken cancellationToken) + { + var started = Stopwatch.StartNew(); + var connectionCount = TagPathsByConnection.Count; + + for (var i = 0; i < _config.InstancesPerSite; i++) + { + var instanceName = InstanceName(i); + var connectionIndex = i % connectionCount; + var connectionName = ConnectionName(connectionIndex); + + var attributes = new List(_config.TagsPerInstance); + for (var t = 0; t < _config.TagsPerInstance; t++) + { + var tagPath = $"{instanceName}.Tag{t:D3}"; + TagPathsByConnection[connectionIndex].Add(tagPath); + attributes.Add(new ResolvedAttribute + { + CanonicalName = $"Tag{t:D3}", + DataType = "Double", + DataSourceReference = tagPath, + BoundDataConnectionId = connectionIndex + 1, + BoundDataConnectionName = connectionName, + BoundDataConnectionProtocol = SimulatedDataConnection.ProtocolName, + }); + } + + var configuration = new FlattenedConfiguration + { + InstanceUniqueName = instanceName, + TemplateId = 1, + SiteId = 1, + Attributes = attributes, + Connections = new Dictionary + { + [connectionName] = new() + { + Protocol = SimulatedDataConnection.ProtocolName, + ConfigurationJson = "{}", + }, + }, + }; + + var configJson = JsonSerializer.Serialize(configuration); + var actor = System.ActorOf( + Props.Create(() => new InstanceActor( + instanceName, configJson, Storage, CompilationService, SharedScriptLibrary, + StreamManager, SiteOptions, NullLogger.Instance, + DataConnectionManager, HealthCollector, null, null)), + instanceName); + _instanceActors.Add(actor); + + // Production staggered-startup pacing (SiteRuntimeOptions defaults): + // batches of StartupBatchSize separated by StartupBatchDelayMs, which is + // exactly what DeploymentManagerActor does on a real site start. + if ((i + 1) % SiteOptions.StartupBatchSize == 0) + await Task.Delay(SiteOptions.StartupBatchDelayMs, cancellationToken); + } + + started.Stop(); + InstanceRampDuration = started.Elapsed; + + HealthCollector.SetInstanceCounts( + _config.InstancesPerSite, _config.InstancesPerSite, 0); + for (var c = 0; c < connectionCount; c++) + { + HealthCollector.UpdateTagResolution( + ConnectionName(c), TagPathsByConnection[c].Count, TagPathsByConnection[c].Count); + } + } + + /// Deterministic instance unique name for an instance index. + /// Zero-based instance index. + /// The instance unique name. + public string InstanceName(int instanceIndex) => $"{SiteId}-inst-{instanceIndex:D4}"; + + /// + /// Attaches live stream subscribers, + /// each built from the production pieces the gRPC server uses: a real + /// writing into a bounded DropOldest channel + /// of the production capacity, subscribed through the real + /// . Only the socket writer is replaced — + /// by a reader task under the harness's control, which is what makes the + /// slow-subscriber scenario possible at all. + /// + /// + /// Attaching at least one subscriber is also load-bearing: + /// PublishAttributeValueChanged short-circuits at zero subscribers, so an + /// unsubscribed site would publish nothing and measure nothing. + /// + /// + /// Histogram that receives end-to-end tag update latencies. + public void AttachStreamProbes(LatencyHistogram latency) + { + var stride = Math.Max(1, _config.InstancesPerSite / Math.Max(1, _config.StreamProbesPerSite)); + for (var p = 0; p < _config.StreamProbesPerSite; p++) + { + var instanceIndex = Math.Min(p * stride, _config.InstancesPerSite - 1); + var instanceName = InstanceName(instanceIndex); + var probe = StreamSubscriberProbe.Attach( + System, StreamManager, instanceName, $"{SiteId}-probe-{p:D2}", latency); + _probes.Add(probe); + } + } + + /// Total tag paths registered across this site's connections. + public int TotalTagPaths => TagPathsByConnection.Sum(list => list.Count); + + /// + public async ValueTask DisposeAsync() + { + foreach (var probe in _probes) + await probe.DisposeAsync(); + + await StoreAndForward.StopAsync(); + await System.Terminate(); + await _localDbProvider.DisposeAsync(); + + try + { + if (Directory.Exists(_dataDirectory)) + Directory.Delete(_dataDirectory, recursive: true); + } + catch (IOException) + { + // Best-effort cleanup of a temp directory; a lingering WAL handle is not + // a harness failure and must not mask the measured result. + } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs new file mode 100644 index 00000000..44b3332b --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/TagUpdateDriver.cs @@ -0,0 +1,175 @@ +using System.Diagnostics; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.LoadHarness; + +/// +/// Drives simulated tag value changes into a site's connections at a fixed nominal +/// rate, standing in for an OPC UA server's notification thread. +/// +/// +/// One driver task per data connection, because that mirrors production: each +/// DataConnectionActor has exactly one adapter feeding it, and every update +/// for that connection funnels through that actor's single mailbox. Sharding the +/// emitters differently would hide the per-connection-actor serialization point, +/// which is one of the things WP-4 needs to characterize. +/// +/// +/// The emitter walks the connection's tag list in slices sized so that one full pass +/// takes , then sleeps out the remainder +/// of each slice's budget. If a slice overruns its budget the driver does NOT try to +/// catch up — it records the shortfall in so the results +/// can say honestly whether the offered load was actually delivered. +/// +/// +public sealed class TagUpdateDriver : IAsyncDisposable +{ + private const int SlicesPerPeriod = 20; + + private readonly List _tasks = new(); + private readonly CancellationTokenSource _cts = new(); + private long _emitted; + private long _skippedNoCallback; + private long _lagTicks; + + /// Total tag value changes handed to adapter callbacks. + public long EmittedCount => Interlocked.Read(ref _emitted); + + /// + /// Emissions skipped because no Instance Actor had subscribed to that connection + /// yet (the adapter callback is captured at subscribe time). Non-zero only during + /// the ramp; a non-zero value in the steady window would mean lost offered load. + /// + public long SkippedNoCallback => Interlocked.Read(ref _skippedNoCallback); + + /// + /// Cumulative seconds by which emit slices overran their time budget, summed + /// across driver tasks. Large values mean the harness itself could not offer the + /// nominal rate and the measured throughput is driver-bound, not system-bound. + /// + public double EmitLagSeconds => Interlocked.Read(ref _lagTicks) / (double)Stopwatch.Frequency; + + /// + /// Starts one emitter task per connection across every site. + /// + /// The sites to drive. + /// Harness configuration supplying the update period. + /// The running driver. + public static TagUpdateDriver Start(IReadOnlyList sites, HarnessConfig config) + { + var driver = new TagUpdateDriver(); + foreach (var site in sites) + { + for (var c = 0; c < site.TagPathsByConnection.Count; c++) + { + var connectionIndex = c; + var tagPaths = site.TagPathsByConnection[connectionIndex].ToArray(); + driver._tasks.Add(Task.Run(() => driver.EmitLoopAsync( + site, connectionIndex, tagPaths, config.TagUpdatePeriod, driver._cts.Token))); + } + } + + return driver; + } + + private async Task EmitLoopAsync( + SiteRuntimeFixture site, + int connectionIndex, + string[] tagPaths, + TimeSpan period, + CancellationToken cancellationToken) + { + if (tagPaths.Length == 0) + return; + + var sliceBudget = period / SlicesPerPeriod; + var sliceSize = Math.Max(1, (int)Math.Ceiling(tagPaths.Length / (double)SlicesPerPeriod)); + var cursor = 0; + var sequence = 0d; + + while (!cancellationToken.IsCancellationRequested) + { + var sliceStart = Stopwatch.GetTimestamp(); + var callback = ResolveCallback(site, connectionIndex); + + if (callback == null) + { + Interlocked.Add(ref _skippedNoCallback, sliceSize); + } + else + { + var end = Math.Min(cursor + sliceSize, tagPaths.Length); + for (var i = cursor; i < end; i++) + { + // Stamped HERE: this instant rides TagValueUpdate.Timestamp all the + // way to the subscriber, so the probe's subtraction is a genuine + // end-to-end latency and not a re-stamped approximation. + callback(tagPaths[i], new TagValue(sequence, QualityCode.Good, DateTimeOffset.UtcNow)); + } + + Interlocked.Add(ref _emitted, end - cursor); + cursor = end; + } + + if (cursor >= tagPaths.Length) + { + cursor = 0; + sequence += 1d; + } + + var elapsed = Stopwatch.GetTimestamp() - sliceStart; + var budgetTicks = (long)(sliceBudget.TotalSeconds * Stopwatch.Frequency); + if (elapsed < budgetTicks) + { + var remaining = TimeSpan.FromSeconds((budgetTicks - elapsed) / (double)Stopwatch.Frequency); + try + { + await Task.Delay(remaining, cancellationToken); + } + catch (OperationCanceledException) + { + return; + } + } + else + { + Interlocked.Add(ref _lagTicks, elapsed - budgetTicks); + } + } + } + + /// + /// Resolves the live adapter callback for a connection, by NAME. The DCL factory + /// appends adapters as its manager actor processes CreateConnectionCommands, so + /// list position does not track connection index. + /// + private static SubscriptionCallback? ResolveCallback(SiteRuntimeFixture site, int connectionIndex) + { + var name = SiteRuntimeFixture.ConnectionName(connectionIndex); + // Last match wins: a reconnect would create a fresh adapter for the same name, + // and only the newest one holds the live subscription callback. + return site.Connections.LastOrDefault(c => c.ConnectionName == name)?.ValueCallback; + } + + private int _disposed; + + /// + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + await _cts.CancelAsync(); + try + { + await Task.WhenAll(_tasks); + } + catch (OperationCanceledException) + { + // Expected on shutdown. + } + + _cts.Dispose(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj new file mode 100644 index 00000000..480b3344 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.LoadHarness/ZB.MOM.WW.ScadaBridge.LoadHarness.csproj @@ -0,0 +1,40 @@ + + + + + net10.0 + Exe + enable + enable + true + false + ZB.MOM.WW.ScadaBridge.LoadHarness + + true + true + + + + + + + + + + + + + diff --git a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs new file mode 100644 index 00000000..b5dbb5fe --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/TargetScale/TargetScaleHarnessSmokeTests.cs @@ -0,0 +1,84 @@ +using ZB.MOM.WW.ScadaBridge.LoadHarness; + +namespace ZB.MOM.WW.ScadaBridge.PerformanceTests.TargetScale; + +/// +/// Keeps the target-scale load harness (deferred-work register #25 / Phase-8 WP-4) +/// honest at CI scale. +/// +/// +/// The full protocol — 10 sites x 500 instances x 75 tags sustained for 20 minutes — +/// deliberately lives in the standalone ZB.MOM.WW.ScadaBridge.LoadHarness +/// executable, NOT here: perf tests in this project run as part of an ordinary +/// dotnet test ZB.MOM.WW.ScadaBridge.slnx (the Category=Performance +/// trait enables a filter, it does not exclude by default), and a 20-minute test +/// would be intolerable there. What this test protects is that the harness still +/// compiles, wires up, and produces coherent measurements — so #25's evidence can be +/// regenerated on demand rather than bit-rotting. +/// +/// +/// Results doc: docs/plans/2026-08-15-target-scale-load-test-results.md. +/// Design memo: docs/plans/2026-08-15-target-scale-load-test-design.md. +/// +/// +public class TargetScaleHarnessSmokeTests +{ + /// + /// Runs the harness at ~1/1000th of target scale for a few seconds and asserts the + /// pipeline is intact end to end: tag updates reach live stream subscribers, the + /// central health aggregator tracks every site, debug snapshots answer, the + /// store-and-forward buffer drains to empty, and a stalled subscriber does not cost + /// the healthy ones any events. + /// + /// A task representing the test run. + [Trait("Category", "Performance")] + [Fact] + public async Task Harness_AtSmokeScale_ProducesCoherentMeasurements() + { + var config = new HarnessConfig + { + Sites = 2, + InstancesPerSite = 10, + TagsPerInstance = 5, + TagUpdatePeriod = TimeSpan.FromSeconds(1), + SettleDuration = TimeSpan.FromSeconds(5), + SustainDuration = TimeSpan.FromSeconds(15), + SampleInterval = TimeSpan.FromSeconds(2), + // Shortened from the production 30 s only because the smoke window is 20 s. + HealthReportInterval = TimeSpan.FromSeconds(2), + DebugProbeInterval = TimeSpan.FromSeconds(2), + SubscribeSettleDuration = TimeSpan.FromSeconds(5), + StreamProbesPerSite = 3, + StoreAndForwardDrainMessages = 200, + SlowSubscriberEvents = 2_000, + ResultsPath = Path.Combine(Path.GetTempPath(), $"loadharness-smoke-{Guid.NewGuid():N}.json"), + }; + + using var cancellation = new CancellationTokenSource(TimeSpan.FromMinutes(10)); + var result = await HarnessRun.ExecuteAsync(config, _ => { }, cancellation.Token); + + // Tag updates flowed all the way through DCL -> InstanceActor -> site stream -> + // StreamRelayActor -> bounded channel -> subscriber. + Assert.True(result.TagUpdateLatency.Count > 0, + "No tag update latency samples — the DCL -> stream -> subscriber path did not carry traffic."); + Assert.True(result.StreamProbeReceived > 0, "Live stream subscribers received nothing."); + Assert.Equal(0, result.DriverSkippedNoCallback); + + // Observability held up. + Assert.Equal(config.Sites, result.SitesTrackedByAggregator); + Assert.True(result.DebugSnapshotsCompleted > 0, "No debug snapshot completed."); + Assert.Equal(0, result.DebugSnapshotTimeouts); + + // Store-and-forward drained completely (register row 50, first half). + Assert.NotNull(result.StoreAndForwardDrain); + Assert.Equal(0, result.StoreAndForwardDrain!.ResidualDepth); + Assert.True(result.StoreAndForwardDrain.DrainPerSecond > 0); + + // A stalled subscriber costs the healthy ones nothing (register row 50, second + // half). This is the design's isolation claim, asserted rather than assumed. + Assert.NotNull(result.SlowSubscriber); + Assert.Equal(1.0, result.SlowSubscriber!.HealthyMinDeliveryRatio, precision: 2); + Assert.True(result.SlowSubscriber.SlowDeliveryRatio < 1.0, + "The deliberately stalled subscriber kept up, so the probe proved nothing."); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj index 02f7b2fa..0ddcfaac 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj +++ b/tests/ZB.MOM.WW.ScadaBridge.PerformanceTests/ZB.MOM.WW.ScadaBridge.PerformanceTests.csproj @@ -31,6 +31,9 @@ + +