From bffbdb24b3fc264b69c0499eee23e45190488c14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?St=C3=A5le=20Pedersen?= Date: Mon, 20 Oct 2025 19:32:25 +0200 Subject: [PATCH 1/5] added a test that usually triggers a deadlock. refactored calculateDatapoints and runChangeDetection removed calls to executeBlocking, only one useage of messageBus remaining --- .../horreum/svc/AlertingServiceImpl.java | 413 ++++++++++-------- .../tools/horreum/svc/EventAggregator.java | 2 +- .../tools/horreum/svc/SchemaServiceImpl.java | 3 - .../tools/horreum/svc/ServiceMediator.java | 5 +- .../tools/horreum/svc/TestServiceImpl.java | 24 +- .../io/hyperfoil/tools/horreum/svc/Util.java | 14 - .../tools/horreum/svc/BaseServiceTest.java | 4 - .../tools/horreum/svc/TestServiceTest.java | 89 ++-- 8 files changed, 289 insertions(+), 265 deletions(-) diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java index 67e4bff60..252b60e22 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java @@ -38,6 +38,7 @@ import jakarta.ws.rs.core.Response; import org.eclipse.microprofile.config.inject.ConfigProperty; +import org.graalvm.polyglot.Value; import org.hibernate.Session; import org.hibernate.query.NativeQuery; import org.hibernate.type.StandardBasicTypes; @@ -54,7 +55,6 @@ import io.hyperfoil.tools.horreum.api.data.changeDetection.ChangeDetectionModelType; import io.hyperfoil.tools.horreum.api.internal.services.AlertingService; import io.hyperfoil.tools.horreum.bus.AsyncEventChannels; -import io.hyperfoil.tools.horreum.bus.BlockingTaskDispatcher; import io.hyperfoil.tools.horreum.changedetection.ChangeDetectionException; import io.hyperfoil.tools.horreum.changedetection.ChangeDetectionModel; import io.hyperfoil.tools.horreum.changedetection.ChangeDetectionModelResolver; @@ -202,9 +202,6 @@ AND EXISTS ( @Inject EntityManager em; - @Inject - BlockingTaskDispatcher messageBus; - @Inject SecurityIdentity identity; @@ -404,7 +401,43 @@ public String fullName() { @Transactional void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Recalculation recalculation) { Set missingValueVariables = new HashSet<>(); - List values = session.createNativeQuery(LOOKUP_VARIABLES, Tuple.class) + + List values = fetchVariableValues(dataset); + if (debug) + logFetchedValuesIfDebug(dataset, values); + + Instant timestamp = resolveTimestamp(dataset); + + Util.evaluateWithCombinationFunction(values, + data -> data.calculation, + data -> data.value, + (data, result) -> handleCalculatedValue(dataset, timestamp, data, result, notify, recalculation, + missingValueVariables), + data -> handleNonCalculatedValue(dataset, timestamp, data, notify, recalculation, missingValueVariables), + (data, exception, code) -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, + "Evaluation of variable %s failed: '%s' Code:
%s
", data.fullName(), exception.getMessage(), + code), + output -> logCalculationMessage(dataset, PersistentLogDAO.DEBUG, + "Output while calculating variable: 
%s
", output)); + + if (!missingValueVariables.isEmpty()) { + MissingValuesEvent event = new MissingValuesEvent(dataset.getInfo(), missingValueVariables, notify); + if (mediator.testMode()) { + mediator.publishEvent(AsyncEventChannels.DATASET_MISSING_VALUES, dataset.testid, event); + } + mediator.missingValuesDataset(event); + } + DataPoint.DatasetProcessedEvent event = new DataPoint.DatasetProcessedEvent(DatasetMapper.fromInfo(dataset.getInfo()), + notify); + if (mediator.testMode()) { + Util.registerTxSynchronization(tm, + txStatus -> mediator.publishEvent(AsyncEventChannels.DATAPOINT_PROCESSED, dataset.testid, event)); + } + mediator.dataPointsProcessed(event); + } + + private List fetchVariableValues(DatasetDAO dataset) { + return session.createNativeQuery(LOOKUP_VARIABLES, Tuple.class) .setParameter(1, dataset.testid) .setParameter(2, dataset.id) .addScalar("variableId", StandardBasicTypes.INTEGER) @@ -424,12 +457,16 @@ void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Reca return data; }) .getResultList(); - if (debug) { - for (VariableData data : values) { - logCalculationMessage(dataset, PersistentLogDAO.DEBUG, "Fetched value for variable %s: 
%s
", - data.fullName(), data.value); - } + } + + private void logFetchedValuesIfDebug(DatasetDAO dataset, List values) { + for (VariableData data : values) { + logCalculationMessage(dataset, PersistentLogDAO.DEBUG, + "Fetched value for variable %s: 
%s
", data.fullName(), data.value); } + } + + private Instant resolveTimestamp(DatasetDAO dataset) { List timestampList = session .createNativeQuery(LOOKUP_TIMESTAMP, Object[].class) .setParameter(1, dataset.testid) @@ -437,96 +474,94 @@ void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Reca .addScalar("timeline_function", StandardBasicTypes.TEXT) .addScalar("value", JsonBinaryType.INSTANCE) .getResultList(); - Instant timestamp = dataset.start; - if (!timestampList.isEmpty()) { - String timestampFunction = (String) timestampList.get(0)[0]; - JsonNode value = (JsonNode) timestampList.get(0)[1]; - if (timestampFunction != null && !timestampFunction.isBlank()) { - value = Util.evaluateOnce(timestampFunction, value, Util::convertToJson, - (code, throwable) -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, - "Evaluation of timestamp failed: '%s' Code: 
%s
", throwable.getMessage(), - code), - output -> logCalculationMessage(dataset, PersistentLogDAO.DEBUG, - "Output while calculating timestamp: 
%s
", output)); + + if (timestampList.isEmpty()) { + return dataset.start; + } + + String timestampFunction = (String) timestampList.get(0)[0]; + JsonNode value = (JsonNode) timestampList.get(0)[1]; + + if (timestampFunction != null && !timestampFunction.isBlank()) { + value = Util.evaluateOnce( + timestampFunction, + value, + Util::convertToJson, + (code, throwable) -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, + "Evaluation of timestamp failed: '%s' Code: 
%s
", throwable.getMessage(), + code), + output -> logCalculationMessage(dataset, PersistentLogDAO.DEBUG, + "Output while calculating timestamp: 
%s
", output)); + } + + Instant ts = Util.toInstant(value); + if (ts == null) { + logCalculationMessage(dataset, PersistentLogDAO.ERROR, + "Cannot parse timestamp, must be number or ISO-8601 timestamp: %s", value); + ts = dataset.start; + } + return ts; + } + + private void handleCalculatedValue(DatasetDAO dataset, Instant timestamp, VariableData data, + Value result, boolean notify, Recalculation recalculation, Set missingValueVariables) { + Double value = Util.toDoubleOrNull(result, + error -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, + "Evaluation of variable %s failed: %s", data.fullName(), error), + info -> logCalculationMessage(dataset, PersistentLogDAO.INFO, + "Evaluation of variable %s: %s", data.fullName(), info)); + + if (value != null) { + createDataPoint(dataset, timestamp, data.variableId, value, notify, recalculation); + return; + } + + if (recalculation != null) { + recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); + } + missingValueVariables.add(data.fullName()); + } + + private void handleNonCalculatedValue(DatasetDAO dataset, Instant timestamp, VariableData data, boolean notify, + Recalculation recalculation, Set missingValueVariables) { + if (data.numLabels > 1) { + logCalculationMessage(dataset, PersistentLogDAO.WARN, + "Variable %s has more than one label (%s) but no calculation function.", data.fullName(), + data.value.fieldNames()); + } + + if (data.value == null || data.value.isNull()) { + logCalculationMessage(dataset, PersistentLogDAO.INFO, + "Null value for variable %s - datapoint is not created", data.fullName()); + if (recalculation != null) { + recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); } - timestamp = Util.toInstant(value); - if (timestamp == null) { - logCalculationMessage(dataset, PersistentLogDAO.ERROR, - "Cannot parse timestamp, must be number or ISO-8601 timestamp: %s", value); - timestamp = dataset.start; + missingValueVariables.add(data.fullName()); + return; + } + + Double value = null; + if (data.value.isNumber()) { + value = data.value.asDouble(); + } else if (data.value.isTextual()) { + try { + value = Double.parseDouble(data.value.asText()); + } catch (NumberFormatException ignored) { + // keep value as null } } - Instant finalTimestamp = timestamp; - Util.evaluateWithCombinationFunction(values, data -> data.calculation, data -> data.value, - (data, result) -> { - Double value = Util.toDoubleOrNull(result, - error -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, - "Evaluation of variable %s failed: %s", data.fullName(), error), - info -> logCalculationMessage(dataset, PersistentLogDAO.INFO, "Evaluation of variable %s: %s", - data.fullName(), info)); - if (value != null) { - createDataPoint(dataset, finalTimestamp, data.variableId, value, notify, recalculation); - } else { - if (recalculation != null) { - recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); - } - missingValueVariables.add(data.fullName()); - } - }, - data -> { - if (data.numLabels > 1) { - logCalculationMessage(dataset, PersistentLogDAO.WARN, - "Variable %s has more than one label (%s) but no calculation function.", data.fullName(), - data.value.fieldNames()); - } - if (data.value == null || data.value.isNull()) { - logCalculationMessage(dataset, PersistentLogDAO.INFO, - "Null value for variable %s - datapoint is not created", data.fullName()); - if (recalculation != null) { - recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); - } - missingValueVariables.add(data.fullName()); - return; - } - Double value = null; - if (data.value.isNumber()) { - value = data.value.asDouble(); - } else if (data.value.isTextual()) { - try { - value = Double.parseDouble(data.value.asText()); - } catch (NumberFormatException e) { - // ignore - } - } - if (value == null) { - logCalculationMessage(dataset, PersistentLogDAO.ERROR, - "Cannot turn %s into a floating-point value for variable %s", data.value, data.fullName()); - if (recalculation != null) { - recalculation.errors++; - } - missingValueVariables.add(data.fullName()); - } else { - createDataPoint(dataset, finalTimestamp, data.variableId, value, notify, recalculation); - } - }, - (data, exception, code) -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, - "Evaluation of variable %s failed: '%s' Code:
%s
", data.fullName(), exception.getMessage(), - code), - output -> logCalculationMessage(dataset, PersistentLogDAO.DEBUG, - "Output while calculating variable: 
%s
", output)); - if (!missingValueVariables.isEmpty()) { - MissingValuesEvent event = new MissingValuesEvent(dataset.getInfo(), missingValueVariables, notify); - if (mediator.testMode()) - mediator.publishEvent(AsyncEventChannels.DATASET_MISSING_VALUES, dataset.testid, event); - mediator.missingValuesDataset(event); + if (value == null) { + logCalculationMessage(dataset, PersistentLogDAO.ERROR, + "Cannot turn %s into a floating-point value for variable %s", data.value, data.fullName()); + if (recalculation != null) { + recalculation.errors++; + } + missingValueVariables.add(data.fullName()); + return; } - DataPoint.DatasetProcessedEvent event = new DataPoint.DatasetProcessedEvent(DatasetMapper.fromInfo(dataset.getInfo()), - notify); - if (mediator.testMode()) - Util.registerTxSynchronization(tm, - txStatus -> mediator.publishEvent(AsyncEventChannels.DATAPOINT_PROCESSED, dataset.testid, event)); - mediator.dataPointsProcessed(event); + + createDataPoint(dataset, timestamp, data.variableId, value, notify, recalculation); } @Transactional @@ -546,12 +581,13 @@ void createDataPoint(DatasetDAO dataset, Instant timestamp, int variableId, doub Parameters.with("dataset", dataset).and("variable", variableDAO)).firstResult(); } if (dataPoint != null) { - DataPoint.Event event = new DataPoint.Event(dataPoint.id, dataset.id, notify); - onNewDataPoint(event, recalculation.lastDatapoint); //Test failure if we do not start a new thread and new tx + onNewDataPoint(dataPoint, recalculation.lastDatapoint, notify); - if (mediator.testMode()) + if (mediator.testMode()) { + DataPoint.Event event = new DataPoint.Event(dataPoint.id, dataset.id, notify); Util.registerTxSynchronization(tm, txStatus -> mediator.publishEvent(AsyncEventChannels.DATAPOINT_NEW, dataset.testid, event)); + } } else { Log.debugf("DataPoint for dataset %d, variable %d, timestamp %s, value %f not found", dataset.id, variableId, timestamp, value); @@ -591,14 +627,13 @@ private void logChangeDetectionMessage(int testId, int datasetId, int level, Str @WithRoles(extras = Roles.HORREUM_SYSTEM) @Transactional - void onNewDataPoint(DataPoint.Event event, boolean lastDatapoint) { - DataPointDAO dataPoint = DataPointDAO.findById(event.dataPointId); + void onNewDataPoint(DataPointDAO dataPoint, boolean lastDatapoint, boolean notify) { if (dataPoint.variable != null) { VariableDAO variable = dataPoint.variable; Log.debugf("Processing new datapoint for dataset %d at %s, variable %d (%s), value %f", - event.datasetId, dataPoint.timestamp, variable.id, variable.name, dataPoint.value); + dataPoint.dataset.id, dataPoint.timestamp, variable.id, variable.name, dataPoint.value); - FingerprintDAO fingerprint = FingerprintDAO. findByIdOptional(event.datasetId).orElse(null); + FingerprintDAO fingerprint = FingerprintDAO. findByIdOptional(dataPoint.dataset.id).orElse(null); JsonNode fpNode = fingerprint != null ? fingerprint.fingerprint : null; Integer fpHash = fingerprint != null ? fingerprint.fpHash : null; @@ -612,10 +647,14 @@ void onNewDataPoint(DataPoint.Event event, boolean lastDatapoint) { return current; } }); - runChangeDetection(variable.id, variable.testId, fpNode, fpHash, event.notify, true, lastDatapoint); + //before we run change detection, lets see how many changes there are for this dataset + long count = ChangeDAO.count("dataset.id", dataPoint.dataset.id); + Log.infof("There are %d changes for dataset %d and variable %d", count, dataPoint.dataset.id, variable.id); + + runChangeDetection(variable.id, variable.testId, fpNode, fpHash, notify, true, lastDatapoint); } else { Log.warnf("Could not process new datapoint for dataset %d when the supplied variable or id reference is null ", - event.datasetId); + dataPoint.dataset.id); } } @@ -627,10 +666,35 @@ void tryRunChangeDetection(int variableId, int testId, JsonNode fingerprint, Int @Transactional void runChangeDetection(int variableId, int testId, JsonNode fingerprint, Integer fpHash, boolean notify, - boolean expectExists, - boolean lastDatapoint) { + boolean expectExists, boolean lastDatapoint) { UpTo valid = validUpTo.get(new VarAndFingerprint(variableId, fpHash)); - Instant nextTimestamp = session.createNativeQuery( + Instant nextTimestamp = fetchNextTimestamp(variableId, fpHash, valid); + + if (nextTimestamp == null) { + Log.debugf("No further datapoints for change detection"); + return; + } + if (valid != null) { + deleteChangesByTimeframe(variableId, fpHash, valid); + } + + Instant changeTimestamp = fetchLastChangeTimestamp(variableId, fpHash, valid); + List dataPoints = fetchDataPoints(variableId, fpHash, changeTimestamp, nextTimestamp); + + if (dataPoints.isEmpty()) { + if (expectExists) { + Log.warn("The published datapoint should be already in the list"); + } + } else { + processChangeDetection(variableId, testId, fingerprint, notify, lastDatapoint, dataPoints); + } + + validateUpTo(variableId, fpHash, nextTimestamp); + tryRunChangeDetection(variableId, testId, fingerprint, fpHash, notify); + } + + private Instant fetchNextTimestamp(int variableId, Integer fpHash, UpTo valid) { + return session.createNativeQuery( "SELECT MIN(timestamp) FROM datapoint dp LEFT JOIN fingerprint fp ON dp.dataset_id = fp.dataset_id " + "WHERE dp.variable_id = :variableId " + "AND (timestamp > :validTimestamp OR (timestamp = :validTimestamp AND :exclusive)) " + @@ -641,46 +705,38 @@ void runChangeDetection(int variableId, int testId, JsonNode fingerprint, Intege .setParameter("exclusive", valid == null || !valid.inclusive) .setParameter("fpHash", fpHash) .getResultStream().filter(Objects::nonNull).findFirst().orElse(null); - if (nextTimestamp == null) { - // this is the exit clause to stops the recursive invocation - Log.debugf("No further datapoints for change detection"); - return; - } + } - // this should happen only after reboot, let's start with last change - // FIXME: this is happening also when updating a single label for a schema - if (valid != null) { - int numDeleted = session.createNativeQuery(DELETE_CHANGES_BY_TIMEFRAME, int.class) - .setParameter("variableId", variableId) - .setParameter("validTimestamp", valid.timestamp, StandardBasicTypes.INSTANT) - .setParameter("exclusive", !valid.inclusive) - .setParameter("fpHash", fpHash) - .executeUpdate(); - Log.debugf("Deleted %d changes %s %s for variable %d, fingerprint %s", numDeleted, valid.inclusive ? ">" : ">=", - valid.timestamp, variableId, fpHash); - } - - var changeQuery = session - .createQuery("SELECT c FROM Change c LEFT JOIN Fingerprint fp ON c.dataset.id = fp.datasetId " + + private void deleteChangesByTimeframe(int variableId, Integer fpHash, UpTo valid) { + int numDeleted = session.createNativeQuery(DELETE_CHANGES_BY_TIMEFRAME, int.class) + .setParameter("variableId", variableId) + .setParameter("validTimestamp", valid.timestamp, StandardBasicTypes.INSTANT) + .setParameter("exclusive", !valid.inclusive) + .setParameter("fpHash", fpHash) + .executeUpdate(); + Log.debugf("Deleted %d changes %s %s for variable %d, fingerprint %s", numDeleted, valid.inclusive ? ">" : ">=", + valid.timestamp, variableId, fpHash); + } + + private Instant fetchLastChangeTimestamp(int variableId, Integer fpHash, UpTo valid) { + var changeQuery = session.createQuery( + "SELECT c FROM Change c LEFT JOIN Fingerprint fp ON c.dataset.id = fp.datasetId " + "WHERE c.variable.id = :variableId AND fp.fpHash = :fpHash " + "AND (c.timestamp < :validTimestamp OR (c.timestamp = :validTimestamp AND :exclusive = TRUE)) " + - "ORDER by c.timestamp DESC", ChangeDAO.class); - changeQuery - .setParameter("variableId", variableId) + "ORDER by c.timestamp DESC", + ChangeDAO.class); + changeQuery.setParameter("variableId", variableId) .setParameter("validTimestamp", valid != null ? valid.timestamp : VERY_DISTANT_FUTURE) .setParameter("exclusive", valid == null || valid.inclusive) .setParameter("fpHash", fpHash); ChangeDAO lastChange = changeQuery.setMaxResults(1).getResultStream().findFirst().orElse(null); + return lastChange != null ? lastChange.timestamp : LONG_TIME_AGO; + } - Instant changeTimestamp = LONG_TIME_AGO; - if (lastChange != null) { - Log.debugf("Filtering DP between %s (change %d) and %s", lastChange.timestamp, lastChange.id, nextTimestamp); - changeTimestamp = lastChange.timestamp; - } - - List dataPoints = session.createQuery( + private List fetchDataPoints(int variableId, Integer fpHash, Instant changeTimestamp, Instant nextTimestamp) { + return session.createQuery( "SELECT dp FROM DataPoint dp LEFT JOIN Fingerprint fp ON dp.dataset.id = fp.datasetId " + - "JOIN dp.dataset " + // ignore datapoints (that were not deleted yet) from deleted datasets + "JOIN dp.dataset " + "WHERE dp.variable.id = :variableId AND dp.timestamp BETWEEN :changeTimestamp AND :nextTimestamp " + "AND fp.fpHash = :fpHash " + "ORDER BY dp.timestamp DESC, dp.dataset.id DESC", @@ -690,50 +746,39 @@ void runChangeDetection(int variableId, int testId, JsonNode fingerprint, Intege .setParameter("nextTimestamp", nextTimestamp) .setParameter("fpHash", fpHash) .getResultList(); - // Last datapoint is already in the list - if (dataPoints.isEmpty()) { - if (expectExists) { - Log.warn("The published datapoint should be already in the list"); + } + + private void processChangeDetection(int variableId, int testId, JsonNode fingerprint, boolean notify, + boolean lastDatapoint, List dataPoints) { + int datasetId = dataPoints.get(0).getDatasetId(); + for (ChangeDetectionDAO detection : ChangeDetectionDAO. find("variable.id", variableId).list()) { + ChangeDetectionModel model = modelResolver.getModel(ChangeDetectionModelType.fromString(detection.model)); + if (model == null) { + logChangeDetectionMessage(variableId, datasetId, PersistentLogDAO.ERROR, + "Cannot find change detection model %s", detection.model); + continue; } - } else { - int datasetId = dataPoints.get(0).getDatasetId(); - for (ChangeDetectionDAO detection : ChangeDetectionDAO. find("variable.id", variableId) - .list()) { - ChangeDetectionModel model = modelResolver.getModel(ChangeDetectionModelType.fromString(detection.model)); - if (model == null) { - logChangeDetectionMessage(variableId, datasetId, PersistentLogDAO.ERROR, - "Cannot find change detection model %s", detection.model); - continue; - } - //Only run bulk models on the last datapoint, otherwise run on every datapoint - if (model.getType() == ModelType.CONTINOUS || (model.getType() == ModelType.BULK && lastDatapoint)) { - try { - model.analyze(dataPoints, detection.config, change -> { - logChangeDetectionMessage(testId, datasetId, PersistentLogDAO.DEBUG, - "Change %s detected using datapoints %s", change, reversedAndLimited(dataPoints)); - em.persist(change); - // Hibernate.initialize(change.dataset.run.id); - String testName = TestDAO. findByIdOptional(testId).map(test -> test.name) - .orElse(""); - Change.Event event = new Change.Event(ChangeMapper.from(change), testId, testName, notify); - if (mediator.testMode()) - Util.registerTxSynchronization(tm, txStatus -> mediator - .publishEvent(AsyncEventChannels.CHANGE_NEW, change.dataset.testid, event)); - mediator.executeBlocking(() -> mediator.newChange(event)); - }); - } catch (ChangeDetectionException e) { - new ChangeDetectionLogDAO(variableId, fingerprint, PersistentLogDAO.ERROR, e.getLocalizedMessage()) - .persist(); - Log.error("An error occurred while running change detection!", e); - } + if (model.getType() == ModelType.CONTINOUS || (model.getType() == ModelType.BULK && lastDatapoint)) { + try { + model.analyze(dataPoints, detection.config, change -> { + logChangeDetectionMessage(testId, datasetId, PersistentLogDAO.DEBUG, + "Change %s detected using datapoints %s", change, reversedAndLimited(dataPoints)); + em.persist(change); + String testName = TestDAO. findByIdOptional(testId).map(test -> test.name).orElse(""); + Change.Event event = new Change.Event(ChangeMapper.from(change), testId, testName, notify); + if (mediator.testMode()) { + Util.registerTxSynchronization(tm, txStatus -> mediator + .publishEvent(AsyncEventChannels.CHANGE_NEW, change.dataset.testid, event)); + } + mediator.newChange(event); + }); + } catch (ChangeDetectionException e) { + new ChangeDetectionLogDAO(variableId, fingerprint, PersistentLogDAO.ERROR, e.getLocalizedMessage()) + .persist(); + Log.error("An error occurred while running change detection!", e); } } } - Util.doAfterCommit(tm, () -> { - validateUpTo(variableId, fpHash, nextTimestamp); - //assume not last datapoint if we have found more - messageBus.executeForTest(testId, () -> tryRunChangeDetection(variableId, testId, fingerprint, fpHash, notify)); - }); } private void validateUpTo(int variableId, int fpHash, Instant timestamp) { @@ -983,9 +1028,7 @@ public void recalculateDatapoints(int testId, boolean notify, throw ServiceException.forbidden("This user cannot trigger the recalculation"); } - messageBus.executeForTest(testId, () -> { - startRecalculation(testId, notify, debug, clearDatapoints == null || clearDatapoints, from, to); - }); + startRecalculation(testId, notify, debug, clearDatapoints == null || clearDatapoints, from, to); } // It doesn't make sense to limit access to particular user when doing the recalculation, @@ -1248,11 +1291,7 @@ public int updateMissingDataRule(int testId, MissingDataRule dto) { } // The recalculations are executed in independent transactions, therefore we need to make sure that // this rule is committed in DB before starting to reevaluate it. - Util.doAfterCommit(tm, () -> { - messageBus.executeForTest(testId, () -> { - recalculateMissingDataRules(testId, rule); - }); - }); + recalculateMissingDataRules(testId, rule); return rule.id; } @@ -1268,7 +1307,7 @@ void recalculateMissingDataRules(int testId, MissingDataRuleDAO rule) { } @WithRoles(extras = Roles.HORREUM_SYSTEM) - @Transactional(Transactional.TxType.REQUIRES_NEW) + @Transactional void recalculateMissingDataRule(int datasetId, Instant timestamp, MissingDataRuleDAO rule) { JsonNode value = (JsonNode) em.createNativeQuery(LOOKUP_LABEL_VALUE_FOR_RULE) .setParameter(1, datasetId).setParameter(2, rule.id) diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/EventAggregator.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/EventAggregator.java index c19c2abcb..e9682553b 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/EventAggregator.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/EventAggregator.java @@ -51,7 +51,7 @@ void handleDatasetChanges() { if (next == null) { return; } else if (next.emitTimestamp() <= now) { - mediator.executeBlocking(() -> mediator.newDatasetChanges(next)); + mediator.newDatasetChanges(next); datasetChanges.remove(next.dataset.id); } else { if (timerId >= 0) { diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java index 73c9247f7..62977edc5 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java @@ -55,7 +55,6 @@ import io.hyperfoil.tools.horreum.api.data.Transformer; import io.hyperfoil.tools.horreum.api.services.SchemaService; import io.hyperfoil.tools.horreum.bus.AsyncEventChannels; -import io.hyperfoil.tools.horreum.bus.BlockingTaskDispatcher; import io.hyperfoil.tools.horreum.entity.ValidationErrorDAO; import io.hyperfoil.tools.horreum.entity.data.DatasetDAO; import io.hyperfoil.tools.horreum.entity.data.LabelDAO; @@ -126,8 +125,6 @@ SELECT substring(jsonb_path_query(schema, '$.**.\"$ref\" ? (! (@ starts with \"# @Inject ServiceMediator mediator; - @Inject - BlockingTaskDispatcher messageBus; @Inject Session session; diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java index 8235e3756..2ad382d74 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java @@ -117,10 +117,6 @@ public class ServiceMediator { public ServiceMediator() { } - void executeBlocking(Runnable runnable) { - Util.executeBlocking(vertx, runnable); - } - boolean testMode() { return testMode; } @@ -157,6 +153,7 @@ void propagatedDatasetDelete(int datasetId) { datasetService.deleteDataset(datasetId); } + @Transactional void newChange(Change.Event event) { actionService.onNewChange(event); aggregator.onNewChange(event); diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/TestServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/TestServiceImpl.java index b7f0f5200..7713a0f74 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/TestServiceImpl.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/TestServiceImpl.java @@ -707,21 +707,19 @@ public void recalculateTestDatasets(int testId) { Log.debugf("Recalculate Datasets for run %d - forcing recalculation for test %d (%s)", runId, testId, test.name); - mediator.executeBlocking(() -> { - int newDatasets = 0; - try { - newDatasets = mediator.transform(runId, true); - } finally { - synchronized (status) { - status.finished++; - status.datasets += newDatasets; - if (status.finished == status.totalRuns) { - Log.infof("Datasets recalculation for test %d (%s) completed", testId, test.name); - recalculations.remove(testId, status); - } + int newDatasets = 0; + try { + newDatasets = mediator.transform(runId, true); + } finally { + synchronized (status) { + status.finished++; + status.datasets += newDatasets; + if (status.finished == status.totalRuns) { + Log.infof("Datasets recalculation for test %d (%s) completed", testId, test.name); + recalculations.remove(testId, status); } } - }); + } } } } diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/Util.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/Util.java index b27494280..a1d911f94 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/Util.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/Util.java @@ -528,20 +528,6 @@ public static String explainCauses(Throwable e) { return causes.toString(); } - public static void executeBlocking(Vertx vertx, Runnable runnable) { - Runnable wrapped = wrapForBlockingExecution(runnable); - vertx.executeBlocking(promise -> { - try { - wrapped.run(); - } catch (Exception e) { - Log.error("Failed to execute blocking task", e); - } finally { - promise.complete(); - } - }, result -> { - }); - } - public static Runnable wrapForBlockingExecution(Runnable runnable) { // CDI needs to be propagated - without that the interceptors wouldn't run. // Without thread context propagation we would get an exception in Run.findById, though the interceptors would be invoked correctly. diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/BaseServiceTest.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/BaseServiceTest.java index 98aa8a2c7..3872bcbe0 100644 --- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/BaseServiceTest.java +++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/BaseServiceTest.java @@ -62,7 +62,6 @@ import io.hyperfoil.tools.horreum.api.services.RunService; import io.hyperfoil.tools.horreum.api.services.TestService; import io.hyperfoil.tools.horreum.bus.AsyncEventChannels; -import io.hyperfoil.tools.horreum.bus.BlockingTaskDispatcher; import io.hyperfoil.tools.horreum.entity.ExperimentProfileDAO; import io.hyperfoil.tools.horreum.entity.FingerprintDAO; import io.hyperfoil.tools.horreum.entity.alerting.*; @@ -100,9 +99,6 @@ public class BaseServiceTest { @Inject protected RoleManager roleManager; - @Inject - BlockingTaskDispatcher messageBus; - @Inject ObjectMapper mapper; diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java index fd2389780..f81366ee5 100644 --- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java +++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java @@ -642,39 +642,44 @@ void testUpdateLabelsOnRuns() throws IOException, InterruptedException { assertEquals(11, labels2.size()); BlockingQueue newDataPoints = serviceMediator.getEventQueue(AsyncEventChannels.DATAPOINT_NEW, testId); + BlockingQueue updates = serviceMediator.getEventQueue(AsyncEventChannels.DATASET_UPDATED_LABELS, testId); List runIds = new ArrayList<>(); for (int i = 1; i < 5; i++) { - - BlockingQueue events = serviceMediator.getEventQueue(AsyncEventChannels.RUN_NEW, testId); - Run run = new Run(); - run.testid = testId; - run.data = mapper.readTree(p.resolve("quarkus_sb_run" + i + ".json").toFile()); - run.start = Instant.parse(run.data.get("timing").get("start").asText()); - run.stop = Instant.parse(run.data.get("timing").get("stop").asText()); - run.owner = "foo-team"; - - Response response = jsonRequest() - .auth() - .oauth2(getUploaderToken()) - .body(run) - .post("/api/run/test"); - assertEquals(202, response.statusCode()); - assertEquals(1, response.getBody().as(List.class).size()); - - runIds.addAll(response.getBody().as(List.class)); - assertFalse(runIds.isEmpty()); - assertNotNull(events.poll(10, TimeUnit.SECONDS)); + for (int j = 1; j < 11; j++) { + BlockingQueue events = serviceMediator.getEventQueue(AsyncEventChannels.RUN_NEW, testId); + Run run = new Run(); + run.testid = testId; + run.data = mapper.readTree(p.resolve("quarkus_sb_run" + i + ".json").toFile()); + run.start = Instant.parse(run.data.get("timing").get("start").asText()).plusSeconds(j); + run.stop = Instant.parse(run.data.get("timing").get("stop").asText()).plusSeconds(j); + run.owner = "foo-team"; + + Response response = jsonRequest() + .auth() + .oauth2(getUploaderToken()) + .body(run) + .post("/api/run/test"); + assertEquals(202, response.statusCode()); + assertEquals(1, response.getBody().as(List.class).size()); + + runIds.addAll(response.getBody().as(List.class)); + assertFalse(runIds.isEmpty()); + assertNotNull(events.poll(10, TimeUnit.SECONDS)); + } } //make sure we've generating datapoints assertNotNull(newDataPoints.poll(10, TimeUnit.SECONDS)); - //we should have 20 datasets now in total - assertEquals(20, DatasetDAO.findAll().count()); + //we should have 200 datasets now in total + assertEquals(200, DatasetDAO.findAll().count()); //give Horreum some time to calculate LabelValues - Thread.sleep(2000); + for (int i = 0; i < 200; i++) { + assertNotNull(updates.poll(20, TimeUnit.SECONDS)); + } + System.out.println("Number of LabelValues: " + LabelValueDAO.count()); - assertEquals(currentNumberOfLabelValues + 220, LabelValueDAO.count()); + assertEquals(currentNumberOfLabelValues + 2200, LabelValueDAO.count()); OptionalInt maxId; try (Stream datapoints = DataPointDAO.findAll().stream()) { @@ -686,21 +691,27 @@ void testUpdateLabelsOnRuns() throws IOException, InterruptedException { .then().statusCode(200).extract().body().jsonPath().getList(".", Integer.class); assertEquals(1, updatedLabels.size()); - Thread.sleep(2000); - try (Stream datapoints = DataPointDAO.findAll().stream()) { - maxId = datapoints.mapToInt(n -> n.id).max(); - } - - //lets update 11 labels - updatedLabels = jsonRequest().body(labels2).put("/api/schema/" + schemaId2 + "/labels") - .then().statusCode(200).extract().body().jsonPath().getList(".", Integer.class); - assertEquals(11, updatedLabels.size()); - - Thread.sleep(2000); - int oldMaxId = maxId.getAsInt(); - try (Stream datapoints = DataPointDAO.findAll().stream()) { - maxId = datapoints.mapToInt(n -> n.id).max(); + //make sure we get all the updates + for (int i = 0; i < 200; i++) { + assertNotNull(updates.poll(20, TimeUnit.SECONDS)); } - assertEquals(oldMaxId + 80, maxId.getAsInt()); + System.out.println("Number of LabelValues after 1 label update: " + LabelValueDAO.count()); + /* + * try (Stream datapoints = DataPointDAO.findAll().stream()) { + * maxId = datapoints.mapToInt(n -> n.id).max(); + * } + * + * //lets update 11 labels + * updatedLabels = jsonRequest().body(labels2).put("/api/schema/" + schemaId2 + "/labels") + * .then().statusCode(200).extract().body().jsonPath().getList(".", Integer.class); + * assertEquals(11, updatedLabels.size()); + * + * Thread.sleep(2000); + * int oldMaxId = maxId.getAsInt(); + * try (Stream datapoints = DataPointDAO.findAll().stream()) { + * maxId = datapoints.mapToInt(n -> n.id).max(); + * } + * assertEquals(oldMaxId + 80, maxId.getAsInt()); + */ } } From ae75f3d67a3892d6b355ad80ebc77f39dbe214b6 Mon Sep 17 00:00:00 2001 From: barreiro Date: Thu, 6 Nov 2025 12:23:08 +0000 Subject: [PATCH 2/5] Offload change detection from datapoint calculation --- .../horreum/svc/AlertingServiceImpl.java | 199 ++++-------------- .../tools/horreum/svc/ServiceMediator.java | 24 +++ .../src/main/resources/application.properties | 16 ++ .../horreum/svc/AlertingServiceTest.java | 4 +- 4 files changed, 86 insertions(+), 157 deletions(-) diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java index 252b60e22..8f81855bf 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java @@ -12,7 +12,6 @@ import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.TreeMap; @@ -232,10 +231,6 @@ AND EXISTS ( // entries can be removed from timer thread while normally this is updated from one of blocking threads private final ConcurrentMap recalcProgress = new ConcurrentHashMap<>(); - // A new datapoint invalidates anything past its timestamp. Any attempt to recalculate starts - // at the timestamp. - private final ConcurrentMap validUpTo = new ConcurrentHashMap<>(); - static { System.setProperty("polyglot.engine.WarnInterpreterOnly", "false"); } @@ -400,12 +395,12 @@ public String fullName() { @Transactional void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Recalculation recalculation) { - Set missingValueVariables = new HashSet<>(); + Set missingValueVariables = new HashSet<>(); List values = fetchVariableValues(dataset); - if (debug) + if (debug) { logFetchedValuesIfDebug(dataset, values); - + } Instant timestamp = resolveTimestamp(dataset); Util.evaluateWithCombinationFunction(values, @@ -421,7 +416,8 @@ void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Reca "Output while calculating variable: 
%s
", output)); if (!missingValueVariables.isEmpty()) { - MissingValuesEvent event = new MissingValuesEvent(dataset.getInfo(), missingValueVariables, notify); + var variableNames = missingValueVariables.stream().map(VariableData::fullName).collect(Collectors.toSet()); + MissingValuesEvent event = new MissingValuesEvent(dataset.getInfo(), variableNames, notify); if (mediator.testMode()) { mediator.publishEvent(AsyncEventChannels.DATASET_MISSING_VALUES, dataset.testid, event); } @@ -434,6 +430,11 @@ void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Reca txStatus -> mediator.publishEvent(AsyncEventChannels.DATAPOINT_PROCESSED, dataset.testid, event)); } mediator.dataPointsProcessed(event); + + // queue change detection for variables not missing + values.stream().filter(v -> !missingValueVariables.contains(v)) + .map(v -> new ServiceMediator.ChangeDetectionEvent(dataset.testid, dataset.id, v.variableId, timestamp, notify)) + .forEach(mediator::queueChangeDetectionEvent); } private List fetchVariableValues(DatasetDAO dataset) { @@ -504,7 +505,7 @@ private Instant resolveTimestamp(DatasetDAO dataset) { } private void handleCalculatedValue(DatasetDAO dataset, Instant timestamp, VariableData data, - Value result, boolean notify, Recalculation recalculation, Set missingValueVariables) { + Value result, boolean notify, Recalculation recalculation, Set missingValueVariables) { Double value = Util.toDoubleOrNull(result, error -> logCalculationMessage(dataset, PersistentLogDAO.ERROR, "Evaluation of variable %s failed: %s", data.fullName(), error), @@ -519,11 +520,11 @@ private void handleCalculatedValue(DatasetDAO dataset, Instant timestamp, Variab if (recalculation != null) { recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); } - missingValueVariables.add(data.fullName()); + missingValueVariables.add(data); } private void handleNonCalculatedValue(DatasetDAO dataset, Instant timestamp, VariableData data, boolean notify, - Recalculation recalculation, Set missingValueVariables) { + Recalculation recalculation, Set missingValueVariables) { if (data.numLabels > 1) { logCalculationMessage(dataset, PersistentLogDAO.WARN, "Variable %s has more than one label (%s) but no calculation function.", data.fullName(), @@ -536,7 +537,7 @@ private void handleNonCalculatedValue(DatasetDAO dataset, Instant timestamp, Var if (recalculation != null) { recalculation.datasetsWithoutValue.put(dataset.id, dataset.getInfo()); } - missingValueVariables.add(data.fullName()); + missingValueVariables.add(data); return; } @@ -557,7 +558,7 @@ private void handleNonCalculatedValue(DatasetDAO dataset, Instant timestamp, Var if (recalculation != null) { recalculation.errors++; } - missingValueVariables.add(data.fullName()); + missingValueVariables.add(data); return; } @@ -581,8 +582,6 @@ void createDataPoint(DatasetDAO dataset, Instant timestamp, int variableId, doub Parameters.with("dataset", dataset).and("variable", variableDAO)).firstResult(); } if (dataPoint != null) { - onNewDataPoint(dataPoint, recalculation.lastDatapoint, notify); - if (mediator.testMode()) { DataPoint.Event event = new DataPoint.Event(dataPoint.id, dataset.id, notify); Util.registerTxSynchronization(tm, @@ -625,112 +624,55 @@ private void logChangeDetectionMessage(int testId, int datasetId, int level, Str level, "changes", msg).persist(); } - @WithRoles(extras = Roles.HORREUM_SYSTEM) @Transactional - void onNewDataPoint(DataPointDAO dataPoint, boolean lastDatapoint, boolean notify) { - if (dataPoint.variable != null) { - VariableDAO variable = dataPoint.variable; - Log.debugf("Processing new datapoint for dataset %d at %s, variable %d (%s), value %f", - dataPoint.dataset.id, dataPoint.timestamp, variable.id, variable.name, dataPoint.value); - - FingerprintDAO fingerprint = FingerprintDAO. findByIdOptional(dataPoint.dataset.id).orElse(null); - JsonNode fpNode = fingerprint != null ? fingerprint.fingerprint : null; - Integer fpHash = fingerprint != null ? fingerprint.fpHash : null; - - VarAndFingerprint key = new VarAndFingerprint(variable.id, fpHash); - Log.debugf("Invalidating variable %d FP %s timestamp %s, current value is %s", variable.id, fingerprint, - dataPoint.timestamp, validUpTo.get(key)); - validUpTo.compute(key, (ignored, current) -> { - if (current == null || !dataPoint.timestamp.isAfter(current.timestamp)) { - return new UpTo(dataPoint.timestamp, false); - } else { - return current; - } - }); - //before we run change detection, lets see how many changes there are for this dataset - long count = ChangeDAO.count("dataset.id", dataPoint.dataset.id); - Log.infof("There are %d changes for dataset %d and variable %d", count, dataPoint.dataset.id, variable.id); - - runChangeDetection(variable.id, variable.testId, fpNode, fpHash, notify, true, lastDatapoint); - } else { - Log.warnf("Could not process new datapoint for dataset %d when the supplied variable or id reference is null ", - dataPoint.dataset.id); - } - } + void runChangeDetection(int testId, int datasetId, int variableId, Instant timestamp, boolean notify, + boolean expectExists, boolean lastDatapoint) { - @WithRoles(extras = Roles.HORREUM_SYSTEM) - @Transactional - void tryRunChangeDetection(int variableId, int testId, JsonNode fingerprint, Integer fpHash, boolean notify) { - runChangeDetection(variableId, testId, fingerprint, fpHash, notify, false, false); - } + Log.debugf("Performing change detection on test {%d} and variable {%d]", testId, variableId); - @Transactional - void runChangeDetection(int variableId, int testId, JsonNode fingerprint, Integer fpHash, boolean notify, - boolean expectExists, boolean lastDatapoint) { - UpTo valid = validUpTo.get(new VarAndFingerprint(variableId, fpHash)); - Instant nextTimestamp = fetchNextTimestamp(variableId, fpHash, valid); + var fingerprint = FingerprintDAO. findByIdOptional(datasetId); + JsonNode fpNode = fingerprint.map(f -> f.fingerprint).orElse(null); + Integer fpHash = fingerprint.map(f -> f.fpHash).orElse(null); - if (nextTimestamp == null) { - Log.debugf("No further datapoints for change detection"); - return; - } - if (valid != null) { - deleteChangesByTimeframe(variableId, fpHash, valid); - } + deleteChangesByTimeframe(variableId, fpHash, timestamp); - Instant changeTimestamp = fetchLastChangeTimestamp(variableId, fpHash, valid); - List dataPoints = fetchDataPoints(variableId, fpHash, changeTimestamp, nextTimestamp); + Instant changeTimestamp = fetchLastChangeTimestamp(variableId, fpHash, timestamp); + List dataPoints = fetchDataPoints(variableId, fpHash, changeTimestamp, timestamp); if (dataPoints.isEmpty()) { if (expectExists) { Log.warn("The published datapoint should be already in the list"); } } else { - processChangeDetection(variableId, testId, fingerprint, notify, lastDatapoint, dataPoints); + processChangeDetection(variableId, testId, fpNode, notify, lastDatapoint, dataPoints); } - - validateUpTo(variableId, fpHash, nextTimestamp); - tryRunChangeDetection(variableId, testId, fingerprint, fpHash, notify); - } - - private Instant fetchNextTimestamp(int variableId, Integer fpHash, UpTo valid) { - return session.createNativeQuery( - "SELECT MIN(timestamp) FROM datapoint dp LEFT JOIN fingerprint fp ON dp.dataset_id = fp.dataset_id " + - "WHERE dp.variable_id = :variableId " + - "AND (timestamp > :validTimestamp OR (timestamp = :validTimestamp AND :exclusive)) " + - "AND fp.fp_hash = :fpHash", - Instant.class) - .setParameter("variableId", variableId) - .setParameter("validTimestamp", valid != null ? valid.timestamp : LONG_TIME_AGO, StandardBasicTypes.INSTANT) - .setParameter("exclusive", valid == null || !valid.inclusive) - .setParameter("fpHash", fpHash) - .getResultStream().filter(Objects::nonNull).findFirst().orElse(null); } - private void deleteChangesByTimeframe(int variableId, Integer fpHash, UpTo valid) { + private void deleteChangesByTimeframe(int variableId, Integer fpHash, Instant timestamp) { int numDeleted = session.createNativeQuery(DELETE_CHANGES_BY_TIMEFRAME, int.class) .setParameter("variableId", variableId) - .setParameter("validTimestamp", valid.timestamp, StandardBasicTypes.INSTANT) - .setParameter("exclusive", !valid.inclusive) + .setParameter("validTimestamp", timestamp, StandardBasicTypes.INSTANT) + .setParameter("exclusive", false) .setParameter("fpHash", fpHash) .executeUpdate(); - Log.debugf("Deleted %d changes %s %s for variable %d, fingerprint %s", numDeleted, valid.inclusive ? ">" : ">=", - valid.timestamp, variableId, fpHash); + Log.debugf("Deleted %d changes > %s for variable %d, fingerprint %s", numDeleted, timestamp, variableId, fpHash); } - private Instant fetchLastChangeTimestamp(int variableId, Integer fpHash, UpTo valid) { - var changeQuery = session.createQuery( - "SELECT c FROM Change c LEFT JOIN Fingerprint fp ON c.dataset.id = fp.datasetId " + - "WHERE c.variable.id = :variableId AND fp.fpHash = :fpHash " + - "AND (c.timestamp < :validTimestamp OR (c.timestamp = :validTimestamp AND :exclusive = TRUE)) " + - "ORDER by c.timestamp DESC", - ChangeDAO.class); - changeQuery.setParameter("variableId", variableId) - .setParameter("validTimestamp", valid != null ? valid.timestamp : VERY_DISTANT_FUTURE) - .setParameter("exclusive", valid == null || valid.inclusive) - .setParameter("fpHash", fpHash); - ChangeDAO lastChange = changeQuery.setMaxResults(1).getResultStream().findFirst().orElse(null); - return lastChange != null ? lastChange.timestamp : LONG_TIME_AGO; + private Instant fetchLastChangeTimestamp(int variableId, Integer fpHash, Instant timestamp) { + return session.createQuery(""" + SELECT c.timestamp FROM Change c LEFT JOIN Fingerprint fp ON c.dataset.id = fp.datasetId + WHERE c.variable.id = :variableId AND fp.fpHash = :fpHash + AND (c.timestamp < :validTimestamp OR (c.timestamp = :validTimestamp AND :exclusive = TRUE)) + ORDER by c.timestamp DESC + """, Instant.class) + .setParameter("variableId", variableId) + .setParameter("validTimestamp", timestamp != null ? timestamp : VERY_DISTANT_FUTURE) + .setParameter("exclusive", false) + .setParameter("fpHash", fpHash) + .setMaxResults(1) + .getResultStream() + .findFirst() + .orElse(LONG_TIME_AGO); } private List fetchDataPoints(int variableId, Integer fpHash, Instant changeTimestamp, Instant nextTimestamp) { @@ -781,17 +723,6 @@ private void processChangeDetection(int variableId, int testId, JsonNode fingerp } } - private void validateUpTo(int variableId, int fpHash, Instant timestamp) { - validUpTo.compute(new VarAndFingerprint(variableId, fpHash), (ignored, current) -> { - Log.debugf("Attempt %s, valid up to %s", timestamp, current); - if (current == null || !current.timestamp.isAfter(timestamp)) { - return new UpTo(timestamp, true); - } else { - return current; - } - }); - } - private String reversedAndLimited(List list) { int maxIndex = Math.min(list.size() - 1, 20); StringBuilder sb = new StringBuilder("["); @@ -1427,46 +1358,4 @@ public Recalculation(boolean lastDatapoint, boolean clearDatapoints) { this.clearDatapoints = clearDatapoints; } } - - static final class VarAndFingerprint { - final int varId; - final Integer fingerprint; - - VarAndFingerprint(int varId, Integer fingerprint) { - this.varId = varId; - this.fingerprint = fingerprint; - } - - @Override - public boolean equals(Object o) { - if (this == o) - return true; - if (o == null || getClass() != o.getClass()) - return false; - VarAndFingerprint that = (VarAndFingerprint) o; - return varId == that.varId && Objects.equals(fingerprint, that.fingerprint); - } - - @Override - public int hashCode() { - return Objects.hash(varId, fingerprint); - } - } - - private static class UpTo { - final Instant timestamp; - final boolean inclusive; - - private UpTo(Instant timestamp, boolean inclusive) { - this.timestamp = timestamp; - this.inclusive = inclusive; - } - - @Override - public String toString() { - return "{ts=" + timestamp + - ", incl=" + inclusive + - '}'; - } - } } diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java index 2ad382d74..b7003a34c 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java @@ -1,5 +1,6 @@ package io.hyperfoil.tools.horreum.svc; +import java.time.Instant; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -97,6 +98,10 @@ public class ServiceMediator { @Channel("dataset-event-out") Emitter dataSetEmitter; + @OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 10000) + @Channel("change-detection-event-out") + Emitter changeDetectionEmitter; + @OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 10000) @Channel("run-recalc-out") Emitter runEmitter; @@ -237,6 +242,22 @@ public void queueRunUpload(String start, String stop, String test, String owner, runUploadEmitter.send(upload); } + @Incoming("change-detection-event-in") + @Blocking("horreum.change-detection.pool") // the default `ordered = true` ensures messages with the same groupID are executed sequentially + @ActivateRequestContext + public void processChangeDetectionEvent(ChangeDetectionEvent event) { + alertingService.runChangeDetection( + event.testId, event.datasetId, event.variableId, event.timestamp, event.notification, true, true); + } + + @Transactional(Transactional.TxType.NOT_SUPPORTED) + void queueChangeDetectionEvent(ChangeDetectionEvent event) { + OutgoingAmqpMetadata meta = OutgoingAmqpMetadata.builder() + .withGroupId(event.testId + "v" + event.variableId) // serialize messages on a combination of test and variable + .build(); + changeDetectionEmitter.send(Message.of(event).addMetadata(meta)); + } + void dataPointsProcessed(DataPoint.DatasetProcessedEvent event) { experimentService.onDatapointsCreated(event); } @@ -346,4 +367,7 @@ public RunUpload(String start, String stop, String test, String owner, } } + public record ChangeDetectionEvent( + Integer testId, Integer datasetId, Integer variableId, Instant timestamp, boolean notification) { + } } diff --git a/horreum-backend/src/main/resources/application.properties b/horreum-backend/src/main/resources/application.properties index f33630c64..8bd3db918 100644 --- a/horreum-backend/src/main/resources/application.properties +++ b/horreum-backend/src/main/resources/application.properties @@ -34,6 +34,7 @@ quarkus.datasource.jdbc.initial-size=3 # thread pool sizes smallrye.messaging.worker.horreum.dataset.pool.max-concurrency=10 +smallrye.messaging.worker.horreum.change-detection.pool.max-concurrency=5 smallrye.messaging.worker.horreum.run.pool.max-concurrency=6 smallrye.messaging.worker.horreum.schema.pool.max-concurrency=5 @@ -55,6 +56,21 @@ mp.messaging.outgoing.dataset-event-out.container-id=horreum-broker mp.messaging.outgoing.dataset-event-out.link-name=dataset-event mp.messaging.outgoing.dataset-event-out.failure-strategy=modified-failed +# change-detection-event incoming +mp.messaging.incoming.change-detection-event-in.connector=smallrye-amqp +mp.messaging.incoming.change-detection-event-in.address=change-detection-event +mp.messaging.incoming.change-detection-event-in.durable=true +mp.messaging.incoming.change-detection-event-in.container-id=horreum-broker +mp.messaging.incoming.change-detection-event-in.link-name=change-detection-event +mp.messaging.incoming.change-detection-event-in.failure-strategy=modified-failed +# change-detection-event outgoing +mp.messaging.outgoing.change-detection-event-out.connector=smallrye-amqp +mp.messaging.outgoing.change-detection-event-out.address=change-detection-event +mp.messaging.outgoing.change-detection-event-out.durable=true +mp.messaging.outgoing.change-detection-event-out.container-id=horreum-broker +mp.messaging.outgoing.change-detection-event-out.link-name=change-detection-event +mp.messaging.outgoing.change-detection-event-out.failure-strategy=modified-failed + # re-calc incoming mp.messaging.incoming.run-recalc-in.connector=smallrye-amqp mp.messaging.incoming.run-recalc-in.address=run-recalc diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/AlertingServiceTest.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/AlertingServiceTest.java index 21f9e68cc..16f7c4ea9 100644 --- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/AlertingServiceTest.java +++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/AlertingServiceTest.java @@ -153,7 +153,7 @@ public void testChangeDetection(TestInfo info) throws InterruptedException { int run4 = uploadRun(ts + 3, ts + 3, runWithValue(2, schema), test.name); assertValue(datapointQueue, 2); - assertNull(changeQueue.poll(50, TimeUnit.MILLISECONDS)); + assertNull(changeQueue.poll(500, TimeUnit.MILLISECONDS)); uploadRun(ts + 4, ts + 4, runWithValue(3, schema), test.name); assertValue(datapointQueue, 3); @@ -186,7 +186,7 @@ public void testChangeDetection(TestInfo info) throws InterruptedException { assertValue(datapointQueue, 1.5); // mean of previous values is 1.5, now the min is 1.5 => no change - assertNull(changeQueue.poll(50, TimeUnit.MILLISECONDS)); + assertNull(changeQueue.poll(500, TimeUnit.MILLISECONDS)); uploadRun(ts + 6, ts + 6, runWithValue(2, schema), test.name); assertValue(datapointQueue, 2); From 60e0825c78a64b49d0bede2e0ce60dd22e7b1afa Mon Sep 17 00:00:00 2001 From: barreiro Date: Fri, 7 Nov 2025 09:07:52 +0000 Subject: [PATCH 3/5] Batch variables on change detection --- .../horreum/svc/AlertingServiceImpl.java | 44 ++++++++++--------- .../tools/horreum/svc/ServiceMediator.java | 7 +-- 2 files changed, 28 insertions(+), 23 deletions(-) diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java index 8f81855bf..6fdb364c6 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/AlertingServiceImpl.java @@ -182,7 +182,7 @@ DISTINCT ON(variable_id) variable_id AS variable, """ DELETE FROM change cc WHERE NOT cc.confirmed - AND cc.variable_id = :variableId + AND cc.variable_id IN :variableIds AND (cc.timestamp > :validTimestamp OR (cc.timestamp = :validTimestamp AND :exclusive)) AND EXISTS ( SELECT 1 FROM fingerprint fp @@ -432,9 +432,10 @@ void calculateDatapoints(DatasetDAO dataset, boolean notify, boolean debug, Reca mediator.dataPointsProcessed(event); // queue change detection for variables not missing - values.stream().filter(v -> !missingValueVariables.contains(v)) - .map(v -> new ServiceMediator.ChangeDetectionEvent(dataset.testid, dataset.id, v.variableId, timestamp, notify)) - .forEach(mediator::queueChangeDetectionEvent); + values.removeIf(missingValueVariables::contains); + var changeDetectionEvent = new ServiceMediator.ChangeDetectionEvent( + dataset.testid, dataset.id, values.stream().map(v -> v.variableId).toList(), timestamp, notify); + mediator.queueChangeDetectionEvent(changeDetectionEvent); } private List fetchVariableValues(DatasetDAO dataset) { @@ -625,47 +626,50 @@ private void logChangeDetectionMessage(int testId, int datasetId, int level, Str } @Transactional - void runChangeDetection(int testId, int datasetId, int variableId, Instant timestamp, boolean notify, + void runChangeDetection(int testId, int datasetId, Collection variableIds, Instant timestamp, boolean notify, boolean expectExists, boolean lastDatapoint) { - Log.debugf("Performing change detection on test {%d} and variable {%d]", testId, variableId); + Log.debugf("Performing change detection on test {%d} and variables %s after %s", testId, variableIds, timestamp); var fingerprint = FingerprintDAO. findByIdOptional(datasetId); JsonNode fpNode = fingerprint.map(f -> f.fingerprint).orElse(null); Integer fpHash = fingerprint.map(f -> f.fpHash).orElse(null); - deleteChangesByTimeframe(variableId, fpHash, timestamp); + deleteChangesByTimeframe(variableIds, fpHash, timestamp); + Instant changeTimestamp = fetchLastChangeTimestamp(variableIds, fpHash, timestamp); - Instant changeTimestamp = fetchLastChangeTimestamp(variableId, fpHash, timestamp); - List dataPoints = fetchDataPoints(variableId, fpHash, changeTimestamp, timestamp); + for (var variableId : variableIds) { + List dataPoints = fetchDataPoints(variableId, fpHash, changeTimestamp, timestamp); - if (dataPoints.isEmpty()) { - if (expectExists) { - Log.warn("The published datapoint should be already in the list"); + if (dataPoints.isEmpty()) { + if (expectExists) { + Log.warn("The published datapoint should be already in the list"); + } + } else { + processChangeDetection(variableId, testId, fpNode, notify, lastDatapoint, dataPoints); } - } else { - processChangeDetection(variableId, testId, fpNode, notify, lastDatapoint, dataPoints); } } - private void deleteChangesByTimeframe(int variableId, Integer fpHash, Instant timestamp) { + private void deleteChangesByTimeframe(Collection variableIds, Integer fpHash, Instant timestamp) { int numDeleted = session.createNativeQuery(DELETE_CHANGES_BY_TIMEFRAME, int.class) - .setParameter("variableId", variableId) + .setParameter("variableIds", variableIds) .setParameter("validTimestamp", timestamp, StandardBasicTypes.INSTANT) .setParameter("exclusive", false) .setParameter("fpHash", fpHash) .executeUpdate(); - Log.debugf("Deleted %d changes > %s for variable %d, fingerprint %s", numDeleted, timestamp, variableId, fpHash); + Log.debugf("Deleted %d changes > %s for variables %d, fingerprint %s", numDeleted, timestamp, variableIds.toString(), + fpHash); } - private Instant fetchLastChangeTimestamp(int variableId, Integer fpHash, Instant timestamp) { + private Instant fetchLastChangeTimestamp(Collection variableIds, Integer fpHash, Instant timestamp) { return session.createQuery(""" SELECT c.timestamp FROM Change c LEFT JOIN Fingerprint fp ON c.dataset.id = fp.datasetId - WHERE c.variable.id = :variableId AND fp.fpHash = :fpHash + WHERE c.variable.id IN :variableIds AND fp.fpHash = :fpHash AND (c.timestamp < :validTimestamp OR (c.timestamp = :validTimestamp AND :exclusive = TRUE)) ORDER by c.timestamp DESC """, Instant.class) - .setParameter("variableId", variableId) + .setParameter("variableIds", variableIds) .setParameter("validTimestamp", timestamp != null ? timestamp : VERY_DISTANT_FUTURE) .setParameter("exclusive", false) .setParameter("fpHash", fpHash) diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java index b7003a34c..a4fdfb6bf 100644 --- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java +++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java @@ -1,6 +1,7 @@ package io.hyperfoil.tools.horreum.svc; import java.time.Instant; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -247,13 +248,13 @@ public void queueRunUpload(String start, String stop, String test, String owner, @ActivateRequestContext public void processChangeDetectionEvent(ChangeDetectionEvent event) { alertingService.runChangeDetection( - event.testId, event.datasetId, event.variableId, event.timestamp, event.notification, true, true); + event.testId, event.datasetId, event.variableIds, event.timestamp, event.notification, true, true); } @Transactional(Transactional.TxType.NOT_SUPPORTED) void queueChangeDetectionEvent(ChangeDetectionEvent event) { OutgoingAmqpMetadata meta = OutgoingAmqpMetadata.builder() - .withGroupId(event.testId + "v" + event.variableId) // serialize messages on a combination of test and variable + .withGroupId(event.testId.toString()) // serialize messages on a combination of test and variable .build(); changeDetectionEmitter.send(Message.of(event).addMetadata(meta)); } @@ -368,6 +369,6 @@ public RunUpload(String start, String stop, String test, String owner, } public record ChangeDetectionEvent( - Integer testId, Integer datasetId, Integer variableId, Instant timestamp, boolean notification) { + Integer testId, Integer datasetId, Collection variableIds, Instant timestamp, boolean notification) { } } From 18a8e188da0b8f9af3dad9d99d85a922246c63b6 Mon Sep 17 00:00:00 2001 From: barreiro Date: Fri, 7 Nov 2025 09:47:45 +0000 Subject: [PATCH 4/5] uncomment TestServiceTest.testUpdateLabelsOnRuns --- .../tools/horreum/svc/TestServiceTest.java | 33 ++++++++----------- 1 file changed, 14 insertions(+), 19 deletions(-) diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java index f81366ee5..dcef73c5f 100644 --- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java +++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/TestServiceTest.java @@ -36,6 +36,7 @@ import io.hyperfoil.tools.horreum.mapper.VariableMapper; import io.hyperfoil.tools.horreum.server.CloseMe; import io.hyperfoil.tools.horreum.test.*; +import io.quarkus.logging.Log; import io.quarkus.test.common.QuarkusTestResource; import io.quarkus.test.junit.QuarkusTest; import io.quarkus.test.junit.TestProfile; @@ -678,7 +679,7 @@ void testUpdateLabelsOnRuns() throws IOException, InterruptedException { assertNotNull(updates.poll(20, TimeUnit.SECONDS)); } - System.out.println("Number of LabelValues: " + LabelValueDAO.count()); + Log.info("Number of LabelValues: " + LabelValueDAO.count()); assertEquals(currentNumberOfLabelValues + 2200, LabelValueDAO.count()); OptionalInt maxId; @@ -695,23 +696,17 @@ void testUpdateLabelsOnRuns() throws IOException, InterruptedException { for (int i = 0; i < 200; i++) { assertNotNull(updates.poll(20, TimeUnit.SECONDS)); } - System.out.println("Number of LabelValues after 1 label update: " + LabelValueDAO.count()); - /* - * try (Stream datapoints = DataPointDAO.findAll().stream()) { - * maxId = datapoints.mapToInt(n -> n.id).max(); - * } - * - * //lets update 11 labels - * updatedLabels = jsonRequest().body(labels2).put("/api/schema/" + schemaId2 + "/labels") - * .then().statusCode(200).extract().body().jsonPath().getList(".", Integer.class); - * assertEquals(11, updatedLabels.size()); - * - * Thread.sleep(2000); - * int oldMaxId = maxId.getAsInt(); - * try (Stream datapoints = DataPointDAO.findAll().stream()) { - * maxId = datapoints.mapToInt(n -> n.id).max(); - * } - * assertEquals(oldMaxId + 80, maxId.getAsInt()); - */ + + Log.info("Number of LabelValues after 1 label update: " + LabelValueDAO.count()); + maxId = DataPointDAO. streamAll().mapToInt(dp -> dp.id).max(); + + //lets update 11 labels + updatedLabels = jsonRequest().body(labels2).put("/api/schema/" + schemaId2 + "/labels") + .then().statusCode(200).extract().body().jsonPath().getList(".", Integer.class); + assertEquals(11, updatedLabels.size()); + + Log.info("Waiting after labels update"); + Thread.sleep(20 * 1000); + assertEquals(maxId.getAsInt() + 800, DataPointDAO. streamAll().mapToInt(dp -> dp.id).max().getAsInt()); } } From a60cce274831183a3ae7fe97479fd91469c7b43f Mon Sep 17 00:00:00 2001 From: barreiro Date: Thu, 27 Nov 2025 14:03:44 +0000 Subject: [PATCH 5/5] fix testEdivisiveModelAnalyze --- .../hyperfoil/tools/horreum/changedetection/EdivisiveTests.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/changedetection/EdivisiveTests.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/changedetection/EdivisiveTests.java index f97e9f77c..ad14ad2fa 100644 --- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/changedetection/EdivisiveTests.java +++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/changedetection/EdivisiveTests.java @@ -194,7 +194,7 @@ public void testEdvisiveModelAnalyze(TestInfo info) throws Exception { int run10 = uploadRun(ts + 4, ts + 4, runWithValue(10, schema), test.name); assertValue(datapointQueue, 10); - Change.Event changeEvent1 = changeQueue.poll(10, TimeUnit.SECONDS); + Change.Event changeEvent1 = changeQueue.poll(40, TimeUnit.SECONDS); assertNotNull(changeEvent1); testSerialization(changeEvent1, Change.Event.class);