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

feat(kpihistory): route long-range KPI queries to hourly rollups by threshold (plan #22 T5)

Browse Source 
GetSeriesAsync now branches on window width: windows wider than
RollupThresholdHours (default 168h/7d) read pre-aggregated hourly rollups
via GetHourlySeriesAsync; windows at or below the threshold read raw
samples via GetRawSeriesAsync (preserving intra-minute detail on 24h/7d).
The boundary is strict greater-than, so exactly 168h stays on the raw
path. KpiSeriesBucketer.Bucket runs unchanged on whichever series (a 90d
rollup can still exceed the point ceiling). Both ctors route identically
via a shared FetchSeriesAsync helper reading _options.RollupThresholdHours,
the same options instance that already supplies DefaultMaxSeriesPoints.
GetSeriesAsync public signature unchanged.

Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
This commit is contained in:
Joseph Doherty
2026-07-10 12:01:59 -04:00
parent 4f145fd62a
commit a10170f365
2 changed files with 184 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.CentralUI/Services/KpiHistoryQueryService.cs
+28 -6
View File
@@ -79,17 +79,39 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
// Test-seam ctor: use the injected repository directly.
if (_injectedRepository is not null)
{
var injectedRaw = await _injectedRepository.GetRawSeriesAsync(
source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(injectedRaw, fromUtc, toUtc, effectiveMax);
var injectedSeries = await FetchSeriesAsync(
_injectedRepository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax);
}
// Production: a fresh scope (and thus a fresh DbContext) per query so a
// chart's auto-load never shares the circuit-scoped context.
await using var serviceScope = _scopeFactory!.CreateAsyncScope();
var repository = serviceScope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
var raw = await repository.GetRawSeriesAsync(
source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(raw, fromUtc, toUtc, effectiveMax);
var series = await FetchSeriesAsync(
repository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax);
}
/// <summary>
/// Routes a series fetch to the raw-sample or hourly-rollup repository method by
/// the requested window width. Windows wider than
/// <see cref="KpiHistoryOptions.RollupThresholdHours"/> read the pre-aggregated
/// hourly rollups (≈60× fewer rows scanned on 30 d/90 d ranges); windows at or below
/// the threshold read raw samples, preserving intra-minute detail on the 24 h/7 d
/// charts. Both paths return the same ascending <see cref="KpiSeriesPoint"/> shape, so
/// 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.
/// </summary>
private Task<IReadOnlyList<KpiSeriesPoint>> FetchSeriesAsync(
IKpiHistoryRepository repository,
string source, string metric, string scope, string? scopeKey,
DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken)
{
var windowHours = (toUtc - fromUtc).TotalHours;
return windowHours > _options.RollupThresholdHours
? repository.GetHourlySeriesAsync(source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken)
: repository.GetRawSeriesAsync(source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
}
}
tests/ZB.MOM.WW.ScadaBridge.CentralUI.Tests/Services/KpiHistoryQueryServiceTests.cs
+156 -2
View File
@@ -20,8 +20,12 @@ public class KpiHistoryQueryServiceTests
private static readonly DateTime From = new(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
private static readonly DateTime To = new(2026, 5, 20, 11, 0, 0, DateTimeKind.Utc);
private static IOptions<KpiHistoryOptions> Options(int defaultMax) =>
Microsoft.Extensions.Options.Options.Create(new KpiHistoryOptions { DefaultMaxSeriesPoints = defaultMax });
private static IOptions<KpiHistoryOptions> Options(int defaultMax, int rollupThresholdHours = 168) =>
Microsoft.Extensions.Options.Options.Create(new KpiHistoryOptions
{
DefaultMaxSeriesPoints = defaultMax,
RollupThresholdHours = rollupThresholdHours,
});
private static IReadOnlyList<KpiSeriesPoint> Series(params (int minute, double value)[] pts) =>
pts.Select(p => new KpiSeriesPoint(From.AddMinutes(p.minute), p.value)).ToList();
@@ -176,4 +180,154 @@ public class KpiHistoryQueryServiceTests
Assert.Equal(2, resolvedRepos.Count);
Assert.NotSame(resolvedRepos[0], resolvedRepos[1]);
}
// ---- Task 5: raw-vs-rollup range routing by RollupThresholdHours ----------------
private static IKpiHistoryRepository RepoReturningEmptyForBoth()
{
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>>(Array.Empty<KpiSeriesPoint>()));
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>>(Array.Empty<KpiSeriesPoint>()));
return repo;
}
[Fact]
public async Task GetSeriesAsync_24hWindow_UsesRawPath()
{
var repo = RepoReturningEmptyForBoth();
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddHours(24);
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200));
await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
await repo.Received(1).GetRawSeriesAsync(
"src", "metric", "global", null, from, to, Arg.Any<CancellationToken>());
await repo.DidNotReceive().GetHourlySeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task GetSeriesAsync_WindowExactlyAtThreshold_UsesRawPath()
{
// Boundary is strictly greater-than: exactly 168 h stays on the raw path.
var repo = RepoReturningEmptyForBoth();
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddHours(168);
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168));
await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
await repo.Received(1).GetRawSeriesAsync(
"src", "metric", "global", null, from, to, Arg.Any<CancellationToken>());
await repo.DidNotReceive().GetHourlySeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task GetSeriesAsync_WindowJustOverThreshold_UsesRollupPath()
{
// 169 h (threshold + 1) crosses the boundary → hourly rollup path.
var repo = RepoReturningEmptyForBoth();
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddHours(169);
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200, rollupThresholdHours: 168));
await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
await repo.Received(1).GetHourlySeriesAsync(
"src", "metric", "global", null, from, to, Arg.Any<CancellationToken>());
await repo.DidNotReceive().GetRawSeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task GetSeriesAsync_90dWindow_UsesRollupPath()
{
var repo = RepoReturningEmptyForBoth();
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddDays(90);
var sut = new KpiHistoryQueryService(repo, Options(defaultMax: 200));
await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
await repo.Received(1).GetHourlySeriesAsync(
"src", "metric", "global", null, from, to, Arg.Any<CancellationToken>());
await repo.DidNotReceive().GetRawSeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
[Fact]
public async Task GetSeriesAsync_RollupPath_StillBucketsWhenOverMax()
{
// A rollup series longer than maxPoints must still be downsampled by the
// bucketer on top of the rollup read — the bucketer path is unchanged.
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddDays(90);
// 6 hourly rollup points, spread across the window in ascending order.
var rollup = Enumerable.Range(0, 6)
.Select(i => new KpiSeriesPoint(from.AddHours(i * 300), i))
.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: 3));
var result = await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
// Routed to the rollup path...
await repo.Received(1).GetHourlySeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
// ...and the bucketer still applied (6 rollup rows > max of 3 → downsampled).
Assert.NotSame(rollup, result);
Assert.True(result.Count <= 3);
}
[Fact]
public async Task GetSeriesAsync_LongWindow_UsesRollupPath_OnProductionCtor()
{
// The production (IServiceScopeFactory) ctor must apply the same threshold
// routing — a long window resolves the repo from a fresh scope and calls the
// hourly rollup method.
var repo = RepoReturningEmptyForBoth();
var services = new ServiceCollection();
services.AddScoped<IKpiHistoryRepository>(_ => repo);
await using var provider = services.BuildServiceProvider();
var sut = new KpiHistoryQueryService(
provider.GetRequiredService<IServiceScopeFactory>(),
Options(defaultMax: 200));
var from = new DateTime(2026, 5, 1, 0, 0, 0, DateTimeKind.Utc);
var to = from.AddDays(30);
await sut.GetSeriesAsync("src", "metric", "global", null, from, to);
await repo.Received(1).GetHourlySeriesAsync(
"src", "metric", "global", null, from, to, Arg.Any<CancellationToken>());
await repo.DidNotReceive().GetRawSeriesAsync(
Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string>(), Arg.Any<string?>(),
Arg.Any<DateTime>(), Arg.Any<DateTime>(), Arg.Any<CancellationToken>());
}
}