diff --git a/CHANGELOG.md b/CHANGELOG.md index 14725c8..046e267 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ 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. diff --git a/src/main/java/com/kingsrook/qbits/dbperformance/alerts/SnapshotEvaluator.java b/src/main/java/com/kingsrook/qbits/dbperformance/alerts/SnapshotEvaluator.java index a8eb0a0..1cd6880 100644 --- a/src/main/java/com/kingsrook/qbits/dbperformance/alerts/SnapshotEvaluator.java +++ b/src/main/java/com/kingsrook/qbits/dbperformance/alerts/SnapshotEvaluator.java @@ -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> history = ctx.window("query_snapshot", 7).stream().collect(Collectors.groupingBy(r -> r.getValueInteger("statementId"))); + Map history = history(ctx, snapshot.getValueInteger("id")); for(QRecord statement : ctx.identities("statement")) { - List 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 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 history(RuleContext ctx, Integer currentSnapshotId) throws Exception + { + Set 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 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. *******************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/dbperformance/alerts/SpikeHistoryBoundedTest.java b/src/test/java/com/kingsrook/qbits/dbperformance/alerts/SpikeHistoryBoundedTest.java new file mode 100644 index 0000000..0e00272 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/dbperformance/alerts/SpikeHistoryBoundedTest.java @@ -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 postQuery(QueryOrGetInputInterface queryInput, List 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 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 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))); + } +}