fix(attribution): sum attribution in the database, page the daily rollup (prod OOM) #16

Merged
KofTwentyTwo merged 2 commits from feature/bounded-attribution-reads into develop 2026-10-16 20:15:34 +00:00
6 changed files with 391 additions and 11 deletions

No files matched your search

+1
View File
@@ -5,6 +5,7 @@ All notable changes are documented here. This project follows Keep a Changelog a
## [Unreleased]
### Fixed
- 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.
@@ -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));
}
}