Better logging of Rule Engine errors

This commit is contained in:
Andrii Shvaika 2020-05-04 00:13:14 +03:00
parent d2919ba30e
commit 08ab9752fc
5 changed files with 61 additions and 29 deletions

View File

@ -163,7 +163,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
String dispatcherName = tenantId.getId().equals(EntityId.NULL_UUID) ? String dispatcherName = tenantId.getId().equals(EntityId.NULL_UUID) ?
DefaultActorService.SYSTEM_RULE_DISPATCHER_NAME : DefaultActorService.TENANT_RULE_DISPATCHER_NAME; DefaultActorService.SYSTEM_RULE_DISPATCHER_NAME : DefaultActorService.TENANT_RULE_DISPATCHER_NAME;
return context.actorOf( return context.actorOf(
Props.create(new RuleNodeActor.ActorCreator(systemContext, tenantId, entityId, ruleNode.getId())) Props.create(new RuleNodeActor.ActorCreator(systemContext, tenantId, entityId, ruleNode.getName(), ruleNode.getId()))
.withDispatcher(dispatcherName), ruleNode.getId().toString()); .withDispatcher(dispatcherName), ruleNode.getId().toString());
} }
@ -200,7 +200,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, msg.getId(), msg); log.trace("[{}][{}] Processing message [{}]: {}", entityId, firstId, msg.getId(), msg);
if (envelope.getRelationTypes() == null || envelope.getRelationTypes().isEmpty()) { if (envelope.getRelationTypes() == null || envelope.getRelationTypes().isEmpty()) {
try { try {
checkActive(); checkActive(envelope.getTbMsg());
RuleNodeId targetId = msg.getRuleNodeId(); RuleNodeId targetId = msg.getRuleNodeId();
RuleNodeCtx targetCtx; RuleNodeCtx targetCtx;
if (targetId == null) { if (targetId == null) {
@ -216,6 +216,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
log.trace("[{}][{}] Rule node does not exist. Probably old message", entityId, targetId); log.trace("[{}][{}] Rule node does not exist. Probably old message", entityId, targetId);
msg.getCallback().onSuccess(); msg.getCallback().onSuccess();
} }
} catch (RuleNodeException rne) {
envelope.getTbMsg().getCallback().onFailure(rne);
} catch (Exception e) { } catch (Exception e) {
envelope.getTbMsg().getCallback().onFailure(new RuleEngineException(e.getMessage())); envelope.getTbMsg().getCallback().onFailure(new RuleEngineException(e.getMessage()));
} }
@ -225,11 +227,15 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
} }
void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) { void onRuleChainToRuleChainMsg(RuleChainToRuleChainMsg envelope) {
checkActive(); try {
if (firstNode != null) { checkActive(envelope.getMsg());
pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType()); if (firstNode != null) {
} else { pushMsgToNode(firstNode, envelope.getMsg(), envelope.getFromRelationType());
envelope.getMsg().getCallback().onSuccess(); } else {
envelope.getMsg().getCallback().onSuccess();
}
} catch (RuleNodeException e) {
log.debug("Rule Chain is not active. Current state [{}] for processor [{}][{}] tenant [{}]", state, entityId.getEntityType(), entityId, tenantId);
} }
} }
@ -239,7 +245,7 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
private void onTellNext(TbMsg msg, RuleNodeId originatorNodeId, Set<String> relationTypes, String failureMessage) { private void onTellNext(TbMsg msg, RuleNodeId originatorNodeId, Set<String> relationTypes, String failureMessage) {
try { try {
checkActive(); checkActive(msg);
EntityId entityId = msg.getOriginator(); EntityId entityId = msg.getOriginator();
TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); TopicPartitionInfo tpi = systemContext.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId);
List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream() List<RuleNodeRelation> relations = nodeRoutes.get(originatorNodeId).stream()
@ -272,6 +278,8 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
putToQueue(tpi, msg, callbackWrapper, target); putToQueue(tpi, msg, callbackWrapper, target);
} }
} }
} catch (RuleNodeException rne) {
msg.getCallback().onFailure(rne);
} catch (Exception e) { } catch (Exception e) {
msg.getCallback().onFailure(new RuleEngineException("onTellNext - " + e.getMessage())); msg.getCallback().onFailure(new RuleEngineException("onTellNext - " + e.getMessage()));
} }
@ -333,4 +341,9 @@ public class RuleChainActorMessageProcessor extends ComponentMsgProcessor<RuleCh
} }
} }
@Override
protected RuleNodeException getInactiveException() {
RuleNode firstRuleNode = firstNode != null ? firstNode.getSelf() : null;
return new RuleNodeException("Rule Chain is not active! Failed to initialize.", ruleChainName, firstRuleNode);
}
} }

