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

Merge branch 'feature/entity-alarm-rules' of github.com:thingsboard/thingsboard into feature/aggregation-cf

pull/14141/head
IrynaMatveieva 9 months ago
parent
269794ec27 4c656fe89d
commit
11e4e2f17b
27 changed files with 274 additions and 119 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. 5
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 32
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  3. 34
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  4. 56
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  5. 14
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java
  6. 22
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java
  7. 1
      application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java
  8. 18
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java
  9. 63
      application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java
  10. 35
      application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java
  11. 2
      application/src/test/resources/logback-test.xml
  12. 4
      common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java
  13. 2
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java
  14. 1
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java
  15. 8
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java
  16. 2
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java
  17. 5
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java
  18. 5
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java
  19. 6
      common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java
  20. 35
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java
  21. 3
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java
  22. 2
      common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java
  23. 29
      common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java
  24. 2
      common/proto/src/main/proto/queue.proto
  25. 2
      dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java
  26. 1
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmRuleState.java
  27. 4
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java

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

@ -860,8 +860,9 @@ public class ActorSystemContext {
if (errorMessage != null) {
eventBuilder.error(errorMessage);
}
ListenableFuture<Void> future = eventService.saveAsync(eventBuilder.build());
CalculatedFieldDebugEvent event = eventBuilder.build();
log.debug("Persisting calculated field debug event: {}", event);
ListenableFuture<Void> future = eventService.saveAsync(event);
Futures.addCallback(future, CALCULATED_FIELD_DEBUG_EVENT_ERROR_CALLBACK, MoreExecutors.directExecutor());
} catch (IllegalArgumentException ex) {
log.warn("Failed to persist calculated field debug message", ex);

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

@ -175,7 +175,6 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
public void process(CalculatedFieldArgumentResetMsg msg) throws CalculatedFieldException {
log.debug("[{}] Processing CF argument reset msg.", entityId);
var ctx = msg.getCtx();
var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
try {
Map<String, Argument> dynamicSourceArgs = ctx.getArguments().entrySet().stream()
.filter(entry -> entry.getValue().hasOwnerSource())
@ -184,7 +183,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
Map<String, ArgumentEntry> fetchedArgs = cfService.fetchArgsFromDb(tenantId, entityId, dynamicSourceArgs);
fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true));
processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), callback, fetchedArgs, null, null);
processArgumentValuesUpdate(ctx, Collections.singletonList(ctx.getCfId()), msg.getCallback(), fetchedArgs, null, null);
} catch (Exception e) {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).cause(e).build();
}
@ -277,7 +276,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
public void process(EntityCalculatedFieldTelemetryMsg msg) throws CalculatedFieldException {
log.trace("[{}] Processing CF telemetry msg: {}", msg.getEntityId(), msg);
var proto = msg.getProto();
var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size());
var numberOfCallbacks = msg.getEntityIdFields().size() + msg.getProfileIdFields().size();
MultipleTbCallback callback = new MultipleTbCallback(numberOfCallbacks, msg.getCallback());
List<CalculatedFieldId> cfIdList = getCalculatedFieldIds(proto);
Set<CalculatedFieldId> cfIdSet = new HashSet<>(cfIdList);
@ -293,11 +292,11 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
log.trace("[{}] Processing CF link telemetry msg: {}", msg.getEntityId(), msg);
var proto = msg.getProto();
var ctx = msg.getCtx();
var callback = new MultipleTbCallback(CALLBACKS_PER_CF, msg.getCallback());
var callback = msg.getCallback();
try {
List<CalculatedFieldId> cfIds = getCalculatedFieldIds(proto);
if (cfIds.contains(ctx.getCfId())) {
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
} else {
if (proto.getTsDataCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArguments(ctx, msg.getEntityId(), proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
@ -308,7 +307,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} else if (proto.getRemovedAttrKeysCount() > 0) {
processArgumentValuesUpdate(ctx, cfIds, callback, mapToArgumentsWithDefaultValue(ctx, msg.getEntityId(), proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto));
} else {
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
}
}
} catch (Exception e) {
@ -317,10 +316,10 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
}
private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection<CalculatedFieldId> cfIds, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
private void process(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, Collection<CalculatedFieldId> cfIds, List<CalculatedFieldId> cfIdList, TbCallback callback) throws CalculatedFieldException {
try {
if (cfIds.contains(ctx.getCfId())) {
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
} else {
if (proto.getTsDataCount() > 0) {
processTelemetry(ctx, proto, cfIdList, callback);
@ -331,7 +330,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
} else if (proto.getRemovedAttrKeysCount() > 0) {
processRemovedAttributes(ctx, proto, cfIdList, callback);
} else {
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
}
}
} catch (Exception e) {
@ -371,27 +370,27 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
msg.getCallback().onSuccess();
}
private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
private void processTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getTsDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
private void processAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArguments(ctx, proto.getScope(), proto.getAttrDataList()), toTbMsgId(proto), toTbMsgType(proto));
}
private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
private void processRemovedTelemetry(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithFetchedValue(ctx, entityId, proto.getRemovedTsKeysList()), toTbMsgId(proto), toTbMsgType(proto));
}
private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback) throws CalculatedFieldException {
private void processRemovedAttributes(CalculatedFieldCtx ctx, CalculatedFieldTelemetryMsgProto proto, List<CalculatedFieldId> cfIdList, TbCallback callback) throws CalculatedFieldException {
processArgumentValuesUpdate(ctx, cfIdList, callback, mapToArgumentsWithDefaultValue(ctx, proto.getScope(), proto.getRemovedAttrKeysList()), toTbMsgId(proto), toTbMsgType(proto));
}
private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, MultipleTbCallback callback,
private void processArgumentValuesUpdate(CalculatedFieldCtx ctx, List<CalculatedFieldId> cfIdList, TbCallback callback,
Map<String, ArgumentEntry> newArgValues, UUID tbMsgId, TbMsgType tbMsgType) throws CalculatedFieldException {
if (newArgValues.isEmpty()) {
log.debug("[{}] No new argument values to process for CF.", ctx.getCfId());
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
}
CalculatedFieldState state = states.get(ctx.getCfId());
boolean justRestored = false;
@ -418,7 +417,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
cfIdList.add(ctx.getCfId());
processStateIfReady(state, updatedArgs, ctx, cfIdList, tbMsgId, tbMsgType, callback);
} else {
callback.onSuccess(CALLBACKS_PER_CF);
callback.onSuccess();
}
} else {
throw CalculatedFieldException.builder().ctx(ctx).eventEntity(entityId).errorMessage(ctx.getSizeExceedsLimitMessage()).build();
@ -459,6 +458,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
private void processStateIfReady(CalculatedFieldState state, Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx,
List<CalculatedFieldId> cfIdList, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException {
callback = new MultipleTbCallback(CALLBACKS_PER_CF, callback);
log.trace("[{}][{}] Processing state if ready. Current args: {}, updated args: {}", entityId, ctx.getCfId(), state.getArguments(), updatedArgs);
CalculatedFieldEntityCtxId ctxId = new CalculatedFieldEntityCtxId(tenantId, ctx.getCfId(), entityId);
boolean stateSizeChecked = false;

34
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
View File

@ -134,7 +134,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
public void stop() {
log.info("[{}] Stopping CF manager actor.", tenantId);
calculatedFields.values().forEach(CalculatedFieldCtx::stop);
calculatedFields.values().forEach(CalculatedFieldCtx::close);
calculatedFields.clear();
entityIdCalculatedFields.clear();
entityIdCalculatedFieldLinks.clear();
@ -398,7 +398,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
log.debug("[{}] Failed to lookup CF by id [{}]", tenantId, cfId);
callback.onSuccess();
} else {
var newCfCtx = getCfCtx(newCf); // fixme wtf? why isn't oldCfCtx closed properly? when to close it?
var newCfCtx = getCfCtx(newCf);
try {
newCfCtx.init();
} catch (Exception e) {
@ -441,14 +441,26 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
return;
}
applyToTargetCfEntityActors(newCfCtx, callback, (id, cb) -> initCfForEntity(id, newCfCtx, stateAction, cb));
applyToTargetCfEntityActors(newCfCtx, new TbCallback() {
@Override
public void onSuccess() {
oldCfCtx.close();
callback.onSuccess();
}
@Override
public void onFailure(Throwable t) {
oldCfCtx.close();
callback.onFailure(t);
}
}, (id, cb) -> initCfForEntity(id, newCfCtx, stateAction, cb));
}
}
}
private void onCfDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
var cfId = new CalculatedFieldId(msg.getEntityId().getId());
var cfCtx = calculatedFields.remove(cfId); // fixme wtf? why isn't ctx closed properly?
var cfCtx = calculatedFields.remove(cfId);
aggCalculatedFields.remove(cfId);
if (cfCtx == null) {
log.debug("[{}] CF was already deleted [{}]", tenantId, cfId);
@ -457,7 +469,19 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
}
entityIdCalculatedFields.get(cfCtx.getEntityId()).remove(cfCtx);
deleteLinks(cfCtx);
applyToTargetCfEntityActors(cfCtx, callback, (id, cb) -> deleteCfForEntity(id, cfId, cb));
applyToTargetCfEntityActors(cfCtx, new TbCallback() {
@Override
public void onSuccess() {
cfCtx.close();
callback.onSuccess();
}
@Override
public void onFailure(Throwable t) {
cfCtx.close();
callback.onFailure(t);
}
}, (id, cb) -> deleteCfForEntity(id, cfId, cb));
}
public void onTelemetryMsg(CalculatedFieldTelemetryMsg msg) {

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

@ -60,6 +60,7 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState;
import org.thingsboard.server.service.telemetry.AlarmSubscriptionService;
import java.io.Closeable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
@ -68,11 +69,10 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@Data
@Slf4j
public class CalculatedFieldCtx {
public class CalculatedFieldCtx implements Closeable {
private CalculatedField calculatedField;
@ -211,15 +211,12 @@ public class CalculatedFieldCtx {
}
case ALARM -> {
AlarmCalculatedFieldConfiguration configuration = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration();
Stream<AlarmRule> rules = configuration.getCreateRules().values().stream();
if (configuration.getClearRule() != null) {
rules = Stream.concat(rules, Stream.of(configuration.getClearRule()));
}
rules.map(rule -> rule.getCondition().getExpression()).forEach(expression -> {
if (expression instanceof TbelAlarmConditionExpression tbelExpression) {
initTbelExpression(tbelExpression.getExpression());
}
});
configuration.getAllRules().map(rule -> rule.getValue().getCondition().getExpression())
.forEach(expression -> {
if (expression instanceof TbelAlarmConditionExpression tbelExpression) {
initTbelExpression(tbelExpression.getExpression());
}
});
initialized = true;
}
case PROPAGATION -> {
@ -294,7 +291,6 @@ public class CalculatedFieldCtx {
public ScheduledFuture<?> scheduleReevaluation(long delayMs, TbActorRef actorCtx) {
log.debug("[{}] Scheduling CF reevaluation in {} ms", cfId, delayMs);
// TODO: use single lazy-loaded instance of CalculatedFieldReevaluateMsg
return systemContext.scheduleMsgWithDelay(actorCtx, new CalculatedFieldReevaluateMsg(tenantId, this), delayMs);
}
@ -578,8 +574,17 @@ public class CalculatedFieldCtx {
if (!Objects.equals(output, other.output)) {
return true;
}
if (cfType == CalculatedFieldType.ALARM && !calculatedField.getName().equals(other.getCalculatedField().getName())) {
return true;
if (cfType == CalculatedFieldType.ALARM) {
if (!calculatedField.getName().equals(other.getCalculatedField().getName())) {
return true;
}
var thisConfig = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration();
var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration();
if (!thisConfig.rulesEqual(otherConfig, AlarmRule::equals)) {
// if the rules have any changes not tracked by hasStateChanges
return true;
}
}
if (scheduledUpdateIntervalMillis != other.scheduledUpdateIntervalMillis) {
return true;
@ -599,8 +604,10 @@ public class CalculatedFieldCtx {
if (cfType == CalculatedFieldType.ALARM) {
var thisConfig = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration();
var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration();
if (!thisConfig.getCreateRules().equals(otherConfig.getCreateRules()) ||
!Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) {
if (!thisConfig.rulesEqual(otherConfig, (thisRule, otherRule) -> {
return thisRule.getCondition().getType() == otherRule.getCondition().getType();
})) {
// reinitializing only if the rule list changed, or if a condition type changed for any rule
return true;
}
}
@ -654,12 +661,17 @@ public class CalculatedFieldCtx {
};
}
public void stop() {
if (tbelExpressions != null) {
tbelExpressions.values().forEach(CalculatedFieldScriptEngine::destroy);
}
if (simpleExpressions != null) {
simpleExpressions.values().forEach(ThreadLocal::remove);
@Override
public void close() {
try {
if (tbelExpressions != null) {
tbelExpressions.values().forEach(CalculatedFieldScriptEngine::destroy);
}
if (simpleExpressions != null) {
simpleExpressions.values().forEach(ThreadLocal::remove);
}
} catch (Exception e) {
log.warn("Failed to stop {}", this, e);
}
}

14
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java
View File

@ -103,6 +103,18 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
this.configuration = getConfiguration(ctx);
this.alarmType = ctx.getCalculatedField().getName();
Map<AlarmSeverity, AlarmRule> createRules = configuration.getCreateRules();
createRules.forEach((severity, rule) -> {
AlarmRuleState ruleState = createRuleStates.get(severity);
if (ruleState != null) {
ruleState.setAlarmRule(rule);
}
});
AlarmRule clearRule = configuration.getClearRule();
if (clearRule != null && clearRuleState != null) {
clearRuleState.setAlarmRule(clearRule);
}
if (currentAlarm != null && !currentAlarm.getType().equals(alarmType)) {
currentAlarm = null;
initialFetchDone = false;
@ -265,7 +277,7 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
clearState(state);
}
AlarmApiCallResult clearResult = ctx.getAlarmService().clearAlarm(
ctx.getTenantId(), currentAlarm.getId(), System.currentTimeMillis(), createDetails(clearRuleState), true
ctx.getTenantId(), currentAlarm.getId(), System.currentTimeMillis(), createDetails(clearRuleState), false
);
if (clearResult.isCleared()) {
result = TbAlarmResult.builder()

22
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java
View File

@ -237,7 +237,15 @@ public class AlarmRuleState {
}
public void clear() {
clearRepeatingConditionState();
clearDurationConditionState();
}
private void clearRepeatingConditionState() {
eventCount = 0L;
}
private void clearDurationConditionState() {
firstEventTs = 0L;
lastEventTs = 0L;
duration = 0L;
@ -289,6 +297,20 @@ public class AlarmRuleState {
public void setAlarmRule(AlarmRule alarmRule) {
this.alarmRule = alarmRule;
this.condition = alarmRule.getCondition();
// clearing state for other condition types (possibly left from a previous condition type)
switch (condition.getType()) {
case SIMPLE -> {
clearRepeatingConditionState();
clearDurationConditionState();
}
case REPEATING -> {
clearDurationConditionState();
}
case DURATION -> {
clearRepeatingConditionState();
}
}
}
public StateInfo getStateInfo() {

1
application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java
View File

@ -257,7 +257,6 @@ public class EntityStateSourcingListener {
if (calculatedFieldCache.hasCalculatedFields(tenantId, alarm.getOriginator(), ctx -> ctx.getCfType() == CalculatedFieldType.ALARM)) {
ToCalculatedFieldMsg msg = ToCalculatedFieldMsg.newBuilder()
.setEventMsg(toProto(event))
.addCfTypes(CalculatedFieldType.ALARM.name())
.build();
tbClusterService.pushMsgToCalculatedFields(tenantId, alarm.getOriginator(), msg, new TbQueueCallback() {
@Override

18
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.service.queue;
import com.google.protobuf.ProtocolStringList;
import jakarta.annotation.PreDestroy;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Value;
@ -28,7 +27,6 @@ import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTeleme
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
@ -61,7 +59,6 @@ import org.thingsboard.server.service.queue.processing.AbstractPartitionBasedCon
import org.thingsboard.server.service.queue.processing.IdMsgPair;
import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService;
import java.util.EnumSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
@ -183,8 +180,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa
}
private void processMsg(ToCalculatedFieldMsg toCfMsg, UUID id, TbCallback callback) {
Set<CalculatedFieldType> cfTypes = getCfTypes(toCfMsg.getCfTypesList());
if (toCfMsg.hasTelemetryMsg()) { // TODO: add CF type filter to the message. or just rename the CF strategy to "Process alarms and calculated fields
if (toCfMsg.hasTelemetryMsg()) {
log.trace("[{}] Forwarding regular telemetry message for processing {}", id, toCfMsg.getTelemetryMsg());
forwardToActorSystem(toCfMsg.getTelemetryMsg(), callback);
} else if (toCfMsg.hasLinkedTelemetryMsg()) {
@ -264,18 +260,6 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractPartitionBa
return TenantId.fromUUID(new UUID(tenantIdMSB, tenantIdLSB));
}
private Set<CalculatedFieldType> getCfTypes(ProtocolStringList cfTypesList) {
Set<CalculatedFieldType> cfTypes;
if (cfTypesList.isEmpty()) {
cfTypes = EnumSet.allOf(CalculatedFieldType.class);
} else {
cfTypes = cfTypesList.stream()
.map(CalculatedFieldType::valueOf)
.collect(Collectors.toSet());
}
return cfTypes;
}
@Override
protected void stopConsumers() {
super.stopConsumers();

63
application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java
View File

@ -458,7 +458,14 @@ public class AlarmRulesTest extends AbstractControllerTest {
schedule = schedule.replace("\"enabled\":false", "\"enabled\":true");
postAttributes(deviceId, AttributeScope.SERVER_SCOPE, "{\"schedule\":" + schedule + "}");
checkAlarmResult(calculatedField, alarmResult -> {
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
// checking multiple debug events due to scheduled reevaluation (which also produces debug events)
CalculatedFieldDebugEvent debugEvent = getDebugEvents(calculatedField.getId(), 5).stream()
.filter(event -> event.getResult() != null)
.findFirst().orElse(null);
assertThat(debugEvent).isNotNull();
TbAlarmResult alarmResult = JacksonUtil.fromString(debugEvent.getResult(), TbAlarmResult.class);
assertThat(alarmResult.isCreated()).isTrue();
assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
@ -638,11 +645,11 @@ public class AlarmRulesTest extends AbstractControllerTest {
AlarmSeverity.CRITICAL, new Condition("return temperature >= 50 && humidity >= 50;", null, null)
);
CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature and Humidity Alarm",
arguments, createRules, null);
AlarmCalculatedFieldConfiguration configuration = (AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration();
configuration.getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails("""
temperature is ${temperature}, humidity is ${humidity}""");
calculatedField = saveCalculatedField(calculatedField);
arguments, createRules, null, configuration -> {
configuration.getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails(
"temperature is ${temperature}, humidity is ${humidity}"
);
});
postTelemetry(deviceId, "{\"temperature\":50}");
postAttributes(deviceId, AttributeScope.SERVER_SCOPE, "{\"humidity\":50}");
@ -653,6 +660,18 @@ public class AlarmRulesTest extends AbstractControllerTest {
assertThat(alarmResult.getAlarm().getDetails().get("data").asText())
.isEqualTo("temperature is 50, humidity is 50");
});
((AlarmCalculatedFieldConfiguration) calculatedField.getConfiguration()).getCreateRules().get(AlarmSeverity.CRITICAL).setAlarmDetails(
"UPDATED temperature is ${temperature}, humidity is ${humidity}"
);
calculatedField = saveCalculatedField(calculatedField);
checkAlarmResult(calculatedField, alarmResult -> {
assertThat(alarmResult.isCreated()).isFalse();
assertThat(alarmResult.isUpdated()).isTrue();
assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
assertThat(alarmResult.getAlarm().getDetails().get("data").asText())
.isEqualTo("UPDATED temperature is 50, humidity is 50");
});
}
@Test
@ -683,7 +702,12 @@ public class AlarmRulesTest extends AbstractControllerTest {
Thread.sleep(10000);
assertThat(getLatestAlarmResult(calculatedField.getId())).isNull();
checkAlarmResult(calculatedField, alarmResult -> {
await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> {
CalculatedFieldDebugEvent debugEvent = getDebugEvents(calculatedField.getId(), 5).stream()
.filter(event -> event.getResult() != null)
.findFirst().orElse(null);
assertThat(debugEvent).isNotNull();
TbAlarmResult alarmResult = JacksonUtil.fromString(debugEvent.getResult(), TbAlarmResult.class);
assertThat(alarmResult.isCreated()).isTrue();
assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
@ -764,24 +788,14 @@ public class AlarmRulesTest extends AbstractControllerTest {
String alarmType,
Map<String, Argument> arguments,
Map<AlarmSeverity, Condition> createConditions,
Condition clearCondition) {
Condition clearCondition,
Consumer<AlarmCalculatedFieldConfiguration>... modifier) {
Map<AlarmSeverity, AlarmRule> createRules = new HashMap<>();
createConditions.forEach((severity, condition) -> {
createRules.put(severity, toAlarmRule(condition));
});
AlarmRule clearRule = clearCondition != null ? toAlarmRule(clearCondition) : null;
CalculatedField calculatedField = createAlarmCf(entityId, alarmType, arguments, createRules, clearRule);
CalculatedFieldDebugEvent debugEvent = await().atMost(TIMEOUT, TimeUnit.SECONDS).until(() -> getDebugEvents(calculatedField.getId(), 1), events -> !events.isEmpty()).get(0);
latestEventId = debugEvent.getId();
return calculatedField;
}
private CalculatedField createAlarmCf(EntityId entityId,
String alarmType,
Map<String, Argument> arguments,
Map<AlarmSeverity, AlarmRule> createRules,
AlarmRule clearRule) {
CalculatedField calculatedField = new CalculatedField();
calculatedField.setEntityId(entityId);
calculatedField.setName(alarmType);
@ -792,7 +806,16 @@ public class AlarmRulesTest extends AbstractControllerTest {
configuration.setClearRule(clearRule);
calculatedField.setConfiguration(configuration);
calculatedField.setDebugSettings(DebugSettings.all());
return saveCalculatedField(calculatedField);
if (modifier.length > 0) {
modifier[0].accept(configuration);
}
CalculatedField savedCalculatedField = saveCalculatedField(calculatedField);
CalculatedFieldDebugEvent debugEvent = await().atMost(TIMEOUT, TimeUnit.SECONDS)
.until(() -> getDebugEvents(savedCalculatedField.getId(), 1),
events -> !events.isEmpty()).get(0);
latestEventId = debugEvent.getId();
return savedCalculatedField;
}
private AlarmRule toAlarmRule(Condition condition) {

35
application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java
View File

@ -17,10 +17,10 @@ package org.thingsboard.server.cf;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.slf4j.Slf4j;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.test.annotation.DirtiesContext;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.DeviceProfile;
@ -64,8 +64,8 @@ import static org.awaitility.Awaitility.await;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
import static org.thingsboard.server.cf.CalculatedFieldIntegrationTest.POLL_INTERVAL;
@Slf4j
@DaoSqlTest
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractControllerTest {
private Tenant savedTenant;
@ -79,10 +79,10 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
private AssetProfile assetProfile;
private Asset asset;
private long deduplicationInterval = 10;
private final long deduplicationInterval = 5;
@Before
public void beforeTest() throws Exception {
public void beforeEach() throws Exception {
loginSysAdmin();
updateDefaultTenantProfileConfig(tenantProfileConfig -> {
@ -135,7 +135,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
createOccupancyCF(assetProfile.getId());
await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
await().alias("create CF and perform initial aggregation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of(
@ -153,7 +153,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device3.getId(), "{\"occupied\":true}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
await().alias("update telemetry and perform aggregation")
.atLeast(deduplicationInterval / 2, TimeUnit.SECONDS)
.atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of(
@ -200,7 +202,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device3.getId(), "{\"occupied\":false}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of(
@ -284,7 +286,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device1.getId(), "{\"occupied\":false}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
await().alias("update telemetry and perform aggregation")
.atLeast(deduplicationInterval / 2, TimeUnit.SECONDS)
.atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of(
@ -323,14 +327,15 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device1.getId(), "{\"occupied\":false}");
await().alias("update telemetry -> no changes").atMost(deduplicationInterval / 2, TimeUnit.SECONDS)
await().alias("update telemetry -> no changes").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(this::checkInitialCalculationValues);
postTelemetry(device2.getId(), "{\"occupied\":false}");
await().alias("create CF and perform initial calculation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
await().alias("create CF and perform initial calculation")
.atLeast(deduplicationInterval / 2, TimeUnit.SECONDS)
.atMost(TIMEOUT, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of(
"freeSpaces", "2",
@ -517,7 +522,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device2.getId(), "{\"temperature\":19.6}");
CalculatedField cf = createAvgTemperatureCF(asset.getId());
await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
await().alias("create avg temp cf and perform initial aggregation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
@ -531,7 +536,7 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
configuration.setMetrics(Map.of("maxTemperature", aggMetric));
saveCalculatedField(cf);
await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.SECONDS)
await().alias("update metrics and perform aggregation").atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24"));
@ -540,7 +545,9 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
postTelemetry(device1.getId(), "{\"temperature\":101.3}");
postTelemetry(device2.getId(), "{\"temperature\":25.8}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
await().alias("update telemetry and perform aggregation")
.atLeast(deduplicationInterval / 2, TimeUnit.SECONDS)
.atMost(TIMEOUT, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26"));

2
application/src/test/resources/logback-test.xml
View File

@ -17,8 +17,6 @@
<logger name="org.eclipse.leshan" level="INFO"/>
<logger name="org.thingsboard.server.controller.AbstractWebTest" level="INFO"/>
<logger name="org.thingsboard.server.service.script" level="INFO"/>
<logger name="org.thingsboard.server.service.cf.ctx.state.alarm" level="TRACE"/>
<logger name="org.thingsboard.server.actors.calculatedField" level="TRACE"/>
<!-- mute TelemetryEdgeSqlTest that causes a lot of randomly generated errors -->
<logger name="org.thingsboard.server.service.edge.rpc.PostgresEdgeGrpcSession" level="OFF"/>

4
common/dao-api/src/main/java/org/thingsboard/server/dao/alarm/AlarmService.java
View File

@ -51,10 +51,6 @@ import java.util.UUID;
public interface AlarmService extends EntityDaoService {
/*
* New API, since 3.5.
*/
/**
* Designed for atomic operations over active alarms.
* Only one active alarm may exist for the pair {originatorId, alarmType}

2
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/AlarmRule.java
View File

@ -15,6 +15,7 @@
*/
package org.thingsboard.server.common.data.alarm.rule;
import com.fasterxml.jackson.annotation.JsonIgnore;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
@ -30,6 +31,7 @@ public class AlarmRule {
private String alarmDetails;
private DashboardId dashboardId;
@JsonIgnore
public boolean requiresScheduledReevaluation() {
return condition.hasSchedule();
}

1
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java
View File

@ -45,6 +45,7 @@ public abstract class AlarmCondition {
@Valid
private AlarmConditionValue<AlarmSchedule> schedule;
@JsonIgnore
public boolean hasSchedule() {
return schedule != null && !(schedule.getStaticValue() instanceof AnyTimeSchedule);
}

8
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmConditionValue.java
View File

@ -15,6 +15,8 @@
*/
package org.thingsboard.server.common.data.alarm.rule.condition;
import com.fasterxml.jackson.annotation.JsonIgnore;
import jakarta.validation.constraints.AssertTrue;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@ -27,4 +29,10 @@ public class AlarmConditionValue<T> {
private T staticValue;
private String dynamicValueArgument;
@JsonIgnore
@AssertTrue(message = "Either staticValue or dynamicValueArgument must be set")
public boolean isValid() {
return staticValue != null ^ dynamicValueArgument != null;
}
}

2
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/AlarmConditionFilter.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.common.data.alarm.rule.condition.expression;
import io.swagger.v3.oas.annotations.media.Schema;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
@ -24,7 +23,6 @@ import org.thingsboard.server.common.data.alarm.rule.condition.expression.predic
import java.io.Serializable;
@Schema
@Data
public class AlarmConditionFilter implements Serializable {

5
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/BooleanFilterPredicate.java
View File

@ -15,13 +15,18 @@
*/
package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue;
@Data
public class BooleanFilterPredicate implements SimpleKeyFilterPredicate<Boolean> {
@NotNull
private BooleanOperation operation;
@Valid
@NotNull
private AlarmConditionValue<Boolean> value;
@Override

5
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/NumericFilterPredicate.java
View File

@ -15,13 +15,18 @@
*/
package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue;
@Data
public class NumericFilterPredicate implements SimpleKeyFilterPredicate<Double> {
@NotNull
private NumericOperation operation;
@Valid
@NotNull
private AlarmConditionValue<Double> value;
@Override

6
common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/expression/predicate/StringFilterPredicate.java
View File

@ -15,13 +15,18 @@
*/
package org.thingsboard.server.common.data.alarm.rule.condition.expression.predicate;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotNull;
import lombok.Data;
import org.thingsboard.server.common.data.alarm.rule.condition.AlarmConditionValue;
@Data
public class StringFilterPredicate implements SimpleKeyFilterPredicate<String> {
@NotNull
private StringOperation operation;
@Valid
@NotNull
private AlarmConditionValue<String> value;
private boolean ignoreCase;
@ -40,4 +45,5 @@ public class StringFilterPredicate implements SimpleKeyFilterPredicate<String> {
IN,
NOT_IN
}
}

35
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java
View File

@ -15,15 +15,24 @@
*/
package org.thingsboard.server.common.data.cf.configuration;
import com.fasterxml.jackson.annotation.JsonIgnore;
import jakarta.validation.Valid;
import jakarta.validation.constraints.NotEmpty;
import lombok.Data;
import org.apache.commons.lang3.tuple.Pair;
import org.thingsboard.server.common.data.alarm.AlarmSeverity;
import org.thingsboard.server.common.data.alarm.rule.AlarmRule;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.util.CollectionsUtil;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.BiPredicate;
import java.util.stream.Stream;
import static java.util.Map.Entry.comparingByKey;
@Data
public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration {
@ -51,15 +60,31 @@ public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculat
return null;
}
@JsonIgnore
@Override
public void validate() {
public boolean requiresScheduledReevaluation() {
return getAllRules().anyMatch(entry -> entry.getValue().requiresScheduledReevaluation());
}
@JsonIgnore
public Stream<Pair<AlarmSeverity, AlarmRule>> getAllRules() {
Stream<Pair<AlarmSeverity, AlarmRule>> rules = createRules.entrySet().stream()
.map(entry -> Pair.of(entry.getKey(), entry.getValue()));
if (clearRule != null) {
rules = Stream.concat(rules, Stream.of(Pair.of(null, clearRule)));
}
return rules.sorted(comparingByKey(Comparator.nullsLast(Comparator.naturalOrder())));
}
@Override
public boolean requiresScheduledReevaluation() {
return createRules.values().stream().anyMatch(AlarmRule::requiresScheduledReevaluation) ||
(clearRule != null && clearRule.requiresScheduledReevaluation());
public boolean rulesEqual(AlarmCalculatedFieldConfiguration other, BiPredicate<AlarmRule, AlarmRule> equalityCheck) {
List<Pair<AlarmSeverity, AlarmRule>> thisRules = this.getAllRules().toList();
List<Pair<AlarmSeverity, AlarmRule>> otherRules = other.getAllRules().toList();
return CollectionsUtil.elementsEqual(thisRules, otherRules, (thisRule, otherRule) -> {
if (!Objects.equals(thisRule.getKey(), otherRule.getKey())) {
return false;
}
return equalityCheck.test(thisRule.getValue(), otherRule.getValue());
});
}
}

3
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java
View File

@ -52,7 +52,7 @@ public interface CalculatedFieldConfiguration {
Output getOutput();
void validate();
default void validate() {}
@JsonIgnore
default List<EntityId> getReferencedEntities() {
@ -74,6 +74,7 @@ public interface CalculatedFieldConfiguration {
.collect(Collectors.toList());
}
@JsonIgnore
default boolean requiresScheduledReevaluation() {
return false;
}

2
common/data/src/main/java/org/thingsboard/server/common/data/event/CalculatedFieldDebugEvent.java
View File

@ -29,7 +29,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import java.util.UUID;
@ToString
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
public class CalculatedFieldDebugEvent extends Event {

29
common/data/src/main/java/org/thingsboard/server/common/data/util/CollectionsUtil.java
View File

@ -18,9 +18,11 @@ package org.thingsboard.server.common.data.util;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiPredicate;
import java.util.stream.Collectors;
public class CollectionsUtil {
@ -95,4 +97,31 @@ public class CollectionsUtil {
return false;
}
public static <T> boolean elementsEqual(Iterable<T> iterable1, Iterable<T> iterable2, BiPredicate<T, T> equalityCheck) {
if (iterable1 instanceof Collection<?> collection1 && iterable2 instanceof Collection<?> collection2) {
if (collection1.size() != collection2.size()) {
return false;
}
}
Iterator<T> iterator1 = iterable1.iterator();
Iterator<T> iterator2 = iterable2.iterator();
while (true) {
if (iterator1.hasNext()) {
if (!iterator2.hasNext()) {
return false;
}
T o1 = iterator1.next();
T o2 = iterator2.next();
if (equalityCheck.test(o1, o2)) {
continue;
} else {
return false;
}
}
return !iterator2.hasNext();
}
}
}

2
common/proto/src/main/proto/queue.proto
View File

@ -1728,12 +1728,10 @@ message ToCalculatedFieldMsg {
CalculatedFieldTelemetryMsgProto telemetryMsg = 1;
CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 2;
EntityActionEventProto eventMsg = 3;
repeated string cfTypes = 4;
}
message ToCalculatedFieldNotificationMsg {
CalculatedFieldLinkedTelemetryMsgProto linkedTelemetryMsg = 1;
repeated string cfTypes = 2;
}
/* Messages that are handled by ThingsBoard RuleEngine Service */

2
dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java
View File

@ -159,7 +159,7 @@ public class BaseAlarmService extends AbstractCachedEntityService<TenantId, Page
@Override
public AlarmApiCallResult clearAlarm(TenantId tenantId, AlarmId alarmId, long clearTs, JsonNode details, boolean pushEvent) {
var result = withPropagated(alarmDao.clearAlarm(tenantId, alarmId, clearTs, details));
if (result.getAlarm() != null) {
if (pushEvent && result.getAlarm() != null) {
eventPublisher.publishEvent(ActionEntityEvent.builder()
.tenantId(tenantId)
.entityId(result.getAlarm().getId())

1
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/AlarmRuleState.java
View File

@ -55,7 +55,6 @@ import static org.thingsboard.server.common.data.StringUtils.splitByCommaWithout
@Data
@Slf4j
class AlarmRuleState {
private final AlarmSeverity severity;

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNode.java
View File

@ -51,14 +51,14 @@ import java.util.concurrent.TimeUnit;
@Slf4j
@RuleNode(
type = ComponentType.ACTION,
name = "device profile (deprecated)", // TODO: add description on why is it deprecated and what to use
name = "device profile (deprecated)",
customRelations = true,
relationTypes = {"Alarm Created", "Alarm Updated", "Alarm Severity Updated", "Alarm Cleared", "Success", "Failure"},
version = 1,
configClazz = TbDeviceProfileNodeConfiguration.class,
nodeDescription = "Process device messages based on device profile settings (deprecated)",
nodeDetails = "Create and clear alarms based on alarm rules defined in device profile. The output relation type is either " +
"'Alarm Created', 'Alarm Updated', 'Alarm Severity Updated' and 'Alarm Cleared' or simply 'Success' if no alarms were affected.",
"'Alarm Created', 'Alarm Updated', 'Alarm Severity Updated' and 'Alarm Cleared' or simply 'Success' if no alarms were affected.",
configDirective = "tbActionNodeDeviceProfileConfig",
docUrl = "https://thingsboard.io/docs/user-guide/rule-engine-2-0/nodes/action/device-profile/"
)

Write Preview
Loading…
Cancel
Save