dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(sitelog): sampled per-run events; interval run summaries; site_events replication policy pinned

Browse Source 
Implements WP3.2 stage (b) per docs/plans/2026-08-15-site-events-policy-design.md.

- Per-run instance-script Started/Completed Info site events are now off by
  default (SiteRuntimeOptions.PerRunScriptEvents=false) instead of firing on
  every run, closing the dominant site_events writer. Gated at the ScriptRunLauncher
  call sites (moved there from ScriptExecutionActor by WP3.1). Error-level events
  (timeout/failure/stuck-watchdog/recursion-limit) remain unconditional.
- ScriptRunSummaryRecorder accumulates per-(instance, script) run counters and a
  new site-only ScriptRunSummaryFlushService emits one aggregate "script" Info
  site event per ScriptRunSummaryIntervalSeconds (default 300s), top-50-script
  breakdown with an "others" rollup, zero-activity intervals emit nothing.
- Per-script opt-in via PerRunScriptEventScripts ("Instance/Script" exact or
  "Instance/*" wildcard), matched by the new pure ScriptRunEventPolicy. All three
  options are read from IOptionsMonitor<SiteRuntimeOptions> per run, so the
  policy is hot-togglable without a restart.
- Fixed the stale "event log is not replicated" comment at AkkaHostedService.cs
  (~905): site_events IS registered in SiteLocalDbSetup.ReplicatedTables — the
  singleton is what makes queries always hit the actively-written copy;
  replication is what gives the singleton history to read after a failover
  (memo Decision (b)). site_events replication itself is unchanged (still
  registered) and already pinned by
  tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SiteLocalDbCdcRegistrationTests.cs.
- Updated Component-SiteEventLogging.md (Volume Policy section, corrected
  Storage/replication rationale) and Component-SiteRuntime.md (Script Run
  Launch + Error Handling sections).
