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

Merge remote-tracking branch 'origin/lts-4.2-fix/cf-processing' into lts-4.3-fix/cf-processing

pull/14891/head
IrynaMatveieva 6 months ago
parent
1fcc419dbc d568ee8536
commit
c998439fae
3 changed files with 147 additions and 28 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. 82
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  3. 91
      application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java

82
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();
@ -179,7 +179,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} catch (Exception e) {
log.debug("[{}][{}] Failed to initialize CF state", entityId, ctx.getCfId(), e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
persistDebugErrorIfEnabled(cfe, msg.getCallback());
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -261,9 +262,10 @@ 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;
persistDebugErrorIfEnabled(cfe, msg.getCallback());
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -273,31 +275,40 @@ 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;
}
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);
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) {
persistDebugErrorIfEnabled(cfe, msg.getCallback());
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
msg.getCallback().onSuccess();
}
public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException {
@ -340,7 +351,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} catch (Exception e) {
log.debug("[{}][{}] Failed to process linked CF telemetry msg: {}", entityId, ctx.getCfId(), msg, e);
if (e instanceof CalculatedFieldException cfe) {
throw cfe;
persistDebugErrorIfEnabled(cfe, callback);
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -366,7 +378,8 @@ 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;
persistDebugErrorIfEnabled(cfe, callback);
return;
}
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -384,7 +397,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();
}
}
@ -786,4 +799,19 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
return null;
}
private void persistDebugErrorIfEnabled(CalculatedFieldException cfe, TbCallback callback) {
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);
}
callback.onSuccess();
}
}

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