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

fix(kpi): query service applies catalog aggregation at the bucketer — rate charts keep one unit across the raw/rollup boundary (plan R2-04 T6)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 10:10:30 -04:00
parent 54417e1ae4
commit 9a1b160e79
2 changed files with 106 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs
+15 -2
View File
@@ -76,12 +76,18 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
{
var effectiveMax = maxPoints ?? _options.DefaultMaxSeriesPoints;
// Per-metric reduction intent: the same catalog that drives the hourly fold drives
// the chart-time bucket reduction, so a Rate series is summed per bucket on BOTH the
// raw-minute and hourly-rollup routes — one unit ("events per chart bucket") on both
// sides of the RollupThresholdHours routing boundary (arch-review 04 round 2, R3).
var aggregation = KpiMetricAggregationCatalog.Resolve(source, metric);
// Test-seam ctor: use the injected repository directly.
if (_injectedRepository is not null)
{
var injectedSeries = await FetchSeriesAsync(
_injectedRepository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax);
return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax, aggregation);
}
// Production: a fresh scope (and thus a fresh DbContext) per query so a
@@ -90,7 +96,7 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
var repository = serviceScope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
var series = await FetchSeriesAsync(
repository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax);
return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax, aggregation);
}
/// <summary>
@@ -103,6 +109,13 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
/// the caller's <see cref="KpiSeriesBucketer.Bucket"/> step is agnostic to the source
/// (a 90 d rollup can still exceed the point ceiling, so bucketing still applies).
/// The boundary is strict: exactly <c>RollupThresholdHours</c> stays on the raw path.
/// <para>
/// The routing boundary no longer changes a Rate chart's magnitude: both paths reduce to
/// per-bucket sums via the catalog-driven aggregation applied in
/// <see cref="GetSeriesAsync"/>. A residual ≤~1.2× native-resolution difference exists only
/// for an unbucketed just-over-threshold rollup series (plotted at native per-hour sums) vs.
/// an at-threshold bucketed raw series — documented, not hidden (arch-review 04 round 2, R3).
/// </para>
/// </summary>
private Task<IReadOnlyList<KpiSeriesPoint>> FetchSeriesAsync(
IKpiHistoryRepository repository,
tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs
+91
View File
@@ -330,4 +330,95 @@ public class KpiHistoryQueryServiceTests
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
// ---- Task 6: catalog-driven aggregation at the bucketer boundary (R3) -----------
[Fact]
public async Task GetSeriesAsync_RateMetric_SumsPerBucket_OnRollupPath()
{
// (SiteCallAudit, FailedLastInterval) is a catalog Rate pair. 400 hourly points of
// value 1 over a 400 h (>168 h) window, maxPoints 200 → 2 points per 2 h bucket →
// every output value is 2.0 (the summed pair), not 1.0 (last value).
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddHours(400);
var rollup = Enumerable.Range(0, 400)
.Select(i => new KpiSeriesPoint(from.AddHours(i), 1d))
.ToList();
var repo = Substitute.For<IKpiHistoryRepository>();
repo.GetHourlySeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(rollup));
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200));
var result = await sut.GetSeriesAsync(
KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval,
"global", null, from, to);
Assert.Equal(200, result.Count);
Assert.All(result, p => Assert.Equal(2.0, p.Value));
}
[Fact]
public async Task GetSeriesAsync_RateMetric_PreservesTotal_AcrossRoutingBoundary()
{
// Same total event count (50) served as per-minute deltas on the raw path
// (window == threshold) and as hourly sums on the rollup path (window just over
// threshold): the summed chart total must be equal on both routes — the boundary
// is visually seamless for a Rate metric.
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var rawDeltas = Enumerable.Range(0, 10)
.Select(i => new KpiSeriesPoint(from.AddMinutes(i * 10), 5d)).ToList(); // total 50
var hourlySums = Enumerable.Range(0, 5)
.Select(i => new KpiSeriesPoint(from.AddHours(i), 10d)).ToList(); // total 50
var repo = Substitute.For<IKpiHistoryRepository>();
repo.GetRawSeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(rawDeltas));
repo.GetHourlySeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(hourlySums));
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168));
var rawResult = await sut.GetSeriesAsync(
KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval,
"global", null, from, from.AddHours(168)); // raw path
var rollupResult = await sut.GetSeriesAsync(
KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval,
"global", null, from, from.AddHours(169)); // rollup path
Assert.Equal(50d, rawResult.Sum(p => p.Value));
Assert.Equal(50d, rollupResult.Sum(p => p.Value));
Assert.Equal(rawResult.Sum(p => p.Value), rollupResult.Sum(p => p.Value));
}
[Fact]
public async Task GetSeriesAsync_GaugeMetric_KeepsLastValuePerBucket()
{
// queueDepth (uncatalogued for this source pairing → Gauge) keeps the existing
// last-value-per-bucket output unchanged. 4 points, maxPoints 2, raw path.
var repo = Substitute.For<IKpiHistoryRepository>();
var raw = Series((0, 1d), (15, 2d), (25, 99d), (35, 5d));
repo.GetRawSeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult(raw));
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 2));
var result = await sut.GetSeriesAsync(
KpiSources.NotificationOutbox, KpiMetrics.NotificationOutbox.QueueDepth,
"global", null, From, To);
Assert.Equal(2, result.Count);
Assert.Equal(99d, result[0].Value); // last value in bucket 0, not a sum
Assert.Equal(5d, result[1].Value);
}
}