This commit is contained in:
Joseph Doherty
2026-08-14 22:56:08 -04:00
parent 799fd041ec
commit c254d0740e
15 changed files with 922 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Scripts/ScriptRunSummaryRecorder.cs
+205
View File
@@ -0,0 +1,205 @@
using System.Collections.Concurrent;
using System.Globalization;
using System.Text.Json;
using ZB.MOM.WW.ScadaBridge.SiteEventLogging;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
/// <summary>
/// WP3.2 (site_events volume policy — docs/plans/2026-08-15-site-events-policy-design.md):
/// accumulates per-(instance, script) run counters between flushes and emits ONE aggregate
/// "script" Info site event per interval instead of the two per-run Started/Completed rows
/// that were the dominant <c>site_events</c> writer. Registered as a process-wide singleton
/// (<c>ServiceCollectionExtensions.AddSiteRuntime</c>); incremented from the same
/// <see cref="ScriptRunLauncher"/> call sites that used to emit the per-run events, flushed
/// on a timer by <c>ScriptRunSummaryFlushService</c>.
///
/// <para>Lock-free: each script's counters are a small class of <c>long</c> fields mutated
/// via <see cref="Interlocked"/>, reached through a <see cref="ConcurrentDictionary{TKey,TValue}"/>
/// that <see cref="FlushAsync"/> atomically swaps out (<see cref="Interlocked.Exchange"/>) so a
/// concurrent increment either lands in the snapshot being flushed or the fresh one that
/// replaces it — never lost, never double-counted, and the flush never blocks a recording
/// call (or vice versa).</para>
/// </summary>
public sealed class ScriptRunSummaryRecorder
{
/// <summary>
/// Per-script breakdown cap in the flushed details JSON — the rest fold into a single
/// "others" rollup entry so a deployment with hundreds of scripts cannot mint an
/// oversized site event row.
/// </summary>
internal const int TopScriptCap = 50;
private sealed class Counters
{
public long Started;
public long Completed;
public long Failed;
public long TimedOut;
public long TotalDurationMs;
public long MaxDurationMs;
}
private ConcurrentDictionary<(string InstanceName, string ScriptName), Counters> _counters = new();
private Counters GetOrAddCounters(string instanceName, string scriptName) =>
Volatile.Read(ref _counters).GetOrAdd((instanceName, scriptName), static _ => new Counters());
/// <summary>Records that a script run began executing (past the shed-at-dequeue check).</summary>
public void RecordStarted(string instanceName, string scriptName)
{
var counters = GetOrAddCounters(instanceName, scriptName);
Interlocked.Increment(ref counters.Started);
}
/// <summary>Records that a script run completed successfully.</summary>
public void RecordCompleted(string instanceName, string scriptName, long durationMs)
{
var counters = GetOrAddCounters(instanceName, scriptName);
Interlocked.Increment(ref counters.Completed);
RecordDuration(counters, durationMs);
}
/// <summary>Records that a script run threw an unhandled exception.</summary>
public void RecordFailed(string instanceName, string scriptName, long durationMs)
{
var counters = GetOrAddCounters(instanceName, scriptName);
Interlocked.Increment(ref counters.Failed);
RecordDuration(counters, durationMs);
}
/// <summary>
/// Records that a script run timed out — including a run shed at dequeue (its deadline
/// was already spent queueing, so it never reached <see cref="RecordStarted"/>).
/// </summary>
public void RecordTimedOut(string instanceName, string scriptName, long durationMs)
{
var counters = GetOrAddCounters(instanceName, scriptName);
Interlocked.Increment(ref counters.TimedOut);
RecordDuration(counters, durationMs);
}
private static void RecordDuration(Counters counters, long durationMs)
{
if (durationMs < 0) durationMs = 0;
Interlocked.Add(ref counters.TotalDurationMs, durationMs);
InterlockedMax(ref counters.MaxDurationMs, durationMs);
}
/// <summary>Lock-free running-max update (no built-in <c>Interlocked.Max(ref long, long)</c>).</summary>
private static void InterlockedMax(ref long location, long candidate)
{
var initial = Volatile.Read(ref location);
while (candidate > initial)
{
var previous = Interlocked.CompareExchange(ref location, candidate, initial);
if (previous == initial) return;
initial = previous;
}
}
/// <summary>
/// Snapshots and resets the accumulated counters, then — if any script ran, failed, or
/// timed out during the interval — emits ONE aggregate "script" Info site event
/// summarizing the window. An interval with zero activity emits no row (the standby,
/// which never runs scripts, therefore never produces a summary row naturally).
/// </summary>
/// <param name="siteEventLogger">The site event logger to emit the aggregate row to.</param>
/// <returns><c>true</c> if a summary row was emitted; <c>false</c> if the interval was idle.</returns>
public async Task<bool> FlushAsync(ISiteEventLogger siteEventLogger)
{
ArgumentNullException.ThrowIfNull(siteEventLogger);
// The replacement dictionary's type argument must name its tuple elements
// IDENTICALLY to the field's declared type — Interlocked.Exchange<T> is generic
// over both the ref and value parameters, and mixing a named-tuple exact bound
// with an unnamed one erases the names from the inferred T (and therefore from
// `snapshot` below).
var snapshot = Interlocked.Exchange(
ref _counters,
new ConcurrentDictionary<(string InstanceName, string ScriptName), Counters>());
var entries = snapshot
.Where(kvp => kvp.Value.Started > 0 || kvp.Value.Failed > 0 || kvp.Value.TimedOut > 0)
.ToList();
if (entries.Count == 0)
return false;
long totalStarted = 0, totalCompleted = 0, totalFailed = 0, totalTimedOut = 0;
foreach (var entry in entries)
{
totalStarted += entry.Value.Started;
totalCompleted += entry.Value.Completed;
totalFailed += entry.Value.Failed;
totalTimedOut += entry.Value.TimedOut;
}
// Rank by total run activity. Started+TimedOut (not just Started): a run shed at
// dequeue never reaches RecordStarted, so ranking on Started alone would silently
// drop a script that does nothing BUT get shed from the breakdown entirely.
var ranked = entries
.OrderByDescending(e => e.Value.Started + e.Value.TimedOut)
.ToList();
var top = ranked.Take(TopScriptCap);
var rest = ranked.Skip(TopScriptCap).ToList();
var breakdown = top.Select(e => new ScriptSummaryEntry(
e.Key.InstanceName,
e.Key.ScriptName,
e.Value.Started,
e.Value.Completed,
e.Value.Failed,
e.Value.TimedOut,
e.Value.Completed > 0 ? e.Value.TotalDurationMs / e.Value.Completed : 0,
e.Value.MaxDurationMs)).ToList();
object details;
if (rest.Count > 0)
{
details = new
{
scripts = breakdown,
others = new
{
scriptCount = rest.Count,
started = rest.Sum(e => e.Value.Started),
completed = rest.Sum(e => e.Value.Completed),
failed = rest.Sum(e => e.Value.Failed),
timedOut = rest.Sum(e => e.Value.TimedOut),
}
};
}
else
{
details = new { scripts = breakdown };
}
var message = string.Format(
CultureInfo.InvariantCulture,
"{0:N0} runs: {1:N0} completed, {2:N0} failed, {3:N0} timed out across {4:N0} scripts",
totalStarted, totalCompleted, totalFailed, totalTimedOut, entries.Count);
var detailsJson = JsonSerializer.Serialize(details, DetailsJsonOptions);
await siteEventLogger.LogEventAsync(
"script", "Info", null, "ScriptRunSummary", message, detailsJson).ConfigureAwait(false);
return true;
}
// Built once per assembly (WP1.5 convention) rather than per flush.
private static readonly JsonSerializerOptions DetailsJsonOptions =
new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
private sealed record ScriptSummaryEntry(
string InstanceName,
string ScriptName,
long Started,
long Completed,
long Failed,
long TimedOut,
long AvgDurationMs,
long MaxDurationMs);
}