diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index 654ae8bf29..9848ac2fe6 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -666,14 +666,6 @@ public class ActorSystemContext { @Getter private long cfCalculationResultTimeout; - @Value("${actors.calculated_fields.check_interval:60}") - @Getter - private long cfCheckInterval; - - @Value("${actors.alarms.reevaluation_interval:60}") - @Getter - private long alarmRulesReevaluationInterval; - @Autowired @Getter private MqttClientSettings mqttClientSettings; diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index b19ad1a8b4..eca0d9447f 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntityRelationPathQuery; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationPathLevel; +import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.msg.CalculatedFieldStatePartitionRestoreMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg; import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg; @@ -144,10 +145,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware calculatedFields.clear(); entityIdCalculatedFields.clear(); entityIdCalculatedFieldLinks.clear(); - if (cfsReevaluationTask != null) { - cfsReevaluationTask.cancel(true); - cfsReevaluationTask = null; - } + cancelReevaluationTask(); ctx.stop(ctx.getSelf()); } @@ -177,6 +175,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void scheduleCfsReevaluation() { + long cfCheckInterval = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getCfReevaluationCheckInterval); cfsReevaluationTask = systemContext.getScheduler().scheduleWithFixedDelay(() -> { try { calculatedFields.values().forEach(cf -> { @@ -190,7 +189,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } catch (Exception e) { log.warn("[{}] Failed to trigger CFs reevaluation", tenantId, e); } - }, systemContext.getCfCheckInterval(), systemContext.getCfCheckInterval(), TimeUnit.SECONDS); + }, cfCheckInterval, cfCheckInterval, TimeUnit.SECONDS); } public void onEntityLifecycleMsg(CalculatedFieldEntityLifecycleMsg msg) throws CalculatedFieldException { @@ -257,6 +256,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onTenantProfileUpdated(ComponentLifecycleMsg msg, TbCallback callback) { + cancelReevaluationTask(); + scheduleCfsReevaluation(); Stream.concat( calculatedFields.values().stream(), entityIdCalculatedFields.values().stream().flatMap(Collection::stream) @@ -875,4 +876,11 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } } + private void cancelReevaluationTask() { + if (cfsReevaluationTask != null) { + cfsReevaluationTask.cancel(true); + cfsReevaluationTask = null; + } + } + } diff --git a/application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java b/application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java index 2c0ce1c938..175b8403ad 100644 --- a/application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java +++ b/application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java @@ -167,6 +167,8 @@ public class SystemInfoController extends BaseController { systemParams.setMinAllowedDeduplicationIntervalInSecForCF(tenantProfileConfiguration.getMinAllowedDeduplicationIntervalInSecForCF()); systemParams.setMinAllowedAggregationIntervalInSecForCF(tenantProfileConfiguration.getMinAllowedAggregationIntervalInSecForCF()); systemParams.setMinAllowedRealtimeAggregationIntervalInSecForCF(tenantProfileConfiguration.getMinAllowedRealtimeAggregationIntervalInSecForCF()); + systemParams.setCfReevaluationCheckInterval(tenantProfileConfiguration.getCfReevaluationCheckInterval()); + systemParams.setAlarmsReevaluationInterval(tenantProfileConfiguration.getAlarmsReevaluationInterval()); systemParams.setTrendzSettings(trendzSettingsService.findTrendzSettings(currentUser.getTenantId())); } systemParams.setMobileQrEnabled(Optional.ofNullable(qrCodeSettingService.findQrCodeSettings(TenantId.SYS_TENANT_ID)) diff --git 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 index ec2d11ebf9..d0f35a8571 100644 --- 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 @@ -58,6 +58,7 @@ import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileCon import org.thingsboard.server.common.data.util.CollectionsUtil; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.relation.RelationService; +import org.thingsboard.server.dao.usagerecord.ApiLimitService; import org.thingsboard.server.dao.util.TimeUtils; import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto; import org.thingsboard.server.service.cf.CalculatedFieldProcessingService; @@ -123,6 +124,8 @@ public class CalculatedFieldCtx implements Closeable { private List relatedEntityArgumentNames; private long scheduledUpdateIntervalMillis; + private long cfCheckReevaluationInterval; + private long alarmReevaluationInterval; private Argument propagationArgument; private boolean applyExpressionForResolvedArguments; @@ -209,9 +212,12 @@ public class CalculatedFieldCtx implements Closeable { this.alarmService = systemContext.getAlarmService(); this.cfProcessingService = systemContext.getCalculatedFieldProcessingService(); - this.maxStateSize = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; - this.maxSingleValueArgumentSize = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxSingleValueArgumentSizeInKBytes) * 1024; - this.realtimeAggregationIntervalMillis = TimeUnit.SECONDS.toMillis(systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedRealtimeAggregationIntervalInSecForCF)); + ApiLimitService apiLimitService = systemContext.getApiLimitService(); + this.maxStateSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; + this.maxSingleValueArgumentSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxSingleValueArgumentSizeInKBytes) * 1024; + this.realtimeAggregationIntervalMillis = TimeUnit.SECONDS.toMillis(apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedRealtimeAggregationIntervalInSecForCF)); + this.cfCheckReevaluationInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getCfReevaluationCheckInterval); + this.alarmReevaluationInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getAlarmsReevaluationInterval); } public boolean requiresScheduledReevaluation() { @@ -241,7 +247,7 @@ public class CalculatedFieldCtx implements Closeable { boolean requiresScheduledReevaluation = calculatedField.getConfiguration().requiresScheduledReevaluation(); if (calculatedField.getConfiguration() instanceof AlarmCalculatedFieldConfiguration) { if (requiresScheduledReevaluation) { - long reevaluationIntervalMillis = TimeUnit.SECONDS.toMillis(systemContext.getAlarmRulesReevaluationInterval()); + long reevaluationIntervalMillis = TimeUnit.SECONDS.toMillis(alarmReevaluationInterval); if (now - lastReevaluationTs >= reevaluationIntervalMillis) { lastReevaluationTs = now; return true; @@ -297,9 +303,12 @@ public class CalculatedFieldCtx implements Closeable { } public void updateTenantProfileProperties() { - this.maxStateSize = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; - this.maxSingleValueArgumentSize = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxSingleValueArgumentSizeInKBytes) * 1024; - this.realtimeAggregationIntervalMillis = TimeUnit.SECONDS.toMillis(systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedRealtimeAggregationIntervalInSecForCF)); + ApiLimitService apiLimitService = systemContext.getApiLimitService(); + this.maxStateSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxStateSizeInKBytes) * 1024; + this.maxSingleValueArgumentSize = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxSingleValueArgumentSizeInKBytes) * 1024; + this.realtimeAggregationIntervalMillis = TimeUnit.SECONDS.toMillis(apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedRealtimeAggregationIntervalInSecForCF)); + this.cfCheckReevaluationInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getCfReevaluationCheckInterval); + this.alarmReevaluationInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getAlarmsReevaluationInterval); } public double evaluateSimpleExpression(Expression expression, CalculatedFieldState state) { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java index a722f688dc..b752950a15 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java @@ -60,7 +60,6 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt private AggInterval interval; private long watermarkDuration; - private long checkInterval; private Map metrics; private boolean produceIntermediateResult; @@ -80,7 +79,6 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt var configuration = (EntityAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); Watermark watermark = configuration.getWatermark(); watermarkDuration = watermark == null ? 0 : TimeUnit.SECONDS.toMillis(watermark.getDuration()); - checkInterval = TimeUnit.SECONDS.toMillis(ctx.getSystemContext().getCfCheckInterval()); interval = configuration.getInterval(); metrics = configuration.getMetrics(); produceIntermediateResult = configuration.isProduceIntermediateResult(); @@ -208,7 +206,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt handleExpiredInterval(intervalEntry, args, results); expiredIntervals.add(intervalEntry); } else if (now - startTs >= intervalEntry.getIntervalDuration()) { - handleActiveInterval(intervalEntry, args, results); + handleActiveInterval(ctx, intervalEntry, args, results); if (watermarkDuration == 0) { expiredIntervals.add(intervalEntry); } @@ -233,11 +231,12 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt }); } - private void handleActiveInterval(AggIntervalEntry intervalEntry, + private void handleActiveInterval(CalculatedFieldCtx ctx, + AggIntervalEntry intervalEntry, Map args, Map> results) { args.forEach((argName, argEntryIntervalStatus) -> { - if (argEntryIntervalStatus.intervalPassed(checkInterval)) { + if (argEntryIntervalStatus.intervalPassed(ctx.getCfCheckReevaluationInterval())) { if (argEntryIntervalStatus.argsUpdated()) { argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastArgsRefreshTs(-1); @@ -261,7 +260,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); argEntryIntervalStatus.setLastArgsRefreshTs(-1); processArgument(intervalEntry, argName, false, results); - } else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) {// TODO: should we return default value when the interval has not ended + } else if (argEntryIntervalStatus.getLastMetricsEvalTs() == -1) { argEntryIntervalStatus.setLastMetricsEvalTs(System.currentTimeMillis()); processArgument(intervalEntry, argName, true, results); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index 8dab52f100..66669a8280 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -538,11 +538,6 @@ 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}" - # Interval in seconds to check calculated fields for re-evaluation interval. 1 minute by default. - check_interval: "${ACTORS_CALCULATED_FIELDS_CHECK_INTERVAL_SEC:60}" - alarms: - # Interval in seconds to re-evaluate Alarm rules that have a time schedule. 1 minute by default. - reevaluation_interval: "${ACTORS_ALARMS_REEVALUATION_INTERVAL_SEC:60}" debug: settings: diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java b/common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java index d1b8bb0562..37ce35fa84 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java @@ -43,5 +43,7 @@ public class SystemParams { long minAllowedDeduplicationIntervalInSecForCF; long minAllowedAggregationIntervalInSecForCF; long minAllowedRealtimeAggregationIntervalInSecForCF; + long cfReevaluationCheckInterval; + long alarmsReevaluationInterval; TrendzSettings trendzSettings; } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java index 3825c3c264..5291ed7a7a 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java @@ -196,6 +196,10 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura private long minAllowedAggregationIntervalInSecForCF = 60; @Schema(example = "300") private long minAllowedRealtimeAggregationIntervalInSecForCF = 300; + @Schema(example = "60") + private long cfReevaluationCheckInterval = 60; + @Schema(example = "60") + private long alarmsReevaluationInterval = 60; @Override public long getProfileThreshold(ApiUsageRecordKey key) {