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

added min aggregation interval to tenant profile

pull/14253/head
IrynaMatveieva 11 months ago
parent
ccb01ba12e
commit
fa4e056fb2
11 changed files with 52 additions and 22 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. 8
      application/src/main/data/upgrade/basic/schema_update.sql
  2. 1
      application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java
  3. 27
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  4. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java
  5. 4
      application/src/main/resources/thingsboard.yml
  6. 6
      application/src/test/java/org/thingsboard/server/cf/EntityAggregationCalculatedFieldTest.java
  7. 1
      common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java
  8. 5
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java
  9. 1
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java
  10. 2
      common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
  11. 17
      dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java

8
application/src/main/data/upgrade/basic/schema_update.sql
View File

@ -46,6 +46,12 @@ SET profile_data = jsonb_set(
WHEN (profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
THEN NULL
ELSE to_jsonb(3600)
END,
'minAggregationIntervalInSecForCF',
CASE
WHEN (profile_data -> 'configuration') ? 'minAggregationIntervalInSecForCF'
THEN NULL
ELSE to_jsonb(60)
END
)
),
@ -59,6 +65,8 @@ WHERE NOT (
(profile_data -> 'configuration') ? 'maxRelatedEntitiesToReturnPerCfArgument'
AND
(profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
AND
(profile_data -> 'configuration') ? 'minAggregationIntervalInSecForCF'
);
-- UPDATE TENANT PROFILE CONFIGURATION END

1
application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java
View File

@ -165,6 +165,7 @@ public class SystemInfoController extends BaseController {
systemParams.setMinAllowedScheduledUpdateIntervalInSecForCF(tenantProfileConfiguration.getMinAllowedScheduledUpdateIntervalInSecForCF());
systemParams.setMaxRelationLevelPerCfArgument(tenantProfileConfiguration.getMaxRelationLevelPerCfArgument());
systemParams.setMinAllowedDeduplicationIntervalInSecForCF(tenantProfileConfiguration.getMinAllowedDeduplicationIntervalInSecForCF());
systemParams.setMinAggregationIntervalInSecForCF(tenantProfileConfiguration.getMinAggregationIntervalInSecForCF());
systemParams.setTrendzSettings(trendzSettingsService.findTrendzSettings(currentUser.getTenantId()));
}
systemParams.setMobileQrEnabled(Optional.ofNullable(qrCodeSettingService.findQrCodeSettings(TenantId.SYS_TENANT_ID))

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

@ -97,7 +97,7 @@ public class CalculatedFieldCtx implements Closeable {
private boolean useLatestTs;
private boolean requiresScheduledReevaluation;
private long lastReevaluationTs;
private long aggCheckInterval;
private ActorSystemContext systemContext;
private TbelInvokeService tbelInvokeService;
@ -199,6 +199,7 @@ public class CalculatedFieldCtx implements Closeable {
if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) {
this.useLatestTs = aggConfig.isUseLatestTs();
}
this.aggCheckInterval = systemContext.getCfCheckInterval();
this.systemContext = systemContext;
this.tbelInvokeService = systemContext.getTbelInvokeService();
this.relationService = systemContext.getRelationService();
@ -213,20 +214,13 @@ public class CalculatedFieldCtx implements Closeable {
public boolean isRequiresScheduledReevaluation() {
if (calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration entityAggregationConfig) {
long now = System.currentTimeMillis();
long cfCheckIntervalMillis = TimeUnit.SECONDS.toMillis(systemContext.getCfCheckInterval());
Watermark watermark = entityAggregationConfig.getWatermark();
if (watermark != null) {
long checkIntervalMillis = TimeUnit.SECONDS.toMillis(watermark.getCheckInterval());
if (cfCheckIntervalMillis == checkIntervalMillis) {
return true;
}
if (now + cfCheckIntervalMillis >= lastReevaluationTs + checkIntervalMillis) {
lastReevaluationTs = now;
return true;
}
if (watermark != null && watermark.getDuration() > 0) {
return true;
}
long intervalEndTsMillis = entityAggregationConfig.getInterval().getCurrentIntervalEndTs();
if (now + cfCheckIntervalMillis >= intervalEndTsMillis) {
long cfCheckIntervalMillis = TimeUnit.SECONDS.toMillis(systemContext.getCfCheckInterval());
long intervalEndTs = entityAggregationConfig.getInterval().getCurrentIntervalEndTs();
if (now + cfCheckIntervalMillis >= intervalEndTs) {
return true;
}
}
@ -635,6 +629,13 @@ public class CalculatedFieldCtx implements Closeable {
&& (thisConfig.getDeduplicationIntervalInSec() != otherConfig.getDeduplicationIntervalInSec() || !thisConfig.getMetrics().equals(otherConfig.getMetrics()))) {
return true;
}
if (calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration thisConfig
&& other.getCalculatedField().getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration otherConfig) {
boolean metricsChanged = thisConfig.getMetrics().equals(otherConfig.getMetrics());
boolean intervalChanged = thisConfig.getInterval().equals(otherConfig.getInterval());
boolean watermarkChanged = thisConfig.getWatermark().equals(otherConfig.getWatermark());
return metricsChanged || intervalChanged || watermarkChanged;
}
return false;
}

2
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/single/EntityAggregationCalculatedFieldState.java
View File

@ -104,7 +104,7 @@ public class EntityAggregationCalculatedFieldState extends BaseCalculatedFieldSt
intervalDuration = configuration.getInterval().getIntervalDurationMillis();
Watermark watermark = configuration.getWatermark();
watermarkDuration = watermark == null ? 0 : TimeUnit.SECONDS.toMillis(watermark.getDuration());
checkInterval = watermark == null ? 0 : TimeUnit.SECONDS.toMillis(watermark.getCheckInterval());
checkInterval = ctx.getAggCheckInterval();
interval = configuration.getInterval();
metrics = configuration.getMetrics();
}

4
application/src/main/resources/thingsboard.yml
View File

@ -529,8 +529,8 @@ 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 re-evaluate calculated fields that have a time schedule. 2 minutes by default.
check_interval: "${ACTORS_CALCULATED_FIELDS_CHECK_INTERVAL_SEC:30}"
# Interval in seconds to re-evaluate calculated fields that have a time schedule. 1 minute by default.
check_interval: "${ACTORS_CALCULATED_FIELDS_CHECK_INTERVAL_SEC:60}"
debug:
settings:

6
application/src/test/java/org/thingsboard/server/cf/EntityAggregationCalculatedFieldTest.java
View File

@ -109,7 +109,7 @@ public class EntityAggregationCalculatedFieldTest extends AbstractControllerTest
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsInInterval_3));
long interval = customInterval.getIntervalDurationMillis();
Watermark watermark = new Watermark(60, 10);
Watermark watermark = new Watermark(60);
CalculatedField totalConsumptionCF = createTotalConsumptionCF(device.getId(), customInterval, watermark);
await().alias("create CF and perform aggregation after interval end")
@ -140,7 +140,7 @@ public class EntityAggregationCalculatedFieldTest extends AbstractControllerTest
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":120}}", tsInInterval_3));
long interval = customInterval.getIntervalDurationMillis();
Watermark watermark = new Watermark(60, 10);
Watermark watermark = new Watermark(60);
CalculatedField totalConsumptionCF = createTotalConsumptionCF(device.getId(), customInterval, watermark);
await().alias("create CF and perform aggregation after interval end")
@ -155,7 +155,7 @@ public class EntityAggregationCalculatedFieldTest extends AbstractControllerTest
postTelemetry(device.getId(), String.format("{\"ts\": \"%s\", \"values\": {\"energy\":300}}", tsInInterval_1));
await().alias("create CF and perform aggregation after interval end")
.atMost(2 * watermark.getCheckInterval(), TimeUnit.SECONDS)
.atMost(2 * 10, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> {
ObjectNode result = getLatestTelemetry(device.getId(), "consumptionPerMin");

1
common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java
View File

@ -41,5 +41,6 @@ public class SystemParams {
int minAllowedScheduledUpdateIntervalInSecForCF;
int maxRelationLevelPerCfArgument;
long minAllowedDeduplicationIntervalInSecForCF;
long minAggregationIntervalInSecForCF;
TrendzSettings trendzSettings;
}

5
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/EntityAggregationCalculatedFieldConfiguration.java
View File

@ -20,6 +20,7 @@ import jakarta.validation.constraints.NotEmpty;
import lombok.Data;
import org.thingsboard.server.common.data.cf.CalculatedFieldType;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.Output;
import org.thingsboard.server.common.data.cf.configuration.aggregation.AggMetric;
@ -49,8 +50,8 @@ public class EntityAggregationCalculatedFieldConfiguration implements ArgumentsB
if (arguments.containsKey("ctx")) {
throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used.");
}
if (arguments.values().stream().anyMatch(Argument::hasTsRollingArgument)) {
throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support TS_ROLLING arguments.");
if (arguments.values().stream().anyMatch(argument -> !ArgumentType.TS_LATEST.equals(argument.getRefEntityKey().getType()))) {
throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' support only TS_LATEST arguments.");
}
}

1
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/single/interval/Watermark.java
View File

@ -25,6 +25,5 @@ import lombok.NoArgsConstructor;
public class Watermark {
private long duration;
private long checkInterval;
}

2
common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
View File

@ -188,6 +188,8 @@ public class DefaultTenantProfileConfiguration implements TenantProfileConfigura
private long maxSingleValueArgumentSizeInKBytes = 2;
@Schema(example = "3600")
private long minAllowedDeduplicationIntervalInSecForCF = 3600;
@Schema(example = "60")
private long minAggregationIntervalInSecForCF = 60;
@Override
public long getProfileThreshold(ApiUsageRecordKey key) {

17
dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java
View File

@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalcula
import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration;
import org.thingsboard.server.common.data.cf.configuration.ScheduledUpdateSupportedCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.aggregation.RelatedEntitiesAggregationCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.aggregation.single.EntityAggregationCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.cf.CalculatedFieldDao;
@ -30,6 +31,7 @@ import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@Component
@ -48,6 +50,7 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
validateSchedulingConfiguration(tenantId, calculatedField);
validateRelationQuerySourceArguments(tenantId, calculatedField);
validateAggregationConfiguration(tenantId, calculatedField);
validateEntityAggregationConfiguration(tenantId, calculatedField);
}
@Override
@ -123,6 +126,20 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
}
}
private void validateEntityAggregationConfiguration(TenantId tenantId, CalculatedField calculatedField) {
if (!(calculatedField.getConfiguration() instanceof EntityAggregationCalculatedFieldConfiguration aggConfiguration)) {
return;
}
long minAggregationIntervalInSec = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAggregationIntervalInSecForCF);
if (minAggregationIntervalInSec <= 0) {
return;
}
if (aggConfiguration.getInterval().getIntervalDurationMillis() > TimeUnit.SECONDS.toMillis(minAggregationIntervalInSec)) {
throw new IllegalArgumentException("Aggregation interval duration is less than configured " +
"minimum allowed aggregation interval in tenant profile: " + minAggregationIntervalInSec + " sec.");
}
}
private static void wrapAsDataValidation(Runnable validation) {
try {
validation.run();

Write Preview
Loading…
Cancel
Save