From b38b47f4093b547502c3e47a72f7f88526e45032 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Thu, 22 Oct 2020 18:51:34 +0300 Subject: [PATCH] JS Invoke Service stats --- .../actors/ruleChain/DefaultTbContext.java | 2 +- .../controller/RuleChainController.java | 2 +- .../server/service/apiusage/ApiFeature.java | 15 ++ .../service/apiusage/TenantApiUsageState.java | 36 +-- .../queue/DefaultTbClusterService.java | 2 +- .../processing/AbstractConsumerService.java | 11 +- .../script/AbstractJsInvokeService.java | 13 +- .../AbstractNashornJsInvokeService.java | 11 +- .../service/script/JsExecutorService.java | 2 +- .../service/script/JsInvokeService.java | 5 +- .../script/NashornJsInvokeService.java | 5 + .../service/script/RemoteJsInvokeService.java | 5 + .../script/RuleNodeJsScriptEngine.java | 11 +- .../transport/DefaultTransportApiService.java | 2 +- .../script/RuleNodeJsScriptEngineTest.java | 250 ------------------ .../script/TestNashornJsInvokeService.java | 57 ---- .../server/common/data/ApiUsageRecordKey.java | 6 +- .../tenant/profile/TenantProfileData.java | 13 - .../MultipleTbQueueCallbackWrapper.java | 2 +- .../transport/mqtt/MqttTransportHandler.java | 2 +- .../common/transport/TransportService.java | 5 - .../DefaultTransportRateLimitFactory.java | 47 ---- .../DefaultTransportRateLimitService.java | 128 +++++---- ...ry.java => EntityTransportRateLimits.java} | 11 +- .../limits/SimpleTransportRateLimit.java | 5 + .../limits/TransportRateLimitService.java | 3 +- .../limits/TransportRateLimitType.java | 48 ---- .../service/DefaultTransportService.java | 32 +-- .../DefaultTransportTenantProfileCache.java | 2 +- .../dao/tenant/TenantProfileServiceImpl.java | 5 +- .../service/BaseTenantProfileServiceTest.java | 5 +- 31 files changed, 209 insertions(+), 534 deletions(-) delete mode 100644 application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java delete mode 100644 application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitFactory.java rename common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/{TransportRateLimitFactory.java => EntityTransportRateLimits.java} (70%) delete mode 100644 common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitType.java diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index c8bb083577..e7a54385a7 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -312,7 +312,7 @@ class DefaultTbContext implements TbContext { @Override public ScriptEngine createJsScriptEngine(String script, String... argNames) { - return new RuleNodeJsScriptEngine(mainCtx.getJsSandbox(), nodeCtx.getSelf().getId(), script, argNames); + return new RuleNodeJsScriptEngine(getTenantId(), mainCtx.getJsSandbox(), nodeCtx.getSelf().getId(), script, argNames); } @Override diff --git a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java index 6d80c05c05..2ca834d85b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java +++ b/application/src/main/java/org/thingsboard/server/controller/RuleChainController.java @@ -347,7 +347,7 @@ public class RuleChainController extends BaseController { String errorText = ""; ScriptEngine engine = null; try { - engine = new RuleNodeJsScriptEngine(jsInvokeService, getCurrentUser().getId(), script, argNames); + engine = new RuleNodeJsScriptEngine(getTenantId(), jsInvokeService, getCurrentUser().getId(), script, argNames); TbMsg inMsg = TbMsg.newMsg(msgType, null, new TbMsgMetaData(metadata), TbMsgDataType.JSON, data); switch (scriptType) { case "update": diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/ApiFeature.java b/application/src/main/java/org/thingsboard/server/service/apiusage/ApiFeature.java index a893a55873..8267d4addd 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/ApiFeature.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/ApiFeature.java @@ -1,3 +1,18 @@ +/** + * Copyright © 2016-2020 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.service.apiusage; public enum ApiFeature { diff --git a/application/src/main/java/org/thingsboard/server/service/apiusage/TenantApiUsageState.java b/application/src/main/java/org/thingsboard/server/service/apiusage/TenantApiUsageState.java index 0e68d02b71..5dcfb34ec3 100644 --- a/application/src/main/java/org/thingsboard/server/service/apiusage/TenantApiUsageState.java +++ b/application/src/main/java/org/thingsboard/server/service/apiusage/TenantApiUsageState.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -21,6 +21,7 @@ import org.springframework.data.util.Pair; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.ApiUsageState; import org.thingsboard.server.common.data.TenantProfile; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; @@ -100,15 +101,18 @@ public class TenantApiUsageState { } public long getProfileThreshold(ApiUsageRecordKey key) { - Object threshold = tenantProfileData.getProperties().get(key.name()); - if (threshold != null) { - if (threshold instanceof String) { - return Long.parseLong((String) threshold); - } else if (threshold instanceof Long) { - return (Long) threshold; - } else if (threshold instanceof Integer) { - return (Integer) threshold; - } + DefaultTenantProfileConfiguration config = (DefaultTenantProfileConfiguration) tenantProfileData.getConfiguration(); + switch (key) { + case TRANSPORT_MSG_COUNT: + return config.getMaxTransportMessages(); + case TRANSPORT_DP_COUNT: + return config.getMaxTransportDataPoints(); + case JS_EXEC_COUNT: + return config.getMaxJSExecutions(); + case RE_EXEC_COUNT: + return config.getMaxREExecutions(); + case STORAGE_DP_COUNT: + return config.getMaxDPStorageDays(); } return 0L; } @@ -155,12 +159,12 @@ public class TenantApiUsageState { public boolean isFeatureEnabled(ApiUsageRecordKey recordKey) { switch (recordKey) { - case MSG_COUNT: - case DP_TRANSPORT_COUNT: + case TRANSPORT_MSG_COUNT: + case TRANSPORT_DP_COUNT: return isTransportEnabled(); case RE_EXEC_COUNT: return isRuleEngineEnabled(); - case DP_STORAGE_COUNT: + case STORAGE_DP_COUNT: return isDbStorageEnabled(); case JS_EXEC_COUNT: return isJsExecEnabled(); @@ -173,8 +177,8 @@ public class TenantApiUsageState { ApiFeature feature = null; boolean currentValue = isFeatureEnabled(recordKey); switch (recordKey) { - case MSG_COUNT: - case DP_TRANSPORT_COUNT: + case TRANSPORT_MSG_COUNT: + case TRANSPORT_DP_COUNT: feature = ApiFeature.TRANSPORT; setTransportEnabled(value); break; @@ -182,7 +186,7 @@ public class TenantApiUsageState { feature = ApiFeature.RE; setRuleEngineEnabled(value); break; - case DP_STORAGE_COUNT: + case STORAGE_DP_COUNT: feature = ApiFeature.DB; setDbStorageEnabled(value); break; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index b26baf8d2e..f08092d32d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 02fe877cd0..9a4d488e0f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; import org.thingsboard.server.common.data.id.TenantProfileId; +import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.ServiceType; @@ -153,10 +154,14 @@ public abstract class AbstractConsumerService scriptIdToNameMap = new ConcurrentHashMap<>(); protected Map blackListedFunctions = new ConcurrentHashMap<>(); + protected AbstractJsInvokeService(TbUsageStatsClient apiUsageStatsClient) { + this.apiUsageStatsClient = apiUsageStatsClient; + } + public void init(long maxRequestsTimeout) { if (maxRequestsTimeout > 0) { timeoutExecutorService = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("nashorn-js-timeout")); @@ -50,7 +58,7 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { } @Override - public ListenableFuture eval(JsScriptType scriptType, String scriptBody, String... argNames) { + public ListenableFuture eval(TenantId tenantId, JsScriptType scriptType, String scriptBody, String... argNames) { UUID scriptId = UUID.randomUUID(); String functionName = "invokeInternal_" + scriptId.toString().replace('-', '_'); String jsScript = generateJsScript(scriptType, functionName, scriptBody, argNames); @@ -58,12 +66,13 @@ public abstract class AbstractJsInvokeService implements JsInvokeService { } @Override - public ListenableFuture invokeFunction(UUID scriptId, Object... args) { + public ListenableFuture invokeFunction(TenantId tenantId, UUID scriptId, Object... args) { String functionName = scriptIdToNameMap.get(scriptId); if (functionName == null) { return Futures.immediateFailedFuture(new RuntimeException("No compiled script found for scriptId: [" + scriptId + "]!")); } if (!isBlackListed(scriptId)) { + apiUsageStatsClient.report(tenantId, ApiUsageRecordKey.JS_EXEC_COUNT, 1); return doInvokeFunction(scriptId, functionName, args); } else { return Futures.immediateFailedFuture( diff --git a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java index 8e6666bbba..17be0602f8 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/AbstractNashornJsInvokeService.java @@ -27,7 +27,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; -import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.queue.usagestats.TbUsageStatsClient; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; @@ -38,7 +38,6 @@ import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -57,9 +56,8 @@ public abstract class AbstractNashornJsInvokeService extends AbstractJsInvokeSer private final FutureCallback evalCallback = new JsStatCallback<>(jsEvalMsgs, jsTimeoutMsgs, jsFailedMsgs); private final FutureCallback invokeCallback = new JsStatCallback<>(jsInvokeMsgs, jsTimeoutMsgs, jsFailedMsgs); - @Autowired @Getter - private JsExecutorService jsExecutor; + private final JsExecutorService jsExecutor; @Value("${js.local.max_requests_timeout:0}") private long maxRequestsTimeout; @@ -67,6 +65,11 @@ public abstract class AbstractNashornJsInvokeService extends AbstractJsInvokeSer @Value("${js.local.stats.enabled:false}") private boolean statsEnabled; + public AbstractNashornJsInvokeService(TbUsageStatsClient apiUsageStatsClient, JsExecutorService jsExecutor) { + super(apiUsageStatsClient); + this.jsExecutor = jsExecutor; + } + @Scheduled(fixedDelayString = "${js.local.stats.print_interval_ms:10000}") public void printStats() { if (statsEnabled) { diff --git a/application/src/main/java/org/thingsboard/server/service/script/JsExecutorService.java b/application/src/main/java/org/thingsboard/server/service/script/JsExecutorService.java index 97e40e6f62..15fd8c7534 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/JsExecutorService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/JsExecutorService.java @@ -27,7 +27,7 @@ public class JsExecutorService extends AbstractListeningExecutor { @Override protected int getThreadPollSize() { - return jsExecutorThreadPoolSize; + return Math.max(jsExecutorThreadPoolSize, 1); } } diff --git a/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java index a9f727e4a7..ae990af66c 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/JsInvokeService.java @@ -17,14 +17,15 @@ package org.thingsboard.server.service.script; import com.google.common.util.concurrent.ListenableFuture; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; import java.util.UUID; public interface JsInvokeService { - ListenableFuture eval(JsScriptType scriptType, String scriptBody, String... argNames); + ListenableFuture eval(TenantId tenantId, JsScriptType scriptType, String scriptBody, String... argNames); - ListenableFuture invokeFunction(UUID scriptId, Object... args); + ListenableFuture invokeFunction(TenantId tenantId, UUID scriptId, Object... args); ListenableFuture release(UUID scriptId); diff --git a/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java index 66a14cc827..c20f3e8c08 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/NashornJsInvokeService.java @@ -19,6 +19,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; +import org.thingsboard.server.queue.usagestats.TbUsageStatsClient; import java.util.concurrent.TimeUnit; @@ -42,6 +43,10 @@ public class NashornJsInvokeService extends AbstractNashornJsInvokeService { @Value("${js.local.max_black_list_duration_sec:60}") private int maxBlackListDurationSec; + public NashornJsInvokeService(TbUsageStatsClient apiUsageStatsClient, JsExecutorService jsExecutor) { + super(apiUsageStatsClient, jsExecutor); + } + @Override protected boolean useJsSandbox() { return useJsSandbox; diff --git a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java index cfb0c8a8ee..2ada577287 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java @@ -30,6 +30,7 @@ import org.thingsboard.server.gen.js.JsInvokeProtos; import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoJsQueueMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.usagestats.TbUsageStatsClient; import javax.annotation.Nullable; import javax.annotation.PostConstruct; @@ -68,6 +69,10 @@ public class RemoteJsInvokeService extends AbstractJsInvokeService { private final AtomicInteger queueFailedMsgs = new AtomicInteger(0); private final AtomicInteger queueTimeoutMsgs = new AtomicInteger(0); + public RemoteJsInvokeService(TbUsageStatsClient apiUsageStatsClient) { + super(apiUsageStatsClient); + } + @Scheduled(fixedDelayString = "${js.remote.stats.print_interval_ms}") public void printStats() { if (statsEnabled) { diff --git a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java index 110f410fad..275e645ce2 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngine.java @@ -25,6 +25,7 @@ import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; @@ -43,13 +44,15 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S private final JsInvokeService sandboxService; private final UUID scriptId; + private final TenantId tenantId; private final EntityId entityId; - public RuleNodeJsScriptEngine(JsInvokeService sandboxService, EntityId entityId, String script, String... argNames) { + public RuleNodeJsScriptEngine(TenantId tenantId, JsInvokeService sandboxService, EntityId entityId, String script, String... argNames) { + this.tenantId = tenantId; this.sandboxService = sandboxService; this.entityId = entityId; try { - this.scriptId = this.sandboxService.eval(JsScriptType.RULE_NODE_SCRIPT, script, argNames).get(); + this.scriptId = this.sandboxService.eval(tenantId, JsScriptType.RULE_NODE_SCRIPT, script, argNames).get(); } catch (Exception e) { Throwable t = e; if (e instanceof ExecutionException) { @@ -203,7 +206,7 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S private JsonNode executeScript(TbMsg msg) throws ScriptException { try { String[] inArgs = prepareArgs(msg); - String eval = sandboxService.invokeFunction(this.scriptId, inArgs[0], inArgs[1], inArgs[2]).get().toString(); + String eval = sandboxService.invokeFunction(tenantId, this.scriptId, inArgs[0], inArgs[1], inArgs[2]).get().toString(); return mapper.readTree(eval); } catch (ExecutionException e) { if (e.getCause() instanceof ScriptException) { @@ -220,7 +223,7 @@ public class RuleNodeJsScriptEngine implements org.thingsboard.rule.engine.api.S private ListenableFuture executeScriptAsync(TbMsg msg) { String[] inArgs = prepareArgs(msg); - return Futures.transformAsync(sandboxService.invokeFunction(this.scriptId, inArgs[0], inArgs[1], inArgs[2]), + return Futures.transformAsync(sandboxService.invokeFunction(tenantId, this.scriptId, inArgs[0], inArgs[1], inArgs[2]), o -> { try { return Futures.immediateFuture(mapper.readTree(o.toString())); diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 6d0b7a0aa8..ab566fb147 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java b/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java deleted file mode 100644 index 60495b119d..0000000000 --- a/application/src/test/java/org/thingsboard/server/service/script/RuleNodeJsScriptEngineTest.java +++ /dev/null @@ -1,250 +0,0 @@ -/** - * Copyright © 2016-2020 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.service.script; - -import com.datastax.oss.driver.api.core.uuid.Uuids; -import com.google.common.collect.Sets; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.thingsboard.rule.engine.api.ScriptEngine; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.RuleNodeId; -import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.TbMsgDataType; -import org.thingsboard.server.common.msg.TbMsgMetaData; - -import javax.script.ScriptException; -import java.util.Map; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; - -import static org.junit.Assert.*; - -public class RuleNodeJsScriptEngineTest { - - private ScriptEngine scriptEngine; - private TestNashornJsInvokeService jsSandboxService; - - private EntityId ruleNodeId = new RuleNodeId(Uuids.timeBased()); - - @Before - public void beforeTest() throws Exception { - jsSandboxService = new TestNashornJsInvokeService(false, 1, 100, 3); - } - - @After - public void afterTest() throws Exception { - jsSandboxService.stop(); - } - - @Test - public void msgCanBeUpdated() throws ScriptException { - String function = "metadata.temp = metadata.temp * 10; return {metadata: metadata};"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, function); - - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "7"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson); - - TbMsg actual = scriptEngine.executeUpdate(msg); - assertEquals("70", actual.getMetaData().getValue("temp")); - scriptEngine.destroy(); - } - - @Test - public void newAttributesCanBeAddedInMsg() throws ScriptException { - String function = "metadata.newAttr = metadata.humidity - msg.passed; return {metadata: metadata};"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, function); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "7"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson); - - TbMsg actual = scriptEngine.executeUpdate(msg); - assertEquals("94", actual.getMetaData().getValue("newAttr")); - scriptEngine.destroy(); - } - - @Test - public void payloadCanBeUpdated() throws ScriptException { - String function = "msg.passed = msg.passed * metadata.temp; msg.bigObj.newProp = 'Ukraine'; return {msg: msg};"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, function); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "7"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\":\"Vit\",\"passed\": 5,\"bigObj\":{\"prop\":42}}"; - - TbMsg msg =TbMsg.newMsg("USER", null, metaData, TbMsgDataType.JSON, rawJson); - - TbMsg actual = scriptEngine.executeUpdate(msg); - - String expectedJson = "{\"name\":\"Vit\",\"passed\":35,\"bigObj\":{\"prop\":42,\"newProp\":\"Ukraine\"}}"; - assertEquals(expectedJson, actual.getData()); - scriptEngine.destroy(); - } - - @Test - public void metadataAccessibleForFilter() throws ScriptException { - String function = "return metadata.humidity < 15;"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, function); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "7"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg("USER", null, metaData, TbMsgDataType.JSON, rawJson); - assertFalse(scriptEngine.executeFilter(msg)); - scriptEngine.destroy(); - } - - @Test - public void dataAccessibleForFilter() throws ScriptException { - String function = "return msg.passed < 15 && msg.name === 'Vit' && metadata.temp == 7 && msg.bigObj.prop == 42;"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, function); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "7"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg( "USER", null, metaData,TbMsgDataType.JSON, rawJson); - assertTrue(scriptEngine.executeFilter(msg)); - scriptEngine.destroy(); - } - - @Test - public void dataAccessibleForSwitch() throws ScriptException { - String jsCode = "function nextRelation(metadata, msg) {\n" + - " if(msg.passed == 5 && metadata.temp == 10)\n" + - " return 'one'\n" + - " else\n" + - " return 'two';\n" + - "};\n" + - "\n" + - "return nextRelation(metadata, msg);"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, jsCode); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "10"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson); - Set actual = scriptEngine.executeSwitch(msg); - assertEquals(Sets.newHashSet("one"), actual); - scriptEngine.destroy(); - } - - @Test - public void multipleRelationsReturnedFromSwitch() throws ScriptException { - String jsCode = "function nextRelation(metadata, msg) {\n" + - " if(msg.passed == 5 && metadata.temp == 10)\n" + - " return ['three', 'one']\n" + - " else\n" + - " return 'two';\n" + - "};\n" + - "\n" + - "return nextRelation(metadata, msg);"; - scriptEngine = new RuleNodeJsScriptEngine(jsSandboxService, ruleNodeId, jsCode); - TbMsgMetaData metaData = new TbMsgMetaData(); - metaData.putValue("temp", "10"); - metaData.putValue("humidity", "99"); - String rawJson = "{\"name\": \"Vit\", \"passed\": 5, \"bigObj\": {\"prop\":42}}"; - - TbMsg msg = TbMsg.newMsg( "USER", null, metaData, TbMsgDataType.JSON, rawJson); - Set actual = scriptEngine.executeSwitch(msg); - assertEquals(Sets.newHashSet("one", "three"), actual); - scriptEngine.destroy(); - } - - @Test - public void concurrentReleasedCorrectly() throws InterruptedException, ExecutionException { - String code = "metadata.temp = metadata.temp * 10; return {metadata: metadata};"; - - int repeat = 1000; - ExecutorService service = Executors.newFixedThreadPool(repeat); - Map scriptIds = new ConcurrentHashMap<>(); - CountDownLatch startLatch = new CountDownLatch(repeat); - CountDownLatch finishLatch = new CountDownLatch(repeat); - AtomicInteger failedCount = new AtomicInteger(0); - - for (int i = 0; i < repeat; i++) { - service.submit(() -> runScript(startLatch, finishLatch, failedCount, scriptIds, code)); - } - - finishLatch.await(); - assertTrue(scriptIds.size() == 1); - assertTrue(failedCount.get() == 0); - - CountDownLatch nextStart = new CountDownLatch(repeat); - CountDownLatch nextFinish = new CountDownLatch(repeat); - for (int i = 0; i < repeat; i++) { - service.submit(() -> runScript(nextStart, nextFinish, failedCount, scriptIds, code)); - } - - nextFinish.await(); - assertTrue(scriptIds.size() == 1); - assertTrue(failedCount.get() == 0); - service.shutdownNow(); - } - - @Test - public void concurrentFailedEvaluationShouldThrowException() throws InterruptedException { - String code = "metadata.temp = metadata.temp * 10; urn {metadata: metadata};"; - - int repeat = 10000; - ExecutorService service = Executors.newFixedThreadPool(repeat); - Map scriptIds = new ConcurrentHashMap<>(); - CountDownLatch startLatch = new CountDownLatch(repeat); - CountDownLatch finishLatch = new CountDownLatch(repeat); - AtomicInteger failedCount = new AtomicInteger(0); - for (int i = 0; i < repeat; i++) { - service.submit(() -> { - service.submit(() -> runScript(startLatch, finishLatch, failedCount, scriptIds, code)); - }); - } - - finishLatch.await(); - assertTrue(scriptIds.isEmpty()); - assertEquals(repeat, failedCount.get()); - service.shutdownNow(); - } - - private void runScript(CountDownLatch startLatch, CountDownLatch finishLatch, AtomicInteger failedCount, - Map scriptIds, String code) { - try { - for (int k = 0; k < 10; k++) { - startLatch.countDown(); - startLatch.await(); - UUID scriptId = jsSandboxService.eval(JsScriptType.RULE_NODE_SCRIPT, code).get(); - scriptIds.put(scriptId, new Object()); - jsSandboxService.invokeFunction(scriptId, "{}", "{}", "TEXT").get(); - jsSandboxService.release(scriptId).get(); - } - } catch (Throwable th) { - failedCount.incrementAndGet(); - } finally { - finishLatch.countDown(); - } - } - -} diff --git a/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java b/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java deleted file mode 100644 index 8a89de5dff..0000000000 --- a/application/src/test/java/org/thingsboard/server/service/script/TestNashornJsInvokeService.java +++ /dev/null @@ -1,57 +0,0 @@ -/** - * Copyright © 2016-2020 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.service.script; - -public class TestNashornJsInvokeService extends AbstractNashornJsInvokeService { - - private boolean useJsSandbox; - private final int monitorThreadPoolSize; - private final long maxCpuTime; - private final int maxErrors; - - public TestNashornJsInvokeService(boolean useJsSandbox, int monitorThreadPoolSize, long maxCpuTime, int maxErrors) { - this.useJsSandbox = useJsSandbox; - this.monitorThreadPoolSize = monitorThreadPoolSize; - this.maxCpuTime = maxCpuTime; - this.maxErrors = maxErrors; - init(); - } - - @Override - protected boolean useJsSandbox() { - return useJsSandbox; - } - - @Override - protected int getMonitorThreadPoolSize() { - return monitorThreadPoolSize; - } - - @Override - protected long getMaxCpuTime() { - return maxCpuTime; - } - - @Override - protected int getMaxErrors() { - return maxErrors; - } - - @Override - protected long getMaxBlacklistDuration() { - return 100000; - } -} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/ApiUsageRecordKey.java b/common/data/src/main/java/org/thingsboard/server/common/data/ApiUsageRecordKey.java index bc730e7e5a..0048e325f3 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/ApiUsageRecordKey.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/ApiUsageRecordKey.java @@ -17,9 +17,9 @@ package org.thingsboard.server.common.data; public enum ApiUsageRecordKey { - MSG_COUNT, - DP_TRANSPORT_COUNT, - DP_STORAGE_COUNT, + TRANSPORT_MSG_COUNT, + TRANSPORT_DP_COUNT, + STORAGE_DP_COUNT, RE_EXEC_COUNT, JS_EXEC_COUNT diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/TenantProfileData.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/TenantProfileData.java index 6336cb4ace..0996fc9850 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/TenantProfileData.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/TenantProfileData.java @@ -28,17 +28,4 @@ public class TenantProfileData { private TenantProfileConfiguration configuration; - @JsonIgnore - private Map properties = new HashMap<>(); - - @JsonAnyGetter - public Map properties() { - return this.properties; - } - - @JsonAnySetter - public void put(String name, Object value) { - this.properties.put(name, value); - } - } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueCallbackWrapper.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueCallbackWrapper.java index ddedb3a606..c28f3637b7 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueCallbackWrapper.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueCallbackWrapper.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index e70da7029e..14712a7cb3 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java index 3b775b9f26..980aa33221 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java @@ -19,7 +19,6 @@ import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.DeviceTransportType; import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; -import org.thingsboard.server.common.transport.limits.TransportRateLimitType; import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetEntityProfileRequestMsg; @@ -62,10 +61,6 @@ public interface TransportService { void process(ProvisionDeviceRequestMsg msg, TransportServiceCallback callback); - boolean checkLimits(SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback); - - boolean checkLimits(SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback, int dataPoints, TransportRateLimitType... limits); - void process(SessionInfoProto sessionInfo, SessionEventMsg msg, TransportServiceCallback callback); void process(SessionInfoProto sessionInfo, PostTelemetryMsg msg, TransportServiceCallback callback); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitFactory.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitFactory.java deleted file mode 100644 index 362502bf17..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitFactory.java +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Copyright © 2016-2020 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.common.transport.limits; - -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; -import org.thingsboard.server.common.msg.tools.TbRateLimits; - -@Slf4j -@Component -public class DefaultTransportRateLimitFactory implements TransportRateLimitFactory { - - private static final DummyTransportRateLimit ALWAYS_TRUE = new DummyTransportRateLimit(); - - @Override - public TransportRateLimit create(TransportRateLimitType type, Object configuration) { - if (!StringUtils.isEmpty(configuration)) { - try { - return new SimpleTransportRateLimit(new TbRateLimits(configuration.toString()), configuration.toString()); - } catch (Exception e) { - log.warn("[{}] Failed to init rate limit with configuration: {}", type, configuration, e); - return ALWAYS_TRUE; - } - } else { - return ALWAYS_TRUE; - } - } - - @Override - public TransportRateLimit createDefault(TransportRateLimitType type) { - return ALWAYS_TRUE; - } -} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java index bc415cd92a..b6f160ad1e 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/DefaultTransportRateLimitService.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -17,76 +17,96 @@ package org.thingsboard.server.common.transport.limits; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; +import org.thingsboard.server.common.msg.tools.TbRateLimits; import org.thingsboard.server.common.transport.TransportTenantProfileCache; import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; import org.thingsboard.server.queue.util.TbTransportComponent; +import java.util.HashSet; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.function.BiConsumer; +import java.util.function.Function; @Service @TbTransportComponent @Slf4j public class DefaultTransportRateLimitService implements TransportRateLimitService { + private final static DummyTransportRateLimit ALLOW = new DummyTransportRateLimit(); private final ConcurrentMap tenantAllowed = new ConcurrentHashMap<>(); - private final ConcurrentMap perTenantLimits = new ConcurrentHashMap<>(); - private final ConcurrentMap perDeviceLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap> tenantDevices = new ConcurrentHashMap<>(); + private final ConcurrentMap perTenantLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap perDeviceLimits = new ConcurrentHashMap<>(); - private final TransportRateLimitFactory rateLimitFactory; private final TransportTenantProfileCache tenantProfileCache; - public DefaultTransportRateLimitService(TransportRateLimitFactory rateLimitFactory, TransportTenantProfileCache tenantProfileCache) { - this.rateLimitFactory = rateLimitFactory; + public DefaultTransportRateLimitService(TransportTenantProfileCache tenantProfileCache) { this.tenantProfileCache = tenantProfileCache; } @Override - public TransportRateLimitType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints, TransportRateLimitType... limits) { + public EntityType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints) { if (!tenantAllowed.getOrDefault(tenantId, Boolean.TRUE)) { - return TransportRateLimitType.TENANT_ADDED_TO_DISABLED_LIST; + return EntityType.TENANT; } - TransportRateLimit[] tenantLimits = getTenantRateLimits(tenantId); - TransportRateLimit[] deviceLimits = getDeviceRateLimits(tenantId, deviceId); - for (TransportRateLimitType limitType : limits) { - TransportRateLimit rateLimit; - if (limitType.isTenantLevel()) { - rateLimit = tenantLimits[limitType.ordinal()]; - } else { - rateLimit = deviceLimits[limitType.ordinal()]; - } - if (!rateLimit.tryConsume(limitType.isMessageLevel() ? 1L : dataPoints)) { - return limitType; - } + if (!checkEntityRateLimit(dataPoints, getTenantRateLimits(tenantId))) { + return EntityType.TENANT; + } + if (!checkEntityRateLimit(dataPoints, getDeviceRateLimits(tenantId, deviceId))) { + return EntityType.DEVICE; } return null; } + private boolean checkEntityRateLimit(int dataPoints, EntityTransportRateLimits tenantLimits) { + if (dataPoints > 0) { + return tenantLimits.getTelemetryMsgRateLimit().tryConsume() && tenantLimits.getTelemetryDataPointsRateLimit().tryConsume(dataPoints); + } else { + return tenantLimits.getRegularMsgRateLimit().tryConsume(); + } + } + @Override public void update(TenantProfileUpdateResult update) { - TransportRateLimit[] newLimits = createTransportRateLimits(update.getProfile()); + EntityTransportRateLimits tenantRateLimitPrototype = createRateLimits(update.getProfile(), true); + EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(update.getProfile(), false); for (TenantId tenantId : update.getAffectedTenants()) { - mergeLimits(tenantId, newLimits); + mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put); + tenantDevices.get(tenantId).forEach(deviceId -> { + mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put); + }); } } @Override public void update(TenantId tenantId) { - mergeLimits(tenantId, fetchProfileAndInit(tenantId)); + EntityTransportRateLimits tenantRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), true); + EntityTransportRateLimits deviceRateLimitPrototype = createRateLimits(tenantProfileCache.get(tenantId), false); + mergeLimits(tenantId, tenantRateLimitPrototype, perTenantLimits::get, perTenantLimits::put); + tenantDevices.get(tenantId).forEach(deviceId -> { + mergeLimits(deviceId, deviceRateLimitPrototype, perDeviceLimits::get, perDeviceLimits::put); + }); } @Override public void remove(TenantId tenantId) { perTenantLimits.remove(tenantId); + tenantDevices.remove(tenantId); } @Override public void remove(DeviceId deviceId) { perDeviceLimits.remove(deviceId); + tenantDevices.values().forEach(set -> set.remove(deviceId)); } @Override @@ -94,48 +114,66 @@ public class DefaultTransportRateLimitService implements TransportRateLimitServi tenantAllowed.put(tenantId, allowed); } - private void mergeLimits(TenantId tenantId, TransportRateLimit[] newRateLimits) { - TransportRateLimit[] oldRateLimits = perTenantLimits.get(tenantId); + private void mergeLimits(T deviceId, EntityTransportRateLimits newRateLimits, + Function getFunction, + BiConsumer putFunction) { + EntityTransportRateLimits oldRateLimits = getFunction.apply(deviceId); if (oldRateLimits == null) { - perTenantLimits.put(tenantId, newRateLimits); + putFunction.accept(deviceId, newRateLimits); } else { - for (int i = 0; i < TransportRateLimitType.values().length; i++) { - TransportRateLimit newLimit = newRateLimits[i]; - TransportRateLimit oldLimit = oldRateLimits[i]; - if (newLimit != null && (oldLimit == null || !oldLimit.getConfiguration().equals(newLimit.getConfiguration()))) { - oldRateLimits[i] = newLimit; - } + EntityTransportRateLimits updated = merge(oldRateLimits, newRateLimits); + if (updated != null) { + putFunction.accept(deviceId, updated); } } } - private TransportRateLimit[] fetchProfileAndInit(TenantId tenantId) { - return perTenantLimits.computeIfAbsent(tenantId, tmp -> createTransportRateLimits(tenantProfileCache.get(tenantId))); + private EntityTransportRateLimits merge(EntityTransportRateLimits oldRateLimits, EntityTransportRateLimits newRateLimits) { + boolean regularUpdate = !oldRateLimits.getRegularMsgRateLimit().getConfiguration().equals(newRateLimits.getRegularMsgRateLimit().getConfiguration()); + boolean telemetryMsgRateUpdate = !oldRateLimits.getTelemetryMsgRateLimit().getConfiguration().equals(newRateLimits.getTelemetryMsgRateLimit().getConfiguration()); + boolean telemetryDataPointUpdate = !oldRateLimits.getTelemetryDataPointsRateLimit().getConfiguration().equals(newRateLimits.getTelemetryDataPointsRateLimit().getConfiguration()); + if (regularUpdate || telemetryMsgRateUpdate || telemetryDataPointUpdate) { + return new EntityTransportRateLimits( + regularUpdate ? newLimit(newRateLimits.getRegularMsgRateLimit().getConfiguration()) : oldRateLimits.getRegularMsgRateLimit(), + telemetryMsgRateUpdate ? newLimit(newRateLimits.getTelemetryMsgRateLimit().getConfiguration()) : oldRateLimits.getTelemetryMsgRateLimit(), + telemetryDataPointUpdate ? newLimit(newRateLimits.getTelemetryDataPointsRateLimit().getConfiguration()) : oldRateLimits.getTelemetryDataPointsRateLimit()); + } else { + return null; + } } - private TransportRateLimit[] createTransportRateLimits(TenantProfile tenantProfile) { + private EntityTransportRateLimits createRateLimits(TenantProfile tenantProfile, boolean tenant) { TenantProfileData profileData = tenantProfile.getProfileData(); - TransportRateLimit[] rateLimits = new TransportRateLimit[TransportRateLimitType.values().length]; - for (TransportRateLimitType type : TransportRateLimitType.values()) { - rateLimits[type.ordinal()] = rateLimitFactory.create(type, profileData.getProperties().get(type.getConfigurationKey())); + DefaultTenantProfileConfiguration profile = (DefaultTenantProfileConfiguration) profileData.getConfiguration(); + if (profile == null) { + return new EntityTransportRateLimits(ALLOW, ALLOW, ALLOW); + } else { + TransportRateLimit regularMsgRateLimit = newLimit(tenant ? profile.getTransportTenantMsgRateLimit() : profile.getTransportDeviceMsgRateLimit()); + TransportRateLimit telemetryMsgRateLimit = newLimit(tenant ? profile.getTransportTenantTelemetryMsgRateLimit() : profile.getTransportDeviceTelemetryMsgRateLimit()); + TransportRateLimit telemetryDpRateLimit = newLimit(tenant ? profile.getTransportTenantTelemetryDataPointsRateLimit() : profile.getTransportTenantTelemetryDataPointsRateLimit()); + return new EntityTransportRateLimits(regularMsgRateLimit, telemetryMsgRateLimit, telemetryDpRateLimit); } - return rateLimits; } - private TransportRateLimit[] getTenantRateLimits(TenantId tenantId) { - TransportRateLimit[] limits = perTenantLimits.get(tenantId); + private static TransportRateLimit newLimit(String config) { + return StringUtils.isEmpty(config) ? ALLOW : new SimpleTransportRateLimit(config); + } + + private EntityTransportRateLimits getTenantRateLimits(TenantId tenantId) { + EntityTransportRateLimits limits = perTenantLimits.get(tenantId); if (limits == null) { - limits = fetchProfileAndInit(tenantId); + limits = createRateLimits(tenantProfileCache.get(tenantId), true); perTenantLimits.put(tenantId, limits); } return limits; } - private TransportRateLimit[] getDeviceRateLimits(TenantId tenantId, DeviceId deviceId) { - TransportRateLimit[] limits = perDeviceLimits.get(deviceId); + private EntityTransportRateLimits getDeviceRateLimits(TenantId tenantId, DeviceId deviceId) { + EntityTransportRateLimits limits = perDeviceLimits.get(deviceId); if (limits == null) { - limits = fetchProfileAndInit(tenantId); + limits = createRateLimits(tenantProfileCache.get(tenantId), false); perDeviceLimits.put(deviceId, limits); + tenantDevices.computeIfAbsent(tenantId, id -> ConcurrentHashMap.newKeySet()).add(deviceId); } return limits; } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitFactory.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/EntityTransportRateLimits.java similarity index 70% rename from common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitFactory.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/EntityTransportRateLimits.java index 54d83610c2..50831ccff7 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitFactory.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/EntityTransportRateLimits.java @@ -15,10 +15,15 @@ */ package org.thingsboard.server.common.transport.limits; -public interface TransportRateLimitFactory { +import lombok.AllArgsConstructor; +import lombok.Data; - TransportRateLimit create(TransportRateLimitType type, Object config); +@Data +@AllArgsConstructor +public class EntityTransportRateLimits { - TransportRateLimit createDefault(TransportRateLimitType type); + private TransportRateLimit regularMsgRateLimit; + private TransportRateLimit telemetryMsgRateLimit; + private TransportRateLimit telemetryDataPointsRateLimit; } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/SimpleTransportRateLimit.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/SimpleTransportRateLimit.java index 3253272ded..0ea920a4ba 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/SimpleTransportRateLimit.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/SimpleTransportRateLimit.java @@ -26,6 +26,11 @@ public class SimpleTransportRateLimit implements TransportRateLimit { @Getter private final String configuration; + public SimpleTransportRateLimit(String configuration) { + this.configuration = configuration; + this.rateLimit = new TbRateLimits(configuration); + } + @Override public boolean tryConsume() { return rateLimit.tryConsume(); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java index 2ab1ec38ac..3918927c0b 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitService.java @@ -15,13 +15,14 @@ */ package org.thingsboard.server.common.transport.limits; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; public interface TransportRateLimitService { - TransportRateLimitType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints, TransportRateLimitType... limits); + EntityType checkLimits(TenantId tenantId, DeviceId deviceId, int dataPoints); void update(TenantProfileUpdateResult update); diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitType.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitType.java deleted file mode 100644 index 038de86e84..0000000000 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/limits/TransportRateLimitType.java +++ /dev/null @@ -1,48 +0,0 @@ -/** - * Copyright © 2016-2020 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.common.transport.limits; - -import lombok.Getter; - -public enum TransportRateLimitType { - - TENANT_ADDED_TO_DISABLED_LIST("general.tenant.disabled", true, false), - TENANT_MAX_MSGS("transport.tenant.msg", true, true), - TENANT_TELEMETRY_MSGS("transport.tenant.telemetry", true, true), - TENANT_MAX_DATA_POINTS("transport.tenant.dataPoints", true, false), - DEVICE_MAX_MSGS("transport.device.msg", false, true), - DEVICE_TELEMETRY_MSGS("transport.device.telemetry", false, true), - DEVICE_MAX_DATA_POINTS("transport.device.dataPoints", false, false); - - @Getter - private final String configurationKey; - @Getter - private final boolean tenantLevel; - @Getter - private final boolean deviceLevel; - @Getter - private final boolean messageLevel; - @Getter - private final boolean dataPointLevel; - - TransportRateLimitType(String configurationKey, boolean tenantLevel, boolean messageLevel) { - this.configurationKey = configurationKey; - this.tenantLevel = tenantLevel; - this.deviceLevel = !tenantLevel; - this.messageLevel = messageLevel; - this.dataPointLevel = !messageLevel; - } -} diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index 326f996d14..f77a3ce44b 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -55,8 +55,6 @@ import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGateway import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.limits.TransportRateLimitService; -import org.thingsboard.server.common.transport.limits.TransportRateLimitType; -import org.thingsboard.server.common.transport.profile.TenantProfileUpdateResult; import org.thingsboard.server.common.transport.util.DataDecodingEncodingService; import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.transport.TransportProtos; @@ -364,7 +362,7 @@ public class DefaultTransportService implements TransportService { for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { dataPoints += tsKv.getKvCount(); } - if (checkLimits(sessionInfo, msg, callback, dataPoints, TELEMETRY)) { + if (checkLimits(sessionInfo, msg, callback, dataPoints)) { reportActivityInternal(sessionInfo); TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB())); DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); @@ -385,7 +383,7 @@ public class DefaultTransportService implements TransportService { @Override public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.PostAttributeMsg msg, TransportServiceCallback callback) { - if (checkLimits(sessionInfo, msg, callback, msg.getKvCount(), TELEMETRY)) { + if (checkLimits(sessionInfo, msg, callback, msg.getKvCount())) { reportActivityInternal(sessionInfo); TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB())); DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); @@ -575,31 +573,23 @@ public class DefaultTransportService implements TransportService { sessions.remove(toSessionId(sessionInfo)); } - private TransportRateLimitType[] DEFAULT = new TransportRateLimitType[]{TransportRateLimitType.TENANT_MAX_MSGS, TransportRateLimitType.DEVICE_MAX_MSGS}; - private TransportRateLimitType[] TELEMETRY = TransportRateLimitType.values(); - - @Override - public boolean checkLimits(TransportProtos.SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback) { - return checkLimits(sessionInfo, msg, callback, 0, DEFAULT); + private boolean checkLimits(TransportProtos.SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback) { + return checkLimits(sessionInfo, msg, callback, 0); } - @Override - public boolean checkLimits(TransportProtos.SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback, int dataPoints, TransportRateLimitType... limits) { + private boolean checkLimits(TransportProtos.SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback, int dataPoints) { if (log.isTraceEnabled()) { log.trace("[{}] Processing msg: {}", toSessionId(sessionInfo), msg); } TenantId tenantId = new TenantId(new UUID(sessionInfo.getTenantIdMSB(), sessionInfo.getTenantIdLSB())); DeviceId deviceId = new DeviceId(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); - TransportRateLimitType limit = rateLimitService.checkLimits(tenantId, deviceId, 0, limits); - if (limit == null) { + EntityType rateLimitedEntityType = rateLimitService.checkLimits(tenantId, deviceId, dataPoints); + if (rateLimitedEntityType == null) { return true; } else { if (callback != null) { - callback.onError(new TbRateLimitsException(limit.isTenantLevel() ? EntityType.TENANT : EntityType.DEVICE)); - } - if (log.isTraceEnabled()) { - log.trace("[{}][{}] {} rateLimit detected: {}", toSessionId(sessionInfo), tenantId, limit, msg); + callback.onError(new TbRateLimitsException(rateLimitedEntityType)); } return false; } @@ -831,8 +821,8 @@ public class DefaultTransportService implements TransportService { @Override public void onSuccess(T msg) { try { - apiUsageStatsClient.report(tenantId, ApiUsageRecordKey.MSG_COUNT, 1); - apiUsageStatsClient.report(tenantId, ApiUsageRecordKey.DP_TRANSPORT_COUNT, dataPoints); + apiUsageStatsClient.report(tenantId, ApiUsageRecordKey.TRANSPORT_MSG_COUNT, 1); + apiUsageStatsClient.report(tenantId, ApiUsageRecordKey.TRANSPORT_DP_COUNT, dataPoints); } finally { callback.onSuccess(msg); } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportTenantProfileCache.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportTenantProfileCache.java index 18c1512a27..6773d09417 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportTenantProfileCache.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportTenantProfileCache.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantProfileServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantProfileServiceImpl.java index 864679ce2a..af327d3505 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantProfileServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/tenant/TenantProfileServiceImpl.java @@ -29,6 +29,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.dao.entity.AbstractEntityService; import org.thingsboard.server.dao.exception.DataValidationException; @@ -145,7 +146,9 @@ public class TenantProfileServiceImpl extends AbstractEntityService implements T defaultTenantProfile = new TenantProfile(); defaultTenantProfile.setDefault(true); defaultTenantProfile.setName("Default"); - defaultTenantProfile.setProfileData(new TenantProfileData()); + TenantProfileData profileData = new TenantProfileData(); + profileData.setConfiguration(new DefaultTenantProfileConfiguration()); + defaultTenantProfile.setProfileData(profileData); defaultTenantProfile.setDescription("Default tenant profile"); defaultTenantProfile.setIsolatedTbCore(false); defaultTenantProfile.setIsolatedTbRuleEngine(false); diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java index 22830bed8c..f93fc2fb1c 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseTenantProfileServiceTest.java @@ -25,6 +25,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantProfileId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.dao.exception.DataValidationException; @@ -262,7 +263,9 @@ public class BaseTenantProfileServiceTest extends AbstractServiceTest { TenantProfile tenantProfile = new TenantProfile(); tenantProfile.setName(name); tenantProfile.setDescription(name + " Test"); - tenantProfile.setProfileData(new TenantProfileData()); + TenantProfileData profileData = new TenantProfileData(); + profileData.setConfiguration(new DefaultTenantProfileConfiguration()); + tenantProfile.setProfileData(profileData); tenantProfile.setDefault(false); tenantProfile.setIsolatedTbCore(false); tenantProfile.setIsolatedTbRuleEngine(false);