From c404dd2cf593e81cf599a21dab482eec031210fd Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Wed, 5 Mar 2025 15:42:26 +0200 Subject: [PATCH 1/3] added check for single value size --- .../CalculatedFieldEntityMessageProcessor.java | 2 +- ...DefaultCalculatedFieldProcessingService.java | 2 +- .../cf/ctx/state/BaseCalculatedFieldState.java | 17 ++++++++++++++++- .../cf/ctx/state/CalculatedFieldCtx.java | 2 ++ .../cf/ctx/state/CalculatedFieldState.java | 4 +++- .../state/ScriptCalculatedFieldStateTest.java | 4 ++-- .../state/SimpleCalculatedFieldStateTest.java | 6 +++--- 7 files changed, 28 insertions(+), 9 deletions(-) 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..c75b47c8a0 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); 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..b2a4321961 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,16 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { } } + @Override + public void checkArgumentSize(String name, ArgumentEntry entry, CalculatedFieldCtx ctx) { + if (entry instanceof TsRollingArgumentEntry) { + return; + } + if (ctx.getMaxSingleValueArgumentSize() > 0 && toSingleValueArgumentProto(name, (SingleValueArgumentEntry) entry).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."); } From 7600401b867e0d4420173707eb7a32e7713170cb Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Thu, 6 Mar 2025 10:53:25 +0200 Subject: [PATCH 2/3] fixed error when try to push telemtry and cf fails --- .../server/service/cf/DefaultCalculatedFieldCache.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) 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..3e5d09e953 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,11 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { CalculatedField calculatedField = getCalculatedField(calculatedFieldId); if (calculatedField != null) { ctx = new CalculatedFieldCtx(calculatedField, tbelInvokeService, apiLimitService); - ctx.init(); + try { + ctx.init(); + } catch (Exception e) { + log.warn("Failed to initialize CF context."); + } calculatedFieldsCtx.put(calculatedFieldId, ctx); log.debug("[{}] Put calculated field ctx into cache: {}", calculatedFieldId, ctx); } From 93090af2ee7e8406aab7cde134a66e4ced51214c Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Thu, 6 Mar 2025 16:35:54 +0200 Subject: [PATCH 3/3] updated check for single value argument size --- .../CalculatedFieldEntityMessageProcessor.java | 3 ++- .../CalculatedFieldManagerMessageProcessor.java | 3 --- .../server/service/cf/DefaultCalculatedFieldCache.java | 5 ----- .../service/cf/ctx/state/BaseCalculatedFieldState.java | 6 ++++-- 4 files changed, 6 insertions(+), 11 deletions(-) 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 c75b47c8a0..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 @@ -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 3e5d09e953..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,11 +117,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache { CalculatedField calculatedField = getCalculatedField(calculatedFieldId); if (calculatedField != null) { ctx = new CalculatedFieldCtx(calculatedField, tbelInvokeService, apiLimitService); - try { - ctx.init(); - } catch (Exception e) { - log.warn("Failed to initialize CF context."); - } 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/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index b2a4321961..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 @@ -91,8 +91,10 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { if (entry instanceof TsRollingArgumentEntry) { return; } - if (ctx.getMaxSingleValueArgumentSize() > 0 && toSingleValueArgumentProto(name, (SingleValueArgumentEntry) entry).getSerializedSize() > ctx.getMaxSingleValueArgumentSize()) { - throw new IllegalArgumentException("Single value size exceeds the maximum allowed limit. The argument will not be used for calculation."); + 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."); + } } }