diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java index a67278cc6d..a4558eb0a4 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultMsgQueueService.java @@ -23,6 +23,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.transport.quota.tenant.TenantQuotaService; import org.thingsboard.server.dao.queue.MsgQueue; import javax.annotation.PostConstruct; @@ -48,6 +49,9 @@ public class DefaultMsgQueueService implements MsgQueueService { @Autowired private MsgQueue msgQueue; + @Autowired + private TenantQuotaService quotaService; + private ScheduledExecutorService cleanupExecutor; private Map pendingCountPerTenant = new ConcurrentHashMap<>(); @@ -70,6 +74,11 @@ public class DefaultMsgQueueService implements MsgQueueService { @Override public ListenableFuture put(TenantId tenantId, TbMsg msg, UUID nodeId, long clusterPartition) { + if(quotaService.isQuotaExceeded(tenantId.getId().toString())) { + log.warn("Tenant TbMsg Quota exceeded for [{}:{}] . Reject", tenantId.getId()); + return Futures.immediateFailedFuture(new RuntimeException("Tenant TbMsg Quota exceeded")); + } + AtomicLong pendingMsgCount = pendingCountPerTenant.computeIfAbsent(tenantId, key -> new AtomicLong()); if (pendingMsgCount.incrementAndGet() < queueMaxSize) { return msgQueue.put(tenantId, msg, nodeId, clusterPartition); diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index e479562df5..3d50258d39 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -131,9 +131,28 @@ quota: whitelist: "${QUOTA_HOST_WHITELIST:localhost,127.0.0.1}" # Array of blacklist hosts blacklist: "${QUOTA_HOST_BLACKLIST:}" - log: - topSize: 10 - intervalMin: 2 + log: + topSize: 10 + intervalMin: 2 + rule: + tenant: + # Max allowed number of API requests in interval for single tenant + limit: "${QUOTA_TENANT_LIMIT:100000}" + # Interval duration + intervalMs: "${QUOTA_TENANT_INTERVAL_MS:60000}" + # Maximum silence duration for tenant after which Tenant removed from QuotaService. Must be bigger than intervalMs + ttlMs: "${QUOTA_TENANT_TTL_MS:60000}" + # Interval for scheduled task that cleans expired records. TTL is used for expiring + cleanPeriodMs: "${QUOTA_TENANT_CLEAN_PERIOD_MS:300000}" + # Enable Host API Limits + enabled: "${QUOTA_TENANT_ENABLED:false}" + # Array of whitelist tenants + whitelist: "${QUOTA_TENANT_WHITELIST:}" + # Array of blacklist tenants + blacklist: "${QUOTA_HOST_BLACKLIST:}" + log: + topSize: 10 + intervalMin: 2 database: type: "${DATABASE_TYPE:sql}" # cassandra OR sql diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaService.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/AbstractQuotaService.java similarity index 73% rename from common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaService.java rename to common/transport/src/main/java/org/thingsboard/server/common/transport/quota/AbstractQuotaService.java index c1f045df8f..9d7bd1720c 100644 --- a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaService.java +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/AbstractQuotaService.java @@ -15,33 +15,24 @@ */ package org.thingsboard.server.common.transport.quota; -import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Service; -import org.thingsboard.server.common.transport.quota.inmemory.HostRequestIntervalRegistry; import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryCleaner; import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryLogger; +import org.thingsboard.server.common.transport.quota.inmemory.KeyBasedIntervalRegistry; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; -/** - * @author Vitaliy Paromskiy - * @version 1.0 - */ -@Service -@Slf4j -public class HostRequestsQuotaService implements QuotaService { +public class AbstractQuotaService implements QuotaService { - private final HostRequestIntervalRegistry requestRegistry; - private final HostRequestLimitPolicy requestsPolicy; + private final KeyBasedIntervalRegistry requestRegistry; + private final RequestLimitPolicy requestsPolicy; private final IntervalRegistryCleaner registryCleaner; private final IntervalRegistryLogger registryLogger; private final boolean enabled; - public HostRequestsQuotaService(HostRequestIntervalRegistry requestRegistry, HostRequestLimitPolicy requestsPolicy, + public AbstractQuotaService(KeyBasedIntervalRegistry requestRegistry, RequestLimitPolicy requestsPolicy, IntervalRegistryCleaner registryCleaner, IntervalRegistryLogger registryLogger, - @Value("${quota.host.enabled}") boolean enabled) { + boolean enabled) { this.requestRegistry = requestRegistry; this.requestsPolicy = requestsPolicy; this.registryCleaner = registryCleaner; diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/RequestLimitPolicy.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/RequestLimitPolicy.java new file mode 100644 index 0000000000..0ff1230b1c --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/RequestLimitPolicy.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2018 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.quota; + + +public abstract class RequestLimitPolicy { + + private final long limit; + + public RequestLimitPolicy(long limit) { + this.limit = limit; + } + + public boolean isValid(long currentValue) { + return currentValue <= limit; + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryCleaner.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryCleaner.java new file mode 100644 index 0000000000..ea75f5dc09 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryCleaner.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2018 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.quota.host; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryCleaner; + +@Component +public class HostIntervalRegistryCleaner extends IntervalRegistryCleaner { + + public HostIntervalRegistryCleaner(HostRequestIntervalRegistry intervalRegistry, + @Value("${quota.host.cleanPeriodMs}") long cleanPeriodMs) { + super(intervalRegistry, cleanPeriodMs); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryLogger.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryLogger.java new file mode 100644 index 0000000000..65767f1648 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostIntervalRegistryLogger.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2018 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.quota.host; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryLogger; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +@Component +@Slf4j +public class HostIntervalRegistryLogger extends IntervalRegistryLogger { + + private final long logIntervalMin; + + public HostIntervalRegistryLogger(@Value("${quota.host.log.topSize}") int topSize, + @Value("${quota.host.log.intervalMin}") long logIntervalMin, + HostRequestIntervalRegistry intervalRegistry) { + super(topSize, logIntervalMin, intervalRegistry); + this.logIntervalMin = logIntervalMin; + } + + protected void log(Map top, int uniqHosts, long requestsCount) { + long rps = requestsCount / TimeUnit.MINUTES.toSeconds(logIntervalMin); + StringBuilder builder = new StringBuilder("Quota Statistic : "); + builder.append("uniqHosts : ").append(uniqHosts).append("; "); + builder.append("requestsCount : ").append(requestsCount).append("; "); + builder.append("RPS : ").append(rps).append(" "); + builder.append("top -> "); + for (Map.Entry host : top.entrySet()) { + builder.append(host.getKey()).append(" : ").append(host.getValue()).append("; "); + } + + log.info(builder.toString()); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestIntervalRegistry.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestIntervalRegistry.java new file mode 100644 index 0000000000..9b3b4614e4 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestIntervalRegistry.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2018 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.quota.host; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.KeyBasedIntervalRegistry; + +/** + * @author Vitaliy Paromskiy + * @version 1.0 + */ +@Component +@Slf4j +public class HostRequestIntervalRegistry extends KeyBasedIntervalRegistry { + + public HostRequestIntervalRegistry(@Value("${quota.host.intervalMs}") long intervalDurationMs, + @Value("${quota.host.ttlMs}") long ttlMs, + @Value("${quota.host.whitelist}") String whiteList, + @Value("${quota.host.blacklist}") String blackList) { + super(intervalDurationMs, ttlMs, whiteList, blackList, "host"); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicy.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestLimitPolicy.java similarity index 78% rename from common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicy.java rename to common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestLimitPolicy.java index cf1c4e85ae..eeef9247e6 100644 --- a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicy.java +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestLimitPolicy.java @@ -13,26 +13,21 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.common.transport.quota; +package org.thingsboard.server.common.transport.quota.host; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.RequestLimitPolicy; /** * @author Vitaliy Paromskiy * @version 1.0 */ @Component -public class HostRequestLimitPolicy { - - private final long limit; +public class HostRequestLimitPolicy extends RequestLimitPolicy { public HostRequestLimitPolicy(@Value("${quota.host.limit}") long limit) { - this.limit = limit; - } - - public boolean isValid(long currentValue) { - return currentValue <= limit; + super(limit); } } diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestsQuotaService.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestsQuotaService.java new file mode 100644 index 0000000000..69342b538b --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/host/HostRequestsQuotaService.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2018 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.quota.host; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.transport.quota.AbstractQuotaService; + +/** + * @author Vitaliy Paromskiy + * @version 1.0 + */ +@Service +@Slf4j +public class HostRequestsQuotaService extends AbstractQuotaService { + + public HostRequestsQuotaService(HostRequestIntervalRegistry requestRegistry, HostRequestLimitPolicy requestsPolicy, + HostIntervalRegistryCleaner registryCleaner, HostIntervalRegistryLogger registryLogger, + @Value("${quota.host.enabled}") boolean enabled) { + super(requestRegistry, requestsPolicy, registryCleaner, registryLogger, enabled); + } + +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryCleaner.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryCleaner.java index a227d2abba..0c510ff815 100644 --- a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryCleaner.java +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryCleaner.java @@ -16,10 +16,7 @@ package org.thingsboard.server.common.transport.quota.inmemory; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Component; -import javax.annotation.PreDestroy; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -28,15 +25,14 @@ import java.util.concurrent.TimeUnit; * @author Vitaliy Paromskiy * @version 1.0 */ -@Component @Slf4j -public class IntervalRegistryCleaner { +public abstract class IntervalRegistryCleaner { - private final HostRequestIntervalRegistry intervalRegistry; + private final KeyBasedIntervalRegistry intervalRegistry; private final long cleanPeriodMs; private ScheduledExecutorService executor; - public IntervalRegistryCleaner(HostRequestIntervalRegistry intervalRegistry, @Value("${quota.host.cleanPeriodMs}") long cleanPeriodMs) { + public IntervalRegistryCleaner(KeyBasedIntervalRegistry intervalRegistry, long cleanPeriodMs) { this.intervalRegistry = intervalRegistry; this.cleanPeriodMs = cleanPeriodMs; } diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLogger.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLogger.java index 8b34a6becc..30399a140f 100644 --- a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLogger.java +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLogger.java @@ -17,8 +17,6 @@ package org.thingsboard.server.common.transport.quota.inmemory; import com.google.common.collect.MinMaxPriorityQueue; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Component; import java.util.Comparator; import java.util.Map; @@ -32,17 +30,15 @@ import java.util.stream.Collectors; * @author Vitaliy Paromskiy * @version 1.0 */ -@Component @Slf4j -public class IntervalRegistryLogger { +public abstract class IntervalRegistryLogger { private final int topSize; - private final HostRequestIntervalRegistry intervalRegistry; + private final KeyBasedIntervalRegistry intervalRegistry; private final long logIntervalMin; private ScheduledExecutorService executor; - public IntervalRegistryLogger(@Value("${quota.log.topSize}") int topSize, @Value("${quota.log.intervalMin}") long logIntervalMin, - HostRequestIntervalRegistry intervalRegistry) { + public IntervalRegistryLogger(int topSize, long logIntervalMin, KeyBasedIntervalRegistry intervalRegistry) { this.topSize = topSize; this.logIntervalMin = logIntervalMin; this.intervalRegistry = intervalRegistry; @@ -79,17 +75,5 @@ public class IntervalRegistryLogger { return topQueue.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); } - private void log(Map top, int uniqHosts, long requestsCount) { - long rps = requestsCount / TimeUnit.MINUTES.toSeconds(logIntervalMin); - StringBuilder builder = new StringBuilder("Quota Statistic : "); - builder.append("uniqHosts : ").append(uniqHosts).append("; "); - builder.append("requestsCount : ").append(requestsCount).append("; "); - builder.append("RPS : ").append(rps).append(" "); - builder.append("top -> "); - for (Map.Entry host : top.entrySet()) { - builder.append(host.getKey()).append(" : ").append(host.getValue()).append("; "); - } - - log.info(builder.toString()); - } + protected abstract void log(Map top, int uniqHosts, long requestsCount); } diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistry.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/KeyBasedIntervalRegistry.java similarity index 71% rename from common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistry.java rename to common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/KeyBasedIntervalRegistry.java index 3782ed22ed..0b0fef85e5 100644 --- a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistry.java +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/inmemory/KeyBasedIntervalRegistry.java @@ -18,22 +18,14 @@ package org.thingsboard.server.common.transport.quota.inmemory; import com.google.common.collect.Sets; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.stereotype.Component; -import javax.annotation.PostConstruct; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; -/** - * @author Vitaliy Paromskiy - * @version 1.0 - */ -@Component @Slf4j -public class HostRequestIntervalRegistry { +public abstract class KeyBasedIntervalRegistry { private final Map hostCounts = new ConcurrentHashMap<>(); private final long intervalDurationMs; @@ -41,23 +33,20 @@ public class HostRequestIntervalRegistry { private final Set whiteList; private final Set blackList; - public HostRequestIntervalRegistry(@Value("${quota.host.intervalMs}") long intervalDurationMs, - @Value("${quota.host.ttlMs}") long ttlMs, - @Value("${quota.host.whitelist}") String whiteList, - @Value("${quota.host.blacklist}") String blackList) { + public KeyBasedIntervalRegistry(long intervalDurationMs, long ttlMs, String whiteList, String blackList, String name) { this.intervalDurationMs = intervalDurationMs; this.ttlMs = ttlMs; this.whiteList = Sets.newHashSet(StringUtils.split(whiteList, ',')); this.blackList = Sets.newHashSet(StringUtils.split(blackList, ',')); + validate(name); } - @PostConstruct - public void init() { + private void validate(String name) { if (ttlMs < intervalDurationMs) { - log.warn("TTL for IntervalRegistry [{}] smaller than interval duration [{}]", ttlMs, intervalDurationMs); + log.warn("TTL for {} IntervalRegistry [{}] smaller than interval duration [{}]", name, ttlMs, intervalDurationMs); } - log.info("Start Host Quota Service with whitelist {}", whiteList); - log.info("Start Host Quota Service with blacklist {}", blackList); + log.info("Start {} KeyBasedIntervalRegistry with whitelist {}", name, whiteList); + log.info("Start {} KeyBasedIntervalRegistry with blacklist {}", name, blackList); } public long tick(String clientHostId) { diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryCleaner.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryCleaner.java new file mode 100644 index 0000000000..c48117069a --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryCleaner.java @@ -0,0 +1,29 @@ +/** + * Copyright © 2016-2018 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.quota.tenant; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryCleaner; + +@Component +public class TenantIntervalRegistryCleaner extends IntervalRegistryCleaner { + + public TenantIntervalRegistryCleaner(TenantMsgsIntervalRegistry intervalRegistry, + @Value("${quota.rule.tenant.cleanPeriodMs}") long cleanPeriodMs) { + super(intervalRegistry, cleanPeriodMs); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryLogger.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryLogger.java new file mode 100644 index 0000000000..c56f457423 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantIntervalRegistryLogger.java @@ -0,0 +1,52 @@ +/** + * Copyright © 2016-2018 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.quota.tenant; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryLogger; + +import java.util.Map; +import java.util.concurrent.TimeUnit; + +@Slf4j +@Component +public class TenantIntervalRegistryLogger extends IntervalRegistryLogger { + + private final long logIntervalMin; + + public TenantIntervalRegistryLogger(@Value("${quota.rule.tenant.log.topSize}") int topSize, + @Value("${quota.rule.tenant.log.intervalMin}") long logIntervalMin, + TenantMsgsIntervalRegistry intervalRegistry) { + super(topSize, logIntervalMin, intervalRegistry); + this.logIntervalMin = logIntervalMin; + } + + protected void log(Map top, int uniqHosts, long requestsCount) { + long rps = requestsCount / TimeUnit.MINUTES.toSeconds(logIntervalMin); + StringBuilder builder = new StringBuilder("Tenant Quota Statistic : "); + builder.append("uniqTenants : ").append(uniqHosts).append("; "); + builder.append("requestsCount : ").append(requestsCount).append("; "); + builder.append("RPS : ").append(rps).append(" "); + builder.append("top -> "); + for (Map.Entry host : top.entrySet()) { + builder.append(host.getKey()).append(" : ").append(host.getValue()).append("; "); + } + + log.info(builder.toString()); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantMsgsIntervalRegistry.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantMsgsIntervalRegistry.java new file mode 100644 index 0000000000..6e8402c701 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantMsgsIntervalRegistry.java @@ -0,0 +1,31 @@ +/** + * Copyright © 2016-2018 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.quota.tenant; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.inmemory.KeyBasedIntervalRegistry; + +@Component +public class TenantMsgsIntervalRegistry extends KeyBasedIntervalRegistry { + + public TenantMsgsIntervalRegistry(@Value("${quota.rule.tenant.intervalMs}") long intervalDurationMs, + @Value("${quota.rule.tenant.ttlMs}") long ttlMs, + @Value("${quota.rule.tenant.whitelist}") String whiteList, + @Value("${quota.rule.tenant.blacklist}") String blackList) { + super(intervalDurationMs, ttlMs, whiteList, blackList, "Rule Tenant"); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantQuotaService.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantQuotaService.java new file mode 100644 index 0000000000..a68860a2e0 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantQuotaService.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2018 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.quota.tenant; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.AbstractQuotaService; + +@Component +public class TenantQuotaService extends AbstractQuotaService { + + public TenantQuotaService(TenantMsgsIntervalRegistry requestRegistry, TenantRequestLimitPolicy requestsPolicy, + TenantIntervalRegistryCleaner registryCleaner, TenantIntervalRegistryLogger registryLogger, + @Value("${quota.rule.tenant.enabled}") boolean enabled) { + super(requestRegistry, requestsPolicy, registryCleaner, registryLogger, enabled); + } +} diff --git a/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantRequestLimitPolicy.java b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantRequestLimitPolicy.java new file mode 100644 index 0000000000..cc32c81ad7 --- /dev/null +++ b/common/transport/src/main/java/org/thingsboard/server/common/transport/quota/tenant/TenantRequestLimitPolicy.java @@ -0,0 +1,28 @@ +/** + * Copyright © 2016-2018 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.quota.tenant; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.thingsboard.server.common.transport.quota.RequestLimitPolicy; + +@Component +public class TenantRequestLimitPolicy extends RequestLimitPolicy { + + public TenantRequestLimitPolicy(@Value("${quota.rule.tenant.limit}") long limit) { + super(limit); + } +} diff --git a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicyTest.java b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicyTest.java index 174d18235d..07e03ef0ac 100644 --- a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicyTest.java +++ b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestLimitPolicyTest.java @@ -16,6 +16,7 @@ package org.thingsboard.server.common.transport.quota; import org.junit.Test; +import org.thingsboard.server.common.transport.quota.host.HostRequestLimitPolicy; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; diff --git a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaServiceTest.java b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaServiceTest.java index 547f0cf0e4..20f8a55d0d 100644 --- a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaServiceTest.java +++ b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/HostRequestsQuotaServiceTest.java @@ -17,9 +17,7 @@ package org.thingsboard.server.common.transport.quota; import org.junit.Before; import org.junit.Test; -import org.thingsboard.server.common.transport.quota.inmemory.HostRequestIntervalRegistry; -import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryCleaner; -import org.thingsboard.server.common.transport.quota.inmemory.IntervalRegistryLogger; +import org.thingsboard.server.common.transport.quota.host.*; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -35,8 +33,8 @@ public class HostRequestsQuotaServiceTest { private HostRequestIntervalRegistry requestRegistry = mock(HostRequestIntervalRegistry.class); private HostRequestLimitPolicy requestsPolicy = mock(HostRequestLimitPolicy.class); - private IntervalRegistryCleaner registryCleaner = mock(IntervalRegistryCleaner.class); - private IntervalRegistryLogger registryLogger = mock(IntervalRegistryLogger.class); + private HostIntervalRegistryCleaner registryCleaner = mock(HostIntervalRegistryCleaner.class); + private HostIntervalRegistryLogger registryLogger = mock(HostIntervalRegistryLogger.class); @Before public void init() { diff --git a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistryTest.java b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistryTest.java index 78b82eec08..b49dd00bc9 100644 --- a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistryTest.java +++ b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/HostRequestIntervalRegistryTest.java @@ -15,11 +15,9 @@ */ package org.thingsboard.server.common.transport.quota.inmemory; -import com.google.common.collect.Sets; import org.junit.Before; import org.junit.Test; - -import java.util.Collections; +import org.thingsboard.server.common.transport.quota.host.HostRequestIntervalRegistry; import static org.junit.Assert.assertEquals; diff --git a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLoggerTest.java b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLoggerTest.java index c9139aeeda..6e51420cd5 100644 --- a/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLoggerTest.java +++ b/common/transport/src/test/java/org/thingsboard/server/common/transport/quota/inmemory/IntervalRegistryLoggerTest.java @@ -18,6 +18,8 @@ package org.thingsboard.server.common.transport.quota.inmemory; import com.google.common.collect.ImmutableMap; import org.junit.Before; import org.junit.Test; +import org.thingsboard.server.common.transport.quota.host.HostIntervalRegistryLogger; +import org.thingsboard.server.common.transport.quota.host.HostRequestIntervalRegistry; import java.util.Collections; import java.util.Map; @@ -37,7 +39,7 @@ public class IntervalRegistryLoggerTest { @Before public void init() { - logger = new IntervalRegistryLogger(3, 10, requestRegistry); + logger = new HostIntervalRegistryLogger(3, 10, requestRegistry); } @Test diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/MsgAck.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java similarity index 94% rename from dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/MsgAck.java rename to dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java index 1b1cd3f467..a1b039a1a7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/MsgAck.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/MsgAck.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.dao.queue.db.nosql; +package org.thingsboard.server.dao.queue.db; import lombok.Data; import lombok.EqualsAndHashCode; diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilter.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java similarity index 92% rename from dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilter.java rename to dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java index c912e8e114..66eaa6d46b 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilter.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/UnprocessedMsgFilter.java @@ -13,10 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.dao.queue.db.nosql; +package org.thingsboard.server.dao.queue.db; import org.springframework.stereotype.Component; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.queue.db.MsgAck; import java.util.Collection; import java.util.List; diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java index ceaab586cf..ce481b78b8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/CassandraMsgQueue.java @@ -26,6 +26,8 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.dao.queue.MsgQueue; +import org.thingsboard.server.dao.queue.db.MsgAck; +import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter; import org.thingsboard.server.dao.queue.db.repository.AckRepository; import org.thingsboard.server.dao.queue.db.repository.MsgRepository; import org.thingsboard.server.dao.util.NoSqlDao; diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java index 31ab074365..6c59c5997c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepository.java @@ -26,7 +26,7 @@ import com.google.common.util.concurrent.ListenableFuture; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.thingsboard.server.dao.nosql.CassandraAbstractDao; -import org.thingsboard.server.dao.queue.db.nosql.MsgAck; +import org.thingsboard.server.dao.queue.db.MsgAck; import org.thingsboard.server.dao.queue.db.repository.AckRepository; import org.thingsboard.server.dao.util.NoSqlDao; diff --git a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java b/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java index 458dba81cb..6fbd2da57e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/queue/db/repository/AckRepository.java @@ -16,7 +16,7 @@ package org.thingsboard.server.dao.queue.db.repository; import com.google.common.util.concurrent.ListenableFuture; -import org.thingsboard.server.dao.queue.db.nosql.MsgAck; +import org.thingsboard.server.dao.queue.db.MsgAck; import java.util.List; import java.util.UUID; diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index 7b2b391b10..ea29f6fa77 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -16,6 +16,7 @@ package org.thingsboard.server.dao.relation; import com.google.common.base.Function; +import com.google.common.util.concurrent.AsyncFunction; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; @@ -176,97 +177,65 @@ public class BaseRelationService implements RelationService { } @Override - public boolean deleteEntityRelations(EntityId entity) { - Cache cache = cacheManager.getCache(RELATIONS_CACHE); - log.trace("Executing deleteEntityRelations [{}]", entity); - validate(entity); - List>> inboundRelationsList = new ArrayList<>(); - for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { - inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup)); - } - ListenableFuture>> inboundRelations = Futures.allAsList(inboundRelationsList); - ListenableFuture> inboundDeletions = Futures.transform(inboundRelations, relations -> - getBooleans(relations, cache, true)); - - ListenableFuture inboundFuture = Futures.transform(inboundDeletions, getListToBooleanFunction()); - boolean inboundDeleteResult = false; + public void deleteEntityRelations(EntityId entityId) { try { - inboundDeleteResult = inboundFuture.get(); + deleteEntityRelationsAsync(entityId).get(); } catch (InterruptedException | ExecutionException e) { - log.error("Error deleting entity inbound relations", e); - } - - List>> outboundRelationsList = new ArrayList<>(); - for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { - outboundRelationsList.add(relationDao.findAllByFrom(entity, typeGroup)); - } - ListenableFuture>> outboundRelations = Futures.allAsList(outboundRelationsList); - Futures.transform(outboundRelations, relations -> getBooleans(relations, cache, false)); - - boolean outboundDeleteResult = relationDao.deleteOutboundRelations(entity); - return inboundDeleteResult && outboundDeleteResult; - } - - private List getBooleans(List> relations, Cache cache, boolean isRemove) { - List results = new ArrayList<>(); - for (List relationList : relations) { - relationList.forEach(relation -> checkFromDeleteSync(cache, results, relation, isRemove)); + throw new RuntimeException(e); } - return results; - } - - private void checkFromDeleteSync(Cache cache, List results, EntityRelation relation, boolean isRemove) { - if (isRemove) { - results.add(relationDao.deleteRelation(relation)); - } - cacheEviction(relation, cache); } @Override - public ListenableFuture deleteEntityRelationsAsync(EntityId entity) { + public ListenableFuture deleteEntityRelationsAsync(EntityId entityId) { Cache cache = cacheManager.getCache(RELATIONS_CACHE); - log.trace("Executing deleteEntityRelationsAsync [{}]", entity); - validate(entity); + log.trace("Executing deleteEntityRelationsAsync [{}]", entityId); + validate(entityId); List>> inboundRelationsList = new ArrayList<>(); for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { - inboundRelationsList.add(relationDao.findAllByTo(entity, typeGroup)); + inboundRelationsList.add(relationDao.findAllByTo(entityId, typeGroup)); } + ListenableFuture>> inboundRelations = Futures.allAsList(inboundRelationsList); + + List>> outboundRelationsList = new ArrayList<>(); + for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { + outboundRelationsList.add(relationDao.findAllByFrom(entityId, typeGroup)); + } + + ListenableFuture>> outboundRelations = Futures.allAsList(outboundRelationsList); + ListenableFuture> inboundDeletions = Futures.transformAsync(inboundRelations, relations -> { - List> results = getListenableFutures(relations, cache, true); + List> results = deleteRelationGroupsAsync(relations, cache, true); return Futures.allAsList(results); }); - ListenableFuture inboundFuture = Futures.transform(inboundDeletions, getListToBooleanFunction()); + ListenableFuture> outboundDeletions = Futures.transformAsync(outboundRelations, + relations -> { + List> results = deleteRelationGroupsAsync(relations, cache, false); + return Futures.allAsList(results); + }); - List>> outboundRelationsList = new ArrayList<>(); - for (RelationTypeGroup typeGroup : RelationTypeGroup.values()) { - outboundRelationsList.add(relationDao.findAllByFrom(entity, typeGroup)); - } - ListenableFuture>> outboundRelations = Futures.allAsList(outboundRelationsList); - Futures.transformAsync(outboundRelations, relations -> { - List> results = getListenableFutures(relations, cache, false); - return Futures.allAsList(results); - }); + ListenableFuture>> deletionsFuture = Futures.allAsList(inboundDeletions, outboundDeletions); - ListenableFuture outboundFuture = relationDao.deleteOutboundRelationsAsync(entity); - return Futures.transform(Futures.allAsList(Arrays.asList(inboundFuture, outboundFuture)), getListToBooleanFunction()); + return Futures.transform(Futures.transformAsync(deletionsFuture, (deletions) -> relationDao.deleteOutboundRelationsAsync(entityId)), result -> null); } - private List> getListenableFutures(List> relations, Cache cache, boolean isRemove) { + private List> deleteRelationGroupsAsync(List> relations, Cache cache, boolean deleteFromDb) { List> results = new ArrayList<>(); for (List relationList : relations) { - relationList.forEach(relation -> checkFromDeleteAsync(cache, results, relation, isRemove)); + relationList.forEach(relation -> results.add(deleteAsync(cache, relation, deleteFromDb))); } return results; } - private void checkFromDeleteAsync(Cache cache, List> results, EntityRelation relation, boolean isRemove) { - if (isRemove) { - results.add(relationDao.deleteRelationAsync(relation)); - } + private ListenableFuture deleteAsync(Cache cache, EntityRelation relation, boolean deleteFromDb) { cacheEviction(relation, cache); + if (deleteFromDb) { + return relationDao.deleteRelationAsync(relation); + } else { + return Futures.immediateFuture(false); + } } private void cacheEviction(EntityRelation relation, Cache cache) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index ca1b959c95..da58b11a25 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -23,6 +23,7 @@ import org.thingsboard.server.common.data.relation.EntityRelationsQuery; import org.thingsboard.server.common.data.relation.RelationTypeGroup; import java.util.List; +import java.util.concurrent.ExecutionException; /** * Created by ashvayka on 27.04.17. @@ -47,9 +48,9 @@ public interface RelationService { ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup); - boolean deleteEntityRelations(EntityId entity); + void deleteEntityRelations(EntityId entity); - ListenableFuture deleteEntityRelationsAsync(EntityId entity); + ListenableFuture deleteEntityRelationsAsync(EntityId entity); List findByFrom(EntityId from, RelationTypeGroup typeGroup); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java index d57aa4598d..a8de6775e6 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java @@ -127,39 +127,35 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple @Override public boolean deleteRelation(EntityRelation relation) { RelationCompositeKey key = new RelationCompositeKey(relation); - boolean relationExistsBeforeDelete = relationRepository.exists(key); - relationRepository.delete(key); - return relationExistsBeforeDelete; + return deleteRelationIfExists(key); } @Override public ListenableFuture deleteRelationAsync(EntityRelation relation) { RelationCompositeKey key = new RelationCompositeKey(relation); return service.submit( - () -> { - boolean relationExistsBeforeDelete = relationRepository.exists(key); - relationRepository.delete(key); - return relationExistsBeforeDelete; - }); + () -> deleteRelationIfExists(key)); } @Override public boolean deleteRelation(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { RelationCompositeKey key = getRelationCompositeKey(from, to, relationType, typeGroup); - boolean relationExistsBeforeDelete = relationRepository.exists(key); - relationRepository.delete(key); - return relationExistsBeforeDelete; + return deleteRelationIfExists(key); } @Override public ListenableFuture deleteRelationAsync(EntityId from, EntityId to, String relationType, RelationTypeGroup typeGroup) { RelationCompositeKey key = getRelationCompositeKey(from, to, relationType, typeGroup); return service.submit( - () -> { - boolean relationExistsBeforeDelete = relationRepository.exists(key); - relationRepository.delete(key); - return relationExistsBeforeDelete; - }); + () -> deleteRelationIfExists(key)); + } + + private boolean deleteRelationIfExists(RelationCompositeKey key) { + boolean relationExistsBeforeDelete = relationRepository.exists(key); + if (relationExistsBeforeDelete) { + relationRepository.delete(key); + } + return relationExistsBeforeDelete; } @Override @@ -167,7 +163,9 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple boolean relationExistsBeforeDelete = relationRepository .findAllByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()) .size() > 0; - relationRepository.deleteByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()); + if (relationExistsBeforeDelete) { + relationRepository.deleteByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()); + } return relationExistsBeforeDelete; } @@ -178,7 +176,9 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple boolean relationExistsBeforeDelete = relationRepository .findAllByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()) .size() > 0; - relationRepository.deleteByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()); + if (relationExistsBeforeDelete) { + relationRepository.deleteByFromIdAndFromType(UUIDConverter.fromTimeUUID(entity.getId()), entity.getEntityType().name()); + } return relationExistsBeforeDelete; }); } diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java index 43c6e727ce..fd9bf21164 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/UnprocessedMsgFilterTest.java @@ -18,6 +18,8 @@ package org.thingsboard.server.dao.queue.db.nosql; import com.google.common.collect.Lists; import org.junit.Test; import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.dao.queue.db.MsgAck; +import org.thingsboard.server.dao.queue.db.UnprocessedMsgFilter; import java.util.Collection; import java.util.List; diff --git a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java index 9b743c90ae..b2f38dc539 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/queue/db/nosql/repository/CassandraAckRepositoryTest.java @@ -21,9 +21,9 @@ import com.google.common.util.concurrent.ListenableFuture; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.test.util.ReflectionTestUtils; -import org.thingsboard.server.dao.queue.db.nosql.MsgAck; import org.thingsboard.server.dao.service.AbstractServiceTest; import org.thingsboard.server.dao.service.DaoNoSqlTest; +import org.thingsboard.server.dao.queue.db.MsgAck; import java.util.List; import java.util.UUID; diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java index a774bfd118..743aefcc40 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/BaseRelationServiceTest.java @@ -96,7 +96,7 @@ public abstract class BaseRelationServiceTest extends AbstractServiceTest { saveRelation(relationA); saveRelation(relationB); - Assert.assertTrue(relationService.deleteEntityRelationsAsync(childId).get()); + Assert.assertNull(relationService.deleteEntityRelationsAsync(childId).get()); Assert.assertFalse(relationService.checkRelation(parentId, childId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.COMMON).get()); diff --git a/dao/src/test/resources/application-test.properties b/dao/src/test/resources/application-test.properties index dbd8b84a4b..f2dab45491 100644 --- a/dao/src/test/resources/application-test.properties +++ b/dao/src/test/resources/application-test.properties @@ -30,4 +30,3 @@ redis.connection.db=0 redis.connection.password= rule.queue.type=memory -rule.queue.max_size=10000 \ No newline at end of file diff --git a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportService.java b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportService.java index 15706d4d6f..6c8437cb03 100644 --- a/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportService.java +++ b/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportService.java @@ -27,6 +27,7 @@ import org.springframework.stereotype.Service; import org.thingsboard.server.common.transport.SessionMsgProcessor; import org.thingsboard.server.common.transport.auth.DeviceAuthService; import org.thingsboard.server.common.transport.quota.QuotaService; +import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaService; import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; import javax.annotation.PostConstruct; @@ -55,7 +56,7 @@ public class CoapTransportService { private DeviceAuthService authService; @Autowired(required = false) - private QuotaService quotaService; + private HostRequestsQuotaService quotaService; @Value("${coap.bind_address}") diff --git a/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapServerTest.java b/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapServerTest.java index 320f06e7f1..48150567a0 100644 --- a/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapServerTest.java +++ b/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapServerTest.java @@ -50,7 +50,7 @@ import org.thingsboard.server.common.msg.session.*; import org.thingsboard.server.common.transport.SessionMsgProcessor; import org.thingsboard.server.common.transport.auth.DeviceAuthResult; import org.thingsboard.server.common.transport.auth.DeviceAuthService; -import org.thingsboard.server.common.transport.quota.QuotaService; +import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaService; import java.util.ArrayList; import java.util.List; @@ -134,8 +134,8 @@ public class CoapServerTest { } @Bean - public static QuotaService quotaService() { - return key -> false; + public static HostRequestsQuotaService quotaService() { + return new HostRequestsQuotaService(null, null, null, null, false); } } diff --git a/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index 930b442447..d26d076b21 100644 --- a/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -36,6 +36,7 @@ import org.thingsboard.server.common.transport.SessionMsgProcessor; import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.common.transport.auth.DeviceAuthService; import org.thingsboard.server.common.transport.quota.QuotaService; +import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaService; import org.thingsboard.server.transport.http.session.HttpSessionCtx; import javax.servlet.http.HttpServletRequest; @@ -61,7 +62,7 @@ public class DeviceApiController { private DeviceAuthService authService; @Autowired(required = false) - private QuotaService quotaService; + private HostRequestsQuotaService quotaService; @RequestMapping(value = "/{deviceToken}/attributes", method = RequestMethod.GET, produces = "application/json") public DeferredResult getDeviceAttributes(@PathVariable("deviceToken") String deviceToken, diff --git a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java index f0129e1b6c..1b37ed4513 100644 --- a/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java +++ b/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportService.java @@ -29,7 +29,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.stereotype.Service; import org.thingsboard.server.common.transport.SessionMsgProcessor; import org.thingsboard.server.common.transport.auth.DeviceAuthService; -import org.thingsboard.server.common.transport.quota.QuotaService; +import org.thingsboard.server.common.transport.quota.host.HostRequestsQuotaService; import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.transport.mqtt.adaptors.MqttTransportAdaptor; @@ -67,7 +67,7 @@ public class MqttTransportService { private MqttSslHandlerProvider sslHandlerProvider; @Autowired(required = false) - private QuotaService quotaService; + private HostRequestsQuotaService quotaService; @Value("${mqtt.bind_address}") private String host;