From 618d350791443d9aaf7ac44228a6a64800ab2162 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Tue, 18 Feb 2025 16:55:28 +0200 Subject: [PATCH] State storage improvements --- ...CalculatedFieldEntityMessageProcessor.java | 59 ++++++++++++------- ...alculatedFieldManagerMessageProcessor.java | 22 +++---- .../AbstractCalculatedFieldStateService.java | 2 +- .../ctx/state/BaseCalculatedFieldState.java | 6 +- .../cf/ctx/state/CalculatedFieldCtx.java | 8 ++- .../cf/ctx/state/CalculatedFieldState.java | 7 ++- 6 files changed, 62 insertions(+), 42 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 7211ac0db9..1089ff849e 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 @@ -107,16 +107,20 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM public void process(EntityInitCalculatedFieldMsg msg) throws CalculatedFieldException { log.info("[{}] Processing entity init CF msg.", msg.getCtx().getCfId()); - var cfCtx = msg.getCtx(); + var ctx = msg.getCtx(); if (msg.isForceReinit()) { - log.info("Force reinitialization of CF: [{}].", cfCtx.getCfId()); - states.remove(cfCtx.getCfId()); + log.info("Force reinitialization of CF: [{}].", ctx.getCfId()); + states.remove(ctx.getCfId()); } try { - var cfState = getOrInitState(cfCtx); - processStateIfReady(cfCtx, Collections.singletonList(cfCtx.getCfId()), cfState, null, null, msg.getCallback()); + var state = getOrInitState(ctx); + if (!state.isSizeExceedsLimit()) { + processStateIfReady(ctx, Collections.singletonList(ctx.getCfId()), state, null, null, msg.getCallback()); + } else { + throw new RuntimeException(ctx.getSizeExceedsLimitMessage()); + } } catch (Exception e) { - throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(entityId).cause(e).build(); + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build(); } } @@ -214,12 +218,16 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM state = getOrInitState(ctx); justRestored = true; } - if (state.updateState(newArgValues) || justRestored) { - cfIdList = new ArrayList<>(cfIdList); - cfIdList.add(ctx.getCfId()); - processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); + if (state.isSizeOk()) { + if (state.updateState(newArgValues) || justRestored) { + cfIdList = new ArrayList<>(cfIdList); + cfIdList.add(ctx.getCfId()); + processStateIfReady(ctx, cfIdList, state, tbMsgId, tbMsgType, callback); + } else { + callback.onSuccess(CALLBACKS_PER_CF); + } } else { - callback.onSuccess(CALLBACKS_PER_CF); + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(); } } @@ -235,7 +243,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM // Alternatively, we can fetch the state outside the actor system and push separate command to create this actor, // but this will significantly complicate the code. state = stateFuture.get(1, TimeUnit.MINUTES); - state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSizeInKBytes()); + state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize()); states.put(ctx.getCfId(), state); } return state; @@ -244,18 +252,29 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM @SneakyThrows private void processStateIfReady(CalculatedFieldCtx ctx, List cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) { CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId); - if (state.isReady() && ctx.isInitialized()) { + boolean stateSizeOk; + if (ctx.isInitialized() && state.isReady()) { CalculatedFieldResult calculationResult = state.performCalculation(ctx).get(5, TimeUnit.SECONDS); - state.checkStateSize(ctxId, ctx.getMaxStateSizeInKBytes()); - cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback); - if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) { - systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null); + state.checkStateSize(ctxId, ctx.getMaxStateSize()); + stateSizeOk = state.isSizeOk(); + if (stateSizeOk) { + cfService.pushMsgToRuleEngine(tenantId, entityId, calculationResult, cfIdList, callback); + if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) { + systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, JacksonUtil.writeValueAsString(calculationResult.getResult()), null); + } } } else { - state.checkStateSize(ctxId, ctx.getMaxStateSizeInKBytes()); - callback.onSuccess(); // State was updated but no calculation performed; + state.checkStateSize(ctxId, ctx.getMaxStateSize()); + stateSizeOk = state.isSizeOk(); + if (stateSizeOk) { + callback.onSuccess(); // State was updated but no calculation performed; + } + } + if (stateSizeOk) { + cfStateService.persistState(ctxId, state, callback); + } else { + throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build(); } - cfStateService.persistState(ctxId, state, callback); } 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 5e005935aa..73a3e755c5 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 @@ -94,17 +94,14 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware this.ctx = ctx; } - public void onFieldInitMsg(CalculatedFieldInitMsg msg) { + public void onFieldInitMsg(CalculatedFieldInitMsg msg) throws CalculatedFieldException { log.info("[{}] Processing CF init message.", msg.getCf().getId()); var cf = msg.getCf(); var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService()); try { cfCtx.init(); } catch (Exception e) { - log.debug("[{}] Failed to initialize CF context.", cf.getId(), e); - if (DebugModeUtil.isDebugAllAvailable(cf)) { - systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e.getMessage()); - } + throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } calculatedFields.put(cf.getId(), cfCtx); // We use copy on write lists to safely pass the reference to another actor for the iteration. @@ -135,7 +132,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } - public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) { + public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException { log.info("Processing entity lifecycle event: [{}] for entity: [{}]", msg.getData().getEvent(), msg.getData().getEntityId()); var entityType = msg.getData().getEntityId().getEntityType(); var event = msg.getData().getEvent(); @@ -220,7 +217,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); } - private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) { + private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { var cfId = new CalculatedFieldId(msg.getEntityId().getId()); if (calculatedFields.containsKey(cfId)) { log.warn("[{}] CF was already initialized [{}]", tenantId, cfId); @@ -235,10 +232,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware try { cfCtx.init(); } catch (Exception e) { - log.debug("[{}] Failed to initialize CF context.", cf.getId(), e); - if (DebugModeUtil.isDebugAllAvailable(cf)) { - systemContext.persistCalculatedFieldDebugEvent(cf.getTenantId(), cf.getId(), cf.getEntityId(), null, null, null, null, e.getMessage()); - } + throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } calculatedFields.put(cf.getId(), cfCtx); // We use copy on write lists to safely pass the reference to another actor for the iteration. @@ -250,7 +244,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } - private void onCfUpdated(ComponentLifecycleMsg msg, TbCallback callback) { + private void onCfUpdated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException { var cfId = new CalculatedFieldId(msg.getEntityId().getId()); var oldCfCtx = calculatedFields.get(cfId); if (oldCfCtx == null) { @@ -265,9 +259,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware try { newCfCtx.init(); } catch (Exception e) { - if (DebugModeUtil.isDebugAllAvailable(newCf)) { - systemContext.persistCalculatedFieldDebugEvent(newCf.getTenantId(), newCf.getId(), newCf.getEntityId(), null, null, null, null, e.getMessage()); - } + throw CalculatedFieldException.builder().ctx(newCfCtx).eventEntity(newCfCtx.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build(); } calculatedFields.put(newCf.getId(), newCfCtx); List oldCfList = entityIdCalculatedFields.get(newCf.getEntityId()); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java index e390d9e55c..14a4103216 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldStateService.java @@ -34,7 +34,7 @@ public abstract class AbstractCalculatedFieldStateService implements CalculatedF @Override public final void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { - if (state.isStateTooLarge()) { + if (state.isSizeExceedsLimit()) { throw new CalculatedFieldStateException("State size exceeds the maximum allowed limit. The state will not be persisted to RocksDB."); } doPersist(stateId, toProto(stateId, state), callback); 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 2b103b567e..0f77027650 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 @@ -31,7 +31,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected List requiredArguments; protected Map arguments; - protected boolean stateTooLarge; + protected boolean sizeExceedsLimit; public BaseCalculatedFieldState(List requiredArguments) { this.requiredArguments = requiredArguments; @@ -75,9 +75,9 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { @Override public void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize) { - if (!stateTooLarge && maxStateSize > 0 && CalculatedFieldUtils.toProto(ctxId, this).getSerializedSize() > maxStateSize) { + if (!sizeExceedsLimit && maxStateSize > 0 && CalculatedFieldUtils.toProto(ctxId, this).getSerializedSize() > maxStateSize) { arguments.clear(); - setStateTooLarge(true); + sizeExceedsLimit = true; } } 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 c3105d6b57..4e4844dbe8 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 @@ -70,7 +70,7 @@ public class CalculatedFieldCtx { private boolean initialized; private long maxDataPointsPerRollingArg; - private long maxStateSizeInKBytes; + private long maxStateSize; public CalculatedFieldCtx(CalculatedField calculatedField, TbelInvokeService tbelInvokeService, ApiLimitService apiLimitService) { this.calculatedField = calculatedField; @@ -103,7 +103,7 @@ public class CalculatedFieldCtx { this.tbelInvokeService = tbelInvokeService; this.maxDataPointsPerRollingArg = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg); - this.maxStateSizeInKBytes = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes); + this.maxStateSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; } public void init() { @@ -224,4 +224,8 @@ public class CalculatedFieldCtx { return typeChanged || argumentsChanged; } + public String getSizeExceedsLimitMessage() { + return "Failed to init CF state. State size exceeds limit of " + (maxStateSize / 1024) + "Kb!"; + } + } 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 0219dd2593..00e8c6653b 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 @@ -51,7 +51,12 @@ public interface CalculatedFieldState { @JsonIgnore boolean isReady(); - boolean isStateTooLarge(); + boolean isSizeExceedsLimit(); + + @JsonIgnore + default boolean isSizeOk() { + return !isSizeExceedsLimit(); + } void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize);