Merge pull request #8414 from thingsboard/feature/singleton-rule-node

Feature/singleton rule node

application/src/main/data/upgrade/3.4.4/schema_update.sql

@@ -614,3 +614,16 @@ END
 $$;
 
 -- TTL DROP PARTITIONS FUNCTIONS UPDATE END
+
+-- RULE NODE SINGLETON MODE SUPPORT
+
+ALTER TABLE rule_node ADD COLUMN IF NOT EXISTS singleton_mode bool DEFAULT false;
+
+UPDATE rule_node SET singleton_mode = true WHERE type IN ('org.thingsboard.rule.engine.mqtt.azure.TbAzureIotHubNode', 'org.thingsboard.rule.engine.mqtt.TbMqttNode');
+
+ALTER TABLE component_descriptor ADD COLUMN IF NOT EXISTS clustering_mode varchar(255) DEFAULT 'ENABLED';
+
+UPDATE component_descriptor SET clustering_mode = 'USER_PREFERENCE' WHERE clazz = 'org.thingsboard.rule.engine.mqtt.TbMqttNode';
+
+UPDATE component_descriptor SET clustering_mode = 'SINGLETON' WHERE clazz = 'org.thingsboard.rule.engine.mqtt.azure.TbAzureIotHubNode';
+

application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@@ -87,6 +87,7 @@ import org.thingsboard.server.dao.timeseries.TimeseriesService;
 import org.thingsboard.server.dao.user.UserService;
 import org.thingsboard.server.dao.widget.WidgetTypeService;
 import org.thingsboard.server.dao.widget.WidgetsBundleService;
+import org.thingsboard.server.queue.discovery.DiscoveryService;
 import org.thingsboard.server.queue.discovery.PartitionService;
 import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
 import org.thingsboard.server.queue.notification.NotificationRuleProcessor;
