diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index c787fc26af..5ac152ca0f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -41,7 +41,9 @@ import org.thingsboard.script.api.js.JsInvokeService; import org.thingsboard.script.api.tbel.TbelInvokeService; import org.thingsboard.server.actors.service.ActorService; import org.thingsboard.server.actors.tenant.DebugTbRateLimits; +import org.thingsboard.server.cache.limits.RateLimitService; import org.thingsboard.server.cluster.TbClusterService; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.event.CalculatedFieldDebugEvent; import org.thingsboard.server.common.data.event.ErrorEvent; import org.thingsboard.server.common.data.event.LifecycleEvent; @@ -50,6 +52,7 @@ import org.thingsboard.server.common.data.event.RuleNodeDebugEvent; import org.thingsboard.server.common.data.id.CalculatedFieldId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.limit.LimitedApi; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.msg.TbActorMsg; @@ -58,6 +61,7 @@ import org.thingsboard.server.common.msg.notification.NotificationRuleProcessor; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.common.msg.tools.TbRateLimits; +import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.stats.TbApiUsageReportClient; import org.thingsboard.server.dao.alarm.AlarmCommentService; import org.thingsboard.server.dao.asset.AssetProfileService; @@ -107,8 +111,8 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; -import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.CalculatedFieldStateService; +import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.component.ComponentDiscoveryService; import org.thingsboard.server.service.edge.rpc.EdgeRpcService; @@ -182,8 +186,6 @@ public class ActorSystemContext { private final ConcurrentMap debugPerTenantLimits = new ConcurrentHashMap<>(); - private final ConcurrentMap cfDebugPerTenantLimits = new ConcurrentHashMap<>(); - public ConcurrentMap getDebugPerTenantLimits() { return debugPerTenantLimits; } @@ -452,6 +454,11 @@ public class ActorSystemContext { @Getter private ApiLimitService apiLimitService; + @Lazy + @Autowired() + @Getter + private RateLimitService rateLimitService; + /** * The following Service will be null if we operate in tb-core mode */ @@ -641,14 +648,6 @@ public class ActorSystemContext { @Getter private String deviceStateNodeRateLimitConfig; - @Value("${actors.calculated_fields.debug_mode_rate_limits_per_tenant.enabled:true}") - @Getter - private boolean cfDebugPerTenantEnabled; - - @Value("${actors.calculated_fields.debug_mode_rate_limits_per_tenant.configuration:50000:3600}") - @Getter - private String cfDebugPerTenantLimitsConfiguration; - @Getter @Setter private TbActorSystem actorSystem; @@ -811,45 +810,40 @@ public class ActorSystemContext { } public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, Map arguments, UUID tbMsgId, TbMsgType tbMsgType, String result, Throwable error) { - if (cfDebugPerTenantEnabled) { - TbRateLimits rateLimits = cfDebugPerTenantLimits.computeIfAbsent(tenantId, id -> new TbRateLimits(cfDebugPerTenantLimitsConfiguration)); - - if (rateLimits.tryConsume()) { - try { - CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder eventBuilder = CalculatedFieldDebugEvent.builder() - .tenantId(tenantId) - .entityId(calculatedFieldId.getId()) - .serviceId(getServiceId()) - .calculatedFieldId(calculatedFieldId) - .eventEntity(entityId); - if (tbMsgId != null) { - eventBuilder.msgId(tbMsgId); - } - if (tbMsgType != null) { - eventBuilder.msgType(tbMsgType.name()); - } - if (arguments != null) { - eventBuilder.arguments(JacksonUtil.toString( - arguments.entrySet().stream() - .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue())) - )); - } - if (result != null) { - eventBuilder.result(result); - } - if (error != null) { - eventBuilder.error(toString(error)); - } - - ListenableFuture future = eventService.saveAsync(eventBuilder.build()); - Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); - } catch (IllegalArgumentException ex) { - log.warn("Failed to persist calculated field debug message", ex); - } - if (log.isTraceEnabled()) { - log.trace("[{}] Tenant level debug mode rate limit detected: {}", tenantId, calculatedFieldId); - } + if (!rateLimitService.checkRateLimit(LimitedApi.CALCULATED_FIELD_DEBUG_EVENTS, tenantId)) { + throw new TbRateLimitsException(EntityType.TENANT); + } + try { + CalculatedFieldDebugEvent.CalculatedFieldDebugEventBuilder eventBuilder = CalculatedFieldDebugEvent.builder() + .tenantId(tenantId) + .entityId(calculatedFieldId.getId()) + .serviceId(getServiceId()) + .calculatedFieldId(calculatedFieldId) + .eventEntity(entityId); + if (tbMsgId != null) { + eventBuilder.msgId(tbMsgId); + } + if (tbMsgType != null) { + eventBuilder.msgType(tbMsgType.name()); } + if (arguments != null) { + eventBuilder.arguments(JacksonUtil.toString( + arguments.entrySet().stream() + .filter(entry -> !entry.getValue().isEmpty()) + .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getValue())) + )); + } + if (result != null) { + eventBuilder.result(result); + } + if (error != null) { + eventBuilder.error(toString(error)); + } + + ListenableFuture future = eventService.saveAsync(eventBuilder.build()); + Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor()); + } catch (IllegalArgumentException ex) { + log.warn("Failed to persist calculated field debug message", ex); } } 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 61f7abb297..bfed515eb5 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 @@ -38,8 +38,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelem import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; import org.thingsboard.server.service.cf.CalculatedFieldResult; -import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.CalculatedFieldStateService; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; @@ -225,9 +225,11 @@ 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()) { try { 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.getResultMap()), null); @@ -240,7 +242,13 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM } else { callback.onSuccess(); // State was updated but no calculation performed; } - cfStateService.persistState(ctx, new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), state, callback); + try { + cfStateService.persistState(ctxId, state, callback); + } catch (Exception e) { + if (DebugModeUtil.isDebugFailuresAvailable(ctx.getCalculatedField())) { + systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArguments(), tbMsgId, tbMsgType, null, e); + } + } } 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 22492f3c4f..898e339426 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 @@ -41,9 +41,9 @@ import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.dao.cf.CalculatedFieldService; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; +import org.thingsboard.server.service.cf.CalculatedFieldStateService; import org.thingsboard.server.service.cf.cache.CalculatedFieldEntityProfileCache; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; -import org.thingsboard.server.service.cf.CalculatedFieldStateService; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.profile.TbAssetProfileCache; import org.thingsboard.server.service.profile.TbDeviceProfileCache; @@ -53,7 +53,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; diff --git a/application/src/main/java/org/thingsboard/server/exception/CalculatedFieldStateException.java b/application/src/main/java/org/thingsboard/server/exception/CalculatedFieldStateException.java new file mode 100644 index 0000000000..6248ac1536 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/exception/CalculatedFieldStateException.java @@ -0,0 +1,24 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.exception; + +public class CalculatedFieldStateException extends Exception { + + public CalculatedFieldStateException(String message) { + super(message); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java index 37211c66c5..9a41dba576 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldStateService.java @@ -16,13 +16,13 @@ package org.thingsboard.server.service.cf; import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.exception.CalculatedFieldStateException; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; -import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; public interface CalculatedFieldStateService { - void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback); + void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) throws CalculatedFieldStateException; void removeState(CalculatedFieldEntityCtxId stateId, TbCallback 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 6b47053f2d..e0e6db68fe 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 @@ -60,8 +60,6 @@ import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.dao.usagerecord.ApiLimitService; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; @@ -92,6 +90,7 @@ import java.util.stream.Collectors; import static org.thingsboard.server.common.data.DataConstants.SCOPE; import static org.thingsboard.server.queue.discovery.HashPartitionService.CALCULATED_FIELD_QUEUE_KEY; +import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; @TbRuleEngineComponent @Service @@ -229,19 +228,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP return builder.build(); } - //TODO: IM: move to utils; - private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { - return CalculatedFieldEntityCtxIdProto.newBuilder() - .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) - .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) - .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) - .setEntityType(ctxId.entityId().getEntityType().name()) - .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) - .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) - .build(); - } - private ListenableFuture fetchKvEntry(TenantId tenantId, EntityId entityId, Argument argument) { return switch (argument.getRefEntityKey().getType()) { case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument); @@ -264,7 +250,7 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP if (kvEntry.isPresent() && kvEntry.get().getValue() != null) { return ArgumentEntry.createSingleValueArgument(kvEntry.get()); } else { - return SingleValueArgumentEntry.EMPTY; + return new SingleValueArgumentEntry(); } }, calculatedFieldCallbackExecutor); } @@ -279,7 +265,7 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP ReadTsKvQuery query = new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, currentTime, 0, limit, Aggregation.NONE); ListenableFuture> tsRollingFuture = timeseriesService.findAll(tenantId, entityId, List.of(query)); - return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? TsRollingArgumentEntry.EMPTY : ArgumentEntry.createTsRollingArgument(tsRolling), calculatedFieldCallbackExecutor); + return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? new TsRollingArgumentEntry(limit, timeWindow) : ArgumentEntry.createTsRollingArgument(tsRolling, limit, timeWindow), calculatedFieldCallbackExecutor); } private KvEntry createDefaultKvEntry(Argument argument) { @@ -311,13 +297,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP }; } - private CalculatedFieldIdProto toProto(CalculatedFieldId cfId) { - return CalculatedFieldIdProto.newBuilder() - .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits()) - .build(); - } - private static class TbCallbackWrapper implements TbQueueCallback { private final TbCallback callback; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java index be471a8ad9..6b097124a5 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java @@ -41,15 +41,14 @@ public interface ArgumentEntry { boolean updateEntry(ArgumentEntry entry); + boolean isEmpty(); + static ArgumentEntry createSingleValueArgument(KvEntry kvEntry) { return new SingleValueArgumentEntry(kvEntry); } - static ArgumentEntry createTsRollingArgument(List kvEntries) { - return new TsRollingArgumentEntry(kvEntries); + static ArgumentEntry createTsRollingArgument(List kvEntries, int limit, long timeWindow) { + return new TsRollingArgumentEntry(kvEntries, limit, timeWindow); } - @JsonIgnore - ArgumentEntry copy(); - } 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 f1ce8038c7..5ea4707ff3 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 @@ -17,6 +17,8 @@ package org.thingsboard.server.service.cf.ctx.state; import lombok.AllArgsConstructor; import lombok.Data; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.utils.CalculatedFieldUtils; import java.util.ArrayList; import java.util.HashMap; @@ -29,6 +31,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected List requiredArguments; protected Map arguments; + protected boolean stateTooLarge; public BaseCalculatedFieldState(List requiredArguments) { this.requiredArguments = requiredArguments; @@ -36,7 +39,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { } public BaseCalculatedFieldState() { - this(new ArrayList<>(), new HashMap<>()); + this(new ArrayList<>(), new HashMap<>(), false); } @Override @@ -52,7 +55,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { ArgumentEntry newEntry = entry.getValue(); ArgumentEntry existingEntry = arguments.get(key); - if (existingEntry == null || existingEntry == SingleValueArgumentEntry.EMPTY || existingEntry == TsRollingArgumentEntry.EMPTY) { + if (existingEntry == null) { validateNewEntry(newEntry); arguments.put(key, newEntry); stateUpdated = true; @@ -67,8 +70,14 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { @Override public boolean isReady() { return arguments.keySet().containsAll(requiredArguments) && - !arguments.containsValue(SingleValueArgumentEntry.EMPTY) && - !arguments.containsValue(TsRollingArgumentEntry.EMPTY); + arguments.values().stream().noneMatch(ArgumentEntry::isEmpty); + } + + @Override + public void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize) { + if (maxStateSize > 0 && CalculatedFieldUtils.toProto(ctxId, this).getSerializedSize() > maxStateSize) { + setStateTooLarge(true); + } } protected abstract void validateNewEntry(ArgumentEntry newEntry); 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 0792b8240d..0219dd2593 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 @@ -21,6 +21,7 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.cf.CalculatedFieldType; import org.thingsboard.server.service.cf.CalculatedFieldResult; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import java.util.List; import java.util.Map; @@ -41,8 +42,6 @@ public interface CalculatedFieldState { Map getArguments(); - List getRequiredArguments(); - void setRequiredArguments(List requiredArguments); boolean updateState(Map argumentValues); @@ -51,4 +50,9 @@ public interface CalculatedFieldState { @JsonIgnore boolean isReady(); + + boolean isStateTooLarge(); + + void checkStateSize(CalculatedFieldEntityCtxId ctxId, long maxStateSize); + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java index a1999d438c..f15c5eb416 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/KafkaCalculatedFieldStateService.java @@ -20,8 +20,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.util.AfterStartUp; -import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.CalculatedFieldStateService; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; @Service @RequiredArgsConstructor @@ -33,7 +33,7 @@ public class KafkaCalculatedFieldStateService implements CalculatedFieldStateSer } @Override - public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { + public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { callback.onSuccess(); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java index 3374eb2660..446182e2e9 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBCalculatedFieldStateService.java @@ -20,31 +20,19 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.calculatedField.CalculatedFieldStateRestoreMsg; -import org.thingsboard.server.common.data.StringUtils; -import org.thingsboard.server.common.data.cf.CalculatedFieldType; -import org.thingsboard.server.common.data.id.CalculatedFieldId; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.EntityIdFactory; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.common.msg.queue.TbCallback; -import org.thingsboard.server.common.util.KvProtoUtil; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; -import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; -import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; -import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto; -import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto; +import org.thingsboard.server.exception.CalculatedFieldStateException; import org.thingsboard.server.queue.util.AfterStartUp; +import org.thingsboard.server.service.cf.CalculatedFieldStateService; import org.thingsboard.server.service.cf.RocksDBService; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; -import org.thingsboard.server.service.cf.CalculatedFieldStateService; import java.util.Map; -import java.util.Optional; -import java.util.TreeMap; -import java.util.UUID; import java.util.stream.Collectors; +import static org.thingsboard.server.utils.CalculatedFieldUtils.fromProto; +import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; + @Service @RequiredArgsConstructor @ConditionalOnProperty(prefix = "zk", value = "enabled", havingValue = "false", matchIfMissing = true) // Queue type in mem or Kafka; @@ -67,12 +55,12 @@ public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateS } @Override - public void persistState(CalculatedFieldCtx ctx, CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) { - CalculatedFieldStateProto stateProto = toProto(stateId, state); - long maxStateSizeInKBytes = ctx.getMaxStateSizeInKBytes(); - if (maxStateSizeInKBytes <= 0 || stateProto.getSerializedSize() <= maxStateSizeInKBytes) { - rocksDBService.put(toProto(stateId), stateProto); + public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) throws CalculatedFieldStateException { + if (state.isStateTooLarge()) { + throw new CalculatedFieldStateException("State size exceeds the maximum allowed limit. The state will not be persisted to RocksDB."); } + + rocksDBService.put(toProto(stateId), toProto(stateId, state)); callback.onSuccess(); } @@ -82,107 +70,4 @@ public class RocksDBCalculatedFieldStateService implements CalculatedFieldStateS callback.onSuccess(); } - private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { - return CalculatedFieldEntityCtxIdProto.newBuilder() - .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) - .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) - .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) - .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) - .setEntityType(ctxId.entityId().getEntityType().name()) - .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) - .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) - .build(); - } - - private CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) { - TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB())); - EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); - CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); - return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId); - } - - private CalculatedFieldStateProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) { - CalculatedFieldStateProto.Builder builder = CalculatedFieldStateProto.newBuilder() - .setId(toProto(stateId)) - .setType(state.getType().name()); - - state.getArguments().forEach((argName, argEntry) -> { - if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { - builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry)); - } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) { - builder.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry)); - } - }); - - return builder.build(); - } - - private SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) { - SingleValueArgumentProto.Builder builder = SingleValueArgumentProto.newBuilder() - .setArgName(argName); - - if (entry != SingleValueArgumentEntry.EMPTY) { - builder.setValue(KvProtoUtil.toTsValueProto(entry.getTs(), entry.getKvEntryValue())); - } - - Optional.ofNullable(entry.getVersion()).ifPresent(builder::setVersion); - - return builder.build(); - } - - private TsValueListProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { - TsValueListProto.Builder builder = TsValueListProto.newBuilder().setKey(argName); - - if (entry != TsRollingArgumentEntry.EMPTY) { - entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(KvProtoUtil.toTsValueProto(ts, value))); - } - - return builder.build(); - } - - private CalculatedFieldState fromProto(CalculatedFieldStateProto proto) { - if (StringUtils.isEmpty(proto.getType())) { - return null; - } - - CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType()); - - CalculatedFieldState state = switch (type) { - case SIMPLE -> new SimpleCalculatedFieldState(); - case SCRIPT -> new ScriptCalculatedFieldState(); - }; - - proto.getSingleValueArgumentsList().forEach(argProto -> - state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); - - if (CalculatedFieldType.SCRIPT.equals(type)) { - proto.getRollingValueArgumentsList().forEach(argProto -> - state.getArguments().put(argProto.getKey(), fromRollingArgumentProto(argProto))); - } - - return state; - } - - private SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) { - if (!proto.hasValue()) { - return (SingleValueArgumentEntry) SingleValueArgumentEntry.EMPTY; - } - TsValueProto tsValueProto = proto.getValue(); - long ts = tsValueProto.getTs(); - BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getArgName(), tsValueProto); - return new SingleValueArgumentEntry(ts, kvEntry, proto.getVersion()); - } - - private TsRollingArgumentEntry fromRollingArgumentProto(TsValueListProto proto) { - if (proto.getTsValueCount() <= 0) { - return (TsRollingArgumentEntry) TsRollingArgumentEntry.EMPTY; - } - TreeMap tsRecords = new TreeMap<>(); - proto.getTsValueList().forEach(tsValueProto -> { - BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getKey(), tsValueProto); - tsRecords.put(tsValueProto.getTs(), kvEntry); - }); - return new TsRollingArgumentEntry(tsRecords); - } - } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java index 4a24d13c93..0525c5eb90 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ScriptCalculatedFieldState.java @@ -22,14 +22,11 @@ import lombok.Data; import lombok.NoArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.cf.CalculatedFieldType; -import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.Output; -import org.thingsboard.server.common.data.kv.BasicKvEntry; import org.thingsboard.server.service.cf.CalculatedFieldResult; import java.util.List; import java.util.Map; -import java.util.TreeMap; @Data @Slf4j @@ -51,16 +48,6 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState { @Override public ListenableFuture performCalculation(CalculatedFieldCtx ctx) { - arguments.forEach((key, argumentEntry) -> { - if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) { - Argument argument = ctx.getArguments().get(key); - TreeMap tsRecords = tsRollingEntry.getTsRecords(); - if (tsRecords.size() > argument.getLimit()) { - tsRecords.pollFirstEntry(); - } - tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - argument.getTimeWindow()); - } - }); Object[] args = ctx.getArgNames().stream() .map(key -> arguments.get(key).getValue()) .toArray(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index d7e5ddd017..f271856ddb 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -32,8 +32,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; @AllArgsConstructor public class SingleValueArgumentEntry implements ArgumentEntry { - public static final ArgumentEntry EMPTY = new SingleValueArgumentEntry(0); - private long ts; private BasicKvEntry kvEntryValue; private Long version; @@ -61,27 +59,19 @@ public class SingleValueArgumentEntry implements ArgumentEntry { this.kvEntryValue = ProtoUtils.basicKvEntryFromKvEntry(entry); } - /** - * Internal constructor to create immutable SingleValueArgumentEntry.EMPTY - * */ - private SingleValueArgumentEntry(int ignored) { - this.ts = System.currentTimeMillis(); - this.kvEntryValue = null; - } - @Override public ArgumentEntryType getType() { return ArgumentEntryType.SINGLE_VALUE; } - @JsonIgnore - public Object getValue() { - return kvEntryValue.getValue(); + @Override + public boolean isEmpty() { + return kvEntryValue == null; } - @Override - public ArgumentEntry copy() { - return new SingleValueArgumentEntry(this.ts, this.kvEntryValue, this.version); + @JsonIgnore + public Object getValue() { + return isEmpty() ? null : kvEntryValue.getValue(); } @Override @@ -98,7 +88,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry { // TODO: should we persist updated ts and version values? BasicKvEntry newValue = singleValueEntry.getKvEntryValue(); - if (this.kvEntryValue.getValue().equals(newValue.getValue())) { + if (this.kvEntryValue != null && this.kvEntryValue.getValue().equals(newValue.getValue())) { return false; } this.kvEntryValue = singleValueEntry.getKvEntryValue(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index ddae0c3513..7b2a743fa9 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -37,21 +37,26 @@ import java.util.stream.Collectors; @Slf4j public class TsRollingArgumentEntry implements ArgumentEntry { - public static final ArgumentEntry EMPTY = new TsRollingArgumentEntry(0); - - private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000; - + private Integer limit; + private Long timeWindow; private TreeMap tsRecords = new TreeMap<>(); - public TsRollingArgumentEntry(List kvEntries) { + public TsRollingArgumentEntry(List kvEntries, int limit, long timeWindow) { kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry)); + this.limit = limit; + this.timeWindow = timeWindow; + } + + public TsRollingArgumentEntry(TreeMap tsRecords, int limit, long timeWindow) { + this.tsRecords = tsRecords; + this.limit = limit; + this.timeWindow = timeWindow; } - /** - * Internal constructor to create immutable TsRollingArgumentEntry.EMPTY - */ - private TsRollingArgumentEntry(int ignored) { + public TsRollingArgumentEntry(int limit, long timeWindow) { this.tsRecords = new TreeMap<>(); + this.limit = limit; + this.timeWindow = timeWindow; } @Override @@ -59,6 +64,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry { return ArgumentEntryType.TS_ROLLING; } + @Override + public boolean isEmpty() { + return tsRecords.isEmpty(); + } + @JsonIgnore @Override public Object getValue() { @@ -70,12 +80,6 @@ public class TsRollingArgumentEntry implements ArgumentEntry { (oldValue, newValue) -> oldValue, TreeMap::new )); - - } - - @Override - public ArgumentEntry copy() { - return new TsRollingArgumentEntry(new TreeMap<>(tsRecords)); } @Override @@ -112,9 +116,10 @@ public class TsRollingArgumentEntry implements ArgumentEntry { private void addTsRecord(Long ts, KvEntry value) { if (NumberUtils.isParsable(value.getValue().toString())) { tsRecords.put(ts, ProtoUtils.basicKvEntryFromKvEntry(value)); - if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) { + if (tsRecords.size() > limit) { tsRecords.pollFirstEntry(); } + tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - timeWindow); } else { throw new IllegalArgumentException("Argument type " + getType() + " only supports numeric values."); } diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java new file mode 100644 index 0000000000..d90cc62c8e --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java @@ -0,0 +1,147 @@ +/** + * Copyright © 2016-2024 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.utils; + +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.cf.CalculatedFieldType; +import org.thingsboard.server.common.data.id.CalculatedFieldId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.kv.BasicKvEntry; +import org.thingsboard.server.common.util.KvProtoUtil; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto; +import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto; +import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto; +import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentEntryProto; +import org.thingsboard.server.gen.transport.TransportProtos.TsValueListProto; +import org.thingsboard.server.gen.transport.TransportProtos.TsValueProto; +import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; + +import java.util.Optional; +import java.util.TreeMap; +import java.util.UUID; + +public class CalculatedFieldUtils { + + public static CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) { + return CalculatedFieldEntityCtxIdProto.newBuilder() + .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits()) + .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits()) + .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits()) + .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits()) + .setEntityType(ctxId.entityId().getEntityType().name()) + .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits()) + .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits()) + .build(); + } + + public static CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) { + TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB())); + EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB())); + CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB())); + return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId); + } + + public static CalculatedFieldStateProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) { + CalculatedFieldStateProto.Builder builder = CalculatedFieldStateProto.newBuilder() + .setId(toProto(stateId)) + .setType(state.getType().name()); + + state.getArguments().forEach((argName, argEntry) -> { + if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { + builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry)); + } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) { + builder.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry)); + } + }); + + return builder.build(); + } + + public static SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) { + SingleValueArgumentProto.Builder builder = SingleValueArgumentProto.newBuilder() + .setArgName(argName); + + if (entry.getKvEntryValue() != null) { + builder.setValue(KvProtoUtil.toTsValueProto(entry.getTs(), entry.getKvEntryValue())); + } + + Optional.ofNullable(entry.getVersion()).ifPresent(builder::setVersion); + + return builder.build(); + } + + public static TsRollingArgumentEntryProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) { + TsValueListProto.Builder tsRecordsBuilder = TsValueListProto.newBuilder() + .setKey(argName); + + entry.getTsRecords().forEach((ts, value) -> tsRecordsBuilder.addTsValue(KvProtoUtil.toTsValueProto(ts, value))); + + return TsRollingArgumentEntryProto.newBuilder() + .setLimit(entry.getLimit()) + .setTimeWindow(entry.getTimeWindow()) + .setRecords(tsRecordsBuilder.build()) + .build(); + } + + public static CalculatedFieldState fromProto(CalculatedFieldStateProto proto) { + if (StringUtils.isEmpty(proto.getType())) { + return null; + } + + CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType()); + + CalculatedFieldState state = switch (type) { + case SIMPLE -> new SimpleCalculatedFieldState(); + case SCRIPT -> new ScriptCalculatedFieldState(); + }; + + proto.getSingleValueArgumentsList().forEach(argProto -> + state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto))); + + if (CalculatedFieldType.SCRIPT.equals(type)) { + proto.getRollingValueArgumentsList().forEach(argProto -> + state.getArguments().put(argProto.getRecords().getKey(), fromRollingArgumentProto(argProto))); + } + + return state; + } + + public static SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) { + TsValueProto tsValueProto = proto.getValue(); + long ts = tsValueProto.getTs(); + BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getArgName(), tsValueProto); + return new SingleValueArgumentEntry(ts, kvEntry, proto.getVersion()); + } + + public static TsRollingArgumentEntry fromRollingArgumentProto(TsRollingArgumentEntryProto proto) { + TsValueListProto records = proto.getRecords(); + String key = records.getKey(); + TreeMap tsRecords = new TreeMap<>(); + records.getTsValueList().forEach(tsValueProto -> { + BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(key, tsValueProto); + tsRecords.put(tsValueProto.getTs(), kvEntry); + }); + return new TsRollingArgumentEntry(tsRecords, proto.getLimit(), proto.getTimeWindow()); + } + +} diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 197f05a5d5..30a3e51d8c 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -510,12 +510,6 @@ actors: js_print_interval_ms: "${ACTORS_JS_STATISTICS_PRINT_INTERVAL_MS:10000}" # Actors statistic persistence frequency in milliseconds persist_frequency: "${ACTORS_STATISTICS_PERSIST_FREQUENCY:3600000}" - calculated_fields: - debug_mode_rate_limits_per_tenant: - # Enable/Disable the rate limit of persisted debug events for all calculated fields per tenant - enabled: "${ACTORS_CF_DEBUG_MODE_RATE_LIMITS_PER_TENANT_ENABLED:true}" - # The value of DEBUG mode rate limit. By default, no more than 50 thousand events per hour - configuration: "${ACTORS_CF_DEBUG_MODE_RATE_LIMITS_PER_TENANT_CONFIGURATION:50000:3600}" debug: settings: 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 bded0d058d..1429020481 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 @@ -189,7 +189,7 @@ public class ScriptCalculatedFieldStateTest { @Test void testIsReadyWhenEmptyEntryPresents() { - state.arguments = new HashMap<>(Map.of("deviceTemperature", TsRollingArgumentEntry.EMPTY, "assetHumidity", assetHumidityArgEntry)); +// state.arguments = new HashMap<>(Map.of("deviceTemperature", TsRollingArgumentEntry.EMPTY, "assetHumidity", assetHumidityArgEntry)); assertThat(state.isReady()).isFalse(); } 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 d803ad2dab..d87e690478 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 @@ -117,10 +117,10 @@ public class SimpleCalculatedFieldStateTest { "key2", key2ArgEntry )); - Map newArgs = Map.of("key3", TsRollingArgumentEntry.EMPTY); - assertThatThrownBy(() -> state.updateState(newArgs)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Rolling argument entry is not supported for simple calculated fields."); +// Map newArgs = Map.of("key3", TsRollingArgumentEntry.EMPTY); +// assertThatThrownBy(() -> state.updateState(newArgs)) +// .isInstanceOf(IllegalArgumentException.class) +// .hasMessage("Rolling argument entry is not supported for simple calculated fields."); } @Test @@ -175,7 +175,7 @@ public class SimpleCalculatedFieldStateTest { "key1", key1ArgEntry, "key2", key2ArgEntry )); - state.getArguments().put("key3", SingleValueArgumentEntry.EMPTY); +// state.getArguments().put("key3", SingleValueArgumentEntry.EMPTY); assertThat(state.isReady()).isFalse(); } diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java index 13651e852d..57e4655d95 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java @@ -20,7 +20,6 @@ import org.junit.jupiter.api.Test; import org.thingsboard.server.common.data.kv.LongDataEntry; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; public class SingleValueArgumentEntryTest { @@ -40,9 +39,9 @@ public class SingleValueArgumentEntryTest { @Test void testUpdateEntryWhenRollingEntryPassed() { - assertThatThrownBy(() -> entry.updateEntry(TsRollingArgumentEntry.EMPTY)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Unsupported argument entry type for single value argument entry: " + ArgumentEntryType.TS_ROLLING); +// assertThatThrownBy(() -> entry.updateEntry(TsRollingArgumentEntry.EMPTY)) +// .isInstanceOf(IllegalArgumentException.class) +// .hasMessage("Unsupported argument entry type for single value argument entry: " + ArgumentEntryType.TS_ROLLING); } @Test diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java index 5a6f5e96e5..513eec3d8d 100644 --- a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntryTest.java @@ -40,7 +40,7 @@ public class TsRollingArgumentEntryTest { values.put(ts - 30, new DoubleDataEntry("key", 12.0)); values.put(ts - 20, new DoubleDataEntry("key", 17.0)); - entry = new TsRollingArgumentEntry(values); +// entry = new TsRollingArgumentEntry(values); } @Test diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java index 6532a4fe0d..35068821ee 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/limit/LimitedApi.java @@ -43,7 +43,8 @@ public enum LimitedApi { TRANSPORT_MESSAGES_PER_GATEWAY("transport messages per gateway", false), TRANSPORT_MESSAGES_PER_GATEWAY_DEVICE("transport messages per gateway device", false), EMAILS("emails sending", true), - WS_SUBSCRIPTIONS("WS subscriptions", false); + WS_SUBSCRIPTIONS("WS subscriptions", false), + CALCULATED_FIELD_DEBUG_EVENTS(DefaultTenantProfileConfiguration::getCalculatedFieldDebugEventsRateLimit, "calculated field debug events", true); private Function configExtractor; @Getter diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 054adbab1b..0139792fa7 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -140,6 +140,7 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long maxDataPointsPerRollingArg; private long maxStateSizeInKBytes; private long maxSingleValueArgumentSizeInKBytes; + private String calculatedFieldDebugEventsRateLimit; @Override public long getProfileThreshold(ApiUsageRecordKey key) { diff --git a/common/proto/src/main/proto/queue.proto b/common/proto/src/main/proto/queue.proto index 4a2f3710ae..349f507f35 100644 --- a/common/proto/src/main/proto/queue.proto +++ b/common/proto/src/main/proto/queue.proto @@ -814,12 +814,18 @@ message SingleValueArgumentProto { int64 version = 3; } +message TsRollingArgumentEntryProto { + TsValueListProto records = 1; + int32 limit = 2; + int64 timeWindow = 3; +} + message CalculatedFieldStateProto { CalculatedFieldEntityCtxIdProto id = 1; // int32 version = 2; string type = 3; repeated SingleValueArgumentProto singleValueArguments = 4; - repeated TsValueListProto rollingValueArguments = 5; + repeated TsRollingArgumentEntryProto rollingValueArguments = 5; } //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.