diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index e42ebb593e..3911714ae0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -242,7 +242,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM justRestored = true; } if (state.isSizeOk()) { - if (state.updateState(newArgValues) || justRestored) { + if (state.updateState(newArgValues, ctx) || justRestored) { cfIdList = new ArrayList<>(cfIdList); cfIdList.add(ctx.getCfId()); processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); @@ -301,7 +301,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } } - private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) { + private void removeStateAndRaiseSizeException(CalculatedFieldEntityCtxId ctxId, CalculatedFieldException ex, TbCallback callback) throws CalculatedFieldException { // We remove the state, but remember that it is over-sized in a local map. cfStateService.removeState(ctxId, new TbCallback() { @Override @@ -314,6 +314,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM callback.onFailure(ex); } }); + throw ex; } private Map mapToArguments(CalculatedFieldCtx ctx, List data) { diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index d26d89626e..e7c963a4a7 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -16,7 +16,6 @@ package org.thingsboard.server.actors.calculatedField; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.common.util.DebugModeUtil; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.TbActorCtx; import org.thingsboard.server.actors.TbActorRef; @@ -51,8 +50,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java index 2282124bd5..64487d9b3e 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java @@ -117,7 +117,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { CalculatedField calculatedField = getCalculatedField(calculatedFieldId); if (calculatedField != null) { ctx = new CalculatedFieldCtx(calculatedField, tbelInvokeService, apiLimitService); - ctx.init(); calculatedFieldsCtx.put(calculatedFieldId, ctx); log.debug("[{}] Put calculated field ctx into cache: {}", calculatedFieldId, ctx); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index f69f5cd50b..2f73788fad 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -136,7 +136,7 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP throw new RuntimeException("Error getting future result for key: " + entry.getKey(), e); } } - ))); + )), ctx); return result; }, calculatedFieldCallbackExecutor); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index d59a58f296..624198e579 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -25,6 +25,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import static org.thingsboard.server.utils.CalculatedFieldUtils.toSingleValueArgumentProto; + @Data @AllArgsConstructor public abstract class BaseCalculatedFieldState implements CalculatedFieldState { @@ -43,7 +45,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { } @Override - public boolean updateState(Map argumentValues) { + public boolean updateState(Map argumentValues, CalculatedFieldCtx ctx) { if (arguments == null) { arguments = new HashMap<>(); } @@ -53,6 +55,9 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { for (Map.Entry entry : argumentValues.entrySet()) { String key = entry.getKey(); ArgumentEntry newEntry = entry.getValue(); + + checkArgumentSize(key, newEntry, ctx); + ArgumentEntry existingEntry = arguments.get(key); if (existingEntry == null || newEntry.isForceResetPrevious()) { @@ -81,6 +86,18 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { } } + @Override + public void checkArgumentSize(String name, ArgumentEntry entry, CalculatedFieldCtx ctx) { + if (entry instanceof TsRollingArgumentEntry) { + return; + } + if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { + if (ctx.getMaxSingleValueArgumentSize() > 0 && toSingleValueArgumentProto(name, singleValueArgumentEntry).getSerializedSize() > ctx.getMaxSingleValueArgumentSize()) { + throw new IllegalArgumentException("Single value size exceeds the maximum allowed limit. The argument will not be used for calculation."); + } + } + } + protected abstract void validateNewEntry(ArgumentEntry newEntry); } 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 d3b7020c25..399e0d9f84 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 @@ -71,6 +71,7 @@ public class CalculatedFieldCtx { private long maxDataPointsPerRollingArg; private long maxStateSize; + private long maxSingleValueArgumentSize; public CalculatedFieldCtx(CalculatedField calculatedField, TbelInvokeService tbelInvokeService, ApiLimitService apiLimitService) { this.calculatedField = calculatedField; @@ -104,6 +105,7 @@ public class CalculatedFieldCtx { this.maxDataPointsPerRollingArg = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg); this.maxStateSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; + this.maxSingleValueArgumentSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxSingleValueArgumentSizeInKBytes) * 1024; } public void init() { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java index 70587d8b1e..501cb5f54c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java @@ -44,7 +44,7 @@ public interface CalculatedFieldState { void setRequiredArguments(List requiredArguments); - boolean updateState(Map argumentValues); + boolean updateState(Map argumentValues, CalculatedFieldCtx ctx); ListenableFuture performCalculation(CalculatedFieldCtx ctx); @@ -60,4 +60,6 @@ public interface CalculatedFieldState { void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize); + void checkArgumentSize(String name, ArgumentEntry entry, CalculatedFieldCtx ctx); + } diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java index 2cced15f59..78144a0150 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldStateTest.java @@ -90,7 +90,7 @@ public class ScriptCalculatedFieldStateTest { state.arguments = new HashMap<>(Map.of("assetHumidity", assetHumidityArgEntry)); Map newArgs = Map.of("deviceTemperature", deviceTemperatureArgEntry); - boolean stateUpdated = state.updateState(newArgs); + boolean stateUpdated = state.updateState(newArgs, ctx); assertThat(stateUpdated).isTrue(); assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf( @@ -107,7 +107,7 @@ public class ScriptCalculatedFieldStateTest { SingleValueArgumentEntry newArgEntry = new SingleValueArgumentEntry(ts, new LongDataEntry("assetHumidity", 41L), 349L); Map newArgs = Map.of("assetHumidity", newArgEntry); - boolean stateUpdated = state.updateState(newArgs); + boolean stateUpdated = state.updateState(newArgs, ctx); assertThat(stateUpdated).isTrue(); assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf( diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java index e179e8e706..f146eb434f 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SimpleCalculatedFieldStateTest.java @@ -88,7 +88,7 @@ public class SimpleCalculatedFieldStateTest { )); Map newArgs = Map.of("key3", key3ArgEntry); - boolean stateUpdated = state.updateState(newArgs); + boolean stateUpdated = state.updateState(newArgs, ctx); assertThat(stateUpdated).isTrue(); assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf( @@ -106,7 +106,7 @@ public class SimpleCalculatedFieldStateTest { SingleValueArgumentEntry newArgEntry = new SingleValueArgumentEntry(System.currentTimeMillis(), new LongDataEntry("key1", 18L), 190L); Map newArgs = Map.of("key1", newArgEntry); - boolean stateUpdated = state.updateState(newArgs); + boolean stateUpdated = state.updateState(newArgs, ctx); assertThat(stateUpdated).isTrue(); assertThat(state.getArguments()).containsExactlyInAnyOrderEntriesOf(Map.of("key1", newArgEntry)); @@ -120,7 +120,7 @@ public class SimpleCalculatedFieldStateTest { )); Map newArgs = Map.of("key3", new TsRollingArgumentEntry(10, 30000L)); - assertThatThrownBy(() -> state.updateState(newArgs)) + assertThatThrownBy(() -> state.updateState(newArgs, ctx)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Rolling argument entry is not supported for simple calculated fields."); }