tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge pull request #14891 from irynamatveieva/lts-4.3-fix/cf-processing 

[LTS-4.3] Fix processing calculated fields
pull/14897/head
Viacheslav Klimov 6 months ago
committed by GitHub
parent
9abf5e5ea7 e72cbfc5df
commit
5965abdc6b
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
5 changed files with 142 additions and 34 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 13
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 10
      application/src/main/java/org/thingsboard/server/actors/calculatedField/AbstractCalculatedFieldActor.java
  3. 60
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  4. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  5. 91
      application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java

13
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View File

@ -42,6 +42,7 @@ import org.thingsboard.rule.engine.api.notification.SlackService;
import org.thingsboard.rule.engine.api.sms.SmsSenderFactory;
import org.thingsboard.script.api.js.JsInvokeService;
import org.thingsboard.script.api.tbel.TbelInvokeService;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldException;
import org.thingsboard.server.actors.service.ActorService;
import org.thingsboard.server.actors.tenant.DebugTbRateLimits;
import org.thingsboard.server.cache.limits.RateLimitService;
@ -831,6 +832,18 @@ public class ActorSystemContext {
Futures.addCallback(future, RULE_CHAIN_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
}
public void persistCalculatedFieldDebugError(CalculatedFieldException cfe) {
String message;
if (cfe.getErrorMessage() != null) {
message = cfe.getErrorMessage();
} else if (cfe.getCause() != null) {
message = cfe.getCause().getMessage();
} else {
message = "N/A";
}
persistCalculatedFieldDebugEvent(cfe.getCtx().getTenantId(), cfe.getCtx().getCfId(), cfe.getEventEntity(), cfe.getArguments(), cfe.getMsgId(), cfe.getMsgType(), null, message);
}
public void persistCalculatedFieldDebugEvent(TenantId tenantId, CalculatedFieldId calculatedFieldId, EntityId entityId, JsonNode arguments, UUID tbMsgId, String tbMsgType, String result, String errorMessage) {
if (checkLimits(tenantId)) {
try {

10
application/src/main/java/org/thingsboard/server/actors/calculatedField/AbstractCalculatedFieldActor.java
View File

@ -41,15 +41,7 @@ public abstract class AbstractCalculatedFieldActor extends ContextAwareActor {
return doProcessCfMsg(cfm);
} catch (CalculatedFieldException cfe) {
if (DebugModeUtil.isDebugFailuresAvailable(cfe.getCtx().getCalculatedField())) {
String message;
if (cfe.getErrorMessage() != null) {
message = cfe.getErrorMessage();
} else if (cfe.getCause() != null) {
message = cfe.getCause().getMessage();
} else {
message = "N/A";
}
systemContext.persistCalculatedFieldDebugEvent(tenantId, cfe.getCtx().getCfId(), cfe.getEventEntity(), cfe.getArguments(), cfe.getMsgId(), cfe.getMsgType(), null, message);
systemContext.persistCalculatedFieldDebugError(cfe);
}
cause = cfe.getCause();
} catch (Exception e) {

60
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
View File

@ -171,7 +171,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
if (state.isSizeOk()) {
processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, msg.getEventType().name(), msg.getCallback());
} else {
throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
}
} else {
msg.getCallback().onSuccess();
@ -261,7 +261,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
}
} catch (Exception e) {
log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e);
log.debug("[{}][{}] Failed to handle relation update", entityId, ctx.getCfId(), e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
}
@ -273,31 +273,39 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
CalculatedFieldCtx ctx = msg.getCalculatedField();
CalculatedFieldId cfId = ctx.getCfId();
CalculatedFieldState state = states.get(cfId);
if (state == null) {
msg.getCallback().onSuccess();
return;
}
if (state instanceof RelatedEntitiesAggregationCalculatedFieldState aggState) {
aggState.cleanupEntityData(msg.getRelatedEntityId());
try {
if (state == null) {
msg.getCallback().onSuccess();
return;
}
if (state instanceof RelatedEntitiesAggregationCalculatedFieldState aggState) {
aggState.cleanupEntityData(msg.getRelatedEntityId());
state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
state.checkStateSize(new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId), ctx.getMaxStateSize());
if (state.isSizeOk()) {
processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, TbMsgType.RELATION_DELETED.name(), msg.getCallback());
} else {
throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
if (state.isSizeOk()) {
processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, TbMsgType.RELATION_DELETED.name(), msg.getCallback());
} else {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
}
return;
}
if (state instanceof PropagationCalculatedFieldState propagationState) {
PropagationArgumentEntry entry = new PropagationArgumentEntry();
entry.setRemoved(msg.getRelatedEntityId());
propagationState.update(Map.of(PROPAGATION_CONFIG_ARGUMENT, entry), ctx);
if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) {
systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArgumentsJson(), null, TbMsgType.RELATION_DELETED.name(), null, null);
}
}
return;
}
if (state instanceof PropagationCalculatedFieldState propagationState) {
PropagationArgumentEntry entry = new PropagationArgumentEntry();
entry.setRemoved(msg.getRelatedEntityId());
propagationState.update(Map.of(PROPAGATION_CONFIG_ARGUMENT, entry), ctx);
if (DebugModeUtil.isDebugAllAvailable(ctx.getCalculatedField())) {
systemContext.persistCalculatedFieldDebugEvent(tenantId, ctx.getCfId(), entityId, state.getArgumentsJson(), null, TbMsgType.RELATION_DELETED.name(), null, null);
msg.getCallback().onSuccess();
} catch (Exception e) {
log.debug("[{}][{}] Failed to handle relation delete", entityId, ctx.getCfId(), e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
msg.getCallback().onSuccess();
}
public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException {
@ -366,7 +374,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} catch (Exception e) {
log.debug("[{}][{}] Failed to process CF telemetry msg: {}", entityId, ctx.getCfId(), proto, e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
if (DebugModeUtil.isDebugFailuresAvailable(cfe.getCtx().getCalculatedField())) {
systemContext.persistCalculatedFieldDebugError(cfe);
}
callback.onSuccess();
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -384,7 +396,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
log.debug("[{}][{}] Reevaluating CF state", entityId, cfId);
processStateIfReady(state, null, ctx, Collections.singletonList(cfId), null, CalculatedFieldEventType.REEVALUATION_MSG.name(), msg.getCallback());
} else {
throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
}
}

2
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
View File

@ -804,7 +804,7 @@ public class CalculatedFieldCtx implements Closeable {
}
public String getSizeExceedsLimitMessage() {
return "Failed to init CF state. State size exceeds limit of " + (maxStateSize / 1024) + "Kb!";
return "State size exceeds limit of " + (maxStateSize / 1024) + "Kb!";
}
public boolean hasCurrentOwnerSourceArguments() {

91
application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
View File

@ -20,10 +20,12 @@ import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.NullNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.junit.Test;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.EntityInfo;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.EventInfo;
import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.asset.Asset;
import org.thingsboard.server.common.data.asset.AssetProfile;
@ -45,7 +47,9 @@ import org.thingsboard.server.common.data.cf.configuration.geofencing.Geofencing
import org.thingsboard.server.common.data.cf.configuration.geofencing.ZoneGroupConfiguration;
import org.thingsboard.server.common.data.debug.DebugSettings;
import org.thingsboard.server.common.data.id.AssetProfileId;
import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.EntitySearchDirection;
import org.thingsboard.server.common.data.relation.RelationPathLevel;
@ -1390,6 +1394,93 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
});
}
@Test
public void testCalculatedFieldsWhenOneIsInvalid() throws Exception {
Device testDevice = createDevice("Test device", "1234567890");
long now = System.currentTimeMillis();
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + AttributeScope.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"a\":5}}", now - TimeUnit.MINUTES.toMillis(3))));
// Script CF - invalid
CalculatedField invalidCF = new CalculatedField();
invalidCF.setEntityId(testDevice.getId());
invalidCF.setType(CalculatedFieldType.SCRIPT);
invalidCF.setName("Script CF");
invalidCF.setDebugSettings(DebugSettings.all());
ScriptCalculatedFieldConfiguration scriptConfig = new ScriptCalculatedFieldConfiguration();
ReferencedEntityKey refEntityKeyA = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
Argument argumentA = new Argument();
argumentA.setRefEntityKey(refEntityKeyA);
scriptConfig.setArguments(Map.of("a", argumentA));
scriptConfig.setExpression("""
return {
"temperature": temp
};
""");
scriptConfig.setOutput(new TimeSeriesOutput());
invalidCF.setConfiguration(scriptConfig);
invalidCF = doPost("/api/calculatedField", invalidCF, CalculatedField.class);
CalculatedFieldId invalidCfId = invalidCF.getId();
await().alias("create invalid CF -> check error").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
PageData<EventInfo> debugEvents = getDebugEvents(tenantId, invalidCfId, 1);
if (!debugEvents.getData().isEmpty()) {
EventInfo eventInfo = debugEvents.getData().get(0);
assertThat(eventInfo.getBody().has("error")).isTrue();
}
});
// Simple CF - valid
CalculatedField validCF = new CalculatedField();
validCF.setEntityId(testDevice.getId());
validCF.setType(CalculatedFieldType.SIMPLE);
validCF.setName("Simple CF");
validCF.setDebugSettings(DebugSettings.all());
SimpleCalculatedFieldConfiguration simpleConfig = new SimpleCalculatedFieldConfiguration();
simpleConfig.setArguments(Map.of("a", argumentA));
simpleConfig.setExpression("a+1");
TimeSeriesOutput simpleOutput = new TimeSeriesOutput();
simpleOutput.setName("a+1");
simpleOutput.setDecimalsByDefault(0);
simpleConfig.setOutput(simpleOutput);
validCF.setConfiguration(simpleConfig);
validCF = doPost("/api/calculatedField", validCF, CalculatedField.class);
CalculatedFieldId validCfId = validCF.getId();
await().alias("create CF -> check initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
PageData<EventInfo> debugEvents = getDebugEvents(tenantId, validCfId, 1);
if (!debugEvents.getData().isEmpty()) {
EventInfo eventInfo = debugEvents.getData().get(0);
assertThat(eventInfo.getBody().has("error")).isFalse();
}
ObjectNode result = getLatestTelemetry(testDevice.getId(), "a+1");
assertThat(result).isNotNull();
assertThat(result.get("a+1").get(0).get("value").asText()).isEqualTo("6");
});
doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + AttributeScope.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":6}"));
await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode result = getLatestTelemetry(testDevice.getId(), "a+1");
assertThat(result).isNotNull();
assertThat(result.get("a+1").get(0).get("value").asText()).isEqualTo("7");
});
}
private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception {
return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class);
}

Write Preview
Loading…
Cancel
Save