@@ -179,6 +180,10 @@ public class ActorSystemContext {
     @Setter
     private ComponentDiscoveryService componentService;
 
+    @Autowired
+    @Getter
+    private DiscoveryService discoveryService;
+
     @Autowired
     @Getter
     private DataDecodingEncodingService encodingService;

application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeActorMessageProcessor.java

@@ -31,7 +31,10 @@ import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.RuleNodeException;
 import org.thingsboard.server.common.msg.queue.RuleNodeInfo;
+import org.thingsboard.server.common.msg.queue.ServiceType;
+import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
 import org.thingsboard.server.common.stats.TbApiUsageReportClient;
+import org.thingsboard.server.gen.transport.TransportProtos;
 
 /**
  * @author Andrew Shvayka
@@ -39,11 +42,10 @@ import org.thingsboard.server.common.stats.TbApiUsageReportClient;
 public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
 
     private final String ruleChainName;
-    private final TbActorRef self;
     private final TbApiUsageReportClient apiUsageClient;
+    private final DefaultTbContext defaultCtx;
     private RuleNode ruleNode;
     private TbNode tbNode;
-    private DefaultTbContext defaultCtx;
     private RuleNodeInfo info;
 
     RuleNodeActorMessageProcessor(TenantId tenantId, String ruleChainName, RuleNodeId ruleNodeId, ActorSystemContext systemContext
@@ -51,7 +53,6 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
         super(systemContext, tenantId, ruleNodeId);
         this.apiUsageClient = systemContext.getApiUsageClient();
         this.ruleChainName = ruleChainName;
-        this.self = self;
         this.ruleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
         this.defaultCtx = new DefaultTbContext(systemContext, ruleChainName, new RuleNodeCtx(tenantId, parent, self, ruleNode));
         this.info = new RuleNodeInfo(ruleNodeId, ruleChainName, ruleNode != null ? ruleNode.getName() : "Unknown");
@@ -59,29 +60,36 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
 
     @Override
     public void start(TbActorCtx context) throws Exception {
-        tbNode = initComponent(ruleNode);
-        if (tbNode != null) {
-            state = ComponentLifecycleState.ACTIVE;
+        if (isMyNodePartition()) {
+            tbNode = initComponent(ruleNode);
+            if (tbNode != null) {
+                state = ComponentLifecycleState.ACTIVE;
+            }
         }
     }
 
     @Override
     public void onUpdate(TbActorCtx context) throws Exception {
-        RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
-        this.info = new RuleNodeInfo(entityId, ruleChainName, newRuleNode != null ? newRuleNode.getName() : "Unknown");
-        boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
-                !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
-        this.ruleNode = newRuleNode;
-        this.defaultCtx.updateSelf(newRuleNode);
-        if (restartRequired) {
-            if (tbNode != null) {
-                tbNode.destroy();
-            }
-            try {
-                start(context);
-            } catch (Exception e) {
-                throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+        if (isMyNodePartition()) {
+            RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
+            this.info = new RuleNodeInfo(entityId, ruleChainName, newRuleNode != null ? newRuleNode.getName() : "Unknown");
+            boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
+                    !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
+            this.ruleNode = newRuleNode;
+            this.defaultCtx.updateSelf(newRuleNode);
+            if (restartRequired) {
+                if (tbNode != null) {
+                    tbNode.destroy();
+                }
+                try {
+                    start(context);
+                } catch (Exception e) {
+                    throw new TbRuleNodeUpdateException("Failed to update rule node", e);
+                }
             }
+        } else if (tbNode != null) {
+            stop(null);
+            tbNode = null;
         }
     }
 
@@ -94,9 +102,16 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     @Override
-    public void onPartitionChangeMsg(PartitionChangeMsg msg) {
+    public void onPartitionChangeMsg(PartitionChangeMsg msg) throws Exception {
         if (tbNode != null) {
-            tbNode.onPartitionChangeMsg(defaultCtx, msg);
+            if (!isMyNodePartition()) {
+                stop(null);
+                tbNode = null;
+            } else {
+                tbNode.onPartitionChangeMsg(defaultCtx, msg);
+            }
+        } else if (isMyNodePartition()) {
+            start(null);
         }
     }
 
@@ -121,23 +136,27 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     }
 
     void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception {
-        msg.getMsg().getCallback().onProcessingStart(info);
-        checkComponentStateActive(msg.getMsg());
-        TbMsg tbMsg = msg.getMsg();
-        int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
-        int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();
-        if (maxRuleNodeExecutionsPerMessage == 0 || ruleNodeCount < maxRuleNodeExecutionsPerMessage) {
-            apiUsageClient.report(tenantId, tbMsg.getCustomerId(), ApiUsageRecordKey.RE_EXEC_COUNT);
-            if (ruleNode.isDebugMode()) {
-                systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType());
-            }
-            try {
-                tbNode.onMsg(msg.getCtx(), msg.getMsg());
-            } catch (Exception e) {
-                msg.getCtx().tellFailure(msg.getMsg(), e);
-            }
+        if (!isMyNodePartition()) {
+            putToNodePartition(msg.getMsg());
         } else {
-            tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            msg.getMsg().getCallback().onProcessingStart(info);
+            checkComponentStateActive(msg.getMsg());
+            TbMsg tbMsg = msg.getMsg();
+            int ruleNodeCount = tbMsg.getAndIncrementRuleNodeCounter();
+            int maxRuleNodeExecutionsPerMessage = getTenantProfileConfiguration().getMaxRuleNodeExecsPerMessage();
+            if (maxRuleNodeExecutionsPerMessage == 0 || ruleNodeCount < maxRuleNodeExecutionsPerMessage) {
+                apiUsageClient.report(tenantId, tbMsg.getCustomerId(), ApiUsageRecordKey.RE_EXEC_COUNT);
+                if (ruleNode.isDebugMode()) {
+                    systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType());
+                }
+                try {
+                    tbNode.onMsg(msg.getCtx(), msg.getMsg());
+                } catch (Exception e) {
+                    msg.getCtx().tellFailure(msg.getMsg(), e);
+                }
+            } else {
+                tbMsg.getCallback().onFailure(new RuleNodeException("Message is processed by more then " + maxRuleNodeExecutionsPerMessage + " rule nodes!", ruleChainName, ruleNode));
+            }
         }
     }
 
@@ -160,4 +179,23 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
     protected RuleNodeException getInactiveException() {
         return new RuleNodeException("Rule Node is not active! Failed to initialize.", ruleChainName, ruleNode);
     }
+
+    private boolean isMyNodePartition() {
+        return !ruleNode.isSingletonMode()
+                || systemContext.getDiscoveryService().isMonolith()
+                || defaultCtx.isLocalEntity(ruleNode.getId());
+    }
+
+    //Message will return after processing. See RuleChainActorMessageProcessor.pushToTarget.
+    private void putToNodePartition(TbMsg source) {
+        TbMsg tbMsg = TbMsg.newMsg(source, source.getQueueName(), source.getRuleChainId(), entityId);
+        TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, tbMsg.getQueueName(), tenantId, ruleNode.getId());
+        TransportProtos.ToRuleEngineMsg toQueueMsg = TransportProtos.ToRuleEngineMsg.newBuilder()
+                .setTenantIdMSB(tenantId.getId().getMostSignificantBits())
+                .setTenantIdLSB(tenantId.getId().getLeastSignificantBits())
+                .setTbMsg(TbMsg.toByteString(tbMsg))
+                .build();
+        systemContext.getClusterService().pushMsgToRuleEngine(tpi, tbMsg.getId(), toQueueMsg, null);
+        defaultCtx.ack(source);
+    }
 }

application/src/main/java/org/thingsboard/server/service/component/AnnotationComponentDiscoveryService.java

@@ -155,6 +155,7 @@ public class AnnotationComponentDiscoveryService implements ComponentDiscoverySe
             RuleNode ruleNodeAnnotation = clazz.getAnnotation(RuleNode.class);
             scannedComponent.setName(ruleNodeAnnotation.name());
             scannedComponent.setScope(ruleNodeAnnotation.scope());
+            scannedComponent.setClusteringMode(ruleNodeAnnotation.clusteringMode());
             NodeDefinition nodeDefinition = prepareNodeDefinition(ruleNodeAnnotation);
             ObjectNode configurationDescriptor = mapper.createObjectNode();
             JsonNode node = mapper.valueToTree(nodeDefinition);

common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentClusteringMode.java

@@ -0,0 +1,29 @@
+/**
+ * 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.common.data.plugin;
+
+/**
+ * The main idea to use this - it's adding the ability to start rule nodes in singleton mode in cluster setup
+ * (singleton rule node will start in only one Rule Engine instance)
+ * USER_PREFERENCE - user has ability to configure clustering mode (enable/disable singleton mode in rule node config)
+ * ENABLE - user doesn't have ability to configure clustering mode (singleton mode is always FALSE in rule node config)
+ * SINGLETON - user doesn't have ability to configure clustering mode (singleton mode is always TRUE in rule node config)
+ */
+public enum ComponentClusteringMode {
+    USER_PREFERENCE,
+    ENABLED,
+    SINGLETON
+}

common/data/src/main/java/org/thingsboard/server/common/data/plugin/ComponentDescriptor.java

@@ -36,15 +36,17 @@ public class ComponentDescriptor extends SearchTextBased<ComponentDescriptorId>
     @Getter @Setter private ComponentType type;
     @ApiModelProperty(position = 4, value = "Scope of the Rule Node. Always set to 'TENANT', since no rule chains on the 'SYSTEM' level yet.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, allowableValues = "TENANT", example = "TENANT")
     @Getter @Setter private ComponentScope scope;
+    @ApiModelProperty(position = 5, value = "Clustering mode of the RuleNode. This mode represents the ability to start Rule Node in multiple microservices.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, allowableValues = "USER_PREFERENCE, ENABLED, SINGLETON", example = "ENABLED")
+    @Getter @Setter private ComponentClusteringMode clusteringMode;
     @Length(fieldName = "name")
-    @ApiModelProperty(position = 5, value = "Name of the Rule Node. Taken from the @RuleNode annotation.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "Custom Rule Node")
+    @ApiModelProperty(position = 6, value = "Name of the Rule Node. Taken from the @RuleNode annotation.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "Custom Rule Node")
     @Getter @Setter private String name;
-    @ApiModelProperty(position = 6, value = "Full name of the Java class that implements the Rule Engine Node interface.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "com.mycompany.CustomRuleNode")
+    @ApiModelProperty(position = 7, value = "Full name of the Java class that implements the Rule Engine Node interface.", accessMode = ApiModelProperty.AccessMode.READ_ONLY, example = "com.mycompany.CustomRuleNode")
     @Getter @Setter private String clazz;
-    @ApiModelProperty(position = 7, value = "Complex JSON object that represents the Rule Node configuration.", accessMode = ApiModelProperty.AccessMode.READ_ONLY)
+    @ApiModelProperty(position = 8, value = "Complex JSON object that represents the Rule Node configuration.", accessMode = ApiModelProperty.AccessMode.READ_ONLY)
     @Getter @Setter private transient JsonNode configurationDescriptor;
     @Length(fieldName = "actions")
-    @ApiModelProperty(position = 8, value = "Rule Node Actions. Deprecated. Always null.", accessMode = ApiModelProperty.AccessMode.READ_ONLY)
+    @ApiModelProperty(position = 9, value = "Rule Node Actions. Deprecated. Always null.", accessMode = ApiModelProperty.AccessMode.READ_ONLY)
     @Getter @Setter private String actions;
 
     public ComponentDescriptor() {

common/data/src/main/java/org/thingsboard/server/common/data/rule/RuleNode.java

@@ -48,7 +48,9 @@ public class RuleNode extends SearchTextBasedWithAdditionalInfo<RuleNodeId> impl
     private String name;
     @ApiModelProperty(position = 6, value = "Enable/disable debug. ", example = "false")
     private boolean debugMode;
-    @ApiModelProperty(position = 7, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
+    @ApiModelProperty(position = 7, value = "Enable/disable singleton mode. ", example = "false")
+    private boolean singletonMode;
+    @ApiModelProperty(position = 8, value = "JSON with the rule node configuration. Structure depends on the rule node implementation.", dataType = "com.fasterxml.jackson.databind.JsonNode")
     private transient JsonNode configuration;
     @JsonIgnore
     private byte[] configurationBytes;
@@ -69,6 +71,7 @@ public class RuleNode extends SearchTextBasedWithAdditionalInfo<RuleNodeId> impl
         this.type = ruleNode.getType();
         this.name = ruleNode.getName();
         this.debugMode = ruleNode.isDebugMode();
+        this.singletonMode = ruleNode.isSingletonMode();
         this.setConfiguration(ruleNode.getConfiguration());
         this.externalId = ruleNode.getExternalId();
     }

dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java

@@ -403,6 +403,7 @@ public class ModelConstants {
     public static final String COMPONENT_DESCRIPTOR_COLUMN_FAMILY_NAME = "component_descriptor";
     public static final String COMPONENT_DESCRIPTOR_TYPE_PROPERTY = "type";
     public static final String COMPONENT_DESCRIPTOR_SCOPE_PROPERTY = "scope";
+    public static final String COMPONENT_DESCRIPTOR_CLUSTERING_MODE_PROPERTY = "clustering_mode";
     public static final String COMPONENT_DESCRIPTOR_NAME_PROPERTY = "name";
     public static final String COMPONENT_DESCRIPTOR_CLASS_PROPERTY = "clazz";
     public static final String COMPONENT_DESCRIPTOR_CONFIGURATION_DESCRIPTOR_PROPERTY = "configuration_descriptor";
@@ -446,6 +447,7 @@ public class ModelConstants {
     public static final String EVENT_MESSAGE_COLUMN_NAME = "e_message";
 
     public static final String DEBUG_MODE = "debug_mode";
+    public static final String SINGLETON_MODE = "singleton_mode";
 
     /**
      * Cassandra rule chain constants.

dao/src/main/java/org/thingsboard/server/dao/model/sql/ComponentDescriptorEntity.java

@@ -23,6 +23,7 @@ import org.hibernate.annotations.TypeDef;
 import org.thingsboard.server.common.data.id.ComponentDescriptorId;
 import org.thingsboard.server.common.data.plugin.ComponentDescriptor;
 import org.thingsboard.server.common.data.plugin.ComponentScope;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.dao.model.BaseSqlEntity;
 import org.thingsboard.server.dao.model.ModelConstants;
@@ -50,6 +51,10 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
     @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_SCOPE_PROPERTY)
     private ComponentScope scope;
 
+    @Enumerated(EnumType.STRING)
+    @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_CLUSTERING_MODE_PROPERTY)
+    private ComponentClusteringMode clusteringMode;
+
     @Column(name = ModelConstants.COMPONENT_DESCRIPTOR_NAME_PROPERTY)
     private String name;
 
@@ -77,6 +82,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         this.actions = component.getActions();
         this.type = component.getType();
         this.scope = component.getScope();
+        this.clusteringMode = component.getClusteringMode();
         this.name = component.getName();
         this.clazz = component.getClazz();
         this.configurationDescriptor = component.getConfigurationDescriptor();
@@ -89,6 +95,7 @@ public class ComponentDescriptorEntity extends BaseSqlEntity<ComponentDescriptor
         data.setCreatedTime(createdTime);
         data.setType(type);
         data.setScope(scope);
+        data.setClusteringMode(clusteringMode);
         data.setName(this.getName());
         data.setClazz(this.getClazz());
         data.setActions(this.getActions());

dao/src/main/java/org/thingsboard/server/dao/model/sql/RuleNodeEntity.java

@@ -64,6 +64,9 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
     @Column(name = ModelConstants.DEBUG_MODE)
     private boolean debugMode;
 
+    @Column(name = ModelConstants.SINGLETON_MODE)
+    private boolean singletonMode;
+
     @Column(name = ModelConstants.EXTERNAL_ID_PROPERTY)
     private UUID externalId;
 
@@ -81,6 +84,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
         this.type = ruleNode.getType();
         this.name = ruleNode.getName();
         this.debugMode = ruleNode.isDebugMode();
+        this.singletonMode = ruleNode.isSingletonMode();
         this.searchText = ruleNode.getName();
         this.configuration = ruleNode.getConfiguration();
         this.additionalInfo = ruleNode.getAdditionalInfo();
@@ -109,6 +113,7 @@ public class RuleNodeEntity extends BaseSqlEntity<RuleNode> implements SearchTex
         ruleNode.setType(type);
         ruleNode.setName(name);
         ruleNode.setDebugMode(debugMode);
+        ruleNode.setSingletonMode(singletonMode);
         ruleNode.setConfiguration(configuration);
         ruleNode.setAdditionalInfo(additionalInfo);
         if (externalId != null) {

dao/src/main/java/org/thingsboard/server/dao/rule/BaseRuleChainService.java

@@ -38,6 +38,7 @@ import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.rule.NodeConnectionInfo;
@@ -50,6 +51,7 @@ import org.thingsboard.server.common.data.rule.RuleChainType;
 import org.thingsboard.server.common.data.rule.RuleChainUpdateResult;
 import org.thingsboard.server.common.data.rule.RuleNode;
 import org.thingsboard.server.common.data.rule.RuleNodeUpdateResult;
+import org.thingsboard.server.common.data.util.ReflectionUtils;
 import org.thingsboard.server.dao.entity.AbstractEntityService;
 import org.thingsboard.server.dao.entity.EntityCountService;
 import org.thingsboard.server.dao.exception.DataValidationException;
@@ -154,6 +156,7 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
         Map<RuleNodeId, Integer> ruleNodeIndexMap = new HashMap<>();
         if (nodes != null) {
             for (RuleNode node : nodes) {
+                setSingletonMode(node);
                 if (node.getId() != null) {
                     ruleNodeIndexMap.put(node.getId(), nodes.indexOf(node));
                 } else {
@@ -783,4 +786,30 @@ public class BaseRuleChainService extends AbstractEntityService implements RuleC
                 }
             };
 
+    private void setSingletonMode(RuleNode ruleNode) {
+        boolean singletonMode;
+        try {
+            ComponentClusteringMode nodeConfigType = ReflectionUtils.getAnnotationProperty(ruleNode.getType(),
+                    "org.thingsboard.rule.engine.api.RuleNode", "clusteringMode");
+
+            switch (nodeConfigType) {
+                case ENABLED:
+                    singletonMode = false;
+                    break;
+                case SINGLETON:
+                    singletonMode = true;
+                    break;
+                case USER_PREFERENCE:
+                default:
+                    singletonMode = ruleNode.isSingletonMode();
+                    break;
+            }
+        } catch (Exception e) {
+            log.warn("Failed to get clustering mode: {}", ExceptionUtils.getRootCauseMessage(e));
+            singletonMode = false;
+        }
+
+        ruleNode.setSingletonMode(singletonMode);
+    }
+
 }

dao/src/main/java/org/thingsboard/server/dao/service/validator/RuleChainDataValidator.java

@@ -30,14 +30,12 @@ import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.common.data.rule.RuleChainMetaData;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 import org.thingsboard.server.common.data.rule.RuleNode;
-import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
 import org.thingsboard.server.common.data.util.ReflectionUtils;
 import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.rule.RuleChainDao;
 import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.dao.service.ConstraintValidator;
 import org.thingsboard.server.dao.service.DataValidator;
-import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
 import org.thingsboard.server.dao.tenant.TenantService;
 
 import java.util.HashMap;
@@ -62,7 +60,7 @@ public class RuleChainDataValidator extends DataValidator<RuleChain> {
 
     @Override
     protected void validateCreate(TenantId tenantId, RuleChain data) {
-         validateNumberOfEntitiesPerTenant(tenantId, EntityType.RULE_CHAIN);
+        validateNumberOfEntitiesPerTenant(tenantId, EntityType.RULE_CHAIN);
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/sql/component/AbstractComponentDescriptorInsertRepository.java

@@ -76,7 +76,8 @@ public abstract class AbstractComponentDescriptorInsertRepository implements Com
                 .setParameter("name", entity.getName())
                 .setParameter("scope", entity.getScope().name())
                 .setParameter("search_text", entity.getSearchText())
-                .setParameter("type", entity.getType().name());
+                .setParameter("type", entity.getType().name())
+                .setParameter("clustering_mode", entity.getClusteringMode().name());
     }
 
     private ComponentDescriptorEntity processSaveOrUpdate(ComponentDescriptorEntity entity, String query) {

dao/src/main/java/org/thingsboard/server/dao/sql/component/SqlComponentDescriptorInsertRepository.java

@@ -44,10 +44,10 @@ public class SqlComponentDescriptorInsertRepository extends AbstractComponentDes
     }
 
     private static String getInsertOrUpdateStatement(String conflictKeyStatement, String updateKeyStatement) {
-        return "INSERT INTO component_descriptor (id, created_time, actions, clazz, configuration_descriptor, name, scope, search_text, type) VALUES (:id, :created_time, :actions, :clazz, :configuration_descriptor, :name, :scope, :search_text, :type) ON CONFLICT " + conflictKeyStatement + " DO UPDATE SET " + updateKeyStatement + " returning *";
+        return "INSERT INTO component_descriptor (id, created_time, actions, clazz, configuration_descriptor, name, scope, search_text, type, clustering_mode) VALUES (:id, :created_time, :actions, :clazz, :configuration_descriptor, :name, :scope, :search_text, :type, :clustering_mode) ON CONFLICT " + conflictKeyStatement + " DO UPDATE SET " + updateKeyStatement + " returning *";
     }
 
     private static String getUpdateStatement(String id) {
-        return "actions = :actions, " + id + ",created_time = :created_time, configuration_descriptor = :configuration_descriptor, name = :name, scope = :scope, search_text = :search_text, type = :type";
+        return "actions = :actions, " + id + ",created_time = :created_time, configuration_descriptor = :configuration_descriptor, name = :name, scope = :scope, search_text = :search_text, type = :type, clustering_mode = :clustering_mode";
     }
 }

dao/src/main/resources/sql/schema-entities.sql

@@ -125,7 +125,8 @@ CREATE TABLE IF NOT EXISTS component_descriptor (
     name varchar(255),
     scope varchar(255),
     search_text varchar(255),
-    type varchar(255)
+    type varchar(255),
+    clustering_mode varchar(255)
 );
 
 CREATE TABLE IF NOT EXISTS customer (
@@ -187,6 +188,7 @@ CREATE TABLE IF NOT EXISTS rule_node (
     type varchar(255),
     name varchar(255),
     debug_mode boolean,
+    singleton_mode boolean,
     search_text varchar(255),
     external_id uuid
 );

rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleNode.java

@@ -16,6 +16,7 @@
 package org.thingsboard.rule.engine.api;
 
 import org.thingsboard.server.common.data.plugin.ComponentScope;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 
@@ -38,6 +39,8 @@ public @interface RuleNode {
 
     Class<? extends NodeConfiguration> configClazz();
 
+    ComponentClusteringMode clusteringMode() default ComponentClusteringMode.ENABLED;
+
     boolean inEnabled() default true;
 
     boolean outEnabled() default true;

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java

@@ -33,6 +33,7 @@ import org.thingsboard.rule.engine.api.util.TbNodeUtils;
 import org.thingsboard.rule.engine.credentials.BasicCredentials;
 import org.thingsboard.rule.engine.credentials.ClientCredentials;
 import org.thingsboard.rule.engine.credentials.CredentialsType;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 import org.thingsboard.server.common.msg.TbMsg;
 import org.thingsboard.server.common.msg.TbMsgMetaData;
@@ -47,6 +48,7 @@ import java.util.concurrent.TimeoutException;
         type = ComponentType.EXTERNAL,
         name = "mqtt",
         configClazz = TbMqttNodeConfiguration.class,
+        clusteringMode = ComponentClusteringMode.USER_PREFERENCE,
         nodeDescription = "Publish messages to the MQTT broker",
         nodeDetails = "Will publish message payload to the MQTT broker with QoS <b>AT_LEAST_ONCE</b>.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/azure/TbAzureIotHubNode.java

@@ -16,9 +16,7 @@
 package org.thingsboard.rule.engine.mqtt.azure;
 
 import io.netty.handler.codec.mqtt.MqttVersion;
-import io.netty.handler.ssl.SslContext;
 import lombok.extern.slf4j.Slf4j;
-import org.thingsboard.server.common.data.StringUtils;
 import org.thingsboard.common.util.AzureIotHubUtil;
 import org.thingsboard.mqtt.MqttClientConfig;
 import org.thingsboard.rule.engine.api.RuleNode;
@@ -26,12 +24,12 @@ import org.thingsboard.rule.engine.api.TbContext;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
-import org.thingsboard.rule.engine.credentials.BasicCredentials;
 import org.thingsboard.rule.engine.credentials.CertPemCredentials;
 import org.thingsboard.rule.engine.credentials.ClientCredentials;
 import org.thingsboard.rule.engine.credentials.CredentialsType;
 import org.thingsboard.rule.engine.mqtt.TbMqttNode;
 import org.thingsboard.rule.engine.mqtt.TbMqttNodeConfiguration;
+import org.thingsboard.server.common.data.plugin.ComponentClusteringMode;
 import org.thingsboard.server.common.data.plugin.ComponentType;
 
 import javax.net.ssl.SSLException;
@@ -41,6 +39,7 @@ import javax.net.ssl.SSLException;
         type = ComponentType.EXTERNAL,
         name = "azure iot hub",
         configClazz = TbAzureIotHubNodeConfiguration.class,
+        clusteringMode = ComponentClusteringMode.SINGLETON,
         nodeDescription = "Publish messages to the Azure IoT Hub",
         nodeDetails = "Will publish message payload to the Azure IoT Hub with QoS <b>AT_LEAST_ONCE</b>.",
         uiResources = {"static/rulenode/rulenode-core-config.js"},

ui-ngx/src/app/modules/home/components/import-export/import-export.service.ts

@@ -467,6 +467,7 @@ export class ImportExportService {
           const ruleChainNode: RuleNode = {
             name: '',
             debugMode: false,
+            singletonMode: false,
             type: 'org.thingsboard.rule.engine.flow.TbRuleChainInputNode',
             configuration: {
               ruleChainId: ruleChainConnection.targetRuleChainId.id

ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.html

@@ -24,7 +24,7 @@
 <form [formGroup]="ruleNodeFormGroup" class="mat-padding">
   <fieldset [disabled]="(isLoading$ | async) || !isEdit || isReadOnly">
     <section>
-      <section fxLayout="column" fxLayout.gt-sm="row">
+      <section class="title-row">
         <mat-form-field fxFlex class="mat-block">
           <mat-label translate>rulenode.name</mat-label>
           <input matInput formControlName="name" required>
@@ -36,9 +36,14 @@
             {{ 'rulenode.name-max-length' | translate }}
           </mat-error>
         </mat-form-field>
-        <mat-checkbox formControlName="debugMode">
-          {{ 'rulenode.debug-mode' | translate }}
-        </mat-checkbox>
+        <section class="node-setting">
+          <mat-slide-toggle formControlName="debugMode">
+            {{ 'rulenode.debug-mode' | translate }}
+          </mat-slide-toggle>
+          <mat-slide-toggle *ngIf="isSingletonEditAllowed()" formControlName="singletonMode">
+            {{ 'rulenode.singleton-mode' | translate }}
+          </mat-slide-toggle >
+        </section>
       </section>
       <tb-rule-node-config #ruleNodeConfigComponent
         formControlName="configuration"

ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.scss

@@ -13,8 +13,30 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+@import './../scss/constants';
+
 :host {
   form {
     overflow-x: hidden !important;
   }
+  .title-row {
+    display: flex;
+    flex-direction: column;
+
+    .node-setting {
+      display: flex;
+      flex-direction: column;
+      gap: 8px;
+      margin-bottom: 5px;
+    }
+
+    @media #{$mat-gt-sm} {
+      flex-direction: row;
+      gap: 8px;
+
+      .node-setting {
+        margin-top: 5px;
+      }
+    }
+  }
 }

ui-ngx/src/app/modules/home/pages/rulechain/rule-node-details.component.ts

@@ -26,6 +26,7 @@ import { RuleChainService } from '@core/http/rule-chain.service';
 import { RuleNodeConfigComponent } from './rule-node-config.component';
 import { Router } from '@angular/router';
 import { RuleChainType } from '@app/shared/models/rule-chain.models';
+import { ComponentClusteringMode } from '@shared/models/component-descriptor.models';
 
 @Component({
   selector: 'tb-rule-node',
@@ -78,6 +79,7 @@ export class RuleNodeDetailsComponent extends PageComponent implements OnInit, O
       this.ruleNodeFormGroup = this.fb.group({
         name: [this.ruleNode.name, [Validators.required, Validators.pattern('(.|\\s)*\\S(.|\\s)*'), Validators.maxLength(255)]],
         debugMode: [this.ruleNode.debugMode, []],
+        singletonMode: [this.ruleNode.singletonMode, []],
         configuration: [this.ruleNode.configuration, [Validators.required]],
         additionalInfo: this.fb.group(
           {
@@ -130,4 +132,8 @@ export class RuleNodeDetailsComponent extends PageComponent implements OnInit, O
       }
     }
   }
+
+  isSingletonEditAllowed() {
+    return this.ruleNode.component.clusteringMode === ComponentClusteringMode.USER_PREFERENCE;
+  }
 }

ui-ngx/src/app/modules/home/pages/rulechain/rulechain-page.component.ts

@@ -89,6 +89,7 @@ import { DebugEventType, EventType } from '@shared/models/event.models';
 import { MatMiniFabButton } from '@angular/material/button';
 import { TbPopoverService } from '@shared/components/popover.service';
 import { VersionControlComponent } from '@home/components/vc/version-control.component';
+import { ComponentClusteringMode } from '@shared/models/component-descriptor.models';
 import Timeout = NodeJS.Timeout;
 
 @Component({
@@ -469,6 +470,7 @@ export class RuleChainPageComponent extends PageComponent
         component: ruleNodeComponent,
         name: '',
         nodeClass: desc.nodeClass,
+        singletonMode: ruleNodeComponent.clusteringMode !== ComponentClusteringMode.ENABLED,
         icon,
         iconUrl,
         x: 30,
@@ -554,6 +556,7 @@ export class RuleChainPageComponent extends PageComponent
         additionalInfo: ruleNode.additionalInfo,
         configuration: ruleNode.configuration,
         debugMode: ruleNode.debugMode,
+        singletonMode: ruleNode.singletonMode,
         x: Math.round(ruleNode.additionalInfo.layoutX),
         y: Math.round(ruleNode.additionalInfo.layoutY),
         component,
@@ -912,7 +915,8 @@ export class RuleChainPageComponent extends PageComponent
             name: node.name,
             configuration: deepClone(node.configuration),
             additionalInfo: node.additionalInfo ? deepClone(node.additionalInfo) : {},
-            debugMode: node.debugMode
+            debugMode: node.debugMode,
+            singletonMode: node.singletonMode
           };
           if (minX === null) {
             minX = node.x;
@@ -983,7 +987,8 @@ export class RuleChainPageComponent extends PageComponent
             name: outputEdge.label,
             configuration: {},
             additionalInfo: {},
-            debugMode: false
+            debugMode: false,
+            singletonMode: false
           };
           outputNode.additionalInfo.layoutX = Math.round(destNode.x);
           outputNode.additionalInfo.layoutY = Math.round(destNode.y);
@@ -1029,6 +1034,7 @@ export class RuleChainPageComponent extends PageComponent
               ruleChainId: ruleChain.id.id
             },
             debugMode: false,
+            singletonMode: false,
             x: Math.round(ruleChainNodeX),
             y: Math.round(ruleChainNodeY),
             nodeClass: descriptor.nodeClass,
@@ -1420,7 +1426,8 @@ export class RuleChainPageComponent extends PageComponent
             name: node.name,
             configuration: node.configuration,
             additionalInfo: node.additionalInfo ? node.additionalInfo : {},
-            debugMode: node.debugMode
+            debugMode: node.debugMode,
+            singletonMode: node.singletonMode
           };
           ruleNode.additionalInfo.layoutX = Math.round(node.x);
           ruleNode.additionalInfo.layoutY = Math.round(node.y);

ui-ngx/src/app/shared/models/component-descriptor.models.ts

@@ -30,9 +30,16 @@ export enum ComponentScope {
   TENANT = 'TENANT'
 }
 
+export enum ComponentClusteringMode {
+  USER_PREFERENCE = 'USER_PREFERENCE',
+  ENABLED = 'ENABLED',
+  SINGLETON = 'SINGLETON'
+}
+
 export interface ComponentDescriptor {
   type: ComponentType | RuleNodeType;
   scope?: ComponentScope;
+  clusteringMode: ComponentClusteringMode;
   name: string;
   clazz: string;
   configurationDescriptor?: any;

ui-ngx/src/app/shared/models/rule-chain.models.ts

@@ -19,7 +19,7 @@ import { TenantId } from '@shared/models/id/tenant-id';
 import { RuleChainId } from '@shared/models/id/rule-chain-id';
 import { RuleNodeId } from '@shared/models/id/rule-node-id';
 import { RuleNode, RuleNodeComponentDescriptor, RuleNodeType } from '@shared/models/rule-node.models';
-import { ComponentType } from '@shared/models/component-descriptor.models';
+import { ComponentClusteringMode, ComponentType } from '@shared/models/component-descriptor.models';
 
 export interface RuleChain extends BaseData<RuleChainId>, ExportableEntity<RuleChainId> {
   tenantId: TenantId;
@@ -64,6 +64,7 @@ export const ruleNodeTypeComponentTypes: ComponentType[] =
 export const unknownNodeComponent: RuleNodeComponentDescriptor = {
   type: RuleNodeType.UNKNOWN,
   name: 'unknown',
+  clusteringMode: ComponentClusteringMode.ENABLED,
   clazz: 'tb.internal.Unknown',
   configurationDescriptor: {
     nodeDefinition: {
@@ -80,6 +81,7 @@ export const unknownNodeComponent: RuleNodeComponentDescriptor = {
 
 export const inputNodeComponent: RuleNodeComponentDescriptor = {
   type: RuleNodeType.INPUT,
+  clusteringMode: ComponentClusteringMode.ENABLED,
   name: 'Input',
   clazz: 'tb.internal.Input'
 };

ui-ngx/src/app/shared/models/rule-node.models.ts

@@ -36,6 +36,7 @@ export interface RuleNode extends BaseData<RuleNodeId> {
   type: string;
   name: string;
   debugMode: boolean;
+  singletonMode: boolean;
   configuration: RuleNodeConfiguration;
   additionalInfo?: any;
 }
@@ -308,6 +309,7 @@ export interface RuleNodeComponentDescriptor extends ComponentDescriptor {
 
 export interface FcRuleNodeType extends FcNode {
   component?: RuleNodeComponentDescriptor;
+  singletonMode?: boolean;
   nodeClass?: string;
   icon?: string;
   iconUrl?: string;

ui-ngx/src/assets/locale/locale.constant-en_US.json

@@ -3304,6 +3304,7 @@
         "deselect-all": "Deselect all",
         "rulenode-details": "Rule node details",
         "debug-mode": "Debug mode",
+        "singleton-mode": "Singleton mode",
         "configuration": "Configuration",
         "link": "Link",
         "link-details": "Rule node link details",