using System.Diagnostics.Metrics; using ZB.MOM.WW.ScadaBridge.Commons.Observability; namespace ZB.MOM.WW.ScadaBridge.LoadHarness.Metrics; /// /// Captures the production scadabridge.site.stream.batch_size histogram via a /// , so a wire-path run can report the actual frame/batch /// distribution the R2 batcher produced — not a client-side reconstruction. The /// instrument is recorded once per emitted gRPC frame (negotiated streams only), with /// the frame's event count as the value, which makes count = frames and /// sum = events exactly. /// /// /// The harness hosts every site's /// in-process, so this listener sees every wire frame the run emits and nothing else — /// no other component in the harness records this instrument. /// /// public sealed class BatchSizeAggregator : IDisposable { /// /// Sizes above this are clamped into the last bucket. The production cap is /// GrpcStreamBatchMaxEvents = 100, so 512 leaves generous headroom for a /// configured-up run without an unbounded array. /// public const int MaxTrackedBatchSize = 512; private readonly MeterListener _listener; private readonly long[] _framesBySize = new long[MaxTrackedBatchSize + 1]; /// Starts listening. Must be created BEFORE the first stream opens. public BatchSizeAggregator() { _listener = new MeterListener { InstrumentPublished = (instrument, listener) => { if (instrument.Meter.Name == ScadaBridgeTelemetry.MeterName && instrument.Name == "scadabridge.site.stream.batch_size") { listener.EnableMeasurementEvents(instrument); } }, }; _listener.SetMeasurementEventCallback((_, value, _, _) => { var size = Math.Clamp(value, 1, MaxTrackedBatchSize); Interlocked.Increment(ref _framesBySize[size]); }); _listener.Start(); } /// Copies the cumulative per-size frame counts. /// Index = events-per-frame (clamped), value = frames emitted at that size. public long[] Snapshot() { var copy = new long[_framesBySize.Length]; for (var i = 0; i < copy.Length; i++) copy[i] = Interlocked.Read(ref _framesBySize[i]); return copy; } /// public void Dispose() => _listener.Dispose(); } /// /// Batch-size distribution over one measurement window, computed from two /// s. Percentiles are exact (integer sizes, /// full counts — nothing is bucketed away). /// /// gRPC frames written in the window. /// Events those frames carried. /// Frames that carried more than one event. /// Events / Frames. /// Median events-per-frame. /// 95th percentile events-per-frame. /// 99th percentile events-per-frame. /// Largest frame observed in the window. public sealed record BatchSizeWindow( long Frames, long Events, long BatchedFrames, double MeanEventsPerFrame, int P50, int P95, int P99, int Max) { /// Computes the window distribution from cumulative start/end snapshots. /// Snapshot at window start. /// Snapshot at window end. /// The distribution of frames emitted between the two snapshots. public static BatchSizeWindow From(long[] start, long[] end) { long frames = 0, events = 0, batched = 0; var max = 0; var delta = new long[end.Length]; for (var size = 1; size < end.Length; size++) { var n = end[size] - start[size]; if (n <= 0) continue; delta[size] = n; frames += n; events += n * size; if (size > 1) batched += n; max = size; } return new BatchSizeWindow( frames, events, batched, frames == 0 ? 0 : events / (double)frames, Percentile(delta, frames, 50), Percentile(delta, frames, 95), Percentile(delta, frames, 99), max); } private static int Percentile(long[] delta, long frames, double percentile) { if (frames == 0) return 0; var target = (long)Math.Ceiling(frames * percentile / 100.0); if (target < 1) target = 1; long cumulative = 0; for (var size = 1; size < delta.Length; size++) { cumulative += delta[size]; if (cumulative >= target) return size; } return delta.Length - 1; } }