From 25ac2f2b487c48723b0f7d4c6ae8837b6b6de14e Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Thu, 16 Oct 2025 16:32:35 +0300 Subject: [PATCH] added more tests and changed state proto --- .../cf/ctx/state/CalculatedFieldCtx.java | 15 +- ...ValuesAggregationCalculatedFieldState.java | 5 +- .../service/cf/ctx/state/aggregation/agg.json | 65 ------ .../server/utils/CalculatedFieldUtils.java | 4 + ...tValuesAggregationCalculatedFieldTest.java | 188 +++++++++++++++++- common/proto/src/main/proto/queue.proto | 1 + 6 files changed, 196 insertions(+), 82 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/agg.json diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 3a9d75b3d4..f69d611b64 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java @@ -581,8 +581,7 @@ public class CalculatedFieldCtx { } if (calculatedField.getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration thisConfig && other.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration otherConfig - && thisConfig.getDeduplicationIntervalMillis() != otherConfig.getDeduplicationIntervalMillis() - && !thisConfig.getMetrics().equals(otherConfig.getMetrics())) { + && (thisConfig.getDeduplicationIntervalMillis() != otherConfig.getDeduplicationIntervalMillis() || !thisConfig.getMetrics().equals(otherConfig.getMetrics()))) { return true; } return false; @@ -596,7 +595,7 @@ public class CalculatedFieldCtx { var thisConfig = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration(); var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration(); if (!thisConfig.getCreateRules().equals(otherConfig.getCreateRules()) || - !Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) { + !Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) { return true; } } @@ -611,7 +610,7 @@ public class CalculatedFieldCtx { private boolean hasGeofencingZoneGroupConfigurationChanges(CalculatedFieldCtx other) { if (calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration thisConfig - && other.calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration otherConfig) { + && other.calculatedField.getConfiguration() instanceof GeofencingCalculatedFieldConfiguration otherConfig) { return !thisConfig.getZoneGroups().equals(otherConfig.getZoneGroups()); } return false; @@ -663,10 +662,10 @@ public class CalculatedFieldCtx { @Override public String toString() { return "CalculatedFieldCtx{" + - "cfId=" + cfId + - ", cfType=" + cfType + - ", entityId=" + entityId + - '}'; + "cfId=" + cfId + + ", cfType=" + cfType + + ", entityId=" + entityId + + '}'; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/LatestValuesAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/LatestValuesAggregationCalculatedFieldState.java index 1df27b4cbd..6ae023ed1e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/LatestValuesAggregationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/LatestValuesAggregationCalculatedFieldState.java @@ -48,6 +48,7 @@ import java.util.Map.Entry; @Getter public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedFieldState { + @Setter private long lastArgsRefreshTs = -1; @Setter private long lastMetricsEvalTs = -1; @@ -74,6 +75,7 @@ public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedF lastArgsRefreshTs = -1; lastMetricsEvalTs = -1; metrics = null; + inputs.clear(); } @Override @@ -104,7 +106,8 @@ public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedF @Override public ListenableFuture performCalculation(Map updatedArgs, CalculatedFieldCtx ctx) throws Exception { - if (!shouldRecalculate()) { + boolean shouldRecalculate = updatedArgs == null || updatedArgs.isEmpty(); + if (!shouldRecalculate() && !shouldRecalculate) { return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() .result(null) .build()); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/agg.json b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/agg.json deleted file mode 100644 index b39092ca74..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/agg.json +++ /dev/null @@ -1,65 +0,0 @@ -{ - "type": "LATEST_VALUES_AGGREGATION", - "name": "Occupied spaces", - "debugSettings": { - "failuresEnabled": true, - "allEnabled": true, - "allEnabledUntil": 1769907492297 - }, - "entityId": { - "entityType": "ASSET_PROFILE", - "id": "bb8ddd40-a8bc-11f0-869b-e9d81fa6eaf1" - }, - "configuration": { - "type": "LATEST_VALUES_AGGREGATION", - "source": { - "relation": { - "direction": "FROM", - "relationType": "Contains" - }, - "entityProfiles": [ - { - "entityType": "DEVICE_PROFILE", - "id": "d7a05580-a4cf-11f0-87cb-2d6683c4fccf" - } - ] - }, - "inputs": { - "oc": { - "key": "occupied", - "type": "TS_LATEST" - } - }, - "deduplicationIntervalMillis": 10000, - "metrics": { - "totalSpaces": { - "function": "COUNT", - "input": { - "type": "function", - "function" : "return 1;" - } - }, - "occupiedSpaces": { - "function": "COUNT", - "filter": "return oc == true", - "input": { - "type": "key", - "key" : "oc" - } - }, - "freeSpaces": { - "function": "COUNT", - "filter": "return oc == false", - "input": { - "type": "key", - "key" : "oc" - } - } - }, - "output": { - "type": "TIME_SERIES", - "decimals": 2 - }, - "useLatestTsFromInputs": "true" - } -} diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java index 19ad4bfb7c..00f4cbde0f 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -96,6 +96,9 @@ public class CalculatedFieldUtils { .setId(toProto(stateId)) .setType(state.getType().name()); + if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) { + builder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs()); + } state.getArguments().forEach((argName, argEntry) -> { if (argEntry instanceof AggArgumentEntry aggArgumentEntry) { aggArgumentEntry.getAggInputs() @@ -242,6 +245,7 @@ public class CalculatedFieldUtils { arguments.forEach((argName, entityInputs) -> { aggState.getArguments().put(argName, new AggArgumentEntry(entityInputs, false)); }); + aggState.setLastArgsRefreshTs(proto.getLastArgsUpdateTs()); } } diff --git a/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java b/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java index a4b43e362b..8c2c2ca68c 100644 --- a/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java @@ -15,10 +15,13 @@ */ package org.thingsboard.server.cf; +import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; +import lombok.extern.slf4j.Slf4j; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.Tenant; @@ -61,6 +64,7 @@ import static org.awaitility.Awaitility.await; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.cf.CalculatedFieldIntegrationTest.POLL_INTERVAL; +@Slf4j @DaoSqlTest public class LatestValuesAggregationCalculatedFieldTest extends AbstractControllerTest { @@ -341,13 +345,17 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll createEntityRelation(asset2.getId(), device3.getId(), "Contains"); createEntityRelation(asset2.getId(), device4.getId(), "Contains"); - postTelemetry(device3.getId(), "{\"occupied\":false}"); - postTelemetry(device4.getId(), "{\"occupied\":true}"); - postTelemetry(device3.getId(), "{\"occupied\":true}"); + long currentTime = System.currentTimeMillis(); + long firstTs = currentTime - 10; + long secondTs = currentTime - 10; + long thirdTs = currentTime - 5; + postTelemetry(device3.getId(), "{\"ts\": " + firstTs + ", \"values\": {\"occupied\":true}}"); + postTelemetry(device4.getId(), "{\"ts\": " + secondTs + ", \"values\": {\"occupied\":true}}"); + postTelemetry(device3.getId(), "{\"ts\": " + thirdTs + ", \"values\": {\"occupied\":true}}"); createOccupancyCF(asset2.getId()); - await().alias("create CF and perform aggregation with default values").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + await().alias("create CF and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( @@ -357,15 +365,15 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll )); }); - doDelete("/api/plugins/telemetry/DEVICE/" + device3.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=0&endTs=" + System.currentTimeMillis(), String.class); - doDelete("/api/plugins/telemetry/DEVICE/" + device4.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=0&endTs=" + System.currentTimeMillis(), String.class); + doDelete("/api/plugins/telemetry/DEVICE/" + device3.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + thirdTs + "&endTs=" + thirdTs + 1, String.class); + doDelete("/api/plugins/telemetry/DEVICE/" + device4.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + secondTs + "&endTs=" + secondTs + 1, String.class); await().alias("delete latest telemetry and perform aggregation with previous or default values").atMost(deduplicationInterval * 2, TimeUnit.MILLISECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset2.getId(), Map.of( - "freeSpaces", "2", - "occupiedSpaces", "0", + "freeSpaces", "1", + "occupiedSpaces", "1", "totalSpaces", "2" )); }); @@ -411,6 +419,140 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll }); } + @Test + public void testUpdateRelationPath_checkAggregation() throws Exception { + CalculatedField cf = createOccupancyCF(asset.getId()); + checkInitialCalculation(); + + Device device3 = createDevice("Device 3", "1234567890333"); + createEntityRelation(asset.getId(), device3.getId(), "Has"); + postTelemetry(device3.getId(), "{\"occupied\":true}"); + + var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); + configuration.setRelation(new RelationPathLevel(EntitySearchDirection.FROM, "Has")); + saveCalculatedField(cf); + + await().alias("update relation path and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + } + + @Test + public void testUpdateArguments_checkAggregation() throws Exception { + CalculatedField cf = createOccupancyCF(asset.getId()); + checkInitialCalculation(); + + postTelemetry(device1.getId(), "{\"occupiedStatus\":false}"); + postTelemetry(device2.getId(), "{\"occupiedStatus\":false}"); + + var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); + Argument argument = new Argument(); + argument.setRefEntityKey(new ReferencedEntityKey("oc", ArgumentType.TS_LATEST, null)); + argument.setDefaultValue("false"); + configuration.setArguments(Map.of("oc", argument)); + saveCalculatedField(cf); + + await().alias("update arguments and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of( + "freeSpaces", "2", + "occupiedSpaces", "0", + "totalSpaces", "2" + )); + }); + } + + @Test + public void testUpdateMetrics_checkAggregation() throws Exception { + postTelemetry(device1.getId(), "{\"temperature\":24.2}"); + postTelemetry(device2.getId(), "{\"temperature\":19.6}"); + CalculatedField cf = createAvgTemperatureCF(asset.getId()); + + await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); + }); + + var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); + AggMetric aggMetric = new AggMetric(); + aggMetric.setInput(new AggKeyInput("temp")); + aggMetric.setFilter("return temp < 100;"); + aggMetric.setFunction(AggFunction.MAX); + configuration.setMetrics(Map.of("maxTemperature", aggMetric)); + saveCalculatedField(cf); + + postTelemetry(device1.getId(), "{\"temperature\":101.3}"); + postTelemetry(device2.getId(), "{\"temperature\":25.8}"); + + await().alias("update metrics and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26")); + }); + } + + @Test + public void testUpdateOutput_checkAggregation() throws Exception { + postTelemetry(device1.getId(), "{\"temperature\":24.2}"); + postTelemetry(device2.getId(), "{\"temperature\":19.6}"); + CalculatedField cf = createAvgTemperatureCF(asset.getId()); + + await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); + }); + + var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); + Output output = new Output(); + output.setType(OutputType.ATTRIBUTES); + output.setScope(AttributeScope.SERVER_SCOPE); + configuration.setOutput(output); + saveCalculatedField(cf); + + await().alias("update output and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ArrayNode avgTemperature = getServerAttributes(asset.getId(), "avgTemperature"); + assertThat(avgTemperature).isNotNull(); + assertThat(avgTemperature.get(0)).isNotNull(); + assertThat(avgTemperature.get(0).get("value").asText()).isEqualTo("24.2"); + }); + } + + @Test + public void testUpdateDeduplicationInterval_checkAggregationNotExecutedUntilDeduplicationInterval() throws Exception { + postTelemetry(device1.getId(), "{\"temperature\":24.2}"); + postTelemetry(device2.getId(), "{\"temperature\":19.6}"); + CalculatedField cf = createAvgTemperatureCF(asset.getId()); + + await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); + }); + + var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); + configuration.setDeduplicationIntervalMillis(2 * deduplicationInterval); + saveCalculatedField(cf); + + postTelemetry(device2.getId(), "{\"temperature\":32.1}"); + + await().alias("update deduplication interval and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28")); + }); + } + private void checkInitialCalculation() { await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) @@ -425,6 +567,32 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2"); } + private CalculatedField createAvgTemperatureCF(EntityId entityId) { + Map arguments = new HashMap<>(); + Argument argument = new Argument(); + argument.setRefEntityKey(new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null)); + argument.setDefaultValue("20"); + arguments.put("temp", argument); + + Map aggMetrics = new HashMap<>(); + + AggMetric avgMetric = new AggMetric(); + avgMetric.setFunction(AggFunction.AVG); + avgMetric.setFilter("return temp >= 20;"); + avgMetric.setInput(new AggKeyInput("temp")); + aggMetrics.put("avgTemperature", avgMetric); + + Output output = new Output(); + output.setType(OutputType.TIME_SERIES); + output.setDecimalsByDefault(0); + + return createAggCf("Average temperature", entityId, + new RelationPathLevel(EntitySearchDirection.FROM, "Contains"), + arguments, + aggMetrics, + output); + } + private CalculatedField createOccupancyCF(EntityId entityId) { Map arguments = new HashMap<>(); Argument argument = new Argument(); @@ -513,4 +681,8 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class); } + private ArrayNode getServerAttributes(EntityId entityId, String... keys) throws Exception { + return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/attributes/SERVER_SCOPE?keys=" + String.join(",", keys), ArrayNode.class); + } + } diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 764f4c3ec9..1e01916e55 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -928,6 +928,7 @@ message CalculatedFieldStateProto { repeated GeofencingArgumentProto geofencingArguments = 5; AlarmStateProto alarmState = 6; repeated AggSingleArgumentEntryProto aggArguments = 7; + int64 lastArgsUpdateTs = 8; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.