Joseph Doherty
2b80a6defa
New ResidualBench tool (eventlog, flatten, audit, audit-contention) and docker/measure-ui-responsiveness.sh. All four residuals pass; results in docs/plans/2026-11-10-wp4-residual-scope-results.md. Findings filed as #122, #123 and #124 (fixed in the following commit).
222 lines
12 KiB
C#
222 lines
12 KiB
C#
using System.Diagnostics;
|
||
using Microsoft.Data.SqlClient;
|
||
using Microsoft.EntityFrameworkCore;
|
||
using ZB.MOM.WW.Audit;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
|
||
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.ResidualBench;
|
||
|
||
/// <summary>
|
||
/// [xc-4] Audit logging does not degrade central performance. Creates a throwaway, fully
|
||
/// migrated configuration database (the production schema: partitioned <c>AuditLog</c>,
|
||
/// clustered <c>(OccurredAtUtc, EventId)</c>, every NC index), seeds it through the real
|
||
/// <see cref="AuditLogRepository.InsertManyIfNotExistsAsync"/> ingest path, then runs a fixed
|
||
/// central read mix while site-shaped ingest packets arrive at increasing offered rates.
|
||
/// </summary>
|
||
internal static class AuditBench
|
||
{
|
||
internal const string DefaultAdmin =
|
||
"Server=localhost,1433;User Id=sa;Password=ScadaBridge_Dev1#;TrustServerCertificate=true;Encrypt=false";
|
||
|
||
private const int PacketSize = 256; // rows per site ingest packet
|
||
|
||
public static async Task<int> RunAsync(BenchArgs args)
|
||
{
|
||
var admin = args.Get("connection-string", DefaultAdmin);
|
||
var seedRows = args.GetInt("seed-rows", 1_000_000);
|
||
var phaseSeconds = args.GetInt("phase-seconds", 60);
|
||
var keep = args.Get("keep", "false") == "true";
|
||
args.RejectUnknown();
|
||
|
||
var dbName = $"ScadaBridgeResidualBench_{Guid.NewGuid():N}"[..40];
|
||
var cs = new SqlConnectionStringBuilder(admin) { InitialCatalog = dbName }.ConnectionString;
|
||
await ExecAsync(admin, $"CREATE DATABASE [{dbName}]; ALTER DATABASE [{dbName}] SET RECOVERY SIMPLE;");
|
||
Console.WriteLine($"# database {dbName}");
|
||
try
|
||
{
|
||
var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
|
||
.UseSqlServer(cs, sql => sql.CommandTimeout(1800)).Options;
|
||
var sw = Stopwatch.StartNew();
|
||
await using (var ctx = new ScadaBridgeDbContext(options)) await ctx.Database.MigrateAsync();
|
||
Console.WriteLine($"migrated in {sw.Elapsed.TotalSeconds:F0} s");
|
||
|
||
// ---- seed: also the unthrottled ingest ceiling on a growing table ----------------
|
||
var now = DateTime.UtcNow;
|
||
sw.Restart();
|
||
var seeded = 0L;
|
||
// ONE writer: production ingest is the AuditLogIngestActor cluster singleton, which
|
||
// awaits each packet inside ReceiveAsync, so central never runs two site-ingest
|
||
// inserts at once. (Four concurrent writers here deadlock — error 1205 — see results doc.)
|
||
var packets = (seedRows + PacketSize - 1) / PacketSize;
|
||
for (var packet = 0; packet < packets; packet++)
|
||
{
|
||
var rows = Enumerable.Range(0, PacketSize)
|
||
.Select(i => Row((long)packet * PacketSize + i, now.AddDays(-45).AddSeconds(45.0 * 86_400 * ((long)packet * PacketSize + i) / seedRows)))
|
||
.ToList();
|
||
seeded += await InsertAsync(options, rows);
|
||
}
|
||
Console.WriteLine($"seeded {seeded:N0} rows (45 days) in {sw.Elapsed.TotalSeconds:F0} s = {seeded / sw.Elapsed.TotalSeconds:N0} rows/s (1 serial writer, {PacketSize}-row packets)");
|
||
Console.WriteLine($"AuditLog size {await ScalarAsync(cs, "SELECT SUM(used_page_count) * 8 / 1024 FROM sys.dm_db_partition_stats WHERE object_id = OBJECT_ID('dbo.AuditLog')")} MB");
|
||
|
||
// ---- phases ------------------------------------------------------------------
|
||
Console.WriteLine();
|
||
var phases = new (string Name, int RowsPerSec)[] { ("no ingest", 0), ("ingest 100 rows/s", 100), ("ingest 500 rows/s", 500), ("ingest 2,000 rows/s", 2_000), ("ingest unthrottled (1 serial writer)", -1) };
|
||
foreach (var (name, rate) in phases)
|
||
await RunPhaseAsync(options, name, rate, TimeSpan.FromSeconds(phaseSeconds));
|
||
}
|
||
finally
|
||
{
|
||
SqlConnection.ClearAllPools();
|
||
if (!keep)
|
||
await ExecAsync(admin, $"ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; DROP DATABASE [{dbName}];");
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
private static readonly (string Name, Func<ScadaBridgeDbContext, Task> Run)[] Reads =
|
||
[
|
||
("Audit Log page 1 (newest 50, no filter)", ctx => new AuditLogRepository(ctx).QueryAsync(new AuditLogQueryFilter(), new AuditLogPaging(50))),
|
||
("channel=ApiOutbound, last 24 h, 50", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||
new AuditLogQueryFilter(Channels: [AuditChannel.ApiOutbound], FromUtc: DateTime.UtcNow.AddDays(-1)), new AuditLogPaging(50))),
|
||
("site=site-07, last 7 d, 100", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||
new AuditLogQueryFilter(SourceSiteIds: ["site-07"], FromUtc: DateTime.UtcNow.AddDays(-7)), new AuditLogPaging(100))),
|
||
("status=Failed, last 24 h, 50", ctx => new AuditLogRepository(ctx).QueryAsync(
|
||
new AuditLogQueryFilter(Statuses: [AuditStatus.Failed], FromUtc: DateTime.UtcNow.AddDays(-1)), new AuditLogPaging(50))),
|
||
("KPI snapshot, 1 h window", ctx => new AuditLogRepository(ctx).GetKpiSnapshotAsync(TimeSpan.FromHours(1))),
|
||
("non-audit: templates + sites list", async ctx =>
|
||
{
|
||
await ctx.Templates.AsNoTracking().Take(100).ToListAsync();
|
||
await ctx.Sites.AsNoTracking().ToListAsync();
|
||
}),
|
||
];
|
||
|
||
private static async Task RunPhaseAsync(DbContextOptions<ScadaBridgeDbContext> options, string name, int rowsPerSec, TimeSpan duration)
|
||
{
|
||
using var cts = new CancellationTokenSource(duration);
|
||
var ingested = 0L;
|
||
var packetLatency = new Samples();
|
||
var next = 10_000_000L + Random.Shared.Next(1_000_000) * 100L;
|
||
|
||
async Task Writer(int writers, int index)
|
||
{
|
||
var sw = Stopwatch.StartNew();
|
||
var sent = 0L;
|
||
while (!cts.IsCancellationRequested)
|
||
{
|
||
if (rowsPerSec > 0)
|
||
{
|
||
// Paced: one packet whenever this writer falls behind its share of the offered rate.
|
||
var due = (long)(sw.Elapsed.TotalSeconds * rowsPerSec / writers);
|
||
if (sent + PacketSize > due) { await Task.Delay(2); continue; }
|
||
}
|
||
var baseId = Interlocked.Add(ref next, PacketSize);
|
||
var rows = Enumerable.Range(0, PacketSize).Select(i => Row(baseId + i, DateTime.UtcNow.AddSeconds(-Random.Shared.Next(0, 30)))).ToList();
|
||
var t = Stopwatch.GetTimestamp();
|
||
Interlocked.Add(ref ingested, await InsertAsync(options, rows));
|
||
packetLatency.Add(Stopwatch.GetElapsedTime(t).TotalMilliseconds);
|
||
sent += PacketSize;
|
||
}
|
||
}
|
||
|
||
var writerCount = rowsPerSec == 0 ? 0 : 1;
|
||
var writers = Enumerable.Range(0, writerCount).Select(i => Task.Run(() => Writer(writerCount, i))).ToList();
|
||
|
||
// One reader looping the read mix back to back — a busy operator's UI plus the health poll.
|
||
var latencies = Reads.ToDictionary(r => r.Name, _ => new Samples());
|
||
var started = Stopwatch.StartNew();
|
||
while (!cts.IsCancellationRequested)
|
||
{
|
||
foreach (var (readName, run) in Reads)
|
||
{
|
||
var t = Stopwatch.GetTimestamp();
|
||
await using var ctx = new ScadaBridgeDbContext(options);
|
||
await run(ctx);
|
||
latencies[readName].Add(Stopwatch.GetElapsedTime(t).TotalMilliseconds);
|
||
}
|
||
}
|
||
await Task.WhenAll(writers);
|
||
var seconds = started.Elapsed.TotalSeconds;
|
||
|
||
Console.WriteLine($"### {name}: achieved ingest {ingested / seconds:N0} rows/s, deadlock retries so far {Interlocked.Read(ref _deadlocks)}" +
|
||
(packetLatency.Count > 0 ? $"; {PacketSize}-row packet {packetLatency.Summary()}" : ""));
|
||
Console.WriteLine("| read | P50 ms | P95 ms | P99 ms | max ms | n |");
|
||
Console.WriteLine("|---|---|---|---|---|---|");
|
||
foreach (var (readName, s) in latencies)
|
||
Console.WriteLine($"| {readName} | {s.Percentile(50):F1} | {s.Percentile(95):F1} | {s.Percentile(99):F1} | {s.Max:F1} | {s.Count} |");
|
||
Console.WriteLine();
|
||
}
|
||
|
||
private static readonly string[] Targets = ["MES.MoveIn", "MES.MoveOut", "Historian.Write", "ERP.Order", "Delmia.Recipe"];
|
||
|
||
/// <summary>Site-shaped row: ~1.5 KB of request/response summary + extra, 10 sites × 500 instances.</summary>
|
||
internal static AuditEvent Row(long i, DateTime occurredAtUtc)
|
||
{
|
||
var channel = (i % 10) switch { < 5 => AuditChannel.ApiOutbound, < 8 => AuditChannel.DbOutbound, < 9 => AuditChannel.Notification, _ => AuditChannel.ApiInbound };
|
||
var kind = channel switch
|
||
{
|
||
AuditChannel.ApiOutbound => i % 3 == 0 ? AuditKind.ApiCallCached : AuditKind.ApiCall,
|
||
AuditChannel.DbOutbound => i % 3 == 0 ? AuditKind.DbWriteCached : AuditKind.DbWrite,
|
||
AuditChannel.Notification => AuditKind.NotifySend,
|
||
_ => AuditKind.InboundRequest,
|
||
};
|
||
var failed = i % 25 == 0;
|
||
var site = $"site-{i % 10:D2}";
|
||
return ScadaBridgeAuditEventFactory.Create(
|
||
channel, kind, failed ? AuditStatus.Failed : AuditStatus.Delivered,
|
||
occurredAtUtc: occurredAtUtc,
|
||
actor: $"script:Machine-{i % 500:D4}",
|
||
target: Targets[i % Targets.Length],
|
||
sourceNode: i % 2 == 0 ? "node-a" : "node-b",
|
||
correlationId: kind is AuditKind.ApiCallCached or AuditKind.DbWriteCached ? Guid.NewGuid() : null,
|
||
executionId: Guid.NewGuid(),
|
||
sourceSiteId: site,
|
||
sourceInstanceId: $"Machine-{i % 500:D4}",
|
||
sourceScript: $"Recipe{i % 11}",
|
||
httpStatus: channel is AuditChannel.ApiOutbound or AuditChannel.ApiInbound ? (failed ? 503 : 200) : null,
|
||
durationMs: (int)(i % 400),
|
||
errorMessage: failed ? "Service Unavailable: upstream timeout after 30000 ms" : null,
|
||
requestSummary: "{\"workOrder\":\"WO-" + i + "\",\"machine\":\"Machine-" + (i % 500).ToString("D4") + "\",\"items\":[" +
|
||
string.Join(",", Enumerable.Range(0, 8).Select(k => "{\"sku\":\"SKU-" + k + "\",\"qty\":" + (k * 3) + "}")) + "]}",
|
||
responseSummary: failed ? null : "{\"accepted\":true,\"reference\":\"" + Guid.NewGuid() + "\",\"message\":\"Recorded\"}",
|
||
extra: "{\"retry\":0,\"bufferDepth\":" + (i % 50) + "}");
|
||
}
|
||
|
||
private static long _deadlocks;
|
||
|
||
/// <summary>One ingest packet through the production repository; a deadlock victim (1205) is retried, as the site would re-send it.</summary>
|
||
private static async Task<int> InsertAsync(DbContextOptions<ScadaBridgeDbContext> options, IReadOnlyList<AuditEvent> rows)
|
||
{
|
||
for (var attempt = 1; ; attempt++)
|
||
{
|
||
try
|
||
{
|
||
await using var ctx = new ScadaBridgeDbContext(options);
|
||
return await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows);
|
||
}
|
||
catch (SqlException ex) when (ex.Number == 1205 && attempt < 5)
|
||
{
|
||
Interlocked.Increment(ref _deadlocks);
|
||
}
|
||
}
|
||
}
|
||
|
||
internal static async Task ExecAsync(string cs, string sql)
|
||
{
|
||
await using var c = new SqlConnection(cs);
|
||
await c.OpenAsync();
|
||
await using var cmd = new SqlCommand(sql, c) { CommandTimeout = 300 };
|
||
await cmd.ExecuteNonQueryAsync();
|
||
}
|
||
|
||
internal static async Task<object?> ScalarAsync(string cs, string sql)
|
||
{
|
||
await using var c = new SqlConnection(cs);
|
||
await c.OpenAsync();
|
||
await using var cmd = new SqlCommand(sql, c) { CommandTimeout = 300 };
|
||
return await cmd.ExecuteScalarAsync();
|
||
}
|
||
}
|