diff --git a/application/src/main/data/upgrade/3.6.2/schema_update.sql b/application/src/main/data/upgrade/3.6.2/schema_update.sql new file mode 100644 index 0000000000..4e62209ea8 --- /dev/null +++ b/application/src/main/data/upgrade/3.6.2/schema_update.sql @@ -0,0 +1,23 @@ +-- +-- 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. +-- + +-- RULE NODE INDEXES UPDATE START + +DROP INDEX IF EXISTS idx_rule_node_type; +DROP INDEX IF EXISTS idx_rule_node_type_configuration_version; +CREATE INDEX IF NOT EXISTS idx_rule_node_type_id_configuration_version ON rule_node(type, id, configuration_version); + +-- RULE NODE INDEXES UPDATE END diff --git a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java index 5ab3b7b28e..8b14d2d8b7 100644 --- a/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java +++ b/application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java @@ -276,6 +276,9 @@ public class ThingsboardInstallService { } else { log.info("Skipping images migration. Run the upgrade with fromVersion as '3.6.2-images' to migrate"); } + case "3.6.2": + log.info("Upgrading ThingsBoard from version 3.6.2 to 3.6.3 ..."); + databaseEntitiesUpgradeService.upgradeDatabase("3.6.2"); //TODO DON'T FORGET to update switch statement in the CacheCleanupService if you need to clear the cache break; default: diff --git a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java index c2444390f6..4a96d0d2c3 100644 --- a/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java +++ b/application/src/main/java/org/thingsboard/server/service/install/SqlDatabaseUpgradeService.java @@ -772,6 +772,9 @@ public class SqlDatabaseUpgradeService implements DatabaseEntitiesUpgradeService } }); break; + case "3.6.2": + updateSchema("3.6.2", 3006002, "3.6.3", 3006003, null); + break; default: throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index c650d53b8f..0561c0c249 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -1398,6 +1398,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java index 532da4ac00..a8a60bc565 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/rule/RuleChainService.java @@ -100,6 +100,7 @@ public interface RuleChainService extends EntityDaoService { PageData findAllRuleNodesByType(String type, PageLink pageLink); + @Deprecated(forRemoval = true, since = "3.6.3") PageData findAllRuleNodesByTypeAndVersionLessThan(String type, int version, PageLink pageLink); PageData findAllRuleNodeIdsByTypeAndVersionLessThan(String type, int version, PageLink pageLink); diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java index ba4eeb6ca4..3ccbc229a0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsAdmin.java @@ -24,11 +24,17 @@ import com.amazonaws.services.sqs.AmazonSQS; import com.amazonaws.services.sqs.AmazonSQSClientBuilder; import com.amazonaws.services.sqs.model.CreateQueueRequest; import com.amazonaws.services.sqs.model.GetQueueUrlResult; +import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.ExecutorProvider; +import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.queue.TbQueueAdmin; import org.thingsboard.server.queue.util.PropertyUtils; import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.function.Function; import java.util.stream.Collectors; @@ -38,6 +44,8 @@ public class TbAwsSqsAdmin implements TbQueueAdmin { private final Map attributes; private final AmazonSQS sqsClient; private final Map queues; + @Getter + private final ExecutorService producerExecutor; public TbAwsSqsAdmin(TbAwsSqsSettings sqsSettings, Map attributes) { this.attributes = attributes; @@ -49,6 +57,7 @@ public class TbAwsSqsAdmin implements TbQueueAdmin { AWSCredentials awsCredentials = new BasicAWSCredentials(sqsSettings.getAccessKeyId(), sqsSettings.getSecretAccessKey()); credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials); } + producerExecutor = ThingsBoardExecutors.newWorkStealingPool(sqsSettings.getThreadPoolSize(), "aws-sqs-queue-executor"); sqsClient = AmazonSQSClientBuilder.standard() .withCredentials(credentialsProvider) @@ -104,5 +113,8 @@ public class TbAwsSqsAdmin implements TbQueueAdmin { if (sqsClient != null) { sqsClient.shutdown(); } + if (producerExecutor != null) { + producerExecutor.shutdownNow(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java index 863e96a714..f5bca724d0 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsProducerTemplate.java @@ -20,15 +20,11 @@ import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.auth.AWSStaticCredentialsProvider; import com.amazonaws.auth.BasicAWSCredentials; import com.amazonaws.auth.DefaultAWSCredentialsProviderChain; -import com.amazonaws.services.sqs.AmazonSQS; -import com.amazonaws.services.sqs.AmazonSQSClientBuilder; +import com.amazonaws.handlers.AsyncHandler; +import com.amazonaws.services.sqs.AmazonSQSAsync; +import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder; import com.amazonaws.services.sqs.model.SendMessageRequest; import com.amazonaws.services.sqs.model.SendMessageResult; -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.ListeningExecutorService; -import com.google.common.util.concurrent.MoreExecutors; import com.google.gson.Gson; import lombok.extern.slf4j.Slf4j; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; @@ -41,19 +37,17 @@ import org.thingsboard.server.queue.common.DefaultTbQueueMsg; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; @Slf4j public class TbAwsSqsProducerTemplate implements TbQueueProducer { private final String defaultTopic; - private final AmazonSQS sqsClient; + private final AmazonSQSAsync sqsClient; private final Gson gson = new Gson(); private final Map queueUrlMap = new ConcurrentHashMap<>(); - private final TbQueueAdmin admin; - private ListeningExecutorService producerExecutor; + private final TbAwsSqsAdmin admin; public TbAwsSqsProducerTemplate(TbQueueAdmin admin, TbAwsSqsSettings sqsSettings, String defaultTopic) { - this.admin = admin; + this.admin = (TbAwsSqsAdmin) admin; this.defaultTopic = defaultTopic; AWSCredentialsProvider credentialsProvider; @@ -64,11 +58,11 @@ public class TbAwsSqsProducerTemplate implements TbQueuePr credentialsProvider = new AWSStaticCredentialsProvider(awsCredentials); } - sqsClient = AmazonSQSClientBuilder.standard() + sqsClient = AmazonSQSAsyncClientBuilder.standard() .withCredentials(credentialsProvider) .withRegion(sqsSettings.getRegion()) + .withExecutorFactory(this.admin::getProducerExecutor) .build(); - producerExecutor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool()); } @Override @@ -91,30 +85,24 @@ public class TbAwsSqsProducerTemplate implements TbQueuePr sendMsgRequest.withMessageGroupId(sqsMsgId); sendMsgRequest.withMessageDeduplicationId(sqsMsgId); - ListenableFuture future = producerExecutor.submit(() -> sqsClient.sendMessage(sendMsgRequest)); - - Futures.addCallback(future, new FutureCallback() { - @Override - public void onSuccess(SendMessageResult result) { + sqsClient.sendMessageAsync(sendMsgRequest, new AsyncHandler() { + @Override public void onError(Exception e) { if (callback != null) { - callback.onSuccess(new AwsSqsTbQueueMsgMetadata(result.getSdkHttpMetadata())); + callback.onFailure(e); } } - @Override - public void onFailure(Throwable t) { + @Override public void onSuccess(SendMessageRequest request, + SendMessageResult sendMessageResult) { if (callback != null) { - callback.onFailure(t); + callback.onSuccess(new AwsSqsTbQueueMsgMetadata(sendMessageResult.getSdkHttpMetadata())); } } - }, producerExecutor); + }); } @Override public void stop() { - if (producerExecutor != null) { - producerExecutor.shutdownNow(); - } if (sqsClient != null) { sqsClient.shutdown(); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java index 89dc9d7e47..f2dc1efdfb 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/sqs/TbAwsSqsSettings.java @@ -42,4 +42,7 @@ public class TbAwsSqsSettings { @Value("${queue.aws_sqs.threads_per_topic}") private int threadsPerTopic; + @Value("${queue.aws_sqs.producer_thread_pool_size:50}") + private int threadPoolSize; + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java b/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java index d8c4ffddc9..3c238305bd 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java +++ b/dao/src/main/java/org/thingsboard/server/dao/service/Validator.java @@ -61,7 +61,7 @@ public class Validator { /** - * This method validate long value. If value isn't possitive than throw + * This method validate long value. If value isn't positive than throw * IncorrectParameterException exception * * @param val the val diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java index f02072b356..de61422527 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/rule/RuleNodeRepository.java @@ -43,7 +43,7 @@ public interface RuleNodeRepository extends JpaRepository Pageable pageable); @Query(nativeQuery = true, value = "SELECT * FROM rule_node r WHERE r.type = :ruleType " + - " AND configuration_version < :version " + + " AND r.configuration_version < :version " + " AND (:searchText IS NULL OR r.configuration ILIKE CONCAT('%', :searchText, '%'))") Page findAllRuleNodesByTypeAndVersionLessThan(@Param("ruleType") String ruleType, @Param("version") int version, diff --git a/dao/src/main/resources/sql/schema-entities-idx.sql b/dao/src/main/resources/sql/schema-entities-idx.sql index df95a50d47..6a95d9e346 100644 --- a/dao/src/main/resources/sql/schema-entities-idx.sql +++ b/dao/src/main/resources/sql/schema-entities-idx.sql @@ -91,9 +91,7 @@ CREATE INDEX IF NOT EXISTS idx_widgets_bundle_external_id ON widgets_bundle(tena CREATE INDEX IF NOT EXISTS idx_rule_node_external_id ON rule_node(rule_chain_id, external_id); -CREATE INDEX IF NOT EXISTS idx_rule_node_type ON rule_node(type); - -CREATE INDEX IF NOT EXISTS idx_rule_node_type_configuration_version ON rule_node(type, configuration_version); +CREATE INDEX IF NOT EXISTS idx_rule_node_type_id_configuration_version ON rule_node(type, id, configuration_version); CREATE INDEX IF NOT EXISTS idx_api_usage_state_entity_id ON api_usage_state(entity_id); diff --git a/msa/vc-executor/src/main/resources/tb-vc-executor.yml b/msa/vc-executor/src/main/resources/tb-vc-executor.yml index 962c5303c7..e4519529c5 100644 --- a/msa/vc-executor/src/main/resources/tb-vc-executor.yml +++ b/msa/vc-executor/src/main/resources/tb-vc-executor.yml @@ -155,6 +155,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds core: "${TB_QUEUE_AWS_SQS_CORE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index ab5deff3f7..f11b1b8ce2 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -281,6 +281,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 74be81e667..b2fa93c434 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -265,6 +265,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml index 10dbac101c..c2891994b1 100644 --- a/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml +++ b/transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml @@ -360,6 +360,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index 0eb8b4315b..519e7a8b32 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -297,6 +297,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}" diff --git a/transport/snmp/src/main/resources/tb-snmp-transport.yml b/transport/snmp/src/main/resources/tb-snmp-transport.yml index d139b641a6..ab9d6ad584 100644 --- a/transport/snmp/src/main/resources/tb-snmp-transport.yml +++ b/transport/snmp/src/main/resources/tb-snmp-transport.yml @@ -250,6 +250,8 @@ queue: region: "${TB_QUEUE_AWS_SQS_REGION:YOUR_REGION}" # Number of threads per each AWS SQS queue in consumer threads_per_topic: "${TB_QUEUE_AWS_SQS_THREADS_PER_TOPIC:1}" + # Thread pool size for aws_sqs queue producer executor provider. Default value equals to AmazonSQSAsyncClient.DEFAULT_THREAD_POOL_SIZE + producer_thread_pool_size: "${TB_QUEUE_AWS_SQS_EXECUTOR_THREAD_POOL_SIZE:50}" queue-properties: # AWS SQS queue properties. VisibilityTimeout in seconds;MaximumMessageSize in bytes;MessageRetentionPeriod in seconds rule-engine: "${TB_QUEUE_AWS_SQS_RE_QUEUE_PROPERTIES:VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800}"