diff --git a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java index c096238a8b..145ae10d06 100644 --- a/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/app/AppActor.java @@ -24,10 +24,9 @@ import akka.actor.Terminated; import com.google.common.collect.BiMap; import com.google.common.collect.HashBiMap; import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor; +import org.thingsboard.server.actors.service.ContextAwareActor; import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.actors.service.DefaultActorService; -import org.thingsboard.server.actors.shared.rulechain.SystemRuleChainManager; import org.thingsboard.server.actors.tenant.TenantActor; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.Tenant; @@ -37,16 +36,14 @@ import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.msg.MsgType; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.aware.TenantAwareMsg; -import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; -import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; -import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.RuleEngineException; import org.thingsboard.server.dao.model.ModelConstants; import org.thingsboard.server.dao.tenant.TenantService; import scala.concurrent.duration.Duration; -public class AppActor extends RuleChainManagerActor { +public class AppActor extends ContextAwareActor { private static final TenantId SYSTEM_TENANT = new TenantId(ModelConstants.NULL_UUID); private final TenantService tenantService; @@ -54,7 +51,7 @@ public class AppActor extends RuleChainManagerActor { private boolean ruleChainsInitialized; private AppActor(ActorSystemContext systemContext) { - super(systemContext, new SystemRuleChainManager(systemContext)); + super(systemContext); this.tenantService = systemContext.getTenantService(); this.tenantActors = HashBiMap.create(); } @@ -80,9 +77,6 @@ public class AppActor extends RuleChainManagerActor { switch (msg.getMsgType()) { case APP_INIT_MSG: break; - case SEND_TO_CLUSTER_MSG: - onPossibleClusterMsg((SendToClusterMsg) msg); - break; case PARTITION_CHANGE_MSG: broadcast(msg); break; @@ -98,7 +92,6 @@ public class AppActor extends RuleChainManagerActor { case DEVICE_NAME_OR_TYPE_UPDATE_TO_DEVICE_ACTOR_MSG: case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG: case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG: - case REMOTE_TO_RULE_CHAIN_TELL_NEXT_MSG: onToDeviceActorMsg((TenantAwareMsg) msg); break; default: @@ -110,7 +103,6 @@ public class AppActor extends RuleChainManagerActor { private void initRuleChainsAndTenantActors() { log.info("Starting main system actor."); try { - initRuleChains(); if (systemContext.isTenantComponentsInitEnabled()) { PageDataIterable tenantIterator = new PageDataIterable<>(tenantService::findTenants, ENTITY_PACK_LIMIT); for (Tenant tenant : tenantIterator) { @@ -125,37 +117,22 @@ public class AppActor extends RuleChainManagerActor { } } - private void onPossibleClusterMsg(SendToClusterMsg msg) { - //TODO 2.5 -// Optional address = systemContext.getRoutingService().resolveById(msg.getEntityId()); -// if (address.isPresent()) { - -// systemContext.getRpcService().tell( -// systemContext.getEncodingService().convertToProtoDataMessage(address.get(), msg.getMsg())); -// } else { - self().tell(msg.getMsg(), ActorRef.noSender()); -// } - } - private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) { if (SYSTEM_TENANT.equals(msg.getTenantId())) { -// this may be a notification about system entities created. -// log.warn("[{}] Invalid service to rule engine msg called. System messages are not supported yet: {}", SYSTEM_TENANT, msg); + msg.getTbMsg().getCallback().onFailure(new RuleEngineException("Message has system tenant id!")); } else { getOrCreateTenantActor(msg.getTenantId()).tell(msg, self()); } } - @Override protected void broadcast(Object msg) { - super.broadcast(msg); tenantActors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender())); } private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) { ActorRef target = null; if (SYSTEM_TENANT.equals(msg.getTenantId())) { - target = getEntityActorRef(msg.getEntityId()); + log.warn("Message has system tenant id: {}", msg); } else { if (msg.getEntityId().getEntityType() == EntityType.TENANT && msg.getEvent() == ComponentLifecycleEvent.DELETED) { diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 307797f864..83d269ef6d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -56,7 +56,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMs import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg; import org.thingsboard.server.service.rpc.ToServerRpcResponseActorMsg; @@ -213,7 +213,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { void process(ActorContext context, TransportToDeviceActorMsgWrapper wrapper) { TransportToDeviceActorMsg msg = wrapper.getMsg(); - TbMsgCallback callback = wrapper.getCallback(); + TbCallback callback = wrapper.getCallback(); if (msg.hasSessionEvent()) { processSessionStateMsgs(msg.getSessionInfo(), msg.getSessionEvent()); } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 7178b94dfa..0e8033fac2 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -110,7 +110,7 @@ class DefaultTbContext implements TbContext { if (nodeCtx.getSelf().isDebugMode()) { relationTypes.forEach(relationType -> mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, relationType, th)); } - nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationTypes, msg), nodeCtx.getSelfActor()); + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), relationTypes, msg, th != null ? th.getMessage() : null), nodeCtx.getSelfActor()); } @Override @@ -139,53 +139,61 @@ class DefaultTbContext implements TbContext { mainCtx.getProducerProvider().getRuleEngineMsgProducer().send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), new SimpleTbQueueCallback(onSuccess, onFailure)); } + @Override + public void enqueueForTellFailure(TbMsg tbMsg, String failureMessage) { + TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); + enqueueForTellNext(tpi, tbMsg, Collections.singleton(TbRelationTypes.FAILURE), failureMessage, null, null); + } + @Override public void enqueueForTellNext(TbMsg tbMsg, String relationType) { TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, null); + enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, null, null); } @Override public void enqueueForTellNext(TbMsg tbMsg, Set relationTypes) { TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, relationTypes, null, null); + enqueueForTellNext(tpi, tbMsg, relationTypes, null, null, null); } @Override public void enqueueForTellNext(TbMsg tbMsg, String relationType, Runnable onSuccess, Consumer onFailure) { TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), onSuccess, onFailure); + enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, Set relationTypes, Runnable onSuccess, Consumer onFailure) { TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, relationTypes, onSuccess, onFailure); + enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, String relationType, Runnable onSuccess, Consumer onFailure) { - TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), onSuccess, onFailure); + TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); + enqueueForTellNext(tpi, tbMsg, Collections.singleton(relationType), null, onSuccess, onFailure); } @Override public void enqueueForTellNext(TbMsg tbMsg, String queueName, Set relationTypes, Runnable onSuccess, Consumer onFailure) { TopicPartitionInfo tpi = mainCtx.resolve(ServiceType.TB_RULE_ENGINE, queueName, getTenantId(), tbMsg.getOriginator()); - enqueueForTellNext(tpi, tbMsg, relationTypes, onSuccess, onFailure); + enqueueForTellNext(tpi, tbMsg, relationTypes, null, onSuccess, onFailure); } - private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg tbMsg, Set relationTypes, Runnable onSuccess, Consumer onFailure) { + private void enqueueForTellNext(TopicPartitionInfo tpi, TbMsg tbMsg, Set relationTypes, String failureMessage, Runnable onSuccess, Consumer onFailure) { RuleChainId ruleChainId = nodeCtx.getSelf().getRuleChainId(); RuleNodeId ruleNodeId = nodeCtx.getSelf().getId(); tbMsg = TbMsg.newMsg(tbMsg, ruleChainId, ruleNodeId); - TransportProtos.ToRuleEngineMsg msg = TransportProtos.ToRuleEngineMsg.newBuilder() + TransportProtos.ToRuleEngineMsg.Builder msg = TransportProtos.ToRuleEngineMsg.newBuilder() .setTenantIdMSB(getTenantId().getId().getMostSignificantBits()) .setTenantIdLSB(getTenantId().getId().getLeastSignificantBits()) .setTbMsg(TbMsg.toByteString(tbMsg)) - .addAllRelationTypes(relationTypes) - .build(); - mainCtx.getProducerProvider().getRuleEngineMsgProducer().send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), new SimpleTbQueueCallback(onSuccess, onFailure)); + .addAllRelationTypes(relationTypes); + if (failureMessage != null) { + msg.setFailureMessage(failureMessage); + } + mainCtx.getProducerProvider().getRuleEngineMsgProducer().send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg.build()), new SimpleTbQueueCallback(onSuccess, onFailure)); } @Override @@ -207,7 +215,8 @@ class DefaultTbContext implements TbContext { if (nodeCtx.getSelf().isDebugMode()) { mainCtx.persistDebugOutput(nodeCtx.getTenantId(), nodeCtx.getSelf().getId(), msg, TbRelationTypes.FAILURE, th); } - nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE), msg), nodeCtx.getSelfActor()); + nodeCtx.getChainActor().tell(new RuleNodeToRuleChainTellNextMsg(nodeCtx.getSelf().getId(), Collections.singleton(TbRelationTypes.FAILURE), + msg, th != null ? th.getMessage() : null), nodeCtx.getSelfActor()); } public void updateSelf(RuleNode self) { diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RemoteToRuleChainTellNextMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RemoteToRuleChainTellNextMsg.java deleted file mode 100644 index d062c5dc49..0000000000 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RemoteToRuleChainTellNextMsg.java +++ /dev/null @@ -1,48 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.actors.ruleChain; - -import lombok.Data; -import org.thingsboard.server.common.data.id.RuleChainId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg; -import org.thingsboard.server.common.msg.aware.TenantAwareMsg; - -import java.io.Serializable; - -/** - * Created by ashvayka on 19.03.18. - */ -@Data -final class RemoteToRuleChainTellNextMsg extends RuleNodeToRuleChainTellNextMsg implements TenantAwareMsg, RuleChainAwareMsg { - - private static final long serialVersionUID = 2459605482321657447L; - private final TenantId tenantId; - private final RuleChainId ruleChainId; - - public RemoteToRuleChainTellNextMsg(RuleNodeToRuleChainTellNextMsg original, TenantId tenantId, RuleChainId ruleChainId) { - super(original.getOriginator(), original.getRelationTypes(), original.getMsg()); - this.tenantId = tenantId; - this.ruleChainId = ruleChainId; - } - - @Override - public MsgType getMsgType() { - return MsgType.REMOTE_TO_RULE_CHAIN_TELL_NEXT_MSG; - } - -} diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java index 015e7a008b..027f4aac9d 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainActor.java @@ -22,6 +22,7 @@ import org.thingsboard.server.actors.service.ComponentActor; import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.common.data.id.RuleChainId; import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; @@ -30,9 +31,9 @@ import scala.concurrent.duration.Duration; public class RuleChainActor extends ComponentActor { - private RuleChainActor(ActorSystemContext systemContext, TenantId tenantId, RuleChainId ruleChainId) { - super(systemContext, tenantId, ruleChainId); - setProcessor(new RuleChainActorMessageProcessor(tenantId, ruleChainId, systemContext, + private RuleChainActor(ActorSystemContext systemContext, TenantId tenantId, RuleChain ruleChain) { + super(systemContext, tenantId, ruleChain.getId()); + setProcessor(new RuleChainActorMessageProcessor(tenantId, ruleChain, systemContext, context().parent(), context().self())); } @@ -46,7 +47,6 @@ public class RuleChainActor extends ComponentActor> nodeRoutes; private final RuleChainService service; private final TbQueueProducer> producer; + private String ruleChainName; private RuleNodeId firstId; private RuleNodeCtx firstNode; private boolean started; - RuleChainActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, ActorSystemContext systemContext + RuleChainActorMessageProcessor(TenantId tenantId, RuleChain ruleChain, ActorSystemContext systemContext , ActorRef parent, ActorRef self) { - super(systemContext, tenantId, ruleChainId); + super(systemContext, tenantId, ruleChain.getId()); + this.ruleChainName = ruleChain.getName(); this.parent = parent; this.self = self; this.nodeActors = new HashMap<>(); @@ -113,6 +117,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor ruleNodeList = service.getRuleChainNodes(tenantId, entityId); log.trace("[{}][{}] Updating rule chain with {} nodes", tenantId, entityId, ruleNodeList.size()); for (RuleNode ruleNode : ruleNodeList) { @@ -194,7 +199,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor relationTypes) { + private void onTellNext(TbMsg msg, RuleNodeId originatorNodeId, Set relationTypes, String failureMessage) { try { checkActive(); EntityId entityId = msg.getOriginator(); @@ -245,9 +250,14 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor actors; + @Getter + protected RuleChain rootChain; + @Getter + protected ActorRef rootChainActor; - public RuleChainManagerActor(ActorSystemContext systemContext, RuleChainManager ruleChainManager) { + public RuleChainManagerActor(ActorSystemContext systemContext, TenantId tenantId) { super(systemContext); - this.ruleChainManager = ruleChainManager; + this.tenantId = tenantId; + this.actors = HashBiMap.create(); this.ruleChainService = systemContext.getRuleChainService(); } protected void initRuleChains() { - ruleChainManager.init(this.context()); + for (RuleChain ruleChain : new PageDataIterable<>(link -> ruleChainService.findTenantRuleChains(tenantId, link), ContextAwareActor.ENTITY_PACK_LIMIT)) { + RuleChainId ruleChainId = ruleChain.getId(); + log.debug("[{}|{}] Creating rule chain actor", ruleChainId.getEntityType(), ruleChain.getId()); + //TODO: remove this cast making UUIDBased subclass of EntityId an interface and vice versa. + ActorRef actorRef = getOrCreateActor(this.context(), ruleChainId, id -> ruleChain); + visit(ruleChain, actorRef); + log.debug("[{}|{}] Rule Chain actor created.", ruleChainId.getEntityType(), ruleChainId.getId()); + } + } + + protected void visit(RuleChain entity, ActorRef actorRef) { + if (entity != null && entity.isRoot()) { + rootChain = entity; + rootChainActor = actorRef; + } + } + + public ActorRef getOrCreateActor(ActorContext context, RuleChainId ruleChainId) { + return getOrCreateActor(context, ruleChainId, eId -> ruleChainService.findRuleChainById(TenantId.SYS_TENANT_ID, eId)); + } + + public ActorRef getOrCreateActor(ActorContext context, RuleChainId ruleChainId, Function provider) { + return actors.computeIfAbsent(ruleChainId, eId -> { + RuleChain ruleChain = provider.apply(eId); + return context.actorOf(Props.create(new RuleChainActor.ActorCreator(systemContext, tenantId, ruleChain)) + .withDispatcher(DefaultActorService.TENANT_RULE_DISPATCHER_NAME), eId.toString()); + }); } protected ActorRef getEntityActorRef(EntityId entityId) { ActorRef target = null; - switch (entityId.getEntityType()) { - case RULE_CHAIN: - target = ruleChainManager.getOrCreateActor(this.context(), (RuleChainId) entityId); - break; + if (entityId.getEntityType() == EntityType.RULE_CHAIN) { + target = getOrCreateActor(this.context(), (RuleChainId) entityId); } return target; } protected void broadcast(Object msg) { - ruleChainManager.broadcast(msg); + actors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender())); } + + public ActorRef get(RuleChainId id) { + return actors.get(id); + } + } diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToRuleChainTellNextMsg.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToRuleChainTellNextMsg.java index 1fda91caf0..2bccc654e0 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToRuleChainTellNextMsg.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleNodeToRuleChainTellNextMsg.java @@ -34,6 +34,7 @@ class RuleNodeToRuleChainTellNextMsg implements TbActorMsg, Serializable { private final RuleNodeId originator; private final Set relationTypes; private final TbMsg msg; + private final String failureMessage; @Override public MsgType getMsgType() { diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/EntityActorsManager.java b/application/src/main/java/org/thingsboard/server/actors/shared/EntityActorsManager.java deleted file mode 100644 index e45f69ca9e..0000000000 --- a/application/src/main/java/org/thingsboard/server/actors/shared/EntityActorsManager.java +++ /dev/null @@ -1,93 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.actors.shared; - -import akka.actor.ActorContext; -import akka.actor.ActorRef; -import akka.actor.Props; -import akka.actor.UntypedActor; -import akka.japi.Creator; -import com.google.common.collect.BiMap; -import com.google.common.collect.HashBiMap; -import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.ruleChain.RuleChainActor; -import org.thingsboard.server.actors.service.ContextAwareActor; -import org.thingsboard.server.common.data.SearchTextBased; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.RuleChainId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.id.UUIDBased; -import org.thingsboard.server.common.data.page.PageDataIterable; - -import java.util.HashMap; -import java.util.Map; - -/** - * Created by ashvayka on 15.03.18. - */ -@Slf4j -public abstract class EntityActorsManager> { - - protected final ActorSystemContext systemContext; - protected final BiMap actors; - - public EntityActorsManager(ActorSystemContext systemContext) { - this.systemContext = systemContext; - this.actors = HashBiMap.create(); - } - - protected abstract TenantId getTenantId(); - - protected abstract String getDispatcherName(); - - protected abstract Creator creator(T entityId); - - protected abstract PageDataIterable.FetchFunction getFetchEntitiesFunction(); - - public void init(ActorContext context) { - for (M entity : new PageDataIterable<>(getFetchEntitiesFunction(), ContextAwareActor.ENTITY_PACK_LIMIT)) { - T entityId = (T) entity.getId(); - log.debug("[{}|{}] Creating entity actor", entityId.getEntityType(), entityId.getId()); - //TODO: remove this cast making UUIDBased subclass of EntityId an interface and vice versa. - ActorRef actorRef = getOrCreateActor(context, entityId); - visit(entity, actorRef); - log.debug("[{}|{}] Entity actor created.", entityId.getEntityType(), entityId.getId()); - } - } - - public void visit(M entity, ActorRef actorRef) { - } - - public ActorRef getOrCreateActor(ActorContext context, T entityId) { - return actors.computeIfAbsent(entityId, eId -> - context.actorOf(Props.create(creator(eId)) - .withDispatcher(getDispatcherName()), eId.toString())); - } - - public void broadcast(Object msg) { - actors.values().forEach(actorRef -> actorRef.tell(msg, ActorRef.noSender())); - } - - public void remove(T id) { - actors.remove(id); - } - - public ActorRef get(T id) { - return actors.get(id); - } - -} diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/RuleChainManager.java b/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/RuleChainManager.java deleted file mode 100644 index ef2b4282d4..0000000000 --- a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/RuleChainManager.java +++ /dev/null @@ -1,59 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.actors.shared.rulechain; - -import akka.actor.ActorRef; -import akka.japi.Creator; -import lombok.Getter; -import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.ruleChain.RuleChainActor; -import org.thingsboard.server.actors.shared.EntityActorsManager; -import org.thingsboard.server.common.data.id.RuleChainId; -import org.thingsboard.server.common.data.rule.RuleChain; -import org.thingsboard.server.dao.rule.RuleChainService; - -/** - * Created by ashvayka on 15.03.18. - */ -@Slf4j -public abstract class RuleChainManager extends EntityActorsManager { - - protected final RuleChainService service; - @Getter - protected RuleChain rootChain; - @Getter - protected ActorRef rootChainActor; - - public RuleChainManager(ActorSystemContext systemContext) { - super(systemContext); - this.service = systemContext.getRuleChainService(); - } - - @Override - public Creator creator(RuleChainId entityId) { - return new RuleChainActor.ActorCreator(systemContext, getTenantId(), entityId); - } - - @Override - public void visit(RuleChain entity, ActorRef actorRef) { - if (entity != null && entity.isRoot()) { - rootChain = entity; - rootChainActor = actorRef; - } - } - -} diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/SystemRuleChainManager.java b/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/SystemRuleChainManager.java deleted file mode 100644 index f2a80a9e82..0000000000 --- a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/SystemRuleChainManager.java +++ /dev/null @@ -1,48 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.actors.shared.rulechain; - -import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.service.DefaultActorService; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.page.PageDataIterable.FetchFunction; -import org.thingsboard.server.common.data.page.TextPageData; -import org.thingsboard.server.common.data.rule.RuleChain; -import org.thingsboard.server.dao.model.ModelConstants; - -import java.util.Collections; - -public class SystemRuleChainManager extends RuleChainManager { - - public SystemRuleChainManager(ActorSystemContext systemContext) { - super(systemContext); - } - - @Override - protected FetchFunction getFetchEntitiesFunction() { - return link -> new TextPageData<>(Collections.emptyList(), link); - } - - @Override - protected TenantId getTenantId() { - return ModelConstants.SYSTEM_TENANT; - } - - @Override - protected String getDispatcherName() { - return DefaultActorService.SYSTEM_RULE_DISPATCHER_NAME; - } -} diff --git a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/TenantRuleChainManager.java b/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/TenantRuleChainManager.java deleted file mode 100644 index cfe862af49..0000000000 --- a/application/src/main/java/org/thingsboard/server/actors/shared/rulechain/TenantRuleChainManager.java +++ /dev/null @@ -1,53 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.actors.shared.rulechain; - -import akka.actor.ActorContext; -import org.thingsboard.server.actors.ActorSystemContext; -import org.thingsboard.server.actors.service.DefaultActorService; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.page.PageDataIterable.FetchFunction; -import org.thingsboard.server.common.data.rule.RuleChain; - -public class TenantRuleChainManager extends RuleChainManager { - - private final TenantId tenantId; - - public TenantRuleChainManager(ActorSystemContext systemContext, TenantId tenantId) { - super(systemContext); - this.tenantId = tenantId; - } - - @Override - public void init(ActorContext context) { - super.init(context); - } - - @Override - protected TenantId getTenantId() { - return tenantId; - } - - @Override - protected String getDispatcherName() { - return DefaultActorService.TENANT_RULE_DISPATCHER_NAME; - } - - @Override - protected FetchFunction getFetchEntitiesFunction() { - return link -> service.findTenantRuleChains(tenantId, link); - } -} diff --git a/application/src/main/java/org/thingsboard/server/actors/stats/StatsActor.java b/application/src/main/java/org/thingsboard/server/actors/stats/StatsActor.java index 1ef33467b2..717dcecc19 100644 --- a/application/src/main/java/org/thingsboard/server/actors/stats/StatsActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/stats/StatsActor.java @@ -24,7 +24,6 @@ import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.Event; import org.thingsboard.server.common.msg.TbActorMsg; -import org.thingsboard.server.common.msg.cluster.ServerAddress; @Slf4j public class StatsActor extends ContextAwareActor { diff --git a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java index 3a719ca857..e54b955419 100644 --- a/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java @@ -30,7 +30,6 @@ import org.thingsboard.server.actors.device.DeviceActorCreator; import org.thingsboard.server.actors.ruleChain.RuleChainManagerActor; import org.thingsboard.server.actors.service.ContextBasedCreator; import org.thingsboard.server.actors.service.DefaultActorService; -import org.thingsboard.server.actors.shared.rulechain.TenantRuleChainManager; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.RuleChainId; @@ -43,6 +42,7 @@ import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg; import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; +import org.thingsboard.server.common.msg.queue.RuleEngineException; import org.thingsboard.server.common.msg.queue.ServiceType; import scala.concurrent.duration.Duration; @@ -51,12 +51,12 @@ import java.util.stream.Collectors; public class TenantActor extends RuleChainManagerActor { - private final TenantId tenantId; private final BiMap deviceActors; + private boolean isRuleEngine; + private boolean isCore; private TenantActor(ActorSystemContext systemContext, TenantId tenantId) { - super(systemContext, new TenantRuleChainManager(systemContext, tenantId)); - this.tenantId = tenantId; + super(systemContext, tenantId); this.deviceActors = HashBiMap.create(); } @@ -69,7 +69,11 @@ public class TenantActor extends RuleChainManagerActor { public void preStart() { log.info("[{}] Starting tenant actor.", tenantId); try { - initRuleChains(); + isRuleEngine = systemContext.getServiceInfoProvider().isService(ServiceType.TB_RULE_ENGINE); + isCore = systemContext.getServiceInfoProvider().isService(ServiceType.TB_CORE); + if (isRuleEngine) { + initRuleChains(); + } log.info("[{}] Tenant actor started.", tenantId); } catch (Exception e) { log.warn("[{}] Unknown failure", tenantId, e); @@ -115,7 +119,6 @@ public class TenantActor extends RuleChainManagerActor { onToDeviceActorMsg((DeviceAwareMsg) msg); break; case RULE_CHAIN_TO_RULE_CHAIN_MSG: - case REMOTE_TO_RULE_CHAIN_TELL_NEXT_MSG: onRuleChainMsg((RuleChainAwareMsg) msg); break; default: @@ -129,16 +132,19 @@ public class TenantActor extends RuleChainManagerActor { } private void onQueueToRuleEngineMsg(QueueToRuleEngineMsg msg) { + if (!isRuleEngine) { + log.warn("RECEIVED INVALID MESSAGE: {}", msg); + } TbMsg tbMsg = msg.getTbMsg(); if (tbMsg.getRuleChainId() == null) { - if (ruleChainManager.getRootChainActor() != null) { - ruleChainManager.getRootChainActor().tell(msg, self()); + if (getRootChainActor() != null) { + getRootChainActor().tell(msg, self()); } else { - tbMsg.getCallback().onFailure(new RuntimeException("No Root Rule Chain available!")); + tbMsg.getCallback().onFailure(new RuleEngineException("No Root Rule Chain available!")); log.info("[{}] No Root Chain: {}", tenantId, msg); } } else { - ActorRef ruleChainActor = ruleChainManager.get(tbMsg.getRuleChainId()); + ActorRef ruleChainActor = get(tbMsg.getRuleChainId()); if (ruleChainActor != null) { ruleChainActor.tell(msg, self()); } else { @@ -150,24 +156,29 @@ public class TenantActor extends RuleChainManagerActor { } private void onRuleChainMsg(RuleChainAwareMsg msg) { - ruleChainManager.getOrCreateActor(context(), msg.getRuleChainId()).tell(msg, self()); + getOrCreateActor(context(), msg.getRuleChainId()).tell(msg, self()); } private void onToDeviceActorMsg(DeviceAwareMsg msg) { + if (!isCore) { + log.warn("RECEIVED INVALID MESSAGE: {}", msg); + } getOrCreateDeviceActor(msg.getDeviceId()).tell(msg, ActorRef.noSender()); } private void onComponentLifecycleMsg(ComponentLifecycleMsg msg) { - ActorRef target = getEntityActorRef(msg.getEntityId()); - if (target != null) { - if (msg.getEntityId().getEntityType() == EntityType.RULE_CHAIN) { - RuleChain ruleChain = systemContext.getRuleChainService(). - findRuleChainById(tenantId, new RuleChainId(msg.getEntityId().getId())); - ruleChainManager.visit(ruleChain, target); + if (isRuleEngine) { + ActorRef target = getEntityActorRef(msg.getEntityId()); + if (target != null) { + if (msg.getEntityId().getEntityType() == EntityType.RULE_CHAIN) { + RuleChain ruleChain = systemContext.getRuleChainService(). + findRuleChainById(tenantId, new RuleChainId(msg.getEntityId().getId())); + visit(ruleChain, target); + } + target.tell(msg, ActorRef.noSender()); + } else { + log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg); } - target.tell(msg, ActorRef.noSender()); - } else { - log.debug("[{}] Invalid component lifecycle msg: {}", tenantId, msg); } } @@ -214,15 +225,12 @@ public class TenantActor extends RuleChainManagerActor { } } - private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), new Function() { - @Override - public SupervisorStrategy.Directive apply(Throwable t) { - log.warn("[{}] Unknown failure", tenantId, t); - if (t instanceof ActorInitializationException) { - return SupervisorStrategy.stop(); - } else { - return SupervisorStrategy.resume(); - } + private final SupervisorStrategy strategy = new OneForOneStrategy(3, Duration.create("1 minute"), t -> { + log.warn("[{}] Unknown failure", tenantId, t); + if (t instanceof ActorInitializationException) { + return SupervisorStrategy.stop(); + } else { + return SupervisorStrategy.resume(); } }); diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index f7d956e146..6e33133522 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -92,10 +92,10 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr try { SessionMetaData sessionMd = internalSessionMap.get(session.getId()); if (sessionMd != null) { - log.info("[{}][{}] Processing {}", sessionMd.sessionRef.getSecurityCtx().getTenantId(), session.getId(), message.getPayload()); + log.trace("[{}][{}] Processing {}", sessionMd.sessionRef.getSecurityCtx().getTenantId(), session.getId(), message.getPayload()); webSocketService.handleWebSocketMsg(sessionMd.sessionRef, message.getPayload()); } else { - log.warn("[{}] Failed to find session", session.getId()); + log.trace("[{}] Failed to find session", session.getId()); session.close(CloseStatus.SERVER_ERROR.withReason("Session not found!")); } } catch (IOException e) { @@ -139,7 +139,7 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr if (sessionMd != null) { processInWebSocketService(sessionMd.sessionRef, SessionEvent.onError(tError)); } else { - log.warn("[{}] Failed to find session", session.getId()); + log.trace("[{}] Failed to find session", session.getId()); } log.trace("[{}] Session transport error", session.getId(), tError); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 434f1d7eae..401262a243 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -25,7 +25,7 @@ import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.gen.transport.TransportProtos.DeviceStateServiceMsgProto; import org.thingsboard.server.gen.transport.TransportProtos.FromDeviceRPCResponseProto; import org.thingsboard.server.gen.transport.TransportProtos.LocalSubscriptionServiceMsgProto; @@ -120,8 +120,8 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService> failedMap = new ConcurrentHashMap<>(); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); pendingMap.forEach((id, msg) -> { - log.info("[{}] Creating main callback for message: {}", id, msg.getValue()); - TbMsgCallback callback = new MsgPackCallback<>(id, processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>(), failedMap); + log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, processingTimeoutLatch, pendingMap, failedMap); try { ToCoreMsg toCoreMsg = msg.getValue(); if (toCoreMsg.hasToSubscriptionMgrMsg()) { @@ -182,7 +182,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService msg, TbMsgCallback callback) { + protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) { ToCoreNotificationMsg toCoreMsg = msg.getValue(); if (toCoreMsg.hasToLocalSubscriptionServiceMsg()) { log.trace("[{}] Forwarding message to local subscription service {}", id, toCoreMsg.getToLocalSubscriptionServiceMsg()); @@ -200,7 +200,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService 0 ? RpcError.values()[proto.getError()] : null; FromDeviceRpcResponse response = new FromDeviceRpcResponse(new UUID(proto.getRequestIdMSB(), proto.getRequestIdLSB()) , proto.getResponse(), error); @@ -215,7 +215,7 @@ public class DefaultTbCoreConsumerService extends AbstractConsumerService>> consumers = new ConcurrentHashMap<>(); private final ConcurrentMap consumerConfigurations = new ConcurrentHashMap<>(); + private final ConcurrentMap consumerStats = new ConcurrentHashMap<>(); public DefaultTbRuleEngineConsumerService(TbRuleEngineProcessingStrategyFactory factory, TbQueueRuleEngineSettings ruleEngineSettings, - TbRuleEngineQueueFactory tbRuleEngineQueueFactory, + TbRuleEngineQueueFactory tbRuleEngineQueueFactory, RuleEngineStatisticsService statisticsService, ActorSystemContext actorContext, DataDecodingEncodingService encodingService) { super(actorContext, encodingService, tbRuleEngineQueueFactory.createToRuleEngineNotificationsMsgConsumer()); + this.statisticsService = statisticsService; this.ruleEngineSettings = ruleEngineSettings; this.tbRuleEngineQueueFactory = tbRuleEngineQueueFactory; this.factory = factory; @@ -92,7 +98,9 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< public void init() { super.init("tb-rule-engine-consumer", "tb-rule-engine-notifications-consumer"); for (TbRuleEngineQueueConfiguration configuration : ruleEngineSettings.getQueues()) { + consumerConfigurations.putIfAbsent(configuration.getName(), configuration); consumers.computeIfAbsent(configuration.getName(), queueName -> tbRuleEngineQueueFactory.createToRuleEngineMsgConsumer(configuration)); + consumerStats.put(configuration.getName(), new TbRuleEngineConsumerStats(configuration.getName())); } } @@ -107,7 +115,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< @Override protected void launchMainConsumers() { - consumers.forEach((queue, consumer) -> launchConsumer(consumer, consumerConfigurations.get(queue))); + consumers.forEach((queue, consumer) -> launchConsumer(consumer, consumerConfigurations.get(queue), consumerStats.get(queue))); } @Override @@ -115,7 +123,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< consumers.values().forEach(TbQueueConsumer::unsubscribe); } - private void launchConsumer(TbQueueConsumer> consumer, TbRuleEngineQueueConfiguration configuration) { + private void launchConsumer(TbQueueConsumer> consumer, TbRuleEngineQueueConfiguration configuration, TbRuleEngineConsumerStats stats) { consumersExecutor.execute(() -> { while (!stopped) { try { @@ -123,7 +131,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< if (msgs.isEmpty()) { continue; } - TbRuleEngineProcessingStrategy strategy = factory.newInstance(configuration.getAckStrategy()); + TbRuleEngineProcessingStrategy strategy = factory.newInstance(configuration.getName(), configuration.getAckStrategy()); TbRuleEngineProcessingDecision decision = null; boolean firstAttempt = true; while (!stopped && (firstAttempt || !decision.isCommit())) { @@ -137,21 +145,21 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } ConcurrentMap> successMap = new ConcurrentHashMap<>(); ConcurrentMap> failedMap = new ConcurrentHashMap<>(); - + ConcurrentMap exceptionsMap = new ConcurrentHashMap<>(); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); allMap.forEach((id, msg) -> { - log.info("[{}] Creating main callback for message: {}", id, msg.getValue()); - TbMsgCallback callback = new MsgPackCallback<>(id, processingTimeoutLatch, allMap, successMap, failedMap); + log.trace("[{}] Creating main callback for message: {}", id, msg.getValue()); + ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); + TenantId tenantId = new TenantId(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); + TbMsgCallback callback = new TbMsgPackCallback<>(id, tenantId, processingTimeoutLatch, allMap, successMap, failedMap, exceptionsMap); try { - ToRuleEngineMsg toRuleEngineMsg = msg.getValue(); - TenantId tenantId = new TenantId(new UUID(toRuleEngineMsg.getTenantIdMSB(), toRuleEngineMsg.getTenantIdLSB())); if (toRuleEngineMsg.getTbMsg() != null && !toRuleEngineMsg.getTbMsg().isEmpty()) { forwardToRuleEngineActor(tenantId, toRuleEngineMsg, callback); } else { callback.onSuccess(); } - } catch (Throwable e) { - callback.onFailure(e); + } catch (Exception e) { + callback.onFailure(new RuleEngineException(e.getMessage())); } }); @@ -159,7 +167,11 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< if (!processingTimeoutLatch.await(packProcessingTimeout, TimeUnit.MILLISECONDS)) { timeout = true; } - decision = strategy.analyze(new TbRuleEngineProcessingResult(timeout, allMap, successMap, failedMap)); + TbRuleEngineProcessingResult result = new TbRuleEngineProcessingResult(timeout, allMap, successMap, failedMap, exceptionsMap); + decision = strategy.analyze(result); + if (statsEnabled) { + stats.log(result, decision.isCommit()); + } } consumer.commit(); } catch (Exception e) { @@ -193,7 +205,7 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< } @Override - protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbMsgCallback callback) throws Exception { + protected void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) throws Exception { ToRuleEngineNotificationMsg nfMsg = msg.getValue(); if (nfMsg.getComponentLifecycleMsg() != null && !nfMsg.getComponentLifecycleMsg().isEmpty()) { Optional actorMsg = encodingService.decode(nfMsg.getComponentLifecycleMsg().toByteArray()); @@ -219,18 +231,18 @@ public class DefaultTbRuleEngineConsumerService extends AbstractConsumerService< relationTypes = new HashSet<>(relationTypesList); } } - msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes); + msg = new QueueToRuleEngineMsg(tenantId, tbMsg, relationTypes, toRuleEngineMsg.getFailureMessage()); actorContext.getAppActor().tell(msg, ActorRef.noSender()); - //TODO: 2.5 before release. -// if (statsEnabled) { -// stats.log(toDeviceActorMsg); -// } } @Scheduled(fixedDelayString = "${queue.rule-engine.stats.print-interval-ms}") public void printStats() { if (statsEnabled) { - stats.printStats(); + long ts = System.currentTimeMillis(); + consumerStats.forEach((queue, stats) -> { + stats.printStats(); + statisticsService.reportQueueStats(ts, stats); + }); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java new file mode 100644 index 0000000000..0470db4cd7 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgPackCallback.java @@ -0,0 +1,78 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.RuleNodeException; +import org.thingsboard.server.common.msg.queue.TbCallback; +import org.thingsboard.server.common.msg.queue.TbMsgCallback; + +import java.util.UUID; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; + +@Slf4j +public class TbMsgPackCallback implements TbMsgCallback { + private final CountDownLatch processingTimeoutLatch; + private final ConcurrentMap ackMap; + private final ConcurrentMap successMap; + private final ConcurrentMap failedMap; + private final UUID id; + private final TenantId tenantId; + private final ConcurrentMap firstExceptions; + + public TbMsgPackCallback(UUID id, TenantId tenantId, + CountDownLatch processingTimeoutLatch, + ConcurrentMap ackMap, + ConcurrentMap successMap, + ConcurrentMap failedMap, + ConcurrentMap firstExceptions) { + this.id = id; + this.tenantId = tenantId; + this.processingTimeoutLatch = processingTimeoutLatch; + this.ackMap = ackMap; + this.successMap = successMap; + this.failedMap = failedMap; + this.firstExceptions = firstExceptions; + } + + @Override + public void onSuccess() { + log.trace("[{}] ON SUCCESS", id); + T msg = ackMap.remove(id); + if (msg != null) { + successMap.put(id, msg); + } + if (msg != null && ackMap.isEmpty()) { + processingTimeoutLatch.countDown(); + } + } + + @Override + public void onFailure(RuleEngineException e) { + log.trace("[{}] ON FAILURE", id, e); + T msg = ackMap.remove(id); + if (msg != null) { + failedMap.put(id, msg); + firstExceptions.putIfAbsent(tenantId, e); + } + if (ackMap.isEmpty()) { + processingTimeoutLatch.countDown(); + } + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbPackCallback.java similarity index 70% rename from application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java rename to application/src/main/java/org/thingsboard/server/service/queue/TbPackCallback.java index 72c5c10458..ba5a883ea0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbPackCallback.java @@ -16,30 +16,26 @@ package org.thingsboard.server.service.queue; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.common.msg.queue.TbCallback; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CountDownLatch; @Slf4j -public class MsgPackCallback implements TbMsgCallback { +public class TbPackCallback implements TbCallback { private final CountDownLatch processingTimeoutLatch; private final ConcurrentMap ackMap; - private final ConcurrentMap successMap; private final ConcurrentMap failedMap; private final UUID id; - public MsgPackCallback(UUID id, CountDownLatch processingTimeoutLatch, - ConcurrentMap ackMap, - ConcurrentMap successMap, - ConcurrentMap failedMap) { + public TbPackCallback(UUID id, + CountDownLatch processingTimeoutLatch, + ConcurrentMap ackMap, + ConcurrentMap failedMap) { this.id = id; this.processingTimeoutLatch = processingTimeoutLatch; this.ackMap = ackMap; - this.successMap = successMap; this.failedMap = failedMap; } @@ -47,9 +43,6 @@ public class MsgPackCallback implements TbMsgCallback { public void onSuccess() { log.trace("[{}] ON SUCCESS", id); T msg = ackMap.remove(id); - if (msg != null) { - successMap.put(id, msg); - } if (msg != null && ackMap.isEmpty()) { processingTimeoutLatch.countDown(); } @@ -57,7 +50,7 @@ public class MsgPackCallback implements TbMsgCallback { @Override public void onFailure(Throwable t) { - log.trace("[{}] ON FAILURE", id); + log.trace("[{}] ON FAILURE", id, t); T msg = ackMap.remove(id); if (msg != null) { failedMap.put(id, msg); 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 27792c5a61..22f4edbbba 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 @@ -15,38 +15,119 @@ */ package org.thingsboard.server.service.queue; +import lombok.Data; import lombok.extern.slf4j.Slf4j; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.RuleNodeException; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.service.queue.processing.TbRuleEngineProcessingResult; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; @Slf4j +@Data public class TbRuleEngineConsumerStats { - private final AtomicInteger totalCounter = new AtomicInteger(0); - private final AtomicInteger postTelemetryCounter = new AtomicInteger(0); - private final AtomicInteger postAttributesCounter = new AtomicInteger(0); - private final AtomicInteger toServerRPCCallRequestCounter = new AtomicInteger(0); + public static final String TOTAL_MSGS = "totalMsgs"; + public static final String SUCCESSFUL_MSGS = "successfulMsgs"; + public static final String TMP_TIMEOUT = "tmpTimeout"; + public static final String TMP_FAILED = "tmpFailed"; + public static final String TIMEOUT_MSGS = "timeoutMsgs"; + public static final String FAILED_MSGS = "failedMsgs"; + public static final String SUCCESSFUL_ITERATIONS = "successfulIterations"; + public static final String FAILED_ITERATIONS = "failedIterations"; - public void log(TransportProtos.TransportToRuleEngineMsg msg) { - totalCounter.incrementAndGet(); - if (msg.hasPostTelemetry()) { - postTelemetryCounter.incrementAndGet(); - } - if (msg.hasPostAttributes()) { - postAttributesCounter.incrementAndGet(); - } - if (msg.hasToServerRPCCallRequest()) { - toServerRPCCallRequestCounter.incrementAndGet(); + private final AtomicInteger totalMsgCounter = new AtomicInteger(0); + private final AtomicInteger successMsgCounter = new AtomicInteger(0); + private final AtomicInteger tmpTimeoutMsgCounter = new AtomicInteger(0); + private final AtomicInteger tmpFailedMsgCounter = new AtomicInteger(0); + + private final AtomicInteger timeoutMsgCounter = new AtomicInteger(0); + private final AtomicInteger failedMsgCounter = new AtomicInteger(0); + + private final AtomicInteger successIterationsCounter = new AtomicInteger(0); + private final AtomicInteger failedIterationsCounter = new AtomicInteger(0); + + private final Map counters = new HashMap<>(); + private final ConcurrentMap tenantStats = new ConcurrentHashMap<>(); + private final ConcurrentMap tenantExceptions = new ConcurrentHashMap<>(); + + private final String queueName; + + public TbRuleEngineConsumerStats(String queueName) { + this.queueName = queueName; + counters.put(TOTAL_MSGS, totalMsgCounter); + counters.put(SUCCESSFUL_MSGS, successMsgCounter); + counters.put(TIMEOUT_MSGS, timeoutMsgCounter); + counters.put(FAILED_MSGS, failedMsgCounter); + + counters.put(TMP_TIMEOUT, tmpTimeoutMsgCounter); + counters.put(TMP_FAILED, tmpFailedMsgCounter); + counters.put(SUCCESSFUL_ITERATIONS, successIterationsCounter); + counters.put(FAILED_ITERATIONS, failedIterationsCounter); + } + + public void log(TbRuleEngineProcessingResult msg, boolean finalIterationForPack) { + int success = msg.getSuccessMap().size(); + int pending = msg.getPendingMap().size(); + int failed = msg.getFailureMap().size(); + totalMsgCounter.addAndGet(success + pending + failed); + successMsgCounter.addAndGet(success); + msg.getSuccessMap().values().forEach(m -> getTenantStats(m).logSuccess()); + if (finalIterationForPack) { + if (pending > 0 || failed > 0) { + timeoutMsgCounter.addAndGet(pending); + failedMsgCounter.addAndGet(failed); + if (pending > 0) { + msg.getPendingMap().values().forEach(m -> getTenantStats(m).logTimeout()); + } + if (failed > 0) { + msg.getFailureMap().values().forEach(m -> getTenantStats(m).logFailed()); + } + failedIterationsCounter.incrementAndGet(); + } else { + successIterationsCounter.incrementAndGet(); + } + } else { + failedIterationsCounter.incrementAndGet(); + tmpTimeoutMsgCounter.addAndGet(pending); + tmpFailedMsgCounter.addAndGet(failed); + if (pending > 0) { + msg.getPendingMap().values().forEach(m -> getTenantStats(m).logTmpTimeout()); + } + if (failed > 0) { + msg.getFailureMap().values().forEach(m -> getTenantStats(m).logTmpFailed()); + } } + msg.getExceptionsMap().forEach(tenantExceptions::putIfAbsent); + } + + private TbTenantRuleEngineStats getTenantStats(TbProtoQueueMsg m) { + ToRuleEngineMsg reMsg = m.getValue(); + return tenantStats.computeIfAbsent(new UUID(reMsg.getTenantIdMSB(), reMsg.getTenantIdLSB()), TbTenantRuleEngineStats::new); } public void printStats() { - int total = totalCounter.getAndSet(0); + int total = totalMsgCounter.get(); if (total > 0) { - log.info("Transport total [{}] telemetry [{}] attributes [{}] toServerRpc [{}]", - total, postTelemetryCounter.getAndSet(0), - postAttributesCounter.getAndSet(0), toServerRPCCallRequestCounter.getAndSet(0)); + StringBuilder stats = new StringBuilder(); + counters.forEach((label, value) -> { + stats.append(label).append(" = [").append(value.get()).append("] "); + }); + log.info("[{}] Stats: {}", queueName, stats); } } + + public void reset() { + counters.values().forEach(counter -> counter.set(0)); + tenantStats.clear(); + tenantExceptions.clear(); + } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbTenantRuleEngineStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbTenantRuleEngineStats.java new file mode 100644 index 0000000000..fb36f5064b --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbTenantRuleEngineStats.java @@ -0,0 +1,92 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.service.queue; + +import lombok.Data; +import lombok.extern.slf4j.Slf4j; + +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +@Slf4j +@Data +public class TbTenantRuleEngineStats { + + private final UUID tenantId; + + private final AtomicInteger totalMsgCounter = new AtomicInteger(0); + private final AtomicInteger successMsgCounter = new AtomicInteger(0); + private final AtomicInteger tmpTimeoutMsgCounter = new AtomicInteger(0); + private final AtomicInteger tmpFailedMsgCounter = new AtomicInteger(0); + + private final AtomicInteger timeoutMsgCounter = new AtomicInteger(0); + private final AtomicInteger failedMsgCounter = new AtomicInteger(0); + + private final Map counters = new HashMap<>(); + + public TbTenantRuleEngineStats(UUID tenantId) { + this.tenantId = tenantId; + counters.put(TbRuleEngineConsumerStats.TOTAL_MSGS, totalMsgCounter); + counters.put(TbRuleEngineConsumerStats.SUCCESSFUL_MSGS, successMsgCounter); + counters.put(TbRuleEngineConsumerStats.TIMEOUT_MSGS, timeoutMsgCounter); + counters.put(TbRuleEngineConsumerStats.FAILED_MSGS, failedMsgCounter); + + counters.put(TbRuleEngineConsumerStats.TMP_TIMEOUT, tmpTimeoutMsgCounter); + counters.put(TbRuleEngineConsumerStats.TMP_FAILED, tmpFailedMsgCounter); + } + + public void logSuccess() { + totalMsgCounter.incrementAndGet(); + successMsgCounter.incrementAndGet(); + } + + public void logFailed() { + totalMsgCounter.incrementAndGet(); + failedMsgCounter.incrementAndGet(); + } + + public void logTimeout() { + totalMsgCounter.incrementAndGet(); + timeoutMsgCounter.incrementAndGet(); + } + + public void logTmpFailed() { + totalMsgCounter.incrementAndGet(); + tmpFailedMsgCounter.incrementAndGet(); + } + + public void logTmpTimeout() { + totalMsgCounter.incrementAndGet(); + tmpTimeoutMsgCounter.incrementAndGet(); + } + + public void printStats() { + int total = totalMsgCounter.get(); + if (total > 0) { + StringBuilder stats = new StringBuilder(); + counters.forEach((label, value) -> { + stats.append(label).append(" = [").append(value.get()).append("]"); + }); + log.info("[{}] Stats: {}", tenantId, stats); + } + } + + public void reset() { + counters.values().forEach(counter -> counter.set(0)); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java index 0df0517232..a238e34337 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java @@ -22,12 +22,12 @@ import org.springframework.context.event.EventListener; import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.TbQueueConsumer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.service.encoding.DataDecodingEncodingService; -import org.thingsboard.server.service.queue.MsgPackCallback; +import org.thingsboard.server.service.queue.TbPackCallback; import javax.annotation.PreDestroy; import java.util.List; @@ -95,8 +95,8 @@ public abstract class AbstractConsumerService> failedMap = new ConcurrentHashMap<>(); CountDownLatch processingTimeoutLatch = new CountDownLatch(1); pendingMap.forEach((id, msg) -> { - log.info("[{}] Creating notification callback for message: {}", id, msg.getValue()); - TbMsgCallback callback = new MsgPackCallback<>(id, processingTimeoutLatch, pendingMap, new ConcurrentHashMap<>(), failedMap); + log.trace("[{}] Creating notification callback for message: {}", id, msg.getValue()); + TbCallback callback = new TbPackCallback<>(id, processingTimeoutLatch, pendingMap, failedMap); try { handleNotification(id, msg, callback); } catch (Throwable e) { @@ -124,7 +124,7 @@ public abstract class AbstractConsumerService msg, TbMsgCallback callback) throws Exception; + protected abstract void handleNotification(UUID id, TbProtoQueueMsg msg, TbCallback callback) throws Exception; @PreDestroy public void destroy() { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java index c17cf54bca..e67936e294 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingResult.java @@ -16,6 +16,8 @@ package org.thingsboard.server.service.queue.processing; import lombok.Getter; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.RuleEngineException; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.queue.common.TbProtoQueueMsg; @@ -25,24 +27,28 @@ import java.util.concurrent.ConcurrentMap; public class TbRuleEngineProcessingResult { @Getter - private boolean success; + private final boolean success; @Getter - private boolean timeout; + private final boolean timeout; @Getter - private ConcurrentMap> pendingMap; + private final ConcurrentMap> pendingMap; @Getter - private ConcurrentMap> successMap; + private final ConcurrentMap> successMap; @Getter - private ConcurrentMap> failureMap; + private final ConcurrentMap> failureMap; + @Getter + private final ConcurrentMap exceptionsMap; public TbRuleEngineProcessingResult(boolean timeout, ConcurrentMap> pendingMap, ConcurrentMap> successMap, - ConcurrentMap> failureMap) { + ConcurrentMap> failureMap, + ConcurrentMap exceptionsMap) { this.timeout = timeout; this.pendingMap = pendingMap; this.successMap = successMap; this.failureMap = failureMap; + this.exceptionsMap = exceptionsMap; this.success = !timeout && pendingMap.isEmpty() && failureMap.isEmpty(); } } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java index 85a44794d8..1096bfbf7e 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/processing/TbRuleEngineProcessingStrategyFactory.java @@ -32,24 +32,25 @@ import java.util.concurrent.TimeUnit; @Slf4j public class TbRuleEngineProcessingStrategyFactory { - public TbRuleEngineProcessingStrategy newInstance(TbRuleEngineQueueAckStrategyConfiguration configuration) { + public TbRuleEngineProcessingStrategy newInstance(String name, TbRuleEngineQueueAckStrategyConfiguration configuration) { switch (configuration.getType()) { case "SKIP_ALL": - return new SkipStrategy(); + return new SkipStrategy(name); case "RETRY_ALL": - return new RetryStrategy(true, true, true, configuration); + return new RetryStrategy(name, true, true, true, configuration); case "RETRY_FAILED": - return new RetryStrategy(false, true, false, configuration); + return new RetryStrategy(name, false, true, false, configuration); case "RETRY_TIMED_OUT": - return new RetryStrategy(false, false, true, configuration); + return new RetryStrategy(name, false, false, true, configuration); case "RETRY_FAILED_AND_TIMED_OUT": - return new RetryStrategy(false, true, true, configuration); + return new RetryStrategy(name, false, true, true, configuration); default: throw new RuntimeException("TbRuleEngineProcessingStrategy with type " + configuration.getType() + " is not supported!"); } } private static class RetryStrategy implements TbRuleEngineProcessingStrategy { + private final String queueName; private final boolean retrySuccessful; private final boolean retryFailed; private final boolean retryTimeout; @@ -60,7 +61,8 @@ public class TbRuleEngineProcessingStrategyFactory { private int initialTotalCount; private int retryCount; - public RetryStrategy(boolean retrySuccessful, boolean retryFailed, boolean retryTimeout, TbRuleEngineQueueAckStrategyConfiguration configuration) { + public RetryStrategy(String queueName, boolean retrySuccessful, boolean retryFailed, boolean retryTimeout, TbRuleEngineQueueAckStrategyConfiguration configuration) { + this.queueName = queueName; this.retrySuccessful = retrySuccessful; this.retryFailed = retryFailed; this.retryTimeout = retryTimeout; @@ -80,10 +82,10 @@ public class TbRuleEngineProcessingStrategyFactory { retryCount++; double failedCount = result.getFailureMap().size() + result.getPendingMap().size(); if (maxRetries > 0 && retryCount > maxRetries) { - log.info("Skip reprocess of the rule engine pack due to max retries"); + log.info("[{}] Skip reprocess of the rule engine pack due to max retries", queueName); return new TbRuleEngineProcessingDecision(true, null); } else if (maxAllowedFailurePercentage > 0 && (failedCount / initialTotalCount) > maxAllowedFailurePercentage) { - log.info("Skip reprocess of the rule engine pack due to max allowed failure percentage"); + log.info("[{}] Skip reprocess of the rule engine pack due to max allowed failure percentage", queueName); return new TbRuleEngineProcessingDecision(true, null); } else { ConcurrentMap> toReprocess = new ConcurrentHashMap<>(initialTotalCount); @@ -96,8 +98,7 @@ public class TbRuleEngineProcessingStrategyFactory { if (retrySuccessful) { result.getSuccessMap().forEach(toReprocess::put); } - log.info("Going to reprocess {} messages", toReprocess.size()); - //TODO: 2.5 Log most popular rule nodes by error count; + log.info("[{}] Going to reprocess {} messages", queueName, toReprocess.size()); if (log.isTraceEnabled()) { toReprocess.forEach((id, msg) -> log.trace("Going to reprocess [{}]: {}", id, msg.getValue())); } @@ -116,9 +117,15 @@ public class TbRuleEngineProcessingStrategyFactory { private static class SkipStrategy implements TbRuleEngineProcessingStrategy { + private final String queueName; + + public SkipStrategy(String name) { + this.queueName = name; + } + @Override public TbRuleEngineProcessingDecision analyze(TbRuleEngineProcessingResult result) { - log.info("Skip reprocess of the rule engine pack"); + log.info("[{}] Reprocessing skipped for {} failed and {} timeout messages", queueName, result.getFailureMap().size(), result.getPendingMap().size()); return new TbRuleEngineProcessingDecision(true, null); } } diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/FromDeviceRpcResponse.java b/application/src/main/java/org/thingsboard/server/service/rpc/FromDeviceRpcResponse.java index 78ea0c5660..c1e5e2e038 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/FromDeviceRpcResponse.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/FromDeviceRpcResponse.java @@ -19,7 +19,6 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; import lombok.ToString; import org.thingsboard.rule.engine.api.RpcError; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import java.util.Optional; import java.util.UUID; diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java b/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java index 3a1ea599e6..3cfe9bc1f4 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/ToDeviceRpcRequestActorMsg.java @@ -22,11 +22,8 @@ import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; -import java.util.Optional; - /** * Created by ashvayka on 16.04.18. */ diff --git a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java index 32980c0064..b4c5719844 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DefaultDeviceStateService.java @@ -55,7 +55,7 @@ import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; @@ -242,7 +242,7 @@ public class DefaultDeviceStateService implements DeviceStateService { } @Override - public void onQueueMsg(TransportProtos.DeviceStateServiceMsgProto proto, TbMsgCallback callback) { + public void onQueueMsg(TransportProtos.DeviceStateServiceMsgProto proto, TbCallback callback) { try { TenantId tenantId = new TenantId(new UUID(proto.getTenantIdMSB(), proto.getTenantIdLSB())); DeviceId deviceId = new DeviceId(new UUID(proto.getDeviceIdMSB(), proto.getDeviceIdLSB())); diff --git a/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java b/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java index c7aadda963..16c699b38a 100644 --- a/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java +++ b/application/src/main/java/org/thingsboard/server/service/state/DeviceStateService.java @@ -20,7 +20,7 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.gen.transport.TransportProtos; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; /** * Created by ashvayka on 01.05.18. @@ -41,6 +41,6 @@ public interface DeviceStateService extends ApplicationListener CALLBACK = new FutureCallback() { + @Override + public void onSuccess(@Nullable Void result) { + + } + + @Override + public void onFailure(Throwable t) { + log.warn("Failed to persist statistics", t); + } + }; + + private final TbServiceInfoProvider serviceInfoProvider; + private final TelemetrySubscriptionService tsService; + private final Lock lock = new ReentrantLock(); + private final AssetService assetService; + private final ConcurrentMap tenantQueueAssets; + + public DefaultRuleEngineStatisticsService(TelemetrySubscriptionService tsService, TbServiceInfoProvider serviceInfoProvider, AssetService assetService) { + this.tsService = tsService; + this.serviceInfoProvider = serviceInfoProvider; + this.assetService = assetService; + this.tenantQueueAssets = new ConcurrentHashMap<>(); + } + + @Override + public void reportQueueStats(long ts, TbRuleEngineConsumerStats ruleEngineStats) { + String queueName = ruleEngineStats.getQueueName(); + ruleEngineStats.getTenantStats().forEach((id, stats) -> { + TenantId tenantId = new TenantId(id); + AssetId serviceAssetId = getServiceAssetId(tenantId, queueName); + if (stats.getTotalMsgCounter().get() > 0) { + List tsList = stats.getCounters().entrySet().stream() + .map(kv -> new BasicTsKvEntry(ts, new LongDataEntry(kv.getKey(), (long) kv.getValue().get()))) + .collect(Collectors.toList()); + if (!tsList.isEmpty()) { + tsService.saveAndNotify(tenantId, serviceAssetId, tsList, CALLBACK); + } + } + }); + ruleEngineStats.getTenantExceptions().forEach((tenantId, e) -> { + TsKvEntry tsKv = new BasicTsKvEntry(ts, new JsonDataEntry("ruleEngineException", e.toJsonString())); + tsService.saveAndNotify(tenantId, getServiceAssetId(tenantId, queueName), Collections.singletonList(tsKv), CALLBACK); + }); + ruleEngineStats.reset(); + } + + private AssetId getServiceAssetId(TenantId tenantId, String queueName) { + TenantQueueKey key = new TenantQueueKey(tenantId, queueName); + AssetId assetId = tenantQueueAssets.get(key); + if (assetId == null) { + lock.lock(); + try { + assetId = tenantQueueAssets.get(key); + if (assetId == null) { + Asset asset = assetService.findAssetByTenantIdAndName(tenantId, queueName + "_" + serviceInfoProvider.getServiceId()); + if (asset == null) { + asset = new Asset(); + asset.setTenantId(tenantId); + asset.setName(queueName + "_" + serviceInfoProvider.getServiceId()); + asset.setType("TbServiceQueue"); + asset = assetService.saveAsset(asset); + } + assetId = asset.getId(); + tenantQueueAssets.put(key, assetId); + } + } finally { + lock.unlock(); + } + } + return assetId; + } + + @Data + private static class TenantQueueKey { + private final TenantId tenantId; + private final String queueName; + } +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java b/application/src/main/java/org/thingsboard/server/service/stats/RuleEngineStatisticsService.java similarity index 72% rename from common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java rename to application/src/main/java/org/thingsboard/server/service/stats/RuleEngineStatisticsService.java index 4b00284c34..88573844c9 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerType.java +++ b/application/src/main/java/org/thingsboard/server/service/stats/RuleEngineStatisticsService.java @@ -13,11 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.common.msg.cluster; +package org.thingsboard.server.service.stats; -/** - * Created by ashvayka on 23.09.18. - */ -public enum ServerType { - CORE +import org.thingsboard.server.service.queue.TbRuleEngineConsumerStats; + +public interface RuleEngineStatisticsService { + + void reportQueueStats(long ts, TbRuleEngineConsumerStats stats); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java index b6955e40a9..0d151c3f46 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java @@ -34,7 +34,7 @@ import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.ReadTsKvQuery; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; @@ -123,7 +123,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void addSubscription(TbSubscription subscription, TbMsgCallback callback) { + public void addSubscription(TbSubscription subscription, TbCallback callback) { log.trace("[{}][{}][{}] Registering remote subscription for entity [{}]", subscription.getServiceId(), subscription.getSessionId(), subscription.getSubscriptionId(), subscription.getEntityId()); TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); @@ -151,7 +151,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void cancelSubscription(String sessionId, int subscriptionId, TbMsgCallback callback) { + public void cancelSubscription(String sessionId, int subscriptionId, TbCallback callback) { log.debug("[{}][{}] Going to remove subscription.", sessionId, subscriptionId); Map sessionSubscriptions = subscriptionsByWsSessionId.get(sessionId); if (sessionSubscriptions != null) { @@ -189,7 +189,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts, TbMsgCallback callback) { + public void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts, TbCallback callback) { onLocalSubUpdate(entityId, s -> { if (TbSubscriptionType.TIMESERIES.equals(s.getType())) { @@ -213,7 +213,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer } @Override - public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbMsgCallback callback) { + public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbCallback callback) { onLocalSubUpdate(entityId, s -> { if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) { @@ -261,7 +261,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer if (subscriptionUpdate != null && !subscriptionUpdate.isEmpty()) { if (serviceId.equals(s.getServiceId())) { SubscriptionUpdate update = new SubscriptionUpdate(s.getSubscriptionId(), subscriptionUpdate); - localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbMsgCallback.EMPTY); + localSubscriptionService.onSubscriptionUpdate(s.getSessionId(), update, TbCallback.EMPTY); } else { TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, s.getServiceId()); toCoreNotificationsProducer.send(tpi, toProto(s, subscriptionUpdate), null); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java index e4d0d838ca..12c5e206e3 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java @@ -35,7 +35,7 @@ import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; @@ -135,7 +135,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer if (currentPartitions.contains(tpi)) { // Subscription is managed on the same server; if (pushToLocalService) { - subscriptionManagerService.addSubscription(subscription, TbMsgCallback.EMPTY); + subscriptionManagerService.addSubscription(subscription, TbCallback.EMPTY); } } else { // Push to the queue; @@ -145,7 +145,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer } @Override - public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbMsgCallback callback) { + public void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback) { TbSubscription subscription = subscriptionsBySessionId .getOrDefault(sessionId, Collections.emptyMap()).get(update.getSubscriptionId()); if (subscription != null) { @@ -177,7 +177,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, subscription.getTenantId(), subscription.getEntityId()); if (currentPartitions.contains(tpi)) { // Subscription is managed on the same server; - subscriptionManagerService.cancelSubscription(sessionId, subscriptionId, TbMsgCallback.EMPTY); + subscriptionManagerService.cancelSubscription(sessionId, subscriptionId, TbCallback.EMPTY); } else { // Push to the queue; TransportProtos.ToCoreMsg toCoreMsg = TbSubscriptionUtils.toCloseSubscriptionProto(subscription); diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java index aa1824c855..e20b707348 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/SubscriptionManagerService.java @@ -21,18 +21,18 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import java.util.List; public interface SubscriptionManagerService extends ApplicationListener { - void addSubscription(TbSubscription subscription, TbMsgCallback callback); + void addSubscription(TbSubscription subscription, TbCallback callback); - void cancelSubscription(String sessionId, int subscriptionId, TbMsgCallback callback); + void cancelSubscription(String sessionId, int subscriptionId, TbCallback callback); - void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts, TbMsgCallback callback); + void onTimeSeriesUpdate(TenantId tenantId, EntityId entityId, List ts, TbCallback callback); - void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbMsgCallback callback); + void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List attributes, TbCallback callback); } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java index b6bee14a32..58fba24250 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java @@ -17,7 +17,7 @@ package org.thingsboard.server.service.subscription; import org.thingsboard.server.queue.discovery.ClusterTopologyChangeEvent; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.service.telemetry.sub.SubscriptionUpdate; public interface TbLocalSubscriptionService { @@ -28,7 +28,7 @@ public interface TbLocalSubscriptionService { void cancelAllSessionSubscriptions(String sessionId); - void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbMsgCallback callback); + void onSubscriptionUpdate(String sessionId, SubscriptionUpdate update, TbCallback callback); void onApplicationEvent(PartitionChangeEvent event); diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 77b0967966..b3c728a25b 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -41,7 +41,7 @@ import org.thingsboard.server.queue.discovery.PartitionChangeEvent; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.service.subscription.SubscriptionManagerService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; @@ -166,7 +166,7 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); if (currentPartitions.contains(tpi)) { if (subscriptionManagerService.isPresent()) { - subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, TbMsgCallback.EMPTY); + subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY); } else { log.warn("Possible misconfiguration because subscriptionManagerService is null!"); } @@ -180,7 +180,7 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); if (currentPartitions.contains(tpi)) { if (subscriptionManagerService.isPresent()) { - subscriptionManagerService.get().onTimeSeriesUpdate(tenantId, entityId, ts, TbMsgCallback.EMPTY); + subscriptionManagerService.get().onTimeSeriesUpdate(tenantId, entityId, ts, TbCallback.EMPTY); } else { log.warn("Possible misconfiguration because subscriptionManagerService is null!"); } diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java index e905559516..d652a3f1c5 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java @@ -17,11 +17,7 @@ package org.thingsboard.server.service.telemetry; import org.springframework.context.ApplicationListener; import org.thingsboard.rule.engine.api.RuleEngineTelemetryService; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import org.thingsboard.server.queue.discovery.PartitionChangeEvent; -import org.thingsboard.server.service.telemetry.sub.SubscriptionState; /** * Created by ashvayka on 27.03.18. diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/Subscription.java b/application/src/main/java/org/thingsboard/server/service/telemetry/sub/Subscription.java deleted file mode 100644 index 7f7db9430d..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/sub/Subscription.java +++ /dev/null @@ -1,80 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.service.telemetry.sub; - -import lombok.AllArgsConstructor; -import lombok.Data; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.msg.cluster.ServerAddress; -import org.thingsboard.server.service.telemetry.TelemetryFeature; - -import java.util.Map; - -@Data -@AllArgsConstructor -public class Subscription { - - private final SubscriptionState sub; - private final boolean local; - private ServerAddress server; - private long startTime; - private long endTime; - - public Subscription(SubscriptionState sub, boolean local, ServerAddress server) { - this(sub, local, server, 0L, 0L); - } - - public String getWsSessionId() { - return getSub().getWsSessionId(); - } - - public int getSubscriptionId() { - return getSub().getSubscriptionId(); - } - - public EntityId getEntityId() { - return getSub().getEntityId(); - } - - public TelemetryFeature getType() { - return getSub().getType(); - } - - public String getScope() { - return getSub().getScope(); - } - - public boolean isAllKeys() { - return getSub().isAllKeys(); - } - - public Map getKeyStates() { - return getSub().getKeyStates(); - } - - public void setKeyState(String key, long ts) { - getSub().getKeyStates().put(key, ts); - } - - @Override - public String toString() { - return "Subscription{" + - "sub=" + sub + - ", local=" + local + - ", server=" + server + - '}'; - } -} diff --git a/application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java b/application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java index 783dd5181a..477f115a96 100644 --- a/application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java +++ b/application/src/main/java/org/thingsboard/server/service/transaction/BaseRuleChainTransactionService.java @@ -22,7 +22,6 @@ import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.rule.engine.api.RuleChainTransactionService; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import org.thingsboard.server.queue.util.TbRuleEngineComponent; import org.thingsboard.server.service.executors.DbCallbackExecutorService; @@ -114,13 +113,6 @@ public class BaseRuleChainTransactionService implements RuleChainTransactionServ // } } - @Override - public void onRemoteTransactionMsg(ServerAddress serverAddress, byte[] data) { - endLocalTransaction(TbMsg.fromBytes(data, null), msg -> { - }, error -> { - }); - } - private void addMsgToQueues(BlockingQueue queue, TbTransactionTask transactionTask) { queue.offer(transactionTask); timeoutQueue.offer(transactionTask); @@ -230,9 +222,4 @@ public class BaseRuleChainTransactionService implements RuleChainTransactionServ callbackExecutor.executeAsync(task); } - private void sendTransactionEventToRemoteServer(TbMsg msg, ServerAddress address) { - log.trace("[{}][{}] Originator is monitored on other server: {}", msg.getTransactionData().getOriginatorId(), msg.getTransactionData().getTransactionId(), address); - //TODO 2.5 -// clusterRpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TRANSACTION_SERVICE_MESSAGE, TbMsg.toByteArray(msg)); - } } diff --git a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java index 16e0dd67f2..ba4e7baab1 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java @@ -23,7 +23,7 @@ import org.thingsboard.server.common.msg.TbActorMsg; import org.thingsboard.server.common.msg.aware.DeviceAwareMsg; import org.thingsboard.server.common.msg.aware.TenantAwareMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; +import org.thingsboard.server.common.msg.queue.TbCallback; import java.io.Serializable; import java.util.UUID; @@ -37,9 +37,9 @@ public class TransportToDeviceActorMsgWrapper implements TbActorMsg, DeviceAware private final TenantId tenantId; private final DeviceId deviceId; private final TransportToDeviceActorMsg msg; - private final TbMsgCallback callback; + private final TbCallback callback; - public TransportToDeviceActorMsgWrapper(TransportToDeviceActorMsg msg, TbMsgCallback callback) { + public TransportToDeviceActorMsgWrapper(TransportToDeviceActorMsg msg, TbCallback callback) { this.msg = msg; this.callback = callback; this.tenantId = new TenantId(new UUID(msg.getSessionInfo().getTenantIdMSB(), msg.getSessionInfo().getTenantIdLSB())); diff --git a/application/src/main/resources/logback.xml b/application/src/main/resources/logback.xml index 3d47db432f..8389fa5d71 100644 --- a/application/src/main/resources/logback.xml +++ b/application/src/main/resources/logback.xml @@ -27,8 +27,8 @@ - - + + diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 04c9205a12..79efacba97 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -561,7 +561,7 @@ queue: poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: - enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:false}" + enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}" print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" queues: # TODO 2.5: specify correct ENV variable names. - @@ -577,7 +577,7 @@ queue: failure-percentage: "${TB_QUEUE_RULE_ENGINE_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RULE_ENGINE_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - - name: "HighPriority" + name: "${TB_QUEUE_RULE_ENGINE_HP_QUEUE_NAME:HighPriority}" topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine.hp}" poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:3}" @@ -594,7 +594,7 @@ queue: poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: - type: "${TB_SERVICE_TYPE:monolith}" # monolith or tb-core or tb-rule-engine or tb-transport + type: "${TB_SERVICE_TYPE:monolith}" # monolith or tb-core or tb-rule-engine # Unique id for this service (autogenerated if empty) id: "${TB_SERVICE_ID:}" tenant_id: "${TB_SERVICE_TENANT_ID:}" # empty or specific tenant id. diff --git a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java index 5ecb5fb415..7fe63827eb 100644 --- a/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/flow/AbstractRuleEngineFlowIntegrationTest.java @@ -35,8 +35,6 @@ import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; -import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; import org.thingsboard.server.controller.AbstractRuleEngineControllerTest; import org.thingsboard.server.dao.attributes.AttributesService; diff --git a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java index bc4db3b535..78129917ef 100644 --- a/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/rules/lifecycle/AbstractRuleEngineLifecycleIntegrationTest.java @@ -39,8 +39,6 @@ import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgDataType; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; -import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; import org.thingsboard.server.controller.AbstractRuleEngineControllerTest; import org.thingsboard.server.dao.attributes.AttributesService; diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index dec6f6b50b..47e09b812f 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -33,11 +33,6 @@ public enum MsgType { APP_INIT_MSG, - /** - * All messages, could be send to cluster - */ - SEND_TO_CLUSTER_MSG, - /** * ADDED/UPDATED/DELETED events for main entities. * diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java index 7230804773..84110ed163 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/TbMsg.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -17,7 +17,6 @@ package org.thingsboard.server.common.msg; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; -import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.extern.slf4j.Slf4j; @@ -29,7 +28,6 @@ import org.thingsboard.server.common.msg.gen.MsgProtos; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import java.io.Serializable; -import java.nio.ByteBuffer; import java.util.UUID; /** diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java b/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java deleted file mode 100644 index 80674f9bce..0000000000 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/ServerAddress.java +++ /dev/null @@ -1,47 +0,0 @@ -/** - * Copyright © 2016-2020 The Thingsboard Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.thingsboard.server.common.msg.cluster; - -import lombok.Data; -import lombok.EqualsAndHashCode; - -import java.io.Serializable; - -/** - * @author Andrew Shvayka - */ -@Data -@EqualsAndHashCode -public class ServerAddress implements Comparable, Serializable { - - private final String host; - private final int port; - private final ServerType serverType; - - @Override - public int compareTo(ServerAddress o) { - int result = this.host.compareTo(o.host); - if (result == 0) { - result = this.port - o.port; - } - return result; - } - - @Override - public String toString() { - return '[' + host + ':' + port + ']'; - } -} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java index 49b30a5992..fca1c51995 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/QueueToRuleEngineMsg.java @@ -33,6 +33,7 @@ public final class QueueToRuleEngineMsg implements TbActorMsg { private final TenantId tenantId; private final TbMsg tbMsg; private final Set relationTypes; + private final String failureMessage; @Override public MsgType getMsgType() { diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleEngineException.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleEngineException.java new file mode 100644 index 0000000000..dd720251bb --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleEngineException.java @@ -0,0 +1,38 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.msg.queue; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class RuleEngineException extends Exception { + protected static final ObjectMapper mapper = new ObjectMapper(); + + public RuleEngineException(String message) { + super(message != null ? message : "Unknown"); + } + + public String toJsonString() { + try { + return mapper.writeValueAsString(mapper.createObjectNode().put("message", getMessage())); + } catch (JsonProcessingException e) { + log.warn("Failed to serialize exception ", e); + throw new RuntimeException(e); + } + } +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java new file mode 100644 index 0000000000..3f437dd1d7 --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/RuleNodeException.java @@ -0,0 +1,58 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.msg.queue; + +import com.fasterxml.jackson.core.JsonProcessingException; +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.RuleNodeId; +import org.thingsboard.server.common.data.rule.RuleNode; + +@Slf4j +public class RuleNodeException extends RuleEngineException { + @Getter + private final String ruleChainName; + @Getter + private final String ruleNodeName; + @Getter + private final RuleChainId ruleChainId; + @Getter + private final RuleNodeId ruleNodeId; + + public RuleNodeException(String message, String ruleChainName, RuleNode ruleNode) { + super(message); + this.ruleChainName = ruleChainName; + this.ruleNodeName = ruleNode.getName(); + this.ruleChainId = ruleNode.getRuleChainId(); + this.ruleNodeId = ruleNode.getId(); + } + + public String toJsonString() { + try { + return mapper.writeValueAsString(mapper.createObjectNode() + .put("ruleNodeId", ruleNodeId.toString()) + .put("ruleChainId", ruleChainId.toString()) + .put("ruleNodeName", ruleNodeName) + .put("ruleChainName", ruleChainName) + .put("message", getMessage())); + } catch (JsonProcessingException e) { + log.warn("Failed to serialize exception ", e); + throw new RuntimeException(e); + } + } + +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbCallback.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbCallback.java new file mode 100644 index 0000000000..4c25f27a3e --- /dev/null +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbCallback.java @@ -0,0 +1,37 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.msg.queue; + +public interface TbCallback { + + TbCallback EMPTY = new TbCallback() { + + @Override + public void onSuccess() { + + } + + @Override + public void onFailure(Throwable t) { + + } + }; + + void onSuccess(); + + void onFailure(Throwable t); + +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbMsgCallback.java b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbMsgCallback.java index 4f0e383fe9..9e8d5ae6b8 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbMsgCallback.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/queue/TbMsgCallback.java @@ -25,13 +25,13 @@ public interface TbMsgCallback { } @Override - public void onFailure(Throwable t) { + public void onFailure(RuleEngineException e) { } }; void onSuccess(); - void onFailure(Throwable t); + void onFailure(RuleEngineException e); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueTbMsgCallbackWrapper.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueTbMsgCallbackWrapper.java index 552eca84a0..f860d8dcc9 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueTbMsgCallbackWrapper.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/MultipleTbQueueTbMsgCallbackWrapper.java @@ -15,6 +15,8 @@ */ package org.thingsboard.server.queue.common; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; @@ -40,6 +42,6 @@ public class MultipleTbQueueTbMsgCallbackWrapper implements TbQueueCallback { @Override public void onFailure(Throwable t) { - tbMsgCallback.onFailure(t); + tbMsgCallback.onFailure(new RuleEngineException(t.getMessage())); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbQueueTbMsgCallbackWrapper.java b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbQueueTbMsgCallbackWrapper.java index c372f64b3d..8bd9c3516b 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/common/TbQueueTbMsgCallbackWrapper.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/common/TbQueueTbMsgCallbackWrapper.java @@ -15,12 +15,12 @@ */ package org.thingsboard.server.queue.common; +import org.thingsboard.server.common.msg.queue.RuleEngineException; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.queue.TbMsgCallback; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; -import java.util.concurrent.atomic.AtomicInteger; - public class TbQueueTbMsgCallbackWrapper implements TbQueueCallback { private final TbMsgCallback tbMsgCallback; @@ -36,6 +36,6 @@ public class TbQueueTbMsgCallbackWrapper implements TbQueueCallback { @Override public void onFailure(Throwable t) { - tbMsgCallback.onFailure(t); + tbMsgCallback.onFailure(new RuleEngineException(t.getMessage())); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java index 1337a1cc74..3558374af5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java @@ -86,24 +86,6 @@ public class ConsistentHashPartitionService implements PartitionService { partitionTopics.put(new ServiceQueue(ServiceType.TB_CORE), coreTopic); } -// public Set getCurrentPartitions(ServiceType serviceType) { -// ServiceInfo currentService = serviceInfoProvider.getServiceInfo(); -// TenantId tenantId = getSystemOrIsolatedTenantId(currentService); -// ServiceQueueKey serviceQueueKey = new ServiceQueueKey(serviceType, tenantId); -// List partitions = myPartitions.get(serviceQueueKey); -// Set topicPartitions = new LinkedHashSet<>(); -// for (Integer partition : partitions) { -// TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder(); -// tpi.topic(partitionTopics.get(serviceType)); -// tpi.partition(partition); -// if (!tenantId.isNullUid()) { -// tpi.tenantId(tenantId); -// } -// topicPartitions.add(tpi.build()); -// } -// return topicPartitions; -// } - @Override public TopicPartitionInfo resolve(ServiceType serviceType, TenantId tenantId, EntityId entityId) { return resolve(new ServiceQueue(serviceType), tenantId, entityId); @@ -131,15 +113,7 @@ public class ConsistentHashPartitionService implements PartitionService { Map> circles = new HashMap<>(); addNode(circles, currentService); for (ServiceInfo other : otherServices) { - TenantId tenantId = getSystemOrIsolatedTenantId(other); addNode(circles, other); - if (!tenantId.isNullUid()) { - isolatedTenants.putIfAbsent(tenantId, new HashSet<>()); - for (String serviceType : other.getServiceTypesList()) { - isolatedTenants.get(tenantId).add(ServiceType.valueOf(serviceType.toUpperCase())); - } - - } } ConcurrentMap> oldPartitions = myPartitions; TenantId myTenantId = getSystemOrIsolatedTenantId(currentService); @@ -214,6 +188,11 @@ public class ConsistentHashPartitionService implements PartitionService { } } + @Override + public Set getIsolatedTenants(ServiceType serviceType) { + throw new RuntimeException("Not Implemented!"); + } + private Map> getServiceKeyListMap(List services) { final Map> currentMap = new HashMap<>(); services.forEach(serviceInfo -> { @@ -280,6 +259,12 @@ public class ConsistentHashPartitionService implements PartitionService { private void addNode(Map> circles, ServiceInfo instance) { TenantId tenantId = getSystemOrIsolatedTenantId(instance); + if (!tenantId.isNullUid()) { + isolatedTenants.putIfAbsent(tenantId, new HashSet<>()); + for (String serviceType : instance.getServiceTypesList()) { + isolatedTenants.get(tenantId).add(ServiceType.valueOf(serviceType.toUpperCase())); + } + } for (String serviceTypeStr : instance.getServiceTypesList()) { ServiceType serviceType = ServiceType.valueOf(serviceTypeStr.toUpperCase()); if (ServiceType.TB_RULE_ENGINE.equals(serviceType)) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java index 15cf33bb82..1c60a03cdd 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/DefaultTbServiceInfoProvider.java @@ -34,6 +34,7 @@ import java.net.UnknownHostException; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.UUID; import java.util.stream.Collectors; @@ -58,6 +59,7 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { private List serviceTypes; private ServiceInfo serviceInfo; + private TenantId isolatedTenant; @PostConstruct public void init() { @@ -80,6 +82,7 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { UUID tenantId; if (!StringUtils.isEmpty(tenantIdStr)) { tenantId = UUID.fromString(tenantIdStr); + isolatedTenant = new TenantId(tenantId); } else { tenantId = TenantId.NULL_UUID; } @@ -103,4 +106,14 @@ public class DefaultTbServiceInfoProvider implements TbServiceInfoProvider { public ServiceInfo getServiceInfo() { return serviceInfo; } + + @Override + public boolean isService(ServiceType serviceType) { + return serviceTypes.contains(serviceType); + } + + @Override + public Optional getIsolatedTenant() { + return Optional.ofNullable(isolatedTenant); + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index e3b3e76b80..7c6900d8a8 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -47,6 +47,8 @@ public interface PartitionService { */ Set getAllServiceIds(ServiceType serviceType); + Set getIsolatedTenants(ServiceType serviceType); + /** * Each Service should start a consumer for messages that target individual service instance based on serviceId. * This topic is likely to have single partition, and is always assigned to the service. diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java index 053dd644c5..fcc7c7a60d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/TbServiceInfoProvider.java @@ -15,12 +15,20 @@ */ package org.thingsboard.server.queue.discovery; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.msg.queue.ServiceType; import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo; +import java.util.Optional; + public interface TbServiceInfoProvider { String getServiceId(); ServiceInfo getServiceInfo(); + boolean isService(ServiceType serviceType); + + Optional getIsolatedTenant(); + } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java index 48eeb5ba2a..95540798fc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java @@ -17,7 +17,9 @@ package org.thingsboard.server.queue.settings; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Configuration; @@ -26,6 +28,7 @@ import java.util.List; @Slf4j @Data +@EnableAutoConfiguration @Configuration @ConfigurationProperties(prefix = "queue.rule-engine") public class TbQueueRuleEngineSettings { diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index 55d376b63b..e36c4e36a9 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -388,6 +388,7 @@ message ToRuleEngineMsg { int64 tenantIdLSB = 2; bytes tbMsg = 3; repeated string relationTypes = 4; + string failureMessage = 5; } message ToRuleEngineNotificationMsg { diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java index bb0210b1ad..1bb56adb1d 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleChainTransactionService.java @@ -16,7 +16,6 @@ package org.thingsboard.rule.engine.api; import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.cluster.ServerAddress; import java.util.function.Consumer; @@ -26,6 +25,4 @@ public interface RuleChainTransactionService { void endTransaction(TbMsg msg, Consumer onSuccess, Consumer onFailure); - void onRemoteTransactionMsg(ServerAddress serverAddress, byte[] bytes); - } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 72a552cb4f..4299aeb867 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -26,10 +26,8 @@ import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.msg.TbMsg; import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.queue.TbMsgCallback; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.asset.AssetService; import org.thingsboard.server.dao.attributes.AttributesService; @@ -116,6 +114,8 @@ public interface TbContext { */ void enqueue(TbMsg msg, String queueName, Runnable onSuccess, Consumer onFailure); + void enqueueForTellFailure(TbMsg msg, String failureMessage); + void enqueueForTellNext(TbMsg msg, String relationType); void enqueueForTellNext(TbMsg msg, Set relationTypes); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbAckNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbAckNode.java new file mode 100644 index 0000000000..9351e970ed --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbAckNode.java @@ -0,0 +1,53 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.flow; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; +import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNode; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TbRelationTypes; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "acknowledge", + configClazz = EmptyNodeConfiguration.class, + nodeDescription = "Acknowledges the incoming message", + nodeDetails = "After acknowledgement, the message is pushed to related rule nodes. Useful if you don't care what happens to this message next.") + +public class TbAckNode implements TbNode { + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) { + ctx.ack(msg); + ctx.tellSuccess(msg); + } + + @Override + public void destroy() { + } +} diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java new file mode 100644 index 0000000000..0364251076 --- /dev/null +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNode.java @@ -0,0 +1,57 @@ +/** + * Copyright © 2016-2020 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.rule.engine.flow; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.rule.engine.api.RuleNode; +import org.thingsboard.rule.engine.api.ScriptEngine; +import org.thingsboard.rule.engine.api.TbContext; +import org.thingsboard.rule.engine.api.TbNode; +import org.thingsboard.rule.engine.api.TbNodeConfiguration; +import org.thingsboard.rule.engine.api.TbNodeException; +import org.thingsboard.rule.engine.api.TbRelationTypes; +import org.thingsboard.rule.engine.api.util.TbNodeUtils; +import org.thingsboard.server.common.data.plugin.ComponentType; +import org.thingsboard.server.common.msg.TbMsg; + +import static org.thingsboard.common.util.DonAsynchron.withCallback; + +@Slf4j +@RuleNode( + type = ComponentType.ACTION, + name = "checkpoint", + configClazz = TbCheckpointNodeConfiguration.class, + nodeDescription = "transfers the message to another queue", + nodeDetails = "After successful transfer incoming message is automatically acknowledged. Queue name is configurable.") + +public class TbCheckpointNode implements TbNode { + + private TbCheckpointNodeConfiguration config; + + @Override + public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { + this.config = TbNodeUtils.convert(configuration, TbCheckpointNodeConfiguration.class); + } + + @Override + public void onMsg(TbContext ctx, TbMsg msg) { + ctx.enqueueForTellNext(msg, config.getQueueName(), TbRelationTypes.SUCCESS, () -> ctx.ack(msg), error -> ctx.tellFailure(msg, error)); + } + + @Override + public void destroy() { + } +} diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/SendToClusterMsg.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java similarity index 53% rename from common/message/src/main/java/org/thingsboard/server/common/msg/cluster/SendToClusterMsg.java rename to rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java index 3326c33aa3..eb5b8e29d9 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/cluster/SendToClusterMsg.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/flow/TbCheckpointNodeConfiguration.java @@ -13,28 +13,20 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.common.msg.cluster; +package org.thingsboard.rule.engine.flow; import lombok.Data; -import org.thingsboard.server.common.data.id.DeviceId; -import org.thingsboard.server.common.data.id.EntityId; -import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.TbActorMsg; +import org.thingsboard.rule.engine.api.NodeConfiguration; @Data -public class SendToClusterMsg implements TbActorMsg { - - private TbActorMsg msg; - private EntityId entityId; - - public SendToClusterMsg(EntityId entityId, TbActorMsg msg) { - this.entityId = entityId; - this.msg = msg; - } +public class TbCheckpointNodeConfiguration implements NodeConfiguration { + private String queueName; @Override - public MsgType getMsgType() { - return MsgType.SEND_TO_CLUSTER_MSG; + public TbCheckpointNodeConfiguration defaultConfiguration() { + TbCheckpointNodeConfiguration configuration = new TbCheckpointNodeConfiguration(); + configuration.setQueueName("HighPriority"); + return configuration; } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java index 88b060fe5c..ad1d399e76 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java @@ -116,8 +116,7 @@ public class TbSendRPCRequestNode implements TbNode { ctx.enqueueForTellNext(next, TbRelationTypes.SUCCESS); } else { TbMsg next = ctx.newMsg(msg.getType(), msg.getOriginator(), msg.getMetaData(), wrap("error", ruleEngineDeviceRpcResponse.getError().get().name())); - ctx.tellFailure(next, new RuntimeException(ruleEngineDeviceRpcResponse.getError().get().name())); - ctx.enqueueForTellNext(next, TbRelationTypes.FAILURE); + ctx.enqueueForTellFailure(next, ruleEngineDeviceRpcResponse.getError().get().name()); } }); ctx.ack(msg); diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index a875d56782..573fe6d5df 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -102,20 +102,44 @@ queue: stats: enabled: "${TB_QUEUE_CORE_STATS_ENABLED:false}" print_interval_ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}" - rule_engine: + rule-engine: topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" - poll_interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" - partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:10}" - pack_processing_timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" + poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" + pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: - enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:false}" - print_interval_ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" + enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}" + print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" + queues: # TODO 2.5: specify correct ENV variable names. + - + name: "Main" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine.main}" + poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" + partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:10}" + pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" + ack-strategy: + type: "${TB_QUEUE_RULE_ENGINE_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT + # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT + retries: "${TB_QUEUE_RULE_ENGINE_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited + failure-percentage: "${TB_QUEUE_RULE_ENGINE_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; + pause-between-retries: "${TB_QUEUE_RULE_ENGINE_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; + - + name: "HighPriority" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine.hp}" + poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" + partitions: "${TB_QUEUE_RULE_ENGINE_PARTITIONS:3}" + pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" + ack-strategy: + type: "${TB_QUEUE_RULE_ENGINE_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT + # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT + retries: "${TB_QUEUE_RULE_ENGINE_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited + failure-percentage: "${TB_QUEUE_RULE_ENGINE_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; + pause-between-retries: "${TB_QUEUE_RULE_ENGINE_STRATEGY_RETRY_PAUSE:1}"# Time in seconds to wait in consumer thread before retries; transport: notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: - type: "${TB_SERVICE_TYPE:tb-transport}" # monolith or tb-core or tb-rule-engine or tb-transport + type: "${TB_SERVICE_TYPE:tb-transport}" # Unique id for this service (autogenerated if empty) id: "${TB_SERVICE_ID:}" tenant_id: "${TB_SERVICE_TENANT_ID:}" # empty or specific tenant id. \ No newline at end of file