|
|
|
@ -36,6 +36,7 @@ import org.springframework.stereotype.Component; |
|
|
|
import org.thingsboard.rule.engine.api.MailService; |
|
|
|
import org.thingsboard.rule.engine.api.RuleChainTransactionService; |
|
|
|
import org.thingsboard.server.actors.service.ActorService; |
|
|
|
import org.thingsboard.server.actors.tenant.DebugTbRateLimits; |
|
|
|
import org.thingsboard.server.common.data.DataConstants; |
|
|
|
import org.thingsboard.server.common.data.Event; |
|
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
|
@ -43,6 +44,7 @@ import org.thingsboard.server.common.data.id.TenantId; |
|
|
|
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|
|
|
import org.thingsboard.server.common.msg.TbMsg; |
|
|
|
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|
|
|
import org.thingsboard.server.common.msg.tools.TbRateLimits; |
|
|
|
import org.thingsboard.server.common.transport.auth.DeviceAuthService; |
|
|
|
import org.thingsboard.server.dao.alarm.AlarmService; |
|
|
|
import org.thingsboard.server.dao.asset.AssetService; |
|
|
|
@ -84,6 +86,8 @@ import java.io.IOException; |
|
|
|
import java.io.PrintWriter; |
|
|
|
import java.io.StringWriter; |
|
|
|
import java.util.Optional; |
|
|
|
import java.util.concurrent.ConcurrentHashMap; |
|
|
|
import java.util.concurrent.ConcurrentMap; |
|
|
|
|
|
|
|
@Slf4j |
|
|
|
@Component |
|
|
|
@ -92,6 +96,12 @@ public class ActorSystemContext { |
|
|
|
|
|
|
|
protected final ObjectMapper mapper = new ObjectMapper(); |
|
|
|
|
|
|
|
private final ConcurrentMap<TenantId, DebugTbRateLimits> debugPerTenantLimits = new ConcurrentHashMap<>(); |
|
|
|
|
|
|
|
public ConcurrentMap<TenantId, DebugTbRateLimits> getDebugPerTenantLimits() { |
|
|
|
return debugPerTenantLimits; |
|
|
|
} |
|
|
|
|
|
|
|
@Getter |
|
|
|
@Setter |
|
|
|
private ActorService actorService; |
|
|
|
@ -291,6 +301,14 @@ public class ActorSystemContext { |
|
|
|
@Getter |
|
|
|
private long sessionReportTimeout; |
|
|
|
|
|
|
|
@Value("${actors.rule.chain.debug_mode_rate_limits_per_tenant.enabled}") |
|
|
|
@Getter |
|
|
|
private boolean debugPerTenantEnabled; |
|
|
|
|
|
|
|
@Value("${actors.rule.chain.debug_mode_rate_limits_per_tenant.configuration}") |
|
|
|
@Getter |
|
|
|
private String debugPerTenantLimitsConfiguration; |
|
|
|
|
|
|
|
@Getter |
|
|
|
@Setter |
|
|
|
private ActorSystem actorSystem; |
|
|
|
@ -318,8 +336,6 @@ public class ActorSystemContext { |
|
|
|
@Getter |
|
|
|
private CassandraBufferedRateExecutor cassandraBufferedRateExecutor; |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
public ActorSystemContext() { |
|
|
|
config = ConfigFactory.parseResources(AKKA_CONF_FILE_NAME).withFallback(ConfigFactory.load()); |
|
|
|
} |
|
|
|
@ -392,46 +408,97 @@ public class ActorSystemContext { |
|
|
|
} |
|
|
|
|
|
|
|
private void persistDebugAsync(TenantId tenantId, EntityId entityId, String type, TbMsg tbMsg, String relationType, Throwable error) { |
|
|
|
try { |
|
|
|
Event event = new Event(); |
|
|
|
event.setTenantId(tenantId); |
|
|
|
event.setEntityId(entityId); |
|
|
|
event.setType(DataConstants.DEBUG_RULE_NODE); |
|
|
|
|
|
|
|
String metadata = mapper.writeValueAsString(tbMsg.getMetaData().getData()); |
|
|
|
|
|
|
|
ObjectNode node = mapper.createObjectNode() |
|
|
|
.put("type", type) |
|
|
|
.put("server", getServerAddress()) |
|
|
|
.put("entityId", tbMsg.getOriginator().getId().toString()) |
|
|
|
.put("entityName", tbMsg.getOriginator().getEntityType().name()) |
|
|
|
.put("msgId", tbMsg.getId().toString()) |
|
|
|
.put("msgType", tbMsg.getType()) |
|
|
|
.put("dataType", tbMsg.getDataType().name()) |
|
|
|
.put("relationType", relationType) |
|
|
|
.put("data", tbMsg.getData()) |
|
|
|
.put("metadata", metadata); |
|
|
|
|
|
|
|
if (error != null) { |
|
|
|
node = node.put("error", toString(error)); |
|
|
|
if (checkLimits(tenantId, tbMsg, error)) { |
|
|
|
try { |
|
|
|
Event event = new Event(); |
|
|
|
event.setTenantId(tenantId); |
|
|
|
event.setEntityId(entityId); |
|
|
|
event.setType(DataConstants.DEBUG_RULE_NODE); |
|
|
|
|
|
|
|
String metadata = mapper.writeValueAsString(tbMsg.getMetaData().getData()); |
|
|
|
|
|
|
|
ObjectNode node = mapper.createObjectNode() |
|
|
|
.put("type", type) |
|
|
|
.put("server", getServerAddress()) |
|
|
|
.put("entityId", tbMsg.getOriginator().getId().toString()) |
|
|
|
.put("entityName", tbMsg.getOriginator().getEntityType().name()) |
|
|
|
.put("msgId", tbMsg.getId().toString()) |
|
|
|
.put("msgType", tbMsg.getType()) |
|
|
|
.put("dataType", tbMsg.getDataType().name()) |
|
|
|
.put("relationType", relationType) |
|
|
|
.put("data", tbMsg.getData()) |
|
|
|
.put("metadata", metadata); |
|
|
|
|
|
|
|
if (error != null) { |
|
|
|
node = node.put("error", toString(error)); |
|
|
|
} |
|
|
|
|
|
|
|
event.setBody(node); |
|
|
|
ListenableFuture<Event> future = eventService.saveAsync(event); |
|
|
|
Futures.addCallback(future, new FutureCallback<Event>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(@Nullable Event event) { |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable th) { |
|
|
|
log.error("Could not save debug Event for Node", th); |
|
|
|
} |
|
|
|
}); |
|
|
|
} catch (IOException ex) { |
|
|
|
log.warn("Failed to persist rule node debug message", ex); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
event.setBody(node); |
|
|
|
ListenableFuture<Event> future = eventService.saveAsync(event); |
|
|
|
Futures.addCallback(future, new FutureCallback<Event>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(@Nullable Event event) { |
|
|
|
private boolean checkLimits(TenantId tenantId, TbMsg tbMsg, Throwable error) { |
|
|
|
if (debugPerTenantEnabled) { |
|
|
|
DebugTbRateLimits debugTbRateLimits = debugPerTenantLimits.computeIfAbsent(tenantId, id -> |
|
|
|
new DebugTbRateLimits(new TbRateLimits(debugPerTenantLimitsConfiguration), false)); |
|
|
|
|
|
|
|
if (!debugTbRateLimits.getTbRateLimits().tryConsume()) { |
|
|
|
if (!debugTbRateLimits.isRuleChainEventSaved()) { |
|
|
|
persistRuleChainDebugModeEvent(tenantId, tbMsg.getRuleChainId(), error); |
|
|
|
debugTbRateLimits.setRuleChainEventSaved(true); |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable th) { |
|
|
|
log.error("Could not save debug Event for Node", th); |
|
|
|
if (log.isTraceEnabled()) { |
|
|
|
log.trace("[{}] Tenant level debug mode rate limit detected: {}", tenantId, tbMsg); |
|
|
|
} |
|
|
|
}); |
|
|
|
} catch (IOException ex) { |
|
|
|
log.warn("Failed to persist rule node debug message", ex); |
|
|
|
return false; |
|
|
|
} |
|
|
|
} |
|
|
|
return true; |
|
|
|
} |
|
|
|
|
|
|
|
private void persistRuleChainDebugModeEvent(TenantId tenantId, EntityId entityId, Throwable error) { |
|
|
|
Event event = new Event(); |
|
|
|
event.setTenantId(tenantId); |
|
|
|
event.setEntityId(entityId); |
|
|
|
event.setType(DataConstants.DEBUG_RULE_CHAIN); |
|
|
|
|
|
|
|
ObjectNode node = mapper.createObjectNode() |
|
|
|
//todo: what fields are needed here?
|
|
|
|
.put("server", getServerAddress()) |
|
|
|
.put("message", "Reached debug mode rate limit!"); |
|
|
|
|
|
|
|
if (error != null) { |
|
|
|
node = node.put("error", toString(error)); |
|
|
|
} |
|
|
|
|
|
|
|
event.setBody(node); |
|
|
|
ListenableFuture<Event> future = eventService.saveAsync(event); |
|
|
|
Futures.addCallback(future, new FutureCallback<Event>() { |
|
|
|
@Override |
|
|
|
public void onSuccess(@Nullable Event event) { |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
public void onFailure(Throwable th) { |
|
|
|
log.error("Could not save debug Event for Rule Chain", th); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
|
|
|
|
public static Exception toException(Throwable error) { |
|
|
|
|
Vladyslav_Prykhodko
7 years ago