Add initial duration alarm condition support for Alarm rules CF

9 months ago · 3e357e5e9b
17 changed files with 219 additions and 85 deletions
--- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
+++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
@ -654,6 +654,10 @@ public class ActorSystemContext {
    @Getter
    private long cfCalculationResultTimeout;

+    @Value("${actors.alarms.reevaluation_interval:60}")
+    @Getter
+    private long alarmsReevaluationInterval;
+
    @Autowired
    @Getter
    private MqttClientSettings mqttClientSettings;
@ -857,7 +861,7 @@ public class ActorSystemContext {

    private boolean checkLimits(TenantId tenantId) {
        if (debugModeRateLimitsConfig.isCalculatedFieldDebugPerTenantLimitsEnabled() &&
-                !rateLimitService.checkRateLimit(LimitedApi.CALCULATED_FIELD_DEBUG_EVENTS, (Object) tenantId, debugModeRateLimitsConfig.getCalculatedFieldDebugPerTenantLimitsConfiguration())) {
+            !rateLimitService.checkRateLimit(LimitedApi.CALCULATED_FIELD_DEBUG_EVENTS, (Object) tenantId, debugModeRateLimitsConfig.getCalculatedFieldDebugPerTenantLimitsConfiguration())) {
            log.trace("[{}] Calculated field debug event limits exceeded!", tenantId);
            return false;
        }
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityActor.java
@ -78,6 +78,9 @@ public class CalculatedFieldEntityActor extends AbstractCalculatedFieldActor {
            case CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG:
                processor.process((EntityCalculatedFieldDynamicArgumentsRefreshMsg) msg);
                break;
+            case CF_REEVALUATE_MSG:
+                processor.process((CalculatedFieldReevaluateMsg) msg);
+                break;
            case CF_ALARM_ACTION_MSG:
                processor.process((CalculatedFieldAlarmActionMsg) msg);
                break;
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
@ -142,7 +142,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
                state.init(ctx);
            }
            if (state.isSizeOk()) {
-                processStateIfReady(ctx, Collections.emptyMap(), Collections.singletonList(ctx.getCfId()), state, null, null, msg.getCallback());
+                processStateIfReady(state, Collections.emptyMap(), ctx, Collections.singletonList(ctx.getCfId()), null, null, msg.getCallback());
            } else {
                throw new RuntimeException(ctx.getSizeExceedsLimitMessage());
            }
@ -257,6 +257,21 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
        msg.getCallback().onSuccess();
    }

+    public void process(CalculatedFieldReevaluateMsg msg) throws CalculatedFieldException {
+        CalculatedFieldId cfId = msg.getCfCtx().getCfId();
+        CalculatedFieldState state = states.get(cfId);
+        if (state == null) {
+            log.debug("[{}][{}] Failed to find CF state for entity to handle {}", entityId, cfId, msg);
+        } else {
+            if (state.isSizeOk()) {
+                log.debug("[{}][{}] Reevaluating CF state", entityId, cfId);
+                processStateIfReady(state, null, msg.getCfCtx(), Collections.singletonList(cfId), null, null, msg.getCallback());
+            } else {
+                throw new RuntimeException(msg.getCfCtx().getSizeExceedsLimitMessage());
+            }
+        }
+    }
+
    public void process(CalculatedFieldAlarmActionMsg msg) {
        log.debug("[{}] Processing alarm action event msg: {}", entityId, msg);
        states.values().forEach(state -> {
@ -312,7 +327,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
            if (!updatedArgs.isEmpty() || justRestored) {
                cfIdList = new ArrayList<>(cfIdList);
                cfIdList.add(ctx.getCfId());
-                processStateIfReady(ctx, updatedArgs, cfIdList, state, tbMsgId, tbMsgType, callback);
+                processStateIfReady(state, updatedArgs, ctx, cfIdList, tbMsgId, tbMsgType, callback);
            } else {
                callback.onSuccess(CALLBACKS_PER_CF);
            }
@ -347,7 +362,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
        return argumentsFuture.get(1, TimeUnit.MINUTES);
    }

-    private void processStateIfReady(CalculatedFieldCtx ctx, Map<String, ArgumentEntry> updatedArgs, List<CalculatedFieldId> cfIdList, CalculatedFieldState state, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException {
+    private void processStateIfReady(CalculatedFieldState state, Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx,
+                                     List<CalculatedFieldId> cfIdList, UUID tbMsgId, TbMsgType tbMsgType, TbCallback callback) throws CalculatedFieldException {
        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;
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
@ -78,6 +78,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
    private final Map<EntityId, List<CalculatedFieldCtx>> entityIdCalculatedFields = new HashMap<>();
    private final Map<EntityId, List<CalculatedFieldLink>> entityIdCalculatedFieldLinks = new HashMap<>();
    private final Map<CalculatedFieldId, ScheduledFuture<?>> cfDynamicArgumentsRefreshTasks = new ConcurrentHashMap<>();
+    private ScheduledFuture<?> cfsReevaluationTask;

    private final CalculatedFieldProcessingService cfExecService;
    private final CalculatedFieldStateService cfStateService;
@ -118,6 +119,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
        entityIdCalculatedFieldLinks.clear();
        cfDynamicArgumentsRefreshTasks.values().forEach(future -> future.cancel(true));
        cfDynamicArgumentsRefreshTasks.clear();
+        if (cfsReevaluationTask != null) {
+            cfsReevaluationTask.cancel(true);
+            cfsReevaluationTask = null;
+        }
        ctx.stop(ctx.getSelf());
    }

@ -125,6 +130,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
        log.debug("[{}] Processing CF actor init message.", msg.getTenantId().getId());
        initEntityProfileCache();
        initCalculatedFields();
+        scheduleCfsReevaluation();
        msg.getCallback().onSuccess();
    }

@ -143,6 +149,23 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
        }
    }

+    private void scheduleCfsReevaluation() {
+        cfsReevaluationTask = systemContext.getScheduler().scheduleWithFixedDelay(() -> {
+            try {
+                calculatedFields.values().forEach(cf -> {
+                    if (cf.isRequiresScheduledReevaluation()) {
+                        applyToTargetCfEntityActors(cf, TbCallback.EMPTY, (entityId, callback) -> {
+                            log.debug("[{}][{}] Pushing scheduled CF reevaluate msg", entityId, cf.getCfId());
+                            getOrCreateActor(entityId).tell(new CalculatedFieldReevaluateMsg(tenantId, cf));
+                        });
+                    }
+                });
+            } catch (Exception e) {
+                log.warn("[{}] Failed to trigger CFs reevaluation", tenantId, e);
+            }
+        }, systemContext.getAlarmsReevaluationInterval(), systemContext.getAlarmsReevaluationInterval(), TimeUnit.SECONDS);
+    }
+
    public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException {
        log.debug("Processing entity lifecycle event: [{}] for entity: [{}]", msg.getData().getEvent(), msg.getData().getEntityId());
        var entityType = msg.getData().getEntityId().getEntityType();
--- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldReevaluateMsg.java
+++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldReevaluateMsg.java
@ -0,0 +1,35 @@
+/**
+ * Copyright © 2016-2025 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.actors.calculatedField;
+
+import lombok.Data;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.msg.MsgType;
+import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
+import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
+
+@Data
+public class CalculatedFieldReevaluateMsg implements ToCalculatedFieldSystemMsg {
+
+    private final TenantId tenantId;
+    private final CalculatedFieldCtx cfCtx;
+
+    @Override
+    public MsgType getMsgType() {
+        return MsgType.CF_REEVALUATE_MSG;
+    }
+
+}
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
@ -77,6 +77,7 @@ public class CalculatedFieldCtx {
    private Output output;
    private String expression;
    private boolean useLatestTs;
+    private boolean requiresScheduledReevaluation;

    private TbelInvokeService tbelInvokeService;
    private RelationService relationService;
@ -140,6 +141,7 @@ public class CalculatedFieldCtx {
                });
            }
        }
+        this.requiresScheduledReevaluation = calculatedField.getConfiguration().requiresScheduledReevaluation();
        this.tbelInvokeService = systemContext.getTbelInvokeService();
        this.relationService = systemContext.getRelationService();
        this.alarmService = systemContext.getAlarmService();
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmCalculatedFieldState.java
@ -117,21 +117,15 @@ public class AlarmCalculatedFieldState extends BaseCalculatedFieldState {

    @Override
    public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx) {
-        if (updatedArgs.isEmpty()) {
-            // FIXME: do we evaluate alarm rule (and increment event count) after arguments or expression change (state reinit)???
-            return Futures.immediateFuture(new AlarmCalculatedFieldResult(null));
-        }
        initCurrentAlarm(ctx);
-        TbAlarmResult result = createOrClearAlarms(state -> state.eval(ctx), ctx);
-        return Futures.immediateFuture(AlarmCalculatedFieldResult.builder()
-                .alarmResult(result)
-                .build());
-    }
-
-    // TODO: harvesting
-    public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, long ts, CalculatedFieldCtx ctx) {
-        initCurrentAlarm(ctx);
-        TbAlarmResult result = createOrClearAlarms(ruleState -> ruleState.eval(ts), ctx);
+        TbAlarmResult result = createOrClearAlarms(state -> {
+            if (updatedArgs != null) {
+                boolean newEvent = !updatedArgs.isEmpty();
+                return state.eval(newEvent, ctx);
+            } else {
+                return state.eval(System.currentTimeMillis());
+            }
+        }, ctx);
        return Futures.immediateFuture(AlarmCalculatedFieldResult.builder()
                .alarmResult(result)
                .build());
--- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java
+++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/alarm/AlarmRuleState.java
@ -64,21 +64,21 @@ public class AlarmRuleState {
        this.state = state;
    }

-    public AlarmEvalResult eval(CalculatedFieldCtx ctx) {
+    public AlarmEvalResult eval(boolean newEvent, CalculatedFieldCtx ctx) { // on event or config change
        boolean active = isActive(state.getLatestTimestamp());
        return switch (condition.getType()) {
-            case SIMPLE -> (active && eval(condition.getExpression(), ctx)) ? AlarmEvalResult.TRUE : AlarmEvalResult.FALSE;
+            case SIMPLE -> evalSimple(active, ctx);
            case DURATION -> evalDuration(active, ctx);
-            case REPEATING -> evalRepeating(active, ctx);
+            case REPEATING -> evalRepeating(active, newEvent, ctx);
        };
    }

-    public AlarmEvalResult eval(long ts) {
+    public AlarmEvalResult eval(long ts) { // on schedule
        switch (condition.getType()) {
-            case SIMPLE:
-            case REPEATING:
+            case SIMPLE, REPEATING -> {
                return AlarmEvalResult.NOT_YET_TRUE;
-            case DURATION:
+            }
+            case DURATION -> {
                long requiredDurationInMs = getRequiredDurationInMs();
                if (requiredDurationInMs > 0 && lastEventTs > 0 && ts > lastEventTs) {
                    long duration = this.duration + (ts - lastEventTs);
@ -88,8 +88,43 @@ public class AlarmRuleState {
                        return AlarmEvalResult.FALSE;
                    }
                }
-            default:
-                return AlarmEvalResult.FALSE;
+            }
+        }
+        return AlarmEvalResult.FALSE;
+    }
+
+    private AlarmEvalResult evalSimple(boolean active, CalculatedFieldCtx ctx) {
+        return (active && eval(condition.getExpression(), ctx)) ?
+                AlarmEvalResult.TRUE : AlarmEvalResult.FALSE;
+    }
+
+    private AlarmEvalResult evalRepeating(boolean active, boolean newEvent, CalculatedFieldCtx ctx) {
+        if (active && eval(condition.getExpression(), ctx)) {
+            if (newEvent) {
+                eventCount++;
+            }
+            long requiredRepeats = getIntValue(((RepeatingAlarmCondition) condition).getCount());
+            return eventCount >= requiredRepeats ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
+        } else {
+            return AlarmEvalResult.FALSE;
+        }
+    }
+
+    private AlarmEvalResult evalDuration(boolean active, CalculatedFieldCtx ctx) {
+        if (active && eval(condition.getExpression(), ctx)) {
+            if (lastEventTs > 0) {
+                if (state.getLatestTimestamp() > lastEventTs) {
+                    duration = duration + (state.getLatestTimestamp() - lastEventTs);
+                    lastEventTs = state.getLatestTimestamp();
+                }
+            } else {
+                lastEventTs = state.getLatestTimestamp();
+                duration = 0L;
+            }
+            long requiredDurationInMs = getRequiredDurationInMs();
+            return duration > requiredDurationInMs ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
+        } else {
+            return AlarmEvalResult.FALSE;
        }
    }

@ -162,42 +197,13 @@ public class AlarmRuleState {
        duration = 0L;
    }

-    private AlarmEvalResult evalRepeating(boolean active, CalculatedFieldCtx ctx) {
-        if (active && eval(condition.getExpression(), ctx)) {
-            eventCount++;
-            long requiredRepeats = getIntValue(((RepeatingAlarmCondition) condition).getCount());
-            return eventCount >= requiredRepeats ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
-        } else {
-            return AlarmEvalResult.FALSE;
-        }
-    }
-
-    private AlarmEvalResult evalDuration(boolean active, CalculatedFieldCtx ctx) {
-        if (active && eval(condition.getExpression(), ctx)) {
-            if (lastEventTs > 0) {
-                if (state.getLatestTimestamp() > lastEventTs) {
-                    duration = duration + (state.getLatestTimestamp() - lastEventTs);
-                    lastEventTs = state.getLatestTimestamp();
-                }
-            } else {
-                lastEventTs = state.getLatestTimestamp();
-                duration = 0L;
-            }
-            long requiredDurationInMs = getRequiredDurationInMs();
-            return duration > requiredDurationInMs ? AlarmEvalResult.TRUE : AlarmEvalResult.NOT_YET_TRUE;
-        } else {
-            return AlarmEvalResult.FALSE;
-        }
-    }
-
    private Integer getIntValue(AlarmConditionValue<Integer> value) {
        return getValue(value, entry -> Optional.ofNullable(KvUtil.getLongValue(entry)).map(Long::intValue).orElse(null));
    }

    private long getRequiredDurationInMs() {
-        // fixme timeUnit??
-
-        return getValue(((DurationAlarmCondition) condition).getValue(), KvUtil::getLongValue);
+        DurationAlarmCondition durationCondition = (DurationAlarmCondition) condition;
+        return durationCondition.getUnit().toMillis(getValue(durationCondition.getValue(), KvUtil::getLongValue));
    }

    private boolean eval(AlarmConditionExpression expression, CalculatedFieldCtx ctx) {
@ -226,7 +232,7 @@ public class AlarmRuleState {
        if (condition.getType() == AlarmConditionType.REPEATING) {
            return new StateInfo(eventCount, null);
        } else if (condition.getType() == AlarmConditionType.DURATION) {
-            return new StateInfo(null, duration);
+            return new StateInfo(null, duration + (System.currentTimeMillis() - lastEventTs));
        } else {
            return StateInfo.EMPTY;
        }
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@ -526,6 +526,9 @@ actors:
      configuration: "${ACTORS_CALCULATED_FIELD_DEBUG_MODE_RATE_LIMITS_PER_TENANT_CONFIGURATION:50000:3600}"
    # Time in seconds to receive calculation result.
    calculation_timeout: "${ACTORS_CALCULATION_TIMEOUT_SEC:5}"
+  alarms:
+    # Interval in seconds to re-evaluate Alarm rules with duration condition
+    reevaluation_interval: "${ACTORS_ALARMS_REEVALUATION_INTERVAL_SEC:60}"

 debug:
  settings:
--- a/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java
+++ b/application/src/test/java/org/thingsboard/server/cf/AlarmRulesTest.java
@ -20,11 +20,11 @@ import org.assertj.core.api.Assertions;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
 import org.thingsboard.common.util.JacksonUtil;
 import org.thingsboard.rule.engine.action.TbAlarmResult;
 import org.thingsboard.server.actors.ActorSystemContext;
-import org.thingsboard.server.common.data.AttributeScope;
 import org.thingsboard.server.common.data.Device;
 import org.thingsboard.server.common.data.alarm.Alarm;
 import org.thingsboard.server.common.data.alarm.AlarmSeverity;
@ -63,6 +63,9 @@ import static org.testcontainers.shaded.org.awaitility.Awaitility.await;

@Slf4j
@DaoSqlTest
+@TestPropertySource(properties = {
+        "actors.alarms.reevaluation_interval=1"
+})
 public class AlarmRulesTest extends AbstractControllerTest {

    @MockitoSpyBean
@ -129,10 +132,10 @@ public class AlarmRulesTest extends AbstractControllerTest {
    }

    /*
-    * todo: state restore (event count)
-    * */
+     * todo: state restore (event count)
+     * */
    @Test
-    public void testCreateAlarmForRepeatingConditionOnTs() throws Exception {
+    public void testCreateAlarmForRepeatingCondition() throws Exception {
        Argument temperatureArgument = new Argument();
        temperatureArgument.setRefEntityKey(new ReferencedEntityKey("temperature", ArgumentType.TS_LATEST, null));
        temperatureArgument.setDefaultValue("0");
@ -161,36 +164,47 @@ public class AlarmRulesTest extends AbstractControllerTest {
            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
            assertThat(alarmResult.getConditionRepeats()).isEqualTo(5);
        });
+
+        for (int i = 0; i < 5; i++) {
+            postTelemetry(deviceId, "{\"temperature\":50}");
+            Thread.sleep(10);
+        }
+        checkAlarmResult(calculatedField, alarmResult -> {
+            assertThat(alarmResult.isSeverityUpdated()).isTrue();
+            assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
+            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
+            assertThat(alarmResult.getConditionRepeats()).isEqualTo(10);
+        });
    }

    @Test
-    public void testCreateAlarmForRepeatingConditionOnAttribute() {
-        Argument temperatureArgument = new Argument();
-        temperatureArgument.setRefEntityKey(new ReferencedEntityKey("temperature", ArgumentType.ATTRIBUTE, AttributeScope.SHARED_SCOPE));
+    public void testCreateAlarmForDurationCondition() throws Exception {
+        Argument argument = new Argument();
+        argument.setRefEntityKey(new ReferencedEntityKey("powerConsumption", ArgumentType.TS_LATEST, null));
+        argument.setDefaultValue("0");
        Map<String, Argument> arguments = Map.of(
-                "temperature", temperatureArgument
+                "powerConsumption", argument
        );

-        Map<AlarmSeverity, String> createRules = Map.of(
-                AlarmSeverity.MAJOR, "return temperature >= 50;",
-                AlarmSeverity.CRITICAL, "return temperature >= 100;"
-        );
-        String clearRule = "return temperature <= 25;";
-//        CalculatedField calculatedField = createAlarmCf(deviceId, "High Temperature Alarm",
-//                arguments, createRules, clearRule);
-    }
-
-    @Test
-    public void testCreateAlarmForDurationCondition() {
-        Argument temperatureArgument = new Argument();
-        temperatureArgument.setRefEntityKey(new ReferencedEntityKey("powerConsumption", ArgumentType.TS_LATEST, null));
-        Map<String, Argument> arguments = Map.of(
-                "powerConsumption", temperatureArgument
+        long createDurationMs = 5000L;
+        Map<AlarmSeverity, Condition> createRules = Map.of(
+                AlarmSeverity.CRITICAL, new Condition("return powerConsumption >= 3000;", null, createDurationMs)
        );
+        long clearDurationMs = 2000L;
+        Condition clearRule = new Condition("return powerConsumption < 3000;", null, createDurationMs);

+        CalculatedField calculatedField = createAlarmCf(deviceId, "High power consumption during 3 seconds",
+                arguments, createRules, clearRule);
+        postTelemetry(deviceId, "{\"powerConsumption\":3500}");
+        Thread.sleep(createDurationMs - 2000);
+        assertThat(getLatestAlarmResult(calculatedField.getId())).isNull();

-//        CalculatedField calculatedField = createAlarmCf(deviceId, "High power consumption during 5 seconds",
-//                arguments, createRules, nu);
+        checkAlarmResult(calculatedField, alarmResult -> {
+            assertThat(alarmResult.isCreated()).isTrue();
+            assertThat(alarmResult.getAlarm().getSeverity()).isEqualTo(AlarmSeverity.CRITICAL);
+            assertThat(alarmResult.getAlarm().getStatus()).isEqualTo(AlarmStatus.ACTIVE_UNACK);
+            assertThat(alarmResult.getConditionDuration()).isBetween(createDurationMs, createDurationMs + 2000);
+        });
    }

    private void checkAlarmResult(CalculatedField calculatedField, Consumer<TbAlarmResult> assertion) {
@ -271,6 +285,7 @@ public class AlarmRulesTest extends AbstractControllerTest {
        } else if (condition.durationMs() != null) {
            DurationAlarmCondition alarmCondition = new DurationAlarmCondition();
            alarmCondition.setExpression(expression);
+            alarmCondition.setUnit(TimeUnit.MILLISECONDS);
            AlarmConditionValue<Long> duration = new AlarmConditionValue<>();
            duration.setStaticValue(condition.durationMs());
            alarmCondition.setValue(duration);
--- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/AlarmCondition.java
@ -41,6 +41,7 @@ public abstract class AlarmCondition {
    @NotNull
    @Valid
    private AlarmConditionExpression expression;
+    @Valid
    private AlarmConditionValue<AlarmSchedule> schedule;

    @JsonIgnore
--- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/DurationAlarmCondition.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/DurationAlarmCondition.java
@ -15,6 +15,8 @@
 */
 package org.thingsboard.server.common.data.alarm.rule.condition;

+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
@ -26,7 +28,10 @@ import java.util.concurrent.TimeUnit;
@ToString(callSuper = true)
 public class DurationAlarmCondition extends AlarmCondition {

+    @NotNull
    private TimeUnit unit;
+    @Valid
+    @NotNull
    private AlarmConditionValue<Long> value;

    @Override
--- a/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/RepeatingAlarmCondition.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/alarm/rule/condition/RepeatingAlarmCondition.java
@ -15,6 +15,8 @@
 */
 package org.thingsboard.server.common.data.alarm.rule.condition;

+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
@ -24,6 +26,8 @@ import lombok.ToString;
@ToString(callSuper = true)
 public class RepeatingAlarmCondition extends AlarmCondition {

+    @Valid
+    @NotNull
    private AlarmConditionValue<Integer> count;

    @Override
--- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java
@ -18,6 +18,8 @@ package org.thingsboard.server.common.data.cf;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@ -64,6 +66,8 @@ public class CalculatedField extends BaseData<CalculatedFieldId> implements HasN
    @Schema(description = "Version of calculated field configuration.", example = "0")
    private int configurationVersion;
    @Schema(implementation = SimpleCalculatedFieldConfiguration.class)
+    @Valid
+    @NotNull
    private CalculatedFieldConfiguration configuration;
    @Getter
    @Setter
--- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AlarmCalculatedFieldConfiguration.java
@ -15,9 +15,12 @@
 */
 package org.thingsboard.server.common.data.cf.configuration;

+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotEmpty;
 import lombok.Data;
 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.condition.AlarmConditionType;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;

 import java.util.List;
@ -26,9 +29,14 @@ import java.util.Map;
@Data
 public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculatedFieldConfiguration {

+    @Valid
+    @NotEmpty
    private Map<String, Argument> arguments;

+    @Valid
+    @NotEmpty
    private Map<AlarmSeverity, AlarmRule> createRules;
+    @Valid
    private AlarmRule clearRule;

    private boolean propagate;
@ -51,4 +59,10 @@ public class AlarmCalculatedFieldConfiguration implements ArgumentsBasedCalculat

    }

+    @Override
+    public boolean requiresScheduledReevaluation() {
+        return createRules.values().stream().anyMatch(rule -> rule.getCondition().getType() == AlarmConditionType.DURATION) ||
+               (clearRule != null && clearRule.getCondition().getType() == AlarmConditionType.DURATION);
+    }
+
 }
--- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java
+++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/CalculatedFieldConfiguration.java
@ -72,4 +72,8 @@ public interface CalculatedFieldConfiguration {
                .collect(Collectors.toList());
    }

+    default boolean requiresScheduledReevaluation() {
+        return false;
+    }
+
 }
--- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
+++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
@ -152,7 +152,8 @@ public enum MsgType {
    CF_ENTITY_DELETE_MSG,

    CF_DYNAMIC_ARGUMENTS_REFRESH_MSG,
-    CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG;
+    CF_ENTITY_DYNAMIC_ARGUMENTS_REFRESH_MSG,
+    CF_REEVALUATE_MSG;

    @Getter
    private final boolean ignoreOnStart;