Compare commits
14
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3dfaf0401e
|
||
|
|
681598be57 | ||
|
|
ea9036509c | ||
|
|
2a2ba58b06 | ||
|
|
61e9c22563 | ||
|
|
65922dd413
|
||
|
|
bdaf35fc3e | ||
|
|
c49b479620
|
||
|
|
de991509ca | ||
|
|
fd0116fbd0
|
||
|
|
744ff04ade
|
||
|
|
fdcad364f1 | ||
|
|
7c71e384c9 | ||
|
|
8a34ac4404
|
No files matched your search
+11
-57
@@ -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+$/
|
||||
@@ -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
|
||||
@@ -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
@@ -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
|
||||
@@ -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)));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user