Compare commits

...
Author SHA1 Message Date
james.maesandCodex 3dfaf0401e ci: centralize branch verification and source scanning
Co-Authored-By: Codex <noreply@openai.com>
2026-10-21 12:10:54 -05:00
james.maesandCodex 681598be57 ci: adopt central orb 0.7.12 and stable gates (#21)
Co-authored-by: Codex <noreply@openai.com>
2026-10-21 08:56:36 -05:00
james.maesandCodex ea9036509c ci: centralize rc release completion in qqq orb (#20)
Co-authored-by: Codex <noreply@openai.com>
2026-10-21 02:25:42 -05:00
james.maesandCodex 2a2ba58b06 ci: adopt shared orb 0.7.11 (#19)
Co-authored-by: Codex <noreply@openai.com>
2026-10-21 00:57:50 -05:00
james.maes 61e9c22563 Merge pull request #18 from QRun-IO/feature/central-ci
ci: adopt shared QQQ orb standard
2026-10-20 18:57:48 -05:00
james.maesandCodex 65922dd413 ci: adopt verified central QQQ orb 0.7.3
Co-Authored-By: Codex <noreply@openai.com>
2026-10-20 18:52:28 -05:00
james.maes bdaf35fc3e Merge pull request #17 from QRun-IO/feature/bounded-spike-history
fix(alerts): page spike history instead of holding 7 days of query snapshots
2026-10-16 15:54:56 -05:00
james.maesandClaude Opus 5.5 c49b479620 fix(alerts): page spike history instead of holding 7 days of query snapshots
SnapshotEvaluator.spikes loaded RuleContext.window("query_snapshot", 7)
as QRecords on every snapshot evaluation (every collection interval on
the worker). kof22 writes ~33,000 query snapshots a day, so by day 7 that
is ~230,000 records per evaluation.

history() now reads the window in BATCH_SIZE keyset pages (only OK and
PARTIAL snapshots, as before) and keeps per statement only the current
interval time, the latest capture time and earlier interval times in a
primitive double array. Median, twelve-sample baseline and the one-hour
absence rule are unchanged (SnapshotEvaluatorTest still passes).

SpikeHistoryBoundedTest counts every query snapshot QQQ materializes:
2,500 samples never exceed one 1,000-row page, failed snapshots and
samples outside the window are excluded, the current interval is found.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 15:24:44 -05:00
james.maes de991509ca Merge pull request #16 from QRun-IO/feature/bounded-attribution-reads
fix(attribution): sum attribution in the database, page the daily rollup (prod OOM)
2026-10-16 15:15:33 -05:00
james.maesandClaude Opus 5.5 fd0116fbd0 fix(analyze): read attribution as database sums in the hourly analysis
dbPerfAnalyze (hourly, on the worker) loaded the full day of
db_perf_attribution through RuleContext.window and cached it for the
rules: ~570,000 rows. kof22 worker working set jumped from ~660 MiB to
~1.84 GiB at every :07-:08 run on 2026-10-16 (2 Gi limit, 1.5 GB heap).

RuleContext.window("attribution", days) now returns AggregateAction sums
(units, statements, rowsCount, errors, totalMs; max maxMs) grouped by
unitKind, unitName, tableName, operation: the only fields the rules
(OVER_AUDITED, HOT_JOB) read. AttributionBoundedReadTest proves no detail
rows are loaded and the totals are unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 15:04:23 -05:00
james.maesandClaude Opus 5.5 744ff04ade fix(attribution): sum attribution in the database, page the daily rollup
The Attribution home widget queried every db_perf_attribution row of the
last 24 hours into QRecords and summed them in Java. kof22 writes about
30,000 attribution rows an hour, so one render loaded ~570,000 rows
(540 MB in PostgreSQL). On 2026-10-16 a single admin request logged
"Large number of records in QueryOutputList" at 50k, 100k and 250k rows
and the web pod died with OutOfMemoryError: Java heap space about 50 s
later, after slowing every other request with GC thrash.

- DbPerfWidgetRenderer.attribution() now uses AggregateAction: SUM of
  statements and totalMs grouped by unitKind, unitName, tableName,
  ordered by time and capped at 5,000 groups (prod has ~1,000).
- RollupService reads the day's attribution detail in 1,000-row keyset
  pages through the new DbPerfStore.forEachPage, so the nightly rollup
  on the worker is bounded by distinct keys instead of the day's rows.
- AttributionBoundedReadTest counts every detail record QQQ materializes
  (post-query customizer): the widget loads none, the rollup never holds
  more than BATCH_SIZE at once, and sums stay correct across pages.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 14:34:58 -05:00
james.maes fdcad364f1 Merge pull request #10 from QRun-IO/ci/release-rc-secrets-inherit
ci(release-rc): pass secrets so the shared workflow can sign the RC tag
2026-10-16 08:04:25 -05:00
james.maes 7c71e384c9 Merge pull request #12 from QRun-IO/fix/collector-failures
fix(collect): no stack overflow on huge statements; every failure closes the run as FAILED
2026-10-15 17:43:04 -05:00
james.maesandClaude Opus 5.5 8a34ac4404 ci(release-rc): pass secrets to the shared workflow so it can sign the RC tag
QRun-IO/.github#2 adds a tag-and-release job that signs vX with the
QRUN_TAG_SIGNING_KEY secret after Maven Central confirms the artifact.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 16:19:54 -05:00
12 changed files with 739 additions and 94 deletions

No files matched your search

+11 -57
View File
@@ -1,60 +1,14 @@
version: 2.1
setup: true
orbs:
qqq-orb: kingsrook/qqq-orb@0.6.9
commands:
configure-package-install:
steps:
- run:
name: Keep package installation unattended
command: |
# These build machines are discarded after the job.
sudo install -d /etc/needrestart/conf.d
printf '%s\n' '$nrconf{restart} = "l";' | sudo tee /etc/needrestart/conf.d/99-qqq-ci.conf >/dev/null
qqq-orb: kingsrook/qqq-orb@0.7.15
workflows:
test_only:
jobs:
- qqq-orb/mvn_test_only:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
filters:
branches:
ignore: /(develop|release\/.*)/
publish_snapshot:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: snapshot
filters:
branches:
only: [develop]
publish_release_candidate:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: release_candidate
filters:
branches:
only: [/release\/.*/]
publish_release:
jobs:
- qqq-orb/mvn_publish:
pre-steps:
- configure-package-install
context: [qqq-maven-registry-credentials]
branch_type: release
filters:
branches:
ignore: /.*/
tags:
only: [/^v\d+\.\d+\.\d+$/]
setup:
jobs:
- qqq-orb/generate_pipeline:
orb_version: 0.7.15
filters:
branches:
ignore: automation/qbit-rc-pending
tags:
only: /^v\d+\.\d+\.\d+$/
+23
View File
@@ -0,0 +1,23 @@
name: CodeQL
'on':
push:
branches:
- main
- develop
- release/**
- hotfix/**
pull_request: {}
schedule:
- cron: 41 16 * * 0
workflow_dispatch: {}
permissions:
contents: read
actions: read
security-events: write
packages: read
concurrency:
group: codeql-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
jobs:
analyze:
uses: QRun-IO/qqq-orb/.github/workflows/qbit-codeql.yml@c288a8864fe9054f68410e1d78f5863aea4c4d22
+62 -13
View File
@@ -1,25 +1,74 @@
# Cut the next release candidate (QRun-IO/qqq docs/QRUN-QBIT-REPO-STANDARD.md, C6):
# run this workflow by hand, or merge a develop PR labelled `release:rc`. It merges develop into
# the release line and the release/* push runs CircleCI publish_release_candidate.
name: Release candidate
on:
'on':
workflow_dispatch:
inputs:
dry_run:
description: Print the plan without merging or creating branches
description: Plan a cut without changing release state
type: boolean
default: false
finalize:
description: Retry finalization of an existing cut; do not cut another RC
type: boolean
default: false
recover:
description: Recover the same published RC after a deployment timeout; never
reupload
type: boolean
default: false
circle_build:
description: Original Circle publisher build number (recovery only)
type: string
default: ''
merge_sha:
description: Existing cut merge_sha (finalize retry only)
type: string
default: ''
bump_sha:
description: Existing cut bump_sha (finalize retry only)
type: string
default: ''
branch:
description: Existing cut branch (finalize retry only)
type: string
default: ''
version:
description: Existing cut version (finalize retry only)
type: string
default: ''
pull_request:
types: [closed]
branches: [develop]
types:
- closed
branches:
- develop
repository_dispatch:
types:
- qbit-rc-published
permissions:
contents: write
actions: read
checks: read
jobs:
release-rc:
if: github.event_name == 'workflow_dispatch' || (github.event.pull_request.merged == true && contains(github.event.pull_request.labels.*.name, 'release:rc'))
uses: QRun-IO/.github/.github/workflows/qbit-release-rc.yml@main
if: github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch'
|| (github.event_name == 'pull_request' && github.event.pull_request.merged
== true && github.event.pull_request.base.ref == 'develop' && contains(github.event.pull_request.labels.*.name,
'release:rc'))
uses: QRun-IO/qqq-orb/.github/workflows/qbit-release-rc.yml@cc30f599e63c8e41629c99d23395bac1adb8bc44
with:
dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run || false
}}
finalize: ${{ github.event_name == 'repository_dispatch' || (github.event_name
== 'workflow_dispatch' && inputs.finalize) }}
recover: ${{ github.event_name == 'workflow_dispatch' && inputs.recover || false
}}
circle_build: ${{ github.event_name == 'workflow_dispatch' && inputs.circle_build
|| '' }}
merge_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.merge_sha
|| inputs.merge_sha || '' }}
bump_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.bump_sha
|| inputs.bump_sha || '' }}
branch: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.branch
|| inputs.branch || '' }}
version: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.version
|| inputs.version || '' }}
secrets: inherit
+16
View File
@@ -0,0 +1,16 @@
profile: qbit
orb_version: 0.7.15
default_branch: develop
java_version: '21'
browsers: true
api_checks: false
docs_only: false
release_on_main: false
release_tag_policy: stable
test_branch_policy: non-release
publish:
snapshot: true
release_candidate: true
release: true
asciidoc: false
release_completion: true
+2
View File
@@ -5,6 +5,8 @@ All notable changes are documented here. This project follows Keep a Changelog a
## [Unreleased]
### Fixed
- Query-spike alerts no longer hold seven days of query snapshots as records on every snapshot evaluation (~230,000 rows at kof22's rate). `SnapshotEvaluator` reads the window in 1,000-row keyset pages and keeps only each statement's current interval time, latest capture time and earlier interval times as primitive doubles; the spike rule (twelve prior samples, median, absence after an hour) is unchanged.
- The Attribution home panel no longer loads a day of attribution detail into the heap. It sums statements and time per unit and table in the database (at most 5,000 groups), so a render holds about a thousand aggregate rows instead of the roughly 570,000 detail rows (540 MB in PostgreSQL) that ran the kof22 web pod out of memory on 2026-10-16. The daily attribution rollup reads the day in 1,000-row keyset pages (`DbPerfStore.forEachPage`), so worker memory is bounded by distinct dimension keys, not by the day's row count. The hourly analysis (`dbPerfAnalyze`) reads attribution the same way: `RuleContext.window("attribution", ...)` returns database sums per unit, table and operation instead of every detail row (it filled the kof22 worker heap to ~1.8 GB every hour).
- Collection no longer stops at RUNNING on large production statement sets: list folding in the SQL normalizer is a linear token scan (the nested-repetition regular expressions overflowed the stack on 60 KB statements with long IN/VALUES lists). Any failure, including an `Error`, now closes the run as FAILED with the exception class, logs it at ERROR (class and frame only, never SQL text) and raises the collection alert; runs left RUNNING by a crash are closed as FAILED after three intervals.
- Only one runtime attributes per JVM: starting attribution on a newer instance (a hot swap, or repeated server starts in tests) stops the previous one, so old instances are never pinned by a forgotten listener and flusher.
- Each table has its own icon (the app icon is now `storage`), so nav entries are distinguishable and do not repeat icons used elsewhere.
@@ -22,9 +22,11 @@ package com.kingsrook.qbits.dbperformance.alerts;
import java.time.Instant;
import java.util.Comparator;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import com.kingsrook.qbits.dbperformance.DbPerfQBitConfig;
import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
@@ -38,6 +40,9 @@ import com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData;
import com.kingsrook.qqq.backend.core.alerts.QAlertTypeMetaData;
import com.kingsrook.qqq.backend.core.alerts.QAlerts;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -159,33 +164,31 @@ public class SnapshotEvaluator
*******************************************************************************/
private void spikes(RuleContext ctx, QRecord snapshot) throws Exception
{
Map<Integer, List<QRecord>> history = ctx.window("query_snapshot", 7).stream().collect(Collectors.groupingBy(r -> r.getValueInteger("statementId")));
Map<Integer, StatementHistory> history = history(ctx, snapshot.getValueInteger("id"));
for(QRecord statement : ctx.identities("statement"))
{
List<QRecord> samples = history.getOrDefault(statement.getValueInteger("id"), List.of());
QRecord current = samples.stream().filter(r -> snapshot.getValueInteger("id").equals(r.getValueInteger("snapshotId"))).findFirst().orElse(null);
if(current == null)
StatementHistory samples = history.getOrDefault(statement.getValueInteger("id"), new StatementHistory());
if(samples.current == null)
{
Instant last = samples.stream().map(r -> r.getValueInstant("capturedAt")).max(Comparator.naturalOrder()).orElse(Instant.EPOCH);
Instant last = samples.last == null ? Instant.EPOCH : samples.last;
if(!last.plusSeconds(3600).isAfter(ctx.now))
{
signal(QUERY_SPIKE, ctx.backend, statement.getValueString("id"), false, "Query absent", "Statement absent from snapshots for at least one hour", statement, "db_perf_statement");
}
continue;
}
List<Double> previous = samples.stream().filter(r -> r.getValueInstant("capturedAt").isBefore(ctx.now))
.map(r -> RuleContext.number(r, "totalMs")).sorted().toList();
if(previous.size() < 12)
double[] previous = samples.sortedPrevious();
if(previous.length < 12)
{
if(RuleContext.number(current, "totalMs") < RuleContext.number(ctx.settings, "spikeMinTotalMs"))
if(samples.current < RuleContext.number(ctx.settings, "spikeMinTotalMs"))
{
signal(QUERY_SPIKE, ctx.backend, statement.getValueString("id"), false, "Query recovered", "Interval time is below the configured floor", statement, "db_perf_statement");
}
continue;
}
int middle = previous.size() / 2;
double median = previous.size() % 2 == 0 ? (previous.get(middle - 1) + previous.get(middle)) / 2 : previous.get(middle);
double time = RuleContext.number(current, "totalMs");
int middle = previous.length / 2;
double median = previous.length % 2 == 0 ? (previous[middle - 1] + previous[middle]) / 2 : previous[middle];
double time = samples.current;
boolean spike = time >= RuleContext.number(ctx.settings, "spikeMinTotalMs") && time >= median * RuleContext.number(ctx.settings, "spikeFactor");
String normalized = SqlNormalizer.normalize(statement.getValueString("queryText"));
String text = normalized.substring(0, Math.min(200, normalized.length()));
@@ -195,6 +198,90 @@ public class SnapshotEvaluator
}
/*******************************************************************************
** Seven days of per-statement interval times, read in keyset pages and kept
** as primitive doubles. kof22 writes ~33,000 query snapshots a day, so the
** window is ~230,000 rows; holding them as QRecords on every snapshot
** evaluation cost hundreds of MB of worker heap.
*******************************************************************************/
static Map<Integer, StatementHistory> history(RuleContext ctx, Integer currentSnapshotId) throws Exception
{
Set<Integer> valid = ctx.window("snapshot", 7).stream().filter(r -> List.of("OK", "PARTIAL").contains(r.getValueString("status")))
.map(r -> r.getValueInteger("id")).collect(Collectors.toSet());
QQueryFilter filter = new QQueryFilter(new QFilterCriteria("capturedAt", QCriteriaOperator.GREATER_THAN, ctx.now.minusSeconds(7 * 86400L)),
new QFilterCriteria("capturedAt", QCriteriaOperator.LESS_THAN_OR_EQUALS, ctx.now));
Map<Integer, StatementHistory> history = new HashMap<>();
DbPerfStore.forEachPage("db_perf_query_snapshot", filter, DbPerfStore.BATCH_SIZE, page ->
{
for(QRecord row : page)
{
Integer snapshotId = row.getValueInteger("snapshotId");
if(!valid.contains(snapshotId))
{
continue;
}
StatementHistory statement = history.computeIfAbsent(row.getValueInteger("statementId"), ignored -> new StatementHistory());
Instant at = row.getValueInstant("capturedAt");
double totalMs = RuleContext.number(row, "totalMs");
if(statement.last == null || at.isAfter(statement.last))
{
statement.last = at;
}
if(snapshotId.equals(currentSnapshotId) && statement.current == null)
{
statement.current = totalMs;
}
if(at.isBefore(ctx.now))
{
statement.add(totalMs);
}
}
});
return (history);
}
/*******************************************************************************
** One statement's samples: the current interval, the latest capture time
** and earlier interval times.
*******************************************************************************/
static final class StatementHistory
{
private Double current;
private Instant last;
private double[] previous = new double[16];
private int size;
/***************************************************************************
** Append an earlier interval time.
***************************************************************************/
void add(double value)
{
if(size == previous.length)
{
previous = Arrays.copyOf(previous, size * 2);
}
previous[size++] = value;
}
/***************************************************************************
** Earlier interval times, ascending.
***************************************************************************/
double[] sortedPrevious()
{
double[] sorted = Arrays.copyOf(previous, size);
Arrays.sort(sorted);
return (sorted);
}
}
/*******************************************************************************
** Stable backend/subject keys make raises and resolutions restart-safe.
*******************************************************************************/
@@ -25,17 +25,27 @@ import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
import com.kingsrook.qbits.dbperformance.collect.DbPerfStore;
import com.kingsrook.qqq.backend.core.actions.tables.AggregateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.Aggregate;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateResult;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.GroupBy;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
/*******************************************************************************
@@ -130,7 +140,7 @@ public class RuleContext
{
filter.withCriteria(new QFilterCriteria("backendName", QCriteriaOperator.EQUALS, backend));
}
List<QRecord> rows = read("db_perf_" + suffix, filter);
List<QRecord> rows = suffix.equals("attribution") ? attributionTotals(filter) : read("db_perf_" + suffix, filter);
if(!Set.of("snapshot", "attribution", "repeat_pattern").contains(suffix))
{
Set<Integer> ids = window("snapshot", days).stream()
@@ -144,6 +154,52 @@ public class RuleContext
}
/*******************************************************************************
** Most attribution groups the rules receive (prod: ~2,000 per day).
*******************************************************************************/
public static final int MAX_ATTRIBUTION_GROUPS = 50_000;
/*******************************************************************************
** Attribution summed in the database per unit, table and operation: the
** fields the rules read (statements, totalMs and the other counters), never
** the day's detail rows. A day is ~570,000 rows in kof22 and loading it made
** every hourly dbPerfAnalyze fill the worker heap (2026-10-16).
*******************************************************************************/
private static List<QRecord> attributionTotals(QQueryFilter filter) throws QException
{
List<String> dimensions = List.of("unitKind", "unitName", "tableName", "operation");
AggregateInput input = new AggregateInput("db_perf_attribution").withFilter(filter).withLimit(MAX_ATTRIBUTION_GROUPS)
.withTransaction(QContext.getQBackendTransaction());
Map<String, GroupBy> groupBys = new LinkedHashMap<>();
for(String dimension : dimensions)
{
GroupBy groupBy = new GroupBy(QFieldType.STRING, dimension);
groupBys.put(dimension, groupBy);
input.withGroupBy(groupBy);
}
Map<String, Aggregate> aggregates = new LinkedHashMap<>();
for(String counter : List.of("units", "statements", "rowsCount", "errors", "totalMs"))
{
aggregates.put(counter, new Aggregate(counter, AggregateOperator.SUM));
}
aggregates.put("maxMs", new Aggregate("maxMs", AggregateOperator.MAX));
aggregates.values().forEach(input::withAggregate);
List<QRecord> rows = new ArrayList<>();
for(AggregateResult result : new AggregateAction().execute(input).getResults())
{
QRecord row = new QRecord();
groupBys.forEach((field, groupBy) -> row.setValue(field, result.getGroupByValue(groupBy)));
aggregates.forEach((field, aggregate) -> row.setValue(field, result.getAggregateValue(aggregate)));
rows.add(row);
}
return (rows);
}
/*******************************************************************************
** Latest inventory only: retired tables and dropped indexes cannot recur.
*******************************************************************************/
@@ -66,6 +66,48 @@ public final class DbPerfStore
}
/*******************************************************************************
** Keyset-paged read for tables that can hold more rows than fit in the heap
** (attribution writes hundreds of thousands per day). At most pageSize
** records are held at once; the consumer must not keep the page.
*******************************************************************************/
public static void forEachPage(String table, QQueryFilter filter, int pageSize, PageConsumer consumer) throws Exception
{
Serializable lastId = 0;
while(true)
{
QQueryFilter page = new QQueryFilter().withSubFilter(filter).withCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, lastId))
.withOrderBy(new QFilterOrderBy("id", true)).withLimit(pageSize);
List<QRecord> rows = query(table, page);
if(rows.isEmpty())
{
return;
}
consumer.accept(rows);
if(rows.size() < pageSize)
{
return;
}
lastId = rows.getLast().getValue("id");
}
}
/*******************************************************************************
** Receives one bounded page from forEachPage.
*******************************************************************************/
@FunctionalInterface
public interface PageConsumer
{
/***************************************************************************
** Handle one page.
***************************************************************************/
void accept(List<QRecord> page) throws Exception;
}
/*******************************************************************************
** First matching record.
*******************************************************************************/
@@ -168,7 +168,19 @@ public class RollupService
{
Map<String, QRecord> groups = new LinkedHashMap<>();
List<String> dimensions = List.of("unitKind", "unitName", "processName", "tableName", "operation", "fingerprint");
for(QRecord row : DbPerfStore.query("db_perf_attribution", range("windowStart", start, end)))
DbPerfStore.forEachPage("db_perf_attribution", range("windowStart", start, end), DbPerfStore.BATCH_SIZE, page -> mergeAttribution(day, page, dimensions, groups));
replace("db_perf_attribution_daily", day, new ArrayList<>(groups.values()));
}
/*******************************************************************************
** Fold one page of attribution detail into the day's groups. Memory is
** bounded by distinct dimension keys, never by the day's detail row count.
*******************************************************************************/
private static void mergeAttribution(LocalDate day, List<QRecord> page, List<String> dimensions, Map<String, QRecord> groups)
{
for(QRecord row : page)
{
StringBuilder key = new StringBuilder();
for(String field : dimensions)
@@ -187,7 +199,6 @@ public class RollupService
sumDecimal(aggregate, row, "totalMs");
aggregate.setValue("maxMs", SnapshotCollector.decimal(aggregate, "maxMs").max(SnapshotCollector.decimal(row, "maxMs")));
}
replace("db_perf_attribution_daily", day, new ArrayList<>(groups.values()));
}
@@ -39,8 +39,15 @@ import com.kingsrook.qbits.dbperformance.collect.SnapshotCollector;
import com.kingsrook.qbits.dbperformance.provider.ConnectionCount;
import com.kingsrook.qbits.dbperformance.provider.DbStatsProviders;
import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.AbstractWidgetRenderer;
import com.kingsrook.qqq.backend.core.actions.tables.AggregateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.Aggregate;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateResult;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.GroupBy;
import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.QFilterOrderByAggregate;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
@@ -59,6 +66,7 @@ import com.kingsrook.qqq.backend.core.model.dashboard.widgets.blocks.text.TextBl
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.blocks.text.TextStyles;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.blocks.text.TextValues;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
import com.kingsrook.qqq.backend.module.rdbms.jdbc.ConnectionManager;
import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSBackendMetaData;
import static com.kingsrook.qbits.dbperformance.api.DbPerformanceData.map;
@@ -280,22 +288,45 @@ public class DbPerfWidgetRenderer extends AbstractWidgetRenderer
/*******************************************************************************
** Most (unit, table) groups summed for the attribution panel. Production
** shows about a thousand; the cap only guards pathological unit names.
*******************************************************************************/
public static final int MAX_ATTRIBUTION_GROUPS = 5000;
/*******************************************************************************
** Aggregate units over 24 hours and show repeated-statement patterns below them.
** The sums run in the database: a day of attribution detail is hundreds of
** thousands of rows (about 540 MB), and loading it into the heap on a widget
** request ran the web pod out of memory (2026-10-16).
*******************************************************************************/
public MultiTableData attribution() throws QException
{
List<QRecord> records = DbPerfStore.query("db_perf_attribution", since("windowEnd"));
GroupBy kindGroup = new GroupBy(QFieldType.STRING, "unitKind");
GroupBy nameGroup = new GroupBy(QFieldType.STRING, "unitName");
GroupBy tableGroup = new GroupBy(QFieldType.STRING, "tableName");
Aggregate statementsSum = new Aggregate("statements", AggregateOperator.SUM);
Aggregate totalMsSum = new Aggregate("totalMs", AggregateOperator.SUM);
List<AggregateResult> results = new AggregateAction().execute(new AggregateInput("db_perf_attribution")
.withFilter(since("windowEnd").withOrderBy(new QFilterOrderByAggregate(totalMsSum, false)))
.withGroupBy(kindGroup).withGroupBy(nameGroup).withGroupBy(tableGroup)
.withAggregate(statementsSum).withAggregate(totalMsSum)
.withLimit(MAX_ATTRIBUTION_GROUPS)
.withTransaction(QContext.getQBackendTransaction())).getResults();
Map<String, Map<String, Object>> grouped = new LinkedHashMap<>();
Map<String, Map<String, Double>> tableTimes = new LinkedHashMap<>();
double total = records.stream().mapToDouble(row -> number(row, "totalMs")).sum();
for(QRecord row : records)
double total = results.stream().mapToDouble(row -> numeric(row.getAggregateValue(totalMsSum))).sum();
for(AggregateResult row : results)
{
String key = row.getValueString("unitKind") + "\n" + row.getValueString("unitName");
Map<String, Object> group = grouped.computeIfAbsent(key, ignored -> map("Kind", row.getValueString("unitKind"), "Name", row.getValueString("unitName"), "Statements", 0.0, "Total ms", 0.0));
group.put("Statements", numeric(group.get("Statements")) + number(row, "statements"));
group.put("Total ms", numeric(group.get("Total ms")) + number(row, "totalMs"));
tableTimes.computeIfAbsent(key, ignored -> new LinkedHashMap<>()).merge(Objects.toString(row.getValueString("tableName"), "Unknown"), number(row, "totalMs"), Double::sum);
String kind = (String) row.getGroupByValue(kindGroup);
String name = (String) row.getGroupByValue(nameGroup);
String key = kind + "\n" + name;
Map<String, Object> group = grouped.computeIfAbsent(key, ignored -> map("Kind", kind, "Name", name, "Statements", 0.0, "Total ms", 0.0));
group.put("Statements", numeric(group.get("Statements")) + numeric(row.getAggregateValue(statementsSum)));
group.put("Total ms", numeric(group.get("Total ms")) + numeric(row.getAggregateValue(totalMsSum)));
tableTimes.computeIfAbsent(key, ignored -> new LinkedHashMap<>()).merge(Objects.toString(row.getGroupByValue(tableGroup), "Unknown"), numeric(row.getAggregateValue(totalMsSum)), Double::sum);
}
grouped.forEach((key, group) ->
{
@@ -0,0 +1,239 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2026. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.dbperformance;
import java.math.BigDecimal;
import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import com.kingsrook.qbits.dbperformance.analyze.RuleContext;
import com.kingsrook.qbits.dbperformance.collect.DbPerfStore;
import com.kingsrook.qbits.dbperformance.collect.RollupService;
import com.kingsrook.qbits.dbperformance.widgets.DbPerfWidgetRenderer;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
import com.kingsrook.qqq.backend.core.model.actions.tables.QueryOrGetInputInterface;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.dashboard.widgets.MultiTableData;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.utils.JsonUtils;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Attribution detail grows by hundreds of thousands of rows a day. On
** 2026-10-16 the home widget loaded a whole day of it into the heap and ran
** the kof22 web pod out of memory, and the hourly analysis filled the worker
** heap the same way. These tests count every attribution detail record QQQ
** materializes, so the widget, the analysis and the daily rollup cannot go
** back to reading the table in one unbounded query.
*******************************************************************************/
class AttributionBoundedReadTest
{
private static final AtomicInteger LOADED = new AtomicInteger();
private static final AtomicInteger MAX_PAGE = new AtomicInteger();
private DbPerfRuntime runtime;
/*******************************************************************************
** Counts the detail records each QueryAction hands back.
*******************************************************************************/
public static class CountingCustomizer implements TableCustomizerInterface
{
/***************************************************************************
** Record the page size, change nothing.
***************************************************************************/
@Override
public List<QRecord> postQuery(QueryOrGetInputInterface queryInput, List<QRecord> records)
{
LOADED.addAndGet(records.size());
MAX_PAGE.accumulateAndGet(records.size(), Math::max);
return (records);
}
}
/*******************************************************************************
** Fresh H2 instance with the counter on the detail table.
*******************************************************************************/
@BeforeEach
void setup() throws Exception
{
runtime = DbPerfRuntime.of(DbPerfTestSupport.instance());
runtime.getInstance().getTable("db_perf_attribution").withCustomizer(TableCustomizers.POST_QUERY_RECORD, new QCodeReference(CountingCustomizer.class));
LOADED.set(0);
MAX_PAGE.set(0);
}
/*******************************************************************************
** The widget sums in the database: no detail rows reach the heap, and the
** totals still match the detail.
*******************************************************************************/
@Test
void widgetSumsInTheDatabase() throws Exception
{
Instant now = Instant.now();
insert(3000, i -> row(now.minusSeconds(60), now, i));
LOADED.set(0);
MultiTableData attribution = new DbPerfWidgetRenderer().attribution();
assertThat(LOADED.get()).as("attribution detail rows loaded by the widget").isZero();
List<Map<String, Object>> units = attribution.getTableDataList().getFirst().getRows();
assertThat(units).hasSize(3);
// unit-0 holds i = 0, 3, 6 ... (1000 rows, 2 statements each), mostly against "patient"
assertThat(((Number) units.getFirst().get("Statements")).doubleValue() + ((Number) units.get(1).get("Statements")).doubleValue()
+ ((Number) units.get(2).get("Statements")).doubleValue()).isEqualTo(6000.0);
assertThat(units).allSatisfy(unit -> assertThat(((Number) unit.get("Statements")).doubleValue()).isEqualTo(2000.0));
assertThat(JsonUtils.toJson(units.getFirst())).contains("top table patient");
assertThat(units.stream().mapToDouble(unit -> ((Number) unit.get("Share %")).doubleValue()).sum()).isCloseTo(100.0, org.assertj.core.data.Offset.offset(0.001));
}
/*******************************************************************************
** Detail older than 24 hours is left out of the sums.
*******************************************************************************/
@Test
void widgetKeepsTheDayWindow() throws Exception
{
Instant now = Instant.now();
insert(3, i -> row(now.minusSeconds(60), now, i));
insert(3, i -> row(now.minusSeconds(3 * 86400), now.minusSeconds(2 * 86400), i));
List<Map<String, Object>> units = new DbPerfWidgetRenderer().attribution().getTableDataList().getFirst().getRows();
assertThat(units).hasSize(3).allSatisfy(unit -> assertThat(((Number) unit.get("Statements")).doubleValue()).isEqualTo(2.0));
}
/*******************************************************************************
** The hourly analysis reads attribution as database sums per unit, table
** and operation: no detail rows, same totals.
*******************************************************************************/
@Test
void analysisSumsInTheDatabase() throws Exception
{
Instant now = Instant.now();
insert(3000, i -> row(now.minusSeconds(60), now, i));
LOADED.set(0);
RuleContext context = new RuleContext(runtime, runtime.getConfig().getBackendName(), new QRecord(), new QRecord(), now.plusSeconds(1));
List<QRecord> rows = context.window("attribution", 1);
assertThat(LOADED.get()).as("attribution detail rows loaded by the analysis").isZero();
assertThat(rows).hasSize(6);
assertThat(rows.stream().mapToDouble(row -> RuleContext.number(row, "statements")).sum()).isEqualTo(6000.0);
assertThat(rows.stream().mapToDouble(row -> RuleContext.number(row, "totalMs")).sum()).isEqualTo(1500 * 10.0 + 1500 * 1.0);
assertThat(rows).allSatisfy(row -> assertThat(row.getValueString("operation")).isEqualTo("SELECT"));
}
/*******************************************************************************
** Keyset pages never exceed the page size, and every row is seen once.
*******************************************************************************/
@Test
void forEachPageIsBounded() throws Exception
{
Instant now = Instant.now();
insert(1050, i -> row(now.minusSeconds(60), now, i));
LOADED.set(0);
List<Integer> sizes = new ArrayList<>();
List<Object> ids = new ArrayList<>();
DbPerfStore.forEachPage("db_perf_attribution", new QQueryFilter(), 100, page ->
{
sizes.add(page.size());
page.forEach(row -> ids.add(row.getValue("id")));
});
assertThat(sizes).hasSize(11).allSatisfy(size -> assertThat(size).isLessThanOrEqualTo(100));
assertThat(ids).hasSize(1050).doesNotHaveDuplicates();
assertThat(MAX_PAGE.get()).isLessThanOrEqualTo(100);
}
/*******************************************************************************
** The daily rollup reads the day in BATCH_SIZE pages and its sums still
** cover every detail row across page boundaries.
*******************************************************************************/
@Test
void rollupReadsTheDayInPages() throws Exception
{
ZoneId zone = ZoneId.of(runtime.getConfig().getTimeZone());
LocalDate day = LocalDate.now(zone).minusDays(1);
Instant start = day.atStartOfDay(zone).toInstant().plusSeconds(3600);
int rows = DbPerfStore.BATCH_SIZE * 2 + 500;
insert(rows, i -> row(start, start.plusSeconds(60), i));
MAX_PAGE.set(0);
LOADED.set(0);
new RollupService(runtime).rollupDay(day);
assertThat(MAX_PAGE.get()).as("largest page of attribution detail held at once").isLessThanOrEqualTo(DbPerfStore.BATCH_SIZE);
assertThat(LOADED.get()).isEqualTo(rows);
List<QRecord> daily = DbPerfStore.query("db_perf_attribution_daily", new QQueryFilter());
assertThat(daily).hasSize(6);
assertThat(daily.stream().mapToLong(row -> row.getValueLong("statements")).sum()).isEqualTo(2L * rows);
}
/*******************************************************************************
** Insert generated rows in batches.
*******************************************************************************/
private static void insert(int count, java.util.function.IntFunction<QRecord> factory) throws Exception
{
List<QRecord> rows = new ArrayList<>();
for(int i = 0; i < count; i++)
{
rows.add(factory.apply(i));
}
DbPerfStore.insert("db_perf_attribution", rows);
}
/*******************************************************************************
** Three units; each unit spends most of its time on "patient".
*******************************************************************************/
private static QRecord row(Instant windowStart, Instant windowEnd, int i)
{
boolean patient = i % 6 < 3;
return (new QRecord().withValue("windowStart", windowStart).withValue("windowEnd", windowEnd).withValue("podName", "pod-" + (i % 2))
.withValue("unitKind", "PROCESS").withValue("unitName", "unit-" + (i % 3)).withValue("tableName", patient ? "patient" : "visit")
.withValue("operation", "SELECT").withValue("fingerprint", "f".repeat(64)).withValue("sqlText", "select 1")
.withValue("units", 1L).withValue("statements", 2L).withValue("rowsCount", 1L).withValue("errors", 0L)
.withValue("totalMs", patient ? BigDecimal.TEN : BigDecimal.ONE).withValue("maxMs", BigDecimal.ONE));
}
}
@@ -0,0 +1,135 @@
/*
* QQQ - Low-code Application Framework for Engineers.
* Copyright (C) 2021-2026. Kingsrook, LLC
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
* contact@kingsrook.com
* https://github.com/Kingsrook/
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.dbperformance.alerts;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import com.kingsrook.qbits.dbperformance.DbPerfRuntime;
import com.kingsrook.qbits.dbperformance.DbPerfTestSupport;
import com.kingsrook.qbits.dbperformance.analyze.RuleContext;
import com.kingsrook.qbits.dbperformance.collect.DbPerfStore;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
import com.kingsrook.qqq.backend.core.model.actions.tables.QueryOrGetInputInterface;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Spike detection reads seven days of query snapshots (~230,000 rows in kof22)
** on every evaluation. It must page them and keep only per-statement numbers.
*******************************************************************************/
class SpikeHistoryBoundedTest
{
private static final AtomicInteger MAX_PAGE = new AtomicInteger();
private static final AtomicInteger LOADED = new AtomicInteger();
/*******************************************************************************
** Records the largest page QQQ hands back.
*******************************************************************************/
public static class CountingCustomizer implements TableCustomizerInterface
{
/***************************************************************************
** Count, change nothing.
***************************************************************************/
@Override
public List<QRecord> postQuery(QueryOrGetInputInterface queryInput, List<QRecord> records)
{
LOADED.addAndGet(records.size());
MAX_PAGE.accumulateAndGet(records.size(), Math::max);
return (records);
}
}
/*******************************************************************************
** 2,500 samples across page boundaries: pages stay at BATCH_SIZE, invalid
** snapshots and the window are honored, the current interval is found.
*******************************************************************************/
@Test
void historyIsPagedAndExact() throws Exception
{
DbPerfRuntime runtime = DbPerfRuntime.of(DbPerfTestSupport.instance());
String backend = runtime.getConfig().getBackendName();
runtime.getInstance().getTable("db_perf_query_snapshot").withCustomizer(TableCustomizers.POST_QUERY_RECORD, new QCodeReference(CountingCustomizer.class));
Instant now = Instant.parse("2026-10-16T20:00:00Z");
QRecord ok = DbPerfStore.insert("db_perf_snapshot", List.of(new QRecord().withValue("backendName", backend).withValue("capturedAt", now.minusSeconds(3600)).withValue("status", "OK"))).getFirst();
QRecord failed = DbPerfStore.insert("db_perf_snapshot", List.of(new QRecord().withValue("backendName", backend).withValue("capturedAt", now.minusSeconds(1800)).withValue("status", "FAILED"))).getFirst();
QRecord current = DbPerfStore.insert("db_perf_snapshot", List.of(new QRecord().withValue("backendName", backend).withValue("capturedAt", now).withValue("status", "OK"))).getFirst();
int one = statement(backend, 1L);
int two = statement(backend, 2L);
List<QRecord> rows = new ArrayList<>();
for(int i = 0; i < 2500; i++)
{
rows.add(sample(ok, i % 2 == 0 ? one : two, now.minusSeconds(3600 + i), i));
}
rows.add(sample(failed, one, now.minusSeconds(1800), 999999));
rows.add(sample(ok, one, now.minusSeconds(8 * 86400L), 999999));
rows.add(sample(current, one, now, 42));
DbPerfStore.insert("db_perf_query_snapshot", rows);
MAX_PAGE.set(0);
LOADED.set(0);
RuleContext ctx = new RuleContext(runtime, backend, new QRecord(), DbPerfStore.first("db_perf_settings", new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter()), now);
Map<Integer, SnapshotEvaluator.StatementHistory> history = SnapshotEvaluator.history(ctx, current.getValueInteger("id"));
assertThat(MAX_PAGE.get()).as("largest page of query snapshots held at once").isLessThanOrEqualTo(DbPerfStore.BATCH_SIZE);
assertThat(history.get(one).sortedPrevious()).hasSize(1250).doesNotContain(999999.0);
assertThat(history.get(two).sortedPrevious()).hasSize(1250);
assertThat(history.get(one).sortedPrevious()[0]).isEqualTo(0.0);
assertThat(history.get(one)).extracting("current").isEqualTo(42.0);
assertThat(history.get(two)).extracting("current").isNull();
}
/*******************************************************************************
** One statement identity; returns its id.
*******************************************************************************/
private static int statement(String backend, long queryId) throws Exception
{
return (DbPerfStore.insert("db_perf_statement", List.of(new QRecord().withValue("backendName", backend).withValue("queryId", queryId)
.withValue("queryText", "select " + queryId).withValue("fingerprint", String.valueOf(queryId).repeat(64).substring(0, 64)))).getFirst().getValueInteger("id"));
}
/*******************************************************************************
** One query snapshot row.
*******************************************************************************/
private static QRecord sample(QRecord snapshot, int statementId, Instant at, int totalMs)
{
return (new QRecord().withValue("snapshotId", snapshot.getValueInteger("id")).withValue("statementId", statementId)
.withValue("capturedAt", at).withValue("totalMs", BigDecimal.valueOf(totalMs)));
}
}