dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #11672 from YevhenBondarenko/fix/rule-engine-oom

Browse Source 
[Rule Engine] fixed RuleEngine OOM
This commit is contained in:
Andrew Shvayka 2024-09-18 17:28:21 +03:00 committed by GitHub
commit c31ca8f841
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 90 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
application/src/main/java/org/thingsboard/server/actors/ruleChain/RuleChainManagerActor.java
View File

@ -95,7 +95,7 @@ public abstract class RuleChainManagerActor extends ContextAwareActor {
() -> {
RuleChain ruleChain = provider.apply(ruleChainId);
if (ruleChain == null) {
return new RuleChainErrorActor.ActorCreator(systemContext, tenantId,
return new RuleChainErrorActor.ActorCreator(systemContext, tenantId, ruleChainId,
new RuleEngineException("Rule Chain with id: " + ruleChainId + " not found!"));
} else {
return new RuleChainActor.ActorCreator(systemContext, tenantId, ruleChain);

14
application/src/main/java/org/thingsboard/server/actors/shared/RuleChainErrorActor.java
View File

@ -19,16 +19,15 @@ import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.TbActor;
import org.thingsboard.server.actors.TbActorId;
import org.thingsboard.server.actors.TbStringActorId;
import org.thingsboard.server.actors.TbEntityActorId;
import org.thingsboard.server.actors.service.ContextAwareActor;
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.msg.TbActorMsg;
import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg;
import org.thingsboard.server.common.msg.queue.RuleEngineException;
import java.util.UUID;
@Slf4j
public class RuleChainErrorActor extends ContextAwareActor {
@ -43,9 +42,8 @@ public class RuleChainErrorActor extends ContextAwareActor {
@Override
protected boolean doProcess(TbActorMsg msg) {
if (msg instanceof RuleChainAwareMsg) {
if (msg instanceof RuleChainAwareMsg rcMsg) {
log.debug("[{}] Reply with {} for message {}", tenantId, error.getMessage(), msg);
var rcMsg = (RuleChainAwareMsg) msg;
rcMsg.getMsg().getCallback().onFailure(error);
return true;
} else {
@ -56,17 +54,19 @@ public class RuleChainErrorActor extends ContextAwareActor {
public static class ActorCreator extends ContextBasedCreator {
private final TenantId tenantId;
private final RuleChainId ruleChainId;
private final RuleEngineException error;
public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleEngineException error) {
public ActorCreator(ActorSystemContext context, TenantId tenantId, RuleChainId ruleChainId, RuleEngineException error) {
super(context);
this.tenantId = tenantId;
this.ruleChainId = ruleChainId;
this.error = error;
}
@Override
public TbActorId createActorId() {
return new TbStringActorId(UUID.randomUUID().toString());
return new TbEntityActorId(ruleChainId);
}
@Override

91
application/src/test/java/org/thingsboard/server/actors/tenant/TenantActorTest.java
View File

@ -18,57 +18,130 @@ package org.thingsboard.server.actors.tenant;
import org.junit.Before;
import org.junit.Test;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.DefaultTbActorSystem;
import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.actors.TbActorMailbox;
import org.thingsboard.server.actors.TbActorRef;
import org.thingsboard.server.actors.TbActorSystem;
import org.thingsboard.server.actors.TbActorSystemSettings;
import org.thingsboard.server.actors.TbEntityActorId;
import org.thingsboard.server.actors.ruleChain.RuleChainActor;
import org.thingsboard.server.actors.ruleChain.RuleChainToRuleChainMsg;
import org.thingsboard.server.actors.shared.RuleChainErrorActor;
import org.thingsboard.server.common.data.ApiUsageState;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg;
import org.thingsboard.server.dao.rule.RuleChainService;
import org.thingsboard.server.dao.tenant.TenantService;
import org.thingsboard.server.queue.discovery.PartitionService;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
import org.thingsboard.server.service.apiusage.TbApiUsageStateService;
import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.thingsboard.server.actors.service.DefaultActorService.RULE_DISPATCHER_NAME;
public class TenantActorTest {
TenantActor tenantActor;
TbActorCtx ctx;
ActorSystemContext systemContext;
RuleChainService ruleChainService;
PartitionService partitionService;
TenantId tenantId = TenantId.SYS_TENANT_ID;
DeviceId deviceId = DeviceId.fromString("78bf9b26-74ef-4af2-9cfb-ad6cf24ad2ec");
RuleChainId ruleChainId = new RuleChainId(UUID.fromString("48cfa2b0-3dca-11ef-8d1a-37c2894cc59c"));
@Before
public void setUp() throws Exception {
systemContext = mock(ActorSystemContext.class);
ctx = mock(TbActorCtx.class);
ruleChainService = mock(RuleChainService.class);
partitionService = mock();
TbServiceInfoProvider serviceInfoProvider = mock(TbServiceInfoProvider.class);
TbApiUsageStateService apiUsageService = mock(TbApiUsageStateService.class);
TenantService tenantService = mock(TenantService.class);
when(systemContext.getRuleChainService()).thenReturn(ruleChainService);
tenantActor = (TenantActor) new TenantActor.ActorCreator(systemContext, tenantId).createActor();
when(systemContext.getTenantService()).thenReturn(mock(TenantService.class));
tenantActor.init(ctx);
tenantActor.cantFindTenant = false;
when(tenantService.findTenantById(tenantId)).thenReturn(mock());
when(systemContext.getTenantService()).thenReturn(tenantService);
when(serviceInfoProvider.isService(ServiceType.TB_CORE)).thenReturn(true);
when(serviceInfoProvider.isService(ServiceType.TB_RULE_ENGINE)).thenReturn(true);
when(systemContext.getServiceInfoProvider()).thenReturn(serviceInfoProvider);
when(partitionService.isManagedByCurrentService(tenantId)).thenReturn(true);
when(systemContext.getPartitionService()).thenReturn(partitionService);
when(systemContext.getApiUsageStateService()).thenReturn(apiUsageService);
when(apiUsageService.getApiUsageState(tenantId)).thenReturn(new ApiUsageState());
}
@Test
public void deleteDeviceTest() {
public void deleteDeviceTest() throws Exception {
TbActorCtx ctx = mock(TbActorCtx.class);
tenantActor.init(ctx);
TbActorRef deviceActorRef = mock(TbActorRef.class);
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 0,true));
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 0, true));
when(ctx.getOrCreateChildActor(any(), any(), any(), any())).thenReturn(deviceActorRef);
ComponentLifecycleMsg componentLifecycleMsg = new ComponentLifecycleMsg(tenantId, deviceId, ComponentLifecycleEvent.DELETED);
tenantActor.doProcess(componentLifecycleMsg);
verify(deviceActorRef).tellWithHighPriority(eq(new DeviceDeleteMsg(tenantId, deviceId)));
reset(ctx, deviceActorRef);
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 1,false));
when(systemContext.resolve(ServiceType.TB_CORE, tenantId, deviceId)).thenReturn(new TopicPartitionInfo("Main", tenantId, 1, false));
tenantActor.doProcess(componentLifecycleMsg);
verify(ctx, never()).getOrCreateChildActor(any(), any(), any(), any());
verify(deviceActorRef, never()).tellWithHighPriority(any());
}
@Test
public void ruleChainErrorActorTest() throws Exception {
TbActorSystemSettings settings = new TbActorSystemSettings(0, 0, 0);
TbActorSystem system = spy(new DefaultTbActorSystem(settings));
system.createDispatcher(RULE_DISPATCHER_NAME, mock());
TbActorMailbox tenantCtx = new TbActorMailbox(system, settings, null, mock(), mock(), null);
tenantActor.init(tenantCtx);
TbMsg msg = mock(TbMsg.class);
when(ruleChainService.findRuleChainById(tenantId, ruleChainId)).thenReturn(new RuleChain(ruleChainId));
RuleChainToRuleChainMsg ruleChainMsg = new RuleChainToRuleChainMsg(ruleChainId, null, msg, null);
tenantActor.doProcess(ruleChainMsg);
verify(system).createChildActor(eq(RULE_DISPATCHER_NAME), any(RuleChainActor.ActorCreator.class), any());
reset(system);
tenantActor.doProcess(ruleChainMsg);
verify(system, never()).createChildActor(any(), any(), any());
//Delete rule-chain
TbActorRef ruleChainActor = system.getActor(new TbEntityActorId(ruleChainId));
assertNotNull(ruleChainActor);
system.stop(ruleChainActor);
when(ruleChainService.findRuleChainById(tenantId, ruleChainId)).thenReturn(null);
tenantActor.doProcess(ruleChainMsg);
verify(system).createChildActor(eq(RULE_DISPATCHER_NAME), any(RuleChainErrorActor.ActorCreator.class), any());
reset(system);
tenantActor.doProcess(ruleChainMsg);
verify(system, never()).createChildActor(any(), any(), any());
system.stop();
}
}