using System.Diagnostics; using System.Reflection; using Microsoft.Data.Sqlite; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.LocalDb; using ZB.MOM.WW.ScadaBridge.Commons.Messages.RemoteQuery; using ZB.MOM.WW.ScadaBridge.SiteEventLogging; namespace ZB.MOM.WW.ScadaBridge.ResidualBench; /// /// [xc-3] Site event logging handles high event volume within the 30-day / 1 GB limits. /// Everything runs over a real ZB.MOM.WW.LocalDb file with the production /// , and /// . /// /// /// Phases: /// /// calibrate — bytes per row on disk for the representative event mix, so /// rows-at-cap and the sustained event rate that fits 30 days inside 1 GB. /// write — writer drain rate and enqueue→durable latency, with and without /// the LocalDb CDC capture triggers (RegisterReplicated), plus storm bursts against /// the bounded write queue. /// full — a table filled to the 1 GB cap with 30 days of history: query /// shapes through the real query service, and writer latency while queries run. /// purge — one real purge tick over a table 10% over the cap. /// /// internal static class EventLogBench { private const int Instances = 500; private const long CapBytes = 1024L * 1024 * 1024; public static async Task RunAsync(BenchArgs args) { var dataDir = args.Get("data-dir", Path.Combine(Path.GetTempPath(), "sb-residual-eventlog")); var calibrationRows = args.GetInt("calibration-rows", 200_000); var writeEvents = args.GetInt("write-events", 200_000); var skipFull = args.Get("skip-full", "false") == "true"; args.RejectUnknown(); Directory.CreateDirectory(dataDir); Console.WriteLine($"# data dir {dataDir}"); // ---- 1. calibrate ------------------------------------------------------------ double bytesPerRow; { var db = Open(Path.Combine(dataDir, "calibrate.db"), out var path); using var logger = NewLogger(db, new SiteEventLogOptions()); var before = UsedBytes(db); Seed(db, calibrationRows, DateTimeOffset.UtcNow.AddDays(-30), DateTimeOffset.UtcNow); var after = UsedBytes(db); bytesPerRow = (after - before) / (double)calibrationRows; var rowsAtCap = (long)(CapBytes / bytesPerRow); var ratePerSec = rowsAtCap / (30.0 * 86_400); Console.WriteLine(); Console.WriteLine("## calibrate"); Console.WriteLine($"rows seeded {calibrationRows:N0}"); Console.WriteLine($"bytes per row (on disk) {bytesPerRow:F1}"); Console.WriteLine($"rows at the 1 GB cap {rowsAtCap:N0}"); Console.WriteLine($"30 days fit in 1 GB at <= {ratePerSec:F3} events/s sustained per site " + $"({ratePerSec * 86_400:N0}/day, {ratePerSec * 86_400 / Instances:F1}/instance/day at {Instances} instances)"); Console.WriteLine($"days retained at 5 ev/s {rowsAtCap / (5.0 * 86_400):F1}"); Console.WriteLine($"days retained at 20 ev/s {rowsAtCap / (20.0 * 86_400):F1}"); logger.Dispose(); Cleanup(db, path); } // ---- 2. write throughput ----------------------------------------------------- Console.WriteLine(); Console.WriteLine("## write"); foreach (var replicated in new[] { false, true }) { var db = Open(Path.Combine(dataDir, $"write-{replicated}.db"), out var path); using var logger = NewLogger(db, new SiteEventLogOptions()); if (replicated) db.RegisterReplicated("site_events"); await MeasureDrainAsync(logger, writeEvents, replicated ? "CDC on " : "CDC off"); foreach (var burst in new[] { 5_000, 10_000, 25_000 }) await MeasureBurstAsync(logger, burst, replicated ? "CDC on " : "CDC off"); logger.Dispose(); Cleanup(db, path); } if (skipFull) return 0; // ---- 3. a table at the cap --------------------------------------------------- Console.WriteLine(); Console.WriteLine("## full (1 GB, 30 days)"); var fullRows = (long)(CapBytes / bytesPerRow); { var db = Open(Path.Combine(dataDir, "full.db"), out var path); var options = new SiteEventLogOptions(); using var logger = NewLogger(db, options); var now = DateTimeOffset.UtcNow; var sw = Stopwatch.StartNew(); Seed(db, fullRows, now.AddDays(-30), now); Console.WriteLine($"seeded {fullRows:N0} rows in {sw.Elapsed.TotalSeconds:F0} s; used {UsedBytes(db) / 1048576.0:F0} MiB, file {new FileInfo(path).Length / 1048576.0:F0} MiB"); var query = new EventLogQueryService(logger, Options.Create(options), NullLogger.Instance); RunQueries(query, now); await MeasureWriterUnderQueriesAsync(logger, query, now); // ---- 4. purge tick over the cap -------------------------------------------- Console.WriteLine(); Console.WriteLine("## purge (cap trim, 10% over)"); Seed(db, fullRows / 10, now.AddDays(-29), now); var purge = new EventLogPurgeService(logger, Options.Create(options), NullLogger.Instance); var runPurge = typeof(EventLogPurgeService).GetMethod("RunPurge", BindingFlags.Instance | BindingFlags.NonPublic) ?? throw new MissingMethodException("EventLogPurgeService.RunPurge"); var countBefore = Count(db); sw.Restart(); runPurge.Invoke(purge, null); var purgeMs = sw.Elapsed.TotalMilliseconds; Console.WriteLine($"rows {countBefore:N0} -> {Count(db):N0} in {purgeMs / 1000:F1} s (ceiling {options.MaxPurgedRowsPerTick:N0}/tick)"); logger.Dispose(); Cleanup(db, path); } return 0; } private static async Task MeasureDrainAsync(SiteEventLogger logger, int total, string label) { // Windows below the queue capacity, so nothing is dropped: measures the writer's drain rate. const int window = 5_000; var latency = new Samples(); var sw = Stopwatch.StartNew(); var failedBefore = logger.FailedWriteCount; for (var offset = 0; offset < total; offset += window) { var tasks = new Task[window]; for (var i = 0; i < window; i++) { var start = Stopwatch.GetTimestamp(); var e = RowFor(offset + i); tasks[i] = logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details) .ContinueWith(_ => latency.Add(Stopwatch.GetElapsedTime(start).TotalMilliseconds), TaskScheduler.Default); } await Task.WhenAll(tasks); } var seconds = sw.Elapsed.TotalSeconds; Console.WriteLine($"{label} drain: {total:N0} events in {seconds:F2} s = {total / seconds:N0} events/s; " + $"failed {logger.FailedWriteCount - failedBefore}; enqueue->durable {latency.Summary()}"); } private static async Task MeasureBurstAsync(SiteEventLogger logger, int burst, string label) { // A storm offered all at once (e.g. every alarm on a site flipping on one data-connection drop). var failedBefore = logger.FailedWriteCount; var sw = Stopwatch.StartNew(); var tasks = new Task[burst]; for (var i = 0; i < burst; i++) { var e = RowFor(i); tasks[i] = logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details); } try { await Task.WhenAll(tasks); } catch { /* dropped events fault their task; counted below */ } Console.WriteLine($"{label} burst {burst,6:N0} at once: all settled in {sw.Elapsed.TotalMilliseconds:F0} ms, " + $"dropped {logger.FailedWriteCount - failedBefore:N0} (queue capacity {new SiteEventLogOptions().WriteQueueCapacity:N0})"); } private static void RunQueries(EventLogQueryService query, DateTimeOffset now) { var shapes = new (string Name, Func Build)[] { ("no filter, page 1 (oldest first)", t => Req(null, null, null, null, null, null, t)), ("last 1 h", t => Req(now.AddHours(-1), null, null, null, null, null, t)), ("last 24 h", t => Req(now.AddDays(-1), null, null, null, null, null, t)), ("severity=Error, 30 d", t => Req(null, null, null, "Error", null, null, t)), ("severity=Error, last 24 h", t => Req(now.AddDays(-1), null, null, "Error", null, null, t)), ("event_type=alarm, last 24 h", t => Req(now.AddDays(-1), null, "alarm", null, null, null, t)), ("instance=Machine-0042, 30 d", t => Req(null, null, null, null, "Machine-0042", null, t)), ("keyword 'timeout' (clamped to 7 d)", t => Req(null, null, null, null, null, "timeout", t)), ("keyword with no match (full 7 d window scan)", t => Req(null, null, null, null, null, "no-such-text", t)), ("keyword 'Machine-0042', last 24 h", t => Req(now.AddDays(-1), null, null, null, null, "Machine-0042", t)), }; Console.WriteLine(); Console.WriteLine("| query (page size 500) | first ms | warm P50 ms | warm max ms | rows | has more |"); Console.WriteLine("|---|---|---|---|---|---|"); foreach (var (name, build) in shapes) { EventLogQueryResponse? response = null; var first = Samples.Time(() => response = query.ExecuteQuery(build(null))); var warm = new Samples(); for (var i = 0; i < 5; i++) warm.Add(Samples.Time(() => query.ExecuteQuery(build(null)))); if (response is { Success: false }) throw new InvalidOperationException($"{name}: {response.ErrorMessage}"); Console.WriteLine($"| {name} | {first:F1} | {warm.Percentile(50):F1} | {warm.Max:F1} | {response!.Entries.Count} | {response.HasMore} |"); } // Paging through: 20 consecutive pages of the last 24 h. string? token = null; var pages = new Samples(); for (var i = 0; i < 20; i++) { EventLogQueryResponse? r = null; pages.Add(Samples.Time(() => r = query.ExecuteQuery(Req(now.AddDays(-1), null, null, null, null, null, token)))); token = r!.ContinuationToken; if (token is null) break; } Console.WriteLine($"| 20 keyset pages, last 24 h | | {pages.Percentile(50):F1} | {pages.Max:F1} | 500/page | |"); } private static async Task MeasureWriterUnderQueriesAsync(SiteEventLogger logger, EventLogQueryService query, DateTimeOffset now) { // Offered 200 events/s (≈ 190x the rate that fits 30 days in 1 GB) for 20 s, first idle, // then with a back-to-back worst-case keyword query loop on another thread. Console.WriteLine(); foreach (var withQueries in new[] { false, true }) { using var cts = new CancellationTokenSource(); var queriesRun = 0; var queryLoop = withQueries ? Task.Run(() => { while (!cts.IsCancellationRequested) { query.ExecuteQuery(Req(null, null, null, null, null, "no-such-text", null)); queriesRun++; } }) : Task.CompletedTask; var latency = new Samples(); var tasks = new List(); var failedBefore = logger.FailedWriteCount; var sw = Stopwatch.StartNew(); var sent = 0; while (sw.Elapsed < TimeSpan.FromSeconds(20)) { var due = (int)(sw.Elapsed.TotalSeconds * 200); for (; sent < due; sent++) { var start = Stopwatch.GetTimestamp(); var e = RowFor(sent); tasks.Add(logger.LogEventAsync(e.Type, e.Severity, e.Instance, e.Source, e.Message, e.Details) .ContinueWith(_ => latency.Add(Stopwatch.GetElapsedTime(start).TotalMilliseconds), TaskScheduler.Default)); } await Task.Delay(5); } await Task.WhenAll(tasks); cts.Cancel(); await queryLoop; Console.WriteLine($"writer at 200 ev/s, {(withQueries ? $"with a back-to-back no-match keyword scan loop ({queriesRun} scans)" : "idle")}: " + $"enqueue->durable {latency.Summary()}; failed {logger.FailedWriteCount - failedBefore}"); } } private static EventLogQueryRequest Req( DateTimeOffset? from, DateTimeOffset? to, string? type, string? severity, string? instance, string? keyword, string? token) => new("bench", "site-bench", from, to, type, severity, instance, keyword, token, 500, DateTimeOffset.UtcNow); // ---- representative event mix -------------------------------------------------------- internal readonly record struct Row(string Type, string Severity, string? Instance, string Source, string Message, string? Details); private static readonly string StackDetails = "System.TimeoutException: The operation has timed out.\n" + string.Concat(Enumerable.Range(0, 8).Select(i => $" at ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts.ScriptRuntimeContext.Step{i}Async(String path, CancellationToken ct)\n")); /// /// Deterministic mix modelled on the post-WP3.2 event stream (per-run script events off, an /// interval summary instead): alarm transitions dominate, then script errors, connection /// changes, S&F and lifecycle. 2-in-3 rows carry an instance; 1-in-10 carry a details payload. /// internal static Row RowFor(long i) { var instance = i % 3 == 2 ? null : $"Machine-{i % Instances:D4}"; var details = i % 10 == 0 ? StackDetails : null; return (i % 20) switch { < 9 => new Row("alarm", i % 9 == 0 ? "Warning" : "Info", instance, $"AlarmActor:HighTemp{i % 7}", $"Alarm 'HighTemp{i % 7}' on '{instance ?? "site"}' changed to {(i % 2 == 0 ? "Active" : "Normal")}", details), < 14 => new Row("script", i % 5 == 0 ? "Error" : "Warning", instance, $"ScriptActor:Recipe{i % 11}", $"Script 'Recipe{i % 11}' failed: timeout waiting for attribute 'Done' after 30000 ms", details), < 17 => new Row("connection", i % 3 == 0 ? "Warning" : "Info", null, $"DataConnectionActor:PLC-{i % 40:D2}", $"Connection 'PLC-{i % 40:D2}' quality changed to {(i % 2 == 0 ? "Bad" : "Good")}", details), < 19 => new Row("store_and_forward", "Warning", instance, "StoreAndForwardService", $"Message parked after 50 retries to target 'MES-{i % 4}'", details), _ => new Row("instance_lifecycle", "Info", instance, "DeploymentManagerActor", $"Instance '{instance ?? "Machine-0000"}' started", null), }; } // ---- LocalDb plumbing ---------------------------------------------------------------- private static ILocalDb Open(string path, out string resolvedPath) { foreach (var suffix in new[] { "", "-wal", "-shm" }) File.Delete(path + suffix); var configuration = new ConfigurationBuilder() .AddInMemoryCollection(new Dictionary { ["LocalDb:Path"] = path }) .Build(); var provider = new ServiceCollection().AddZbLocalDb(configuration).BuildServiceProvider(); resolvedPath = path; Providers[path] = provider; return provider.GetRequiredService(); } private static readonly Dictionary Providers = new(); private static void Cleanup(ILocalDb db, string path) { if (Providers.Remove(path, out var provider)) provider.Dispose(); foreach (var suffix in new[] { "", "-wal", "-shm" }) { try { File.Delete(path + suffix); } catch { /* best effort */ } } } private static SiteEventLogger NewLogger(ILocalDb db, SiteEventLogOptions options) => new(Options.Create(options), NullLogger.Instance, db); private static long UsedBytes(ILocalDb db) { using var c = db.CreateConnection(); long Scalar(string sql) { using var cmd = c.CreateCommand(); cmd.CommandText = sql; return (long)cmd.ExecuteScalar()!; } using (var cp = c.CreateCommand()) { cp.CommandText = "PRAGMA wal_checkpoint(TRUNCATE)"; cp.ExecuteNonQuery(); } return (Scalar("PRAGMA page_count") - Scalar("PRAGMA freelist_count")) * Scalar("PRAGMA page_size"); } private static long Count(ILocalDb db) { using var c = db.CreateConnection(); using var cmd = c.CreateCommand(); cmd.CommandText = "SELECT COUNT(*) FROM site_events"; return (long)cmd.ExecuteScalar()!; } /// /// Bulk-seeds rows with timestamps spread evenly over [from, to] — the logger always stamps /// "now", so history has to be written directly. Same columns and formats as the logger. /// private static void Seed(ILocalDb db, long rows, DateTimeOffset from, DateTimeOffset to) { using var c = db.CreateConnection(); var span = (to - from).Ticks; const int batch = 20_000; for (long offset = 0; offset < rows; offset += batch) { using var tx = c.BeginTransaction(); using var cmd = c.CreateCommand(); cmd.Transaction = tx; cmd.CommandText = """ INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details) VALUES ($id, $ts, $type, $sev, $inst, $src, $msg, $det) """; var pId = cmd.Parameters.Add("$id", SqliteType.Text); var pTs = cmd.Parameters.Add("$ts", SqliteType.Text); var pType = cmd.Parameters.Add("$type", SqliteType.Text); var pSev = cmd.Parameters.Add("$sev", SqliteType.Text); var pInst = cmd.Parameters.Add("$inst", SqliteType.Text); var pSrc = cmd.Parameters.Add("$src", SqliteType.Text); var pMsg = cmd.Parameters.Add("$msg", SqliteType.Text); var pDet = cmd.Parameters.Add("$det", SqliteType.Text); cmd.Prepare(); var end = Math.Min(rows, offset + batch); for (var i = offset; i < end; i++) { var r = RowFor(i); pId.Value = Guid.NewGuid().ToString("N"); pTs.Value = from.AddTicks(span / rows * i).ToUniversalTime().ToString("o"); pType.Value = r.Type; pSev.Value = r.Severity; pInst.Value = (object?)r.Instance ?? DBNull.Value; pSrc.Value = r.Source; pMsg.Value = r.Message; pDet.Value = (object?)r.Details ?? DBNull.Value; cmd.ExecuteNonQuery(); } tx.Commit(); } } }