Browse Source

Alarm rules CF: fixes for config update handling; fix schedule parsing

pull/14193/head
VIacheslavKlimov 12 months ago
parent
commit
8a95f2399a
  1. 1
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 6
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  3. 18
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  4. 16
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java
  5. 54
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java

1
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@ -155,7 +155,6 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
initState(state, ctx); initState(state, ctx);
} else { } else {
state.setCtx(ctx, actorCtx); state.setCtx(ctx, actorCtx);
state.init();
} }
if (state.isSizeOk()) { if (state.isSizeOk()) {
processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback()); processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback());

6
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@ -330,11 +330,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
StateAction stateAction; StateAction stateAction;
if (newCfCtx.getCfType() != oldCfCtx.getCfType()) { if (newCfCtx.getCfType() != oldCfCtx.getCfType()) {
stateAction = StateAction.RECREATE; stateAction = StateAction.RECREATE; // completely recreate state, then calculate
} else if (newCfCtx.hasStateChanges(oldCfCtx)) { } else if (newCfCtx.hasStateChanges(oldCfCtx)) {
stateAction = StateAction.REINIT; stateAction = StateAction.REINIT; // refetch arguments, call state.init, then calculate
} else if (newCfCtx.hasContextOnlyChanges(oldCfCtx)) { } else if (newCfCtx.hasContextOnlyChanges(oldCfCtx)) {
stateAction = StateAction.REPROCESS; stateAction = StateAction.REPROCESS; // call state.setCtx, then calculate
} else { } else {
callback.onSuccess(); callback.onSuccess();
return; return;

18
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java

@ -63,8 +63,8 @@ import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream; import java.util.stream.Stream;
@Data @Data
@ -479,11 +479,11 @@ public class CalculatedFieldCtx {
return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId); return new CalculatedFieldEntityCtxId(tenantId, cfId, entityId);
} }
public boolean hasContextOnlyChanges(CalculatedFieldCtx other) { // has changes that do not require state reinit and will be picked up by the state on the fly public boolean hasContextOnlyChanges(CalculatedFieldCtx other) {
if (calculatedField.getConfiguration() instanceof ExpressionBasedCalculatedFieldConfiguration && !expression.equals(other.expression)) { if (calculatedField.getConfiguration() instanceof ExpressionBasedCalculatedFieldConfiguration && !expression.equals(other.expression)) {
return true; return true;
} }
if (!output.equals(other.output)) { if (!Objects.equals(output, other.output)) {
return true; return true;
} }
if (cfType == CalculatedFieldType.ALARM && !calculatedField.getName().equals(other.getCalculatedField().getName())) { if (cfType == CalculatedFieldType.ALARM && !calculatedField.getName().equals(other.getCalculatedField().getName())) {
@ -495,9 +495,8 @@ public class CalculatedFieldCtx {
return false; return false;
} }
public boolean hasStateChanges(CalculatedFieldCtx other) { // has changes that require state reinit (will trigger state.reset() and re-fetch arguments) public boolean hasStateChanges(CalculatedFieldCtx other) {
boolean hasChanges = !arguments.equals(other.arguments); if (!arguments.equals(other.arguments)) {
if (hasChanges) {
return true; return true;
} }
if (cfType == CalculatedFieldType.ALARM) { if (cfType == CalculatedFieldType.ALARM) {
@ -505,14 +504,13 @@ public class CalculatedFieldCtx {
var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration(); var otherConfig = (AlarmCalculatedFieldConfiguration) other.getCalculatedField().getConfiguration();
if (!thisConfig.getCreateRules().equals(otherConfig.getCreateRules()) || if (!thisConfig.getCreateRules().equals(otherConfig.getCreateRules()) ||
!Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) { !Objects.equals(thisConfig.getClearRule(), otherConfig.getClearRule())) {
hasChanges = true; return true;
} }
// TODO: implement rules update logic!
} }
if (hasGeofencingZoneGroupConfigurationChanges(other)) { if (hasGeofencingZoneGroupConfigurationChanges(other)) {
hasChanges = true; return true;
} }
return hasChanges; return false;
} }
private boolean hasGeofencingZoneGroupConfigurationChanges(CalculatedFieldCtx other) { private boolean hasGeofencingZoneGroupConfigurationChanges(CalculatedFieldCtx other) {

16
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java

@ -77,8 +77,8 @@ import static org.thingsboard.server.service.cf.ctx.state.alarm.AlarmEvalResult.
@Slf4j @Slf4j
public class AlarmCalculatedFieldState extends BaseCalculatedFieldState { public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
private String alarmType;
private AlarmCalculatedFieldConfiguration configuration; private AlarmCalculatedFieldConfiguration configuration;
private String alarmType;
@Getter @Getter
private final Map<AlarmSeverity, AlarmRuleState> createRuleStates = new TreeMap<>(Comparator.comparing(Enum::ordinal)); private final Map<AlarmSeverity, AlarmRuleState> createRuleStates = new TreeMap<>(Comparator.comparing(Enum::ordinal));
@ -97,8 +97,13 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
@Override @Override
public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) { public void setCtx(CalculatedFieldCtx ctx, TbActorRef actorCtx) {
super.setCtx(ctx, actorCtx); super.setCtx(ctx, actorCtx);
this.alarmType = ctx.getCalculatedField().getName();
this.configuration = getConfiguration(ctx); this.configuration = getConfiguration(ctx);
this.alarmType = ctx.getCalculatedField().getName();
if (currentAlarm != null && !currentAlarm.getType().equals(alarmType)) {
currentAlarm = null;
initialFetchDone = false;
}
} }
@Override @Override
@ -170,10 +175,7 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
@Override @Override
public void reset() { public void reset() {
super.reset(); super.reset();
createRuleStates.values().forEach(AlarmRuleState::clear); configuration = null;
if (clearRuleState != null) {
clearRuleState.clear();
}
} }
@Override @Override
@ -502,7 +504,7 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {
SingleValueArgumentEntry entry = getArgument(argument); SingleValueArgumentEntry entry = getArgument(argument);
value = mapper.apply(entry.getKvEntryValue()); value = mapper.apply(entry.getKvEntryValue());
if (value == null) { if (value == null) {
throw new IllegalArgumentException("No value found for argument " + argument); throw new IllegalArgumentException("No proper value found for argument " + argument);
} }
} }
return value; return value;

54
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java

@ -15,10 +15,11 @@
*/ */
package org.thingsboard.server.service.cf.ctx.state.alarm; package org.thingsboard.server.service.cf.ctx.state.alarm;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.Data; import lombok.Data;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.KvUtil; import org.thingsboard.common.util.KvUtil;
import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.alarm.AlarmSeverity;
import org.thingsboard.server.common.data.alarm.rule.AlarmRule; import org.thingsboard.server.common.data.alarm.rule.AlarmRule;
import org.thingsboard.server.common.data.alarm.rule.condition.AlarmCondition; import org.thingsboard.server.common.data.alarm.rule.condition.AlarmCondition;
@ -28,6 +29,8 @@ import org.thingsboard.server.common.data.alarm.rule.condition.DurationAlarmCond
import org.thingsboard.server.common.data.alarm.rule.condition.RepeatingAlarmCondition; import org.thingsboard.server.common.data.alarm.rule.condition.RepeatingAlarmCondition;
import org.thingsboard.server.common.data.alarm.rule.condition.expression.AlarmConditionExpression; import org.thingsboard.server.common.data.alarm.rule.condition.expression.AlarmConditionExpression;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.AlarmSchedule; import org.thingsboard.server.common.data.alarm.rule.condition.schedule.AlarmSchedule;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.AlarmScheduleType;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.AnyTimeSchedule;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.CustomTimeSchedule; import org.thingsboard.server.common.data.alarm.rule.condition.schedule.CustomTimeSchedule;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.CustomTimeScheduleItem; import org.thingsboard.server.common.data.alarm.rule.condition.schedule.CustomTimeScheduleItem;
import org.thingsboard.server.common.data.alarm.rule.condition.schedule.SpecificTimeSchedule; import org.thingsboard.server.common.data.alarm.rule.condition.schedule.SpecificTimeSchedule;
@ -109,8 +112,12 @@ public class AlarmRuleState {
eventCount++; eventCount++;
} }
long requiredRepeats = getIntValue(((RepeatingAlarmCondition) condition).getCount()); long requiredRepeats = getIntValue(((RepeatingAlarmCondition) condition).getCount());
long leftRepeats = requiredRepeats - eventCount; if (requiredRepeats > 0) {
return leftRepeats <= 0 ? AlarmEvalResult.TRUE : AlarmEvalResult.notYetTrue(leftRepeats, 0); long leftRepeats = requiredRepeats - eventCount;
return leftRepeats <= 0 ? AlarmEvalResult.TRUE : AlarmEvalResult.notYetTrue(leftRepeats, 0);
} else {
return AlarmEvalResult.NOT_YET_TRUE;
}
} else { } else {
return AlarmEvalResult.FALSE; return AlarmEvalResult.FALSE;
} }
@ -132,11 +139,15 @@ public class AlarmRuleState {
} }
duration = lastEventTs - firstEventTs; duration = lastEventTs - firstEventTs;
long requiredDuration = getRequiredDurationInMs(); long requiredDuration = getRequiredDurationInMs();
long leftDuration = requiredDuration - duration; if (requiredDuration > 0) {
if (leftDuration <= 0) { long leftDuration = requiredDuration - duration;
return AlarmEvalResult.TRUE; if (leftDuration <= 0) {
return AlarmEvalResult.TRUE;
} else {
return AlarmEvalResult.notYetTrue(0, leftDuration);
}
} else { } else {
return AlarmEvalResult.notYetTrue(0, leftDuration); return AlarmEvalResult.NOT_YET_TRUE;
} }
} else { } else {
return AlarmEvalResult.FALSE; return AlarmEvalResult.FALSE;
@ -148,13 +159,14 @@ public class AlarmRuleState {
return true; return true;
} }
AlarmSchedule schedule = state.resolveValue(condition.getSchedule(), entry -> Optional.ofNullable(KvUtil.getStringValue(entry)) AlarmSchedule schedule = state.resolveValue(condition.getSchedule(), entry -> Optional.ofNullable(KvUtil.getStringValue(entry))
.map(str -> JsonConverter.parse(str, AlarmSchedule.class)) .map(this::parseSchedule).orElse(null));
.orElse(null)); boolean active = switch (schedule.getType()) {
return switch (schedule.getType()) {
case ANY_TIME -> true; case ANY_TIME -> true;
case SPECIFIC_TIME -> isActiveSpecific((SpecificTimeSchedule) schedule, eventTs); case SPECIFIC_TIME -> isActiveSpecific((SpecificTimeSchedule) schedule, eventTs);
case CUSTOM -> isActiveCustom((CustomTimeSchedule) schedule, eventTs); case CUSTOM -> isActiveCustom((CustomTimeSchedule) schedule, eventTs);
}; };
log.trace("Alarm rule active = {} for schedule {}", active, schedule);
return active;
} }
private boolean isActiveSpecific(SpecificTimeSchedule schedule, long eventTs) { private boolean isActiveSpecific(SpecificTimeSchedule schedule, long eventTs) {
@ -221,6 +233,28 @@ public class AlarmRuleState {
return eventCount == 0L && firstEventTs == 0L && lastEventTs == 0L && durationCheckFuture == null; return eventCount == 0L && firstEventTs == 0L && lastEventTs == 0L && durationCheckFuture == null;
} }
private AlarmSchedule parseSchedule(String str) {
ObjectNode json = (ObjectNode) JacksonUtil.toJsonNode(str);
if (json.isEmpty()) {
return new AnyTimeSchedule(); // only if valid json, fail otherwise
}
if (!json.hasNonNull("type")) {
// deducting the schedule type
AlarmScheduleType type;
if (json.hasNonNull("daysOfWeek")) {
type = AlarmScheduleType.SPECIFIC_TIME;
} else if (json.hasNonNull("items")) {
type = AlarmScheduleType.CUSTOM;
} else {
throw new IllegalArgumentException("Failed to parse alarm schedule from '" + str + "'");
}
json.put("type", type.name());
}
return JacksonUtil.treeToValue(json, AlarmSchedule.class);
}
private Integer getIntValue(AlarmConditionValue<Integer> value) { private Integer getIntValue(AlarmConditionValue<Integer> value) {
return state.resolveValue(value, entry -> Optional.ofNullable(KvUtil.getLongValue(entry)).map(Long::intValue).orElse(null)); return state.resolveValue(value, entry -> Optional.ofNullable(KvUtil.getLongValue(entry)).map(Long::intValue).orElse(null));
} }

Loading…
Cancel
Save