From 6db3171ffb499323accd772b42e82ba76446149f Mon Sep 17 00:00:00 2001 From: ViacheslavKlimov Date: Tue, 25 Jul 2023 15:27:54 +0300 Subject: [PATCH] Refactoring --- .../server/controller/QueueController.java | 1 - .../DefaultTbRuleEngineConsumerService.java | 2 +- .../queue/TbRuleEngineConsumerStats.java | 13 +- .../server/controller/AbstractWebTest.java | 1 + .../controller/BaseTenantControllerTest.java | 170 ++++++++++++- .../BaseTenantProfileControllerTest.java | 17 -- .../notification/NotificationRuleApiTest.java | 4 +- .../service/queue/QueueServiceTest.java | 225 ------------------ .../queue/discovery/HashPartitionService.java | 2 +- .../profile/tenant-profile.component.ts | 12 +- 10 files changed, 181 insertions(+), 266 deletions(-) delete mode 100644 application/src/test/java/org/thingsboard/server/service/queue/QueueServiceTest.java diff --git a/application/src/main/java/org/thingsboard/server/controller/QueueController.java b/application/src/main/java/org/thingsboard/server/controller/QueueController.java index 0e79ae7955..faeea1a4aa 100644 --- a/application/src/main/java/org/thingsboard/server/controller/QueueController.java +++ b/application/src/main/java/org/thingsboard/server/controller/QueueController.java @@ -126,7 +126,6 @@ public class QueueController extends BaseController { @PreAuthorize("hasAnyAuthority('SYS_ADMIN')") @RequestMapping(value = "/queues", params = {"serviceType"}, method = RequestMethod.POST) @ResponseBody - public Queue saveQueue(@ApiParam(value = "A JSON value representing the queue.") @RequestBody Queue queue, @ApiParam(value = QUEUE_SERVICE_TYPE_DESCRIPTION, allowableValues = QUEUE_SERVICE_TYPE_ALLOWABLE_VALUES, required = true) diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 3d02a1f60e..598b41035f 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -152,7 +152,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< private void initConsumer(Queue configuration) { QueueKey queueKey = new QueueKey(ServiceType.TB_RULE_ENGINE, configuration); consumerConfigurations.putIfAbsent(queueKey, configuration); - consumerStats.putIfAbsent(queueKey, new TbRuleEngineConsumerStats(configuration.getName(), statsFactory)); + consumerStats.putIfAbsent(queueKey, new TbRuleEngineConsumerStats(configuration, statsFactory)); if (!configuration.isConsumerPerPartition()) { consumers.computeIfAbsent(queueKey, queueName -> tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration)); } else { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java index 77e805eb62..2904c299ce 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java @@ -18,6 +18,7 @@ package org.thingsboard.server.service.queue; import io.micrometer.core.instrument.Timer; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.queue.Queue; import org.thingsboard.server.common.msg.queue.RuleEngineException; import org.thingsboard.server.common.stats.StatsCounter; import org.thingsboard.server.common.stats.StatsFactory; @@ -63,9 +64,11 @@ public class TbRuleEngineConsumerStats { private final ConcurrentMap tenantExceptions = new ConcurrentHashMap<>(); private final String queueName; + private final TenantId tenantId; - public TbRuleEngineConsumerStats(String queueName, StatsFactory statsFactory) { - this.queueName = queueName; + public TbRuleEngineConsumerStats(Queue queue, StatsFactory statsFactory) { + this.queueName = queue.getName(); + this.tenantId = queue.getTenantId(); this.statsFactory = statsFactory; String statsKey = StatsType.RULE_ENGINE.getName() + "." + queueName; @@ -156,7 +159,11 @@ public class TbRuleEngineConsumerStats { counters.forEach(counter -> { stats.append(counter.getName()).append(" = [").append(counter.get()).append("] "); }); - log.info("[{}] Stats: {}", queueName, stats); + if (tenantId.isSysTenantId()) { + log.info("[{}] Stats: {}", queueName, stats); + } else { + log.info("[{}][{}] Stats: {}", queueName, tenantId, stats); + } } } diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java index 449164776a..d3a77e28d1 100644 --- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java @@ -57,6 +57,7 @@ import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilde import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.context.WebApplicationContext; +import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.rule.engine.api.MailService; import org.thingsboard.server.actors.DefaultTbActorSystem; import org.thingsboard.server.actors.TbActorId; diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java index 2e42009e5b..85e3fa35bd 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseTenantControllerTest.java @@ -27,15 +27,20 @@ import org.junit.Before; import org.junit.Test; import org.mockito.ArgumentMatcher; import org.mockito.Mockito; +import org.springframework.boot.test.mock.mockito.SpyBean; import org.springframework.test.context.TestPropertySource; import org.springframework.test.web.servlet.ResultActions; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.TenantInfo; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.page.PageLink; @@ -49,6 +54,12 @@ import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; +import org.thingsboard.server.common.msg.TbMsg; +import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.queue.discovery.PartitionService; import java.util.ArrayList; import java.util.Comparator; @@ -57,12 +68,19 @@ import java.util.List; import java.util.Map; import java.util.Random; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.function.Predicate; import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import static org.hamcrest.Matchers.containsString; +import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @TestPropertySource(properties = { @@ -78,6 +96,11 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { ListeningExecutorService executor; + @SpyBean + private PartitionService partitionService; + @SpyBean + private ActorSystemContext actorContext; + @Before public void setUp() throws Exception { executor = MoreExecutors.listeningDecorator(ThingsBoardExecutors.newWorkStealingPool(8, getClass())); @@ -85,6 +108,12 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { @After public void tearDown() throws Exception { + loginSysAdmin(); + for (Queue queue : doGetTypedWithPageLink("/api/queues?serviceType=TB_RULE_ENGINE&", new TypeReference>() {}, new PageLink(100)).getData()) { + if (!queue.getName().equals(DataConstants.MAIN_QUEUE_NAME)) { + doDelete("/api/queues/" + queue.getId()).andExpect(status().isOk()); + } + } executor.shutdownNow(); } @@ -518,10 +547,139 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { doDelete("/api/tenant/" + tenant.getId().getId().toString()).andExpect(status().isOk()); } + @Test + public void testUpdateTenantProfileToIsolated() throws Exception { + loginSysAdmin(); + doPost("/api/queues?serviceType=TB_RULE_ENGINE", new Queue(TenantId.SYS_TENANT_ID, getQueueConfig(DataConstants.HP_QUEUE_NAME, DataConstants.HP_QUEUE_TOPIC))).andExpect(status().isOk()); + TenantProfile tenantProfile = new TenantProfile(); + tenantProfile.setName("Test profile"); + TenantProfileData tenantProfileData = new TenantProfileData(); + tenantProfileData.setConfiguration(new DefaultTenantProfileConfiguration()); + tenantProfile.setProfileData(tenantProfileData); + tenantProfile.setIsolatedTbRuleEngine(false); + tenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class); + createDifferentTenant(); + loginSysAdmin(); + savedDifferentTenant.setTenantProfileId(tenantProfile.getId()); + savedDifferentTenant = doPost("/api/tenant", savedDifferentTenant, Tenant.class); + TenantId tenantId = differentTenantId; + + loginDifferentTenant(); + DeviceProfile hpQueueProfile = createDeviceProfile("HighPriority profile"); + hpQueueProfile.setDefaultQueueName(DataConstants.HP_QUEUE_NAME); + hpQueueProfile = doPost("/api/deviceProfile", hpQueueProfile, DeviceProfile.class); + Device hpQueueDevice = createDevice("HP", hpQueueProfile.getName(), "HP"); + + DeviceProfile mainQueueProfile = createDeviceProfile("Main profile"); + mainQueueProfile.setDefaultQueueName(DataConstants.MAIN_QUEUE_NAME); + mainQueueProfile = doPost("/api/deviceProfile", mainQueueProfile, DeviceProfile.class); + Device mainQueueDevice = createDevice("Main", mainQueueProfile.getName(), "Main"); + + verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, tenantId, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.HP_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(TenantId.SYS_TENANT_ID); + }); + verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, tenantId, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.MAIN_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(TenantId.SYS_TENANT_ID); + }); + + loginSysAdmin(); + tenantProfile.setIsolatedTbRuleEngine(true); + tenantProfile.getProfileData().setQueueConfiguration(List.of( + getQueueConfig(DataConstants.MAIN_QUEUE_NAME, DataConstants.MAIN_QUEUE_TOPIC) + )); + tenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class); + + loginDifferentTenant(); + verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, tenantId, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.MAIN_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); + }); + verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, tenantId, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.HP_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(TenantId.SYS_TENANT_ID); + }); + + loginSysAdmin(); + tenantProfile.setIsolatedTbRuleEngine(true); + tenantProfile.getProfileData().setQueueConfiguration(List.of( + getQueueConfig(DataConstants.MAIN_QUEUE_NAME, DataConstants.MAIN_QUEUE_TOPIC), + getQueueConfig(DataConstants.HP_QUEUE_NAME, DataConstants.HP_QUEUE_TOPIC) + )); + tenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class); + + loginDifferentTenant(); + verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, tenantId, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.HP_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); + }); + verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, tenantId, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { + doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); + }, usedTpi -> { + assertThat(usedTpi.getTopic()).isEqualTo(DataConstants.MAIN_QUEUE_TOPIC); + assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); + }); + } + + private void verifyUsedQueueAndMessage(String queue, TenantId tenantId, EntityId entityId, String msgType, Runnable action, Consumer tpiAssert) { + await().atMost(15, TimeUnit.SECONDS) + .untilAsserted(() -> { + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue, tenantId, entityId); + tpiAssert.accept(tpi); + }); + action.run(); + TbMsg tbMsg = awaitTbMsg(msg -> msg.getOriginator().equals(entityId) + && msg.getType().equals(msgType), 10000); + assertThat(tbMsg.getQueueName()).isEqualTo(queue); + + TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue, tenantId, entityId); + tpiAssert.accept(tpi); + } + + protected TbMsg awaitTbMsg(Predicate predicate, int timeoutMillis) { + AtomicReference tbMsgCaptor = new AtomicReference<>(); + verify(actorContext, timeout(timeoutMillis).atLeastOnce()).tell(argThat(actorMsg -> { + if (!(actorMsg instanceof QueueToRuleEngineMsg)) { + return false; + } + TbMsg tbMsg = ((QueueToRuleEngineMsg) actorMsg).getMsg(); + if (predicate.test(tbMsg)) { + tbMsgCaptor.set(tbMsg); + return true; + } + return false; + })); + return tbMsgCaptor.get(); + } + private void addQueueConfig(TenantProfile tenantProfile, String queueName) { + TenantProfileQueueConfiguration queueConfiguration = getQueueConfig(queueName, "tb_rule_engine." + queueName.toLowerCase()); + TenantProfileData profileData = tenantProfile.getProfileData(); + + List configs = profileData.getQueueConfiguration(); + if (configs == null) { + configs = new ArrayList<>(); + } + configs.add(queueConfiguration); + profileData.setQueueConfiguration(configs); + tenantProfile.setProfileData(profileData); + } + + private TenantProfileQueueConfiguration getQueueConfig(String queueName, String topic) { TenantProfileQueueConfiguration queueConfiguration = new TenantProfileQueueConfiguration(); queueConfiguration.setName(queueName); - queueConfiguration.setTopic("tb_rule_engine." + queueName.toLowerCase()); + queueConfiguration.setTopic(topic); queueConfiguration.setPollInterval(25); queueConfiguration.setPartitions(1 + new Random().nextInt(99)); queueConfiguration.setConsumerPerPartition(true); @@ -537,15 +695,7 @@ public abstract class BaseTenantControllerTest extends AbstractControllerTest { processingStrategy.setPauseBetweenRetries(3); processingStrategy.setMaxPauseBetweenRetries(3); queueConfiguration.setProcessingStrategy(processingStrategy); - TenantProfileData profileData = tenantProfile.getProfileData(); - - List configs = profileData.getQueueConfiguration(); - if (configs == null) { - configs = new ArrayList<>(); - } - configs.add(queueConfiguration); - profileData.setQueueConfiguration(configs); - tenantProfile.setProfileData(profileData); + return queueConfiguration; } private List getQueuesFromConfig(List queueConfiguration, List queues) { diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java index d4116a873d..e8c3baada9 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseTenantProfileControllerTest.java @@ -165,23 +165,6 @@ public abstract class BaseTenantProfileControllerTest extends AbstractController testBroadcastEntityStateChangeEventNeverTenantProfile(); } - @Test - public void testSaveSameTenantProfileWithDifferentIsolatedTbRuleEngine() throws Exception { - loginSysAdmin(); - TenantProfile tenantProfile = this.createTenantProfile("Tenant Profile"); - TenantProfile savedTenantProfile = doPost("/api/tenantProfile", tenantProfile, TenantProfile.class); - savedTenantProfile.setIsolatedTbRuleEngine(true); - addMainQueueConfig(savedTenantProfile); - - Mockito.reset(tbClusterService); - - doPost("/api/tenantProfile", savedTenantProfile) - .andExpect(status().isBadRequest()) - .andExpect(statusReason(containsString("Can't update isolatedTbRuleEngine property"))); - - testBroadcastEntityStateChangeEventNeverTenantProfile(); - } - @Test public void testDeleteTenantProfileWithExistingTenant() throws Exception { loginSysAdmin(); diff --git a/application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java b/application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java index b03a914d1f..f0ece07cdc 100644 --- a/application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java +++ b/application/src/test/java/org/thingsboard/server/service/notification/NotificationRuleApiTest.java @@ -301,7 +301,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest { @Test public void testNotificationRuleProcessing_entitiesLimit() throws Exception { int limit = 5; - updateDefaultTenantProfile(profileConfiguration -> { + updateDefaultTenantProfileConfig(profileConfiguration -> { profileConfiguration.setMaxDevices(limit); profileConfiguration.setMaxAssets(limit); profileConfiguration.setMaxCustomers(limit); @@ -396,7 +396,7 @@ public class NotificationRuleApiTest extends AbstractNotificationApiTest { @Test public void testNotificationRequestsPerRuleRateLimits() throws Exception { int notificationRequestsLimit = 10; - updateDefaultTenantProfile(profileConfiguration -> { + updateDefaultTenantProfileConfig(profileConfiguration -> { profileConfiguration.setTenantNotificationRequestsPerRuleRateLimit(notificationRequestsLimit + ":300"); }); diff --git a/application/src/test/java/org/thingsboard/server/service/queue/QueueServiceTest.java b/application/src/test/java/org/thingsboard/server/service/queue/QueueServiceTest.java deleted file mode 100644 index 4d2c50add3..0000000000 --- a/application/src/test/java/org/thingsboard/server/service/queue/QueueServiceTest.java +++ /dev/null @@ -1,225 +0,0 @@ -/** - * Copyright © 2016-2023 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.queue; - -import org.junit.Before; -import org.junit.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.mock.mockito.SpyBean; -import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.common.data.DataConstants; -import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.DeviceProfile; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.queue.ProcessingStrategy; -import org.thingsboard.server.common.data.queue.ProcessingStrategyType; -import org.thingsboard.server.common.data.queue.Queue; -import org.thingsboard.server.common.data.queue.SubmitStrategy; -import org.thingsboard.server.common.data.queue.SubmitStrategyType; -import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; -import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; -import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.controller.AbstractControllerTest; -import org.thingsboard.server.dao.queue.QueueService; -import org.thingsboard.server.dao.service.DaoSqlTest; -import org.thingsboard.server.queue.discovery.PartitionService; -import org.thingsboard.server.service.entitiy.queue.TbQueueService; -import org.thingsboard.server.service.profile.TbDeviceProfileCache; - -import java.util.List; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; -import java.util.function.Predicate; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; -import static org.mockito.ArgumentMatchers.argThat; -import static org.mockito.Mockito.timeout; -import static org.mockito.Mockito.verify; - -@DaoSqlTest -public class QueueServiceTest extends AbstractControllerTest { - - @SpyBean - private ActorSystemContext actorContext; - @Autowired - private TbQueueService tbQueueService; - @Autowired - private QueueService queueService; - @SpyBean - private PartitionService partitionService; - @Autowired - private TbDeviceProfileCache deviceProfileCache; - - @Before - public void beforeEach() { - Queue mainQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.MAIN_QUEUE_NAME); - if (mainQueue == null) { - mainQueue = new Queue(TenantId.SYS_TENANT_ID, getMainQueueConfig()); - tbQueueService.saveQueue(mainQueue); - } - - Queue hpQueue = queueService.findQueueByTenantIdAndName(TenantId.SYS_TENANT_ID, DataConstants.HP_QUEUE_NAME); - if (hpQueue == null) { - hpQueue = new Queue(TenantId.SYS_TENANT_ID, getHighPriorityQueueConfig()); - tbQueueService.saveQueue(hpQueue); - } - } - - @Test - public void testQueuesUpdateOnTenantProfileUpdate() throws Exception { - loginTenantAdmin(); - DeviceProfile hpQueueProfile = createDeviceProfile("HighPriority profile"); - hpQueueProfile.setDefaultQueueName(DataConstants.HP_QUEUE_NAME); - hpQueueProfile = doPost("/api/deviceProfile", hpQueueProfile, DeviceProfile.class); - Device hpQueueDevice = createDevice("HP", hpQueueProfile.getName(), "HP"); - deviceProfileCache.evict(tenantId, hpQueueProfile.getId()); - - DeviceProfile mainQueueProfile = createDeviceProfile("Main profile"); - mainQueueProfile.setDefaultQueueName(DataConstants.MAIN_QUEUE_NAME); - mainQueueProfile = doPost("/api/deviceProfile", mainQueueProfile, DeviceProfile.class); - Device mainQueueDevice = createDevice("Main", mainQueueProfile.getName(), "Main"); - - verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTenantId()).get().isEqualTo(TenantId.SYS_TENANT_ID); - }); - verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTenantId()).get().isEqualTo(TenantId.SYS_TENANT_ID); - }); - - updateDefaultTenantProfile(tenantProfile -> { - tenantProfile.setIsolatedTbRuleEngine(true); - tenantProfile.getProfileData().setQueueConfiguration(List.of( - getMainQueueConfig() - )); - }); - - verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); - }); - verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTopic()).endsWith("main"); - assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); - }); - - updateDefaultTenantProfile(tenantProfile -> { - tenantProfile.setIsolatedTbRuleEngine(true); - tenantProfile.getProfileData().setQueueConfiguration(List.of( - getMainQueueConfig(), getHighPriorityQueueConfig() - )); - }); - - verifyUsedQueueAndMessage(DataConstants.HP_QUEUE_NAME, hpQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + hpQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTopic()).endsWith("hp"); - assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); - }); - verifyUsedQueueAndMessage(DataConstants.MAIN_QUEUE_NAME, mainQueueDevice.getId(), DataConstants.ATTRIBUTES_UPDATED, () -> { - doPost("/api/plugins/telemetry/DEVICE/" + mainQueueDevice.getId() + "/attributes/SERVER_SCOPE", "{\"test\":123}", String.class); - }, usedTpi -> { - assertThat(usedTpi.getTenantId()).get().isEqualTo(tenantId); - }); - } - - private void verifyUsedQueueAndMessage(String queue, EntityId entityId, String msgType, Runnable action, Consumer tpiAssert) { - await().atMost(15, TimeUnit.SECONDS) - .untilAsserted(() -> { - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue, tenantId, entityId); - tpiAssert.accept(tpi); - }); - action.run(); - TbMsg tbMsg = awaitTbMsg(msg -> msg.getOriginator().equals(entityId) - && msg.getType().equals(msgType), 10000); - assertThat(tbMsg.getQueueName()).isEqualTo(queue); - - TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, queue, tenantId, entityId); - tpiAssert.accept(tpi); - } - - protected TbMsg awaitTbMsg(Predicate predicate, int timeoutMillis) { - AtomicReference tbMsgCaptor = new AtomicReference<>(); - verify(actorContext, timeout(timeoutMillis).atLeastOnce()).tell(argThat(actorMsg -> { - if (!(actorMsg instanceof QueueToRuleEngineMsg)) { - return false; - } - TbMsg tbMsg = ((QueueToRuleEngineMsg) actorMsg).getMsg(); - if (predicate.test(tbMsg)) { - tbMsgCaptor.set(tbMsg); - return true; - } - return false; - })); - return tbMsgCaptor.get(); - } - - private TenantProfileQueueConfiguration getHighPriorityQueueConfig() { - TenantProfileQueueConfiguration hpQueueConfig = new TenantProfileQueueConfiguration(); - hpQueueConfig.setName(DataConstants.HP_QUEUE_NAME); - hpQueueConfig.setTopic(DataConstants.HP_QUEUE_TOPIC); - hpQueueConfig.setPollInterval(25); - hpQueueConfig.setPartitions(10); - hpQueueConfig.setConsumerPerPartition(true); - hpQueueConfig.setPackProcessingTimeout(2000); - SubmitStrategy highPriorityQueueSubmitStrategy = new SubmitStrategy(); - highPriorityQueueSubmitStrategy.setType(SubmitStrategyType.BURST); - highPriorityQueueSubmitStrategy.setBatchSize(100); - hpQueueConfig.setSubmitStrategy(highPriorityQueueSubmitStrategy); - ProcessingStrategy highPriorityQueueProcessingStrategy = new ProcessingStrategy(); - highPriorityQueueProcessingStrategy.setType(ProcessingStrategyType.RETRY_FAILED_AND_TIMED_OUT); - highPriorityQueueProcessingStrategy.setRetries(0); - highPriorityQueueProcessingStrategy.setFailurePercentage(0); - highPriorityQueueProcessingStrategy.setPauseBetweenRetries(5); - highPriorityQueueProcessingStrategy.setMaxPauseBetweenRetries(5); - hpQueueConfig.setProcessingStrategy(highPriorityQueueProcessingStrategy); - return hpQueueConfig; - } - - private TenantProfileQueueConfiguration getMainQueueConfig() { - TenantProfileQueueConfiguration mainQueue = new TenantProfileQueueConfiguration(); - mainQueue.setName(DataConstants.MAIN_QUEUE_NAME); - mainQueue.setTopic(DataConstants.MAIN_QUEUE_TOPIC); - mainQueue.setPollInterval(25); - mainQueue.setPartitions(10); - mainQueue.setConsumerPerPartition(true); - mainQueue.setPackProcessingTimeout(2000); - SubmitStrategy mainQueueSubmitStrategy = new SubmitStrategy(); - mainQueueSubmitStrategy.setType(SubmitStrategyType.BURST); - mainQueueSubmitStrategy.setBatchSize(1000); - mainQueue.setSubmitStrategy(mainQueueSubmitStrategy); - ProcessingStrategy mainQueueProcessingStrategy = new ProcessingStrategy(); - mainQueueProcessingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); - mainQueueProcessingStrategy.setRetries(3); - mainQueueProcessingStrategy.setFailurePercentage(0); - mainQueueProcessingStrategy.setPauseBetweenRetries(3); - mainQueueProcessingStrategy.setMaxPauseBetweenRetries(3); - mainQueue.setProcessingStrategy(mainQueueProcessingStrategy); - return mainQueue; - } - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java index 31f517c236..a8954b2b33 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java @@ -198,7 +198,7 @@ public class HashPartitionService implements PartitionService { if (!MAIN_QUEUE_NAME.equals(queueName) && !partitionSizesMap.containsKey(queueKey)) { queueKey = new QueueKey(serviceType, TenantId.SYS_TENANT_ID); } - log.warn("Using queue {} instead of isolated {}", queueKey, queueName); + log.warn("Using queue {} instead of isolated {} for tenant {}", queueKey, queueName, isolatedOrSystemTenantId); } } return resolve(queueKey, entityId); diff --git a/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts b/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts index e96a237033..fa8b1636ff 100644 --- a/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts +++ b/ui-ngx/src/app/modules/home/components/profile/tenant-profile.component.ts @@ -56,10 +56,10 @@ export class TenantProfileComponent extends EntityComponent { const mainQueue = [ { id: guid(), - consumerPerPartition: false, + consumerPerPartition: true, name: 'Main', packProcessingTimeout: 2000, - partitions: 1, + partitions: 2, pollInterval: 25, processingStrategy: { failurePercentage: 0, @@ -82,8 +82,8 @@ export class TenantProfileComponent extends EntityComponent { name: 'HighPriority', topic: 'tb_rule_engine.hp', pollInterval: 25, - partitions: 1, - consumerPerPartition: false, + partitions: 2, + consumerPerPartition: true, packProcessingTimeout: 2000, submitStrategy: { type: 'BURST', @@ -105,8 +105,8 @@ export class TenantProfileComponent extends EntityComponent { name: 'SequentialByOriginator', topic: 'tb_rule_engine.sq', pollInterval: 25, - partitions: 1, - consumerPerPartition: false, + partitions: 2, + consumerPerPartition: true, packProcessingTimeout: 2000, submitStrategy: { type: 'SEQUENTIAL_BY_ORIGINATOR',