View File

@ -27,12 +27,14 @@ import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessageProcessor> { public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessageProcessor> {
private final String ruleChainName;
private final RuleChainId ruleChainId; private final RuleChainId ruleChainId;
private RuleNodeActor(ActorSystemContext systemContext, TenantId tenantId, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { private RuleNodeActor(ActorSystemContext systemContext, TenantId tenantId, RuleChainId ruleChainId, String ruleChainName, RuleNodeId ruleNodeId) {
super(systemContext, tenantId, ruleNodeId); super(systemContext, tenantId, ruleNodeId);
this.ruleChainName = ruleChainName;
this.ruleChainId = ruleChainId; this.ruleChainId = ruleChainId;
setProcessor(new RuleNodeActorMessageProcessor(tenantId, ruleChainId, ruleNodeId, systemContext, setProcessor(new RuleNodeActorMessageProcessor(tenantId, this.ruleChainName, ruleNodeId, systemContext,
context().parent(), context().self())); context().parent(), context().self()));
} }
@ -96,19 +98,21 @@ public class RuleNodeActor extends ComponentActor<RuleNodeId, RuleNodeActorMessa
private final TenantId tenantId; private final TenantId tenantId;
private final RuleChainId ruleChainId; private final RuleChainId ruleChainId;
private final String ruleChainName;
private final RuleNodeId ruleNodeId; private final RuleNodeId ruleNodeId;
public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleChainId ruleChainId, RuleNodeId ruleNodeId) { public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleChainId ruleChainId, String ruleChainName, RuleNodeId ruleNodeId) {
super(context); super(context);
this.tenantId = tenantId; this.tenantId = tenantId;
this.ruleChainId = ruleChainId; this.ruleChainId = ruleChainId;
this.ruleChainName = ruleChainName;
this.ruleNodeId = ruleNodeId; this.ruleNodeId = ruleNodeId;
} }
@Override @Override
public RuleNodeActor create() throws Exception { public RuleNodeActor create() throws Exception {
return new RuleNodeActor(context, tenantId, ruleChainId, ruleNodeId); return new RuleNodeActor(context, tenantId, ruleChainId, ruleChainName, ruleNodeId);
} }
} }

View File

@ -17,37 +17,33 @@ package org.thingsboard.server.actors.ruleChain;
import akka.actor.ActorContext; import akka.actor.ActorContext;
import akka.actor.ActorRef; import akka.actor.ActorRef;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.server.actors.ActorSystemContext; import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.shared.ComponentMsgProcessor; import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.RuleNodeId; import org.thingsboard.server.common.data.id.RuleNodeId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleState; import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
import org.thingsboard.server.common.data.rule.RuleNode; import org.thingsboard.server.common.data.rule.RuleNode;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.dao.rule.RuleChainService; import org.thingsboard.server.common.msg.queue.RuleNodeException;
/** /**
* @author Andrew Shvayka * @author Andrew Shvayka
*/ */
public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> { public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNodeId> {
private final ActorRef parent; private final String ruleChainName;
private final ActorRef self; private final ActorRef self;
private final RuleChainService service;
private RuleNode ruleNode; private RuleNode ruleNode;
private TbNode tbNode; private TbNode tbNode;
private DefaultTbContext defaultCtx; private DefaultTbContext defaultCtx;
RuleNodeActorMessageProcessor(TenantId tenantId, RuleChainId ruleChainId, RuleNodeId ruleNodeId, ActorSystemContext systemContext RuleNodeActorMessageProcessor(TenantId tenantId, String ruleChainName, RuleNodeId ruleNodeId, ActorSystemContext systemContext
, ActorRef parent, ActorRef self) { , ActorRef parent, ActorRef self) {
super(systemContext, tenantId, ruleNodeId); super(systemContext, tenantId, ruleNodeId);
this.parent = parent; this.ruleChainName = ruleChainName;
this.self = self; this.self = self;
this.service = systemContext.getRuleChainService();
this.ruleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId); this.ruleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
this.defaultCtx = new DefaultTbContext(systemContext, new RuleNodeCtx(tenantId, parent, self, ruleNode)); this.defaultCtx = new DefaultTbContext(systemContext, new RuleNodeCtx(tenantId, parent, self, ruleNode));
} }
@ -63,8 +59,8 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
@Override @Override
public void onUpdate(ActorContext context) throws Exception { public void onUpdate(ActorContext context) throws Exception {
RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId); RuleNode newRuleNode = systemContext.getRuleChainService().findRuleNodeById(tenantId, entityId);
boolean restartRequired = !(ruleNode.getType().equals(newRuleNode.getType()) boolean restartRequired = state != ComponentLifecycleState.ACTIVE ||
&& ruleNode.getConfiguration().equals(newRuleNode.getConfiguration())); !(ruleNode.getType().equals(newRuleNode.getType()) && ruleNode.getConfiguration().equals(newRuleNode.getConfiguration()));
this.ruleNode = newRuleNode; this.ruleNode = newRuleNode;
this.defaultCtx.updateSelf(newRuleNode); this.defaultCtx.updateSelf(newRuleNode);
if (restartRequired) { if (restartRequired) {
@ -91,7 +87,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
} }
public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception { public void onRuleToSelfMsg(RuleNodeToSelfMsg msg) throws Exception {
checkActive(); checkActive(msg.getMsg());
if (ruleNode.isDebugMode()) { if (ruleNode.isDebugMode()) {
systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), "Self"); systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), "Self");
} }
@ -103,7 +99,7 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
} }
void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception { void onRuleChainToRuleNodeMsg(RuleChainToRuleNodeMsg msg) throws Exception {
checkActive(); checkActive(msg.getMsg());
if (ruleNode.isDebugMode()) { if (ruleNode.isDebugMode()) {
systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType()); systemContext.persistDebugInput(tenantId, entityId, msg.getMsg(), msg.getFromRelationType());
} }
@ -129,4 +125,8 @@ public class RuleNodeActorMessageProcessor extends ComponentMsgProcessor<RuleNod
return tbNode; return tbNode;
} }
@Override
protected RuleNodeException getInactiveException() {
return new RuleNodeException("Rule Node is not active! Failed to initialize.", ruleChainName, ruleNode);
}
} }

View File

@ -22,7 +22,10 @@ import org.thingsboard.server.actors.stats.StatsPersistTick;
import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleState; import org.thingsboard.server.common.data.plugin.ComponentLifecycleState;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import org.thingsboard.server.common.msg.queue.RuleNodeException;
@Slf4j @Slf4j
public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor { public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor {
@ -74,11 +77,17 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency); schedulePeriodicMsgWithDelay(context, new StatsPersistTick(), statsPersistFrequency, statsPersistFrequency);
} }
protected void checkActive() { protected void checkActive(TbMsg tbMsg) throws RuleNodeException {
if (state != ComponentLifecycleState.ACTIVE) { if (state != ComponentLifecycleState.ACTIVE) {
log.debug("Component is not active. Current state [{}] for processor [{}][{}] tenant [{}]", state, entityId.getEntityType(), entityId, tenantId); log.debug("Component is not active. Current state [{}] for processor [{}][{}] tenant [{}]", state, entityId.getEntityType(), entityId, tenantId);
throw new IllegalStateException("Rule chain is not active! " + entityId + " - " + tenantId); RuleNodeException ruleNodeException = getInactiveException();
if (tbMsg != null) {
tbMsg.getCallback().onFailure(ruleNodeException);
}
throw ruleNodeException;
} }
} }
abstract protected RuleNodeException getInactiveException();
} }

View File

@ -36,9 +36,15 @@ public class RuleNodeException extends RuleEngineException {
public RuleNodeException(String message, String ruleChainName, RuleNode ruleNode) { public RuleNodeException(String message, String ruleChainName, RuleNode ruleNode) {
super(message); super(message);
this.ruleChainName = ruleChainName; this.ruleChainName = ruleChainName;
this.ruleNodeName = ruleNode.getName(); if (ruleNode != null) {
this.ruleChainId = ruleNode.getRuleChainId(); this.ruleNodeName = ruleNode.getName();
this.ruleNodeId = ruleNode.getId(); this.ruleChainId = ruleNode.getRuleChainId();
this.ruleNodeId = ruleNode.getId();
} else {
ruleNodeName = "Unknown";
ruleChainId = new RuleChainId(RuleChainId.NULL_UUID);
ruleNodeId = new RuleNodeId(RuleNodeId.NULL_UUID);
}
} }
public String toJsonString() { public String toJsonString() {