From f951c64767c504142bdaaab1ad69711d995ec73c Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 17 Oct 2025 11:14:07 +0300 Subject: [PATCH] added state to proto and fixed deduplication logic --- ...ValuesAggregationCalculatedFieldState.java | 23 +++---- .../state/aggregation/function/new_agg.json | 60 ------------------- .../server/utils/CalculatedFieldUtils.java | 16 +++-- ...tValuesAggregationCalculatedFieldTest.java | 16 ++++- common/proto/src/main/proto/queue.proto | 8 ++- 5 files changed, 42 insertions(+), 81 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/new_agg.json 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 6ae023ed1e..f3560faafa 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 @@ -106,21 +106,22 @@ public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedF @Override public ListenableFuture performCalculation(Map updatedArgs, CalculatedFieldCtx ctx) throws Exception { - boolean shouldRecalculate = updatedArgs == null || updatedArgs.isEmpty(); - if (!shouldRecalculate() && !shouldRecalculate) { + boolean cfUpdated = updatedArgs != null && updatedArgs.isEmpty(); + if (shouldRecalculate() || cfUpdated) { + Output output = ctx.getOutput(); + ObjectNode aggResult = aggregateMetrics(output); + lastMetricsEvalTs = System.currentTimeMillis(); + ctx.scheduleReevaluation(deduplicationInterval, actorCtx); + return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() + .type(output.getType()) + .scope(output.getScope()) + .result(createResultJson(ctx.isUseLatestTs(), aggResult)) + .build()); + } else { return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() .result(null) .build()); } - Output output = ctx.getOutput(); - ObjectNode aggResult = aggregateMetrics(output); - lastMetricsEvalTs = System.currentTimeMillis(); - ctx.scheduleReevaluation(deduplicationInterval, actorCtx); - return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder() - .type(output.getType()) - .scope(output.getScope()) - .result(createResultJson(ctx.isUseLatestTs(), aggResult)) - .build()); } private boolean shouldRecalculate() { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/new_agg.json b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/new_agg.json deleted file mode 100644 index c6b841b673..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/function/new_agg.json +++ /dev/null @@ -1,60 +0,0 @@ -{ - "type": "LATEST_VALUES_AGGREGATION", - "name": "Occupied spaces", - "debugSettings": { - "failuresEnabled": true, - "allEnabled": true, - "allEnabledUntil": 1769907492297 - }, - "entityId": { - "entityType": "ASSET", - "id": "f8ad0800-a9a6-11f0-bbe6-459b63b420fe" - }, - "configuration": { - "type": "LATEST_VALUES_AGGREGATION", - "relation": { - "direction": "FROM", - "relationType": "Contains" - }, - "arguments": { - "oc": { - "refEntityKey": { - "key": "occupied", - "type": "TS_LATEST" - }, - "defaultValue": "false" - } - }, - "deduplicationIntervalMillis": 20000, - "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 00f4cbde0f..f8a7d17543 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -35,6 +35,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdPro import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.GeofencingZoneProto; +import org.thingsboard.server.gen.transport.TransportProtos.LatestValuesAggregationStateProto; import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto; import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentProto; @@ -96,13 +97,11 @@ public class CalculatedFieldUtils { .setId(toProto(stateId)) .setType(state.getType().name()); - if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) { - builder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs()); - } + LatestValuesAggregationStateProto.Builder aggBuilder = LatestValuesAggregationStateProto.newBuilder(); state.getArguments().forEach((argName, argEntry) -> { if (argEntry instanceof AggArgumentEntry aggArgumentEntry) { aggArgumentEntry.getAggInputs() - .forEach((entityId, entry) -> builder.addAggArguments(toAggSingleArgumentProto(argName, entityId, entry))); + .forEach((entityId, entry) -> aggBuilder.addAggArguments(toAggSingleArgumentProto(argName, entityId, entry))); } else if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry)); } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) { @@ -120,6 +119,10 @@ public class CalculatedFieldUtils { alarmStateProto.setClearRuleState(toAlarmRuleStateProto(alarmState.getClearRuleState())); } } + if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) { + aggBuilder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs()); + builder.setLatestValuesAggregationState(aggBuilder.build()); + } return builder.build(); } @@ -237,15 +240,16 @@ public class CalculatedFieldUtils { } case LATEST_VALUES_AGGREGATION -> { LatestValuesAggregationCalculatedFieldState aggState = (LatestValuesAggregationCalculatedFieldState) state; + LatestValuesAggregationStateProto aggregationStateProto = proto.getLatestValuesAggregationState(); Map> arguments = new HashMap<>(); - proto.getAggArgumentsList().forEach(argProto -> { + aggregationStateProto.getAggArgumentsList().forEach(argProto -> { AggSingleEntityArgumentEntry entry = fromAggSingleValueArgumentProto(argProto); arguments.computeIfAbsent(argProto.getValue().getArgName(), name -> new HashMap<>()).put(entry.getEntityId(), entry); }); arguments.forEach((argName, entityInputs) -> { aggState.getArguments().put(argName, new AggArgumentEntry(entityInputs, false)); }); - aggState.setLastArgsRefreshTs(proto.getLastArgsUpdateTs()); + aggState.setLastArgsRefreshTs(aggregationStateProto.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 8c2c2ca68c..46b8b78d57 100644 --- a/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java @@ -489,10 +489,16 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll configuration.setMetrics(Map.of("maxTemperature", aggMetric)); saveCalculatedField(cf); + await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24")); + }); + postTelemetry(device1.getId(), "{\"temperature\":101.3}"); postTelemetry(device2.getId(), "{\"temperature\":25.8}"); - await().alias("update metrics and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) + await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26")); @@ -544,9 +550,15 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll configuration.setDeduplicationIntervalMillis(2 * deduplicationInterval); saveCalculatedField(cf); + await().alias("update deduplication interval and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); + }); + postTelemetry(device2.getId(), "{\"temperature\":32.1}"); - await().alias("update deduplication interval and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS) + await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .untilAsserted(() -> { verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28")); diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 1e01916e55..b59e01128b 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -920,6 +920,11 @@ message AggSingleArgumentEntryProto { SingleValueArgumentProto value = 2; } +message LatestValuesAggregationStateProto { + int64 lastArgsUpdateTs = 1; + repeated AggSingleArgumentEntryProto aggArguments = 2; +} + message CalculatedFieldStateProto { CalculatedFieldEntityCtxIdProto id = 1; string type = 2; @@ -927,8 +932,7 @@ message CalculatedFieldStateProto { repeated TsRollingArgumentProto rollingValueArguments = 4; repeated GeofencingArgumentProto geofencingArguments = 5; AlarmStateProto alarmState = 6; - repeated AggSingleArgumentEntryProto aggArguments = 7; - int64 lastArgsUpdateTs = 8; + LatestValuesAggregationStateProto latestValuesAggregationState = 7; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.