Browse Source

added minDeduplicationInterval to tenant profile config

pull/14141/head
IrynaMatveieva 12 months ago
parent
commit
37039a995d
  1. 8
      application/src/main/data/upgrade/basic/schema_update.sql
  2. 1
      application/src/main/java/org/thingsboard/server/controller/SystemInfoController.java
  3. 2
      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/LatestValuesAggregationCalculatedFieldState.java
  5. 64
      application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java
  6. 1
      common/data/src/main/java/org/thingsboard/server/common/data/SystemParams.java
  7. 2
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java
  8. 2
      common/data/src/main/java/org/thingsboard/server/common/data/tenant/profile/DefaultTenantProfileConfiguration.java
  9. 15
      dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java

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

@ -34,6 +34,12 @@ SET profile_data = jsonb_set(
WHEN (profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument' WHEN (profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument'
THEN NULL THEN NULL
ELSE to_jsonb(10) ELSE to_jsonb(10)
END,
'minAllowedDeduplicationIntervalInSecForCF',
CASE
WHEN (profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
THEN NULL
ELSE to_jsonb(3600)
END END
) )
), ),
@ -43,6 +49,8 @@ WHERE NOT (
(profile_data -> 'configuration') ? 'minAllowedScheduledUpdateIntervalInSecForCF' (profile_data -> 'configuration') ? 'minAllowedScheduledUpdateIntervalInSecForCF'
AND AND
(profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument' (profile_data -> 'configuration') ? 'maxRelationLevelPerCfArgument'
AND
(profile_data -> 'configuration') ? 'minAllowedDeduplicationIntervalInSecForCF'
); );
-- UPDATE TENANT PROFILE CONFIGURATION END -- UPDATE TENANT PROFILE CONFIGURATION END

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

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

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

@ -581,7 +581,7 @@ public class CalculatedFieldCtx {
} }
if (calculatedField.getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration thisConfig if (calculatedField.getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration thisConfig
&& other.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration otherConfig && other.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration otherConfig
&& (thisConfig.getDeduplicationIntervalMillis() != otherConfig.getDeduplicationIntervalMillis() || !thisConfig.getMetrics().equals(otherConfig.getMetrics()))) { && (thisConfig.getDeduplicationIntervalInSec() != otherConfig.getDeduplicationIntervalInSec() || !thisConfig.getMetrics().equals(otherConfig.getMetrics()))) {
return true; return true;
} }
return false; return false;

2
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/LatestValuesAggregationCalculatedFieldState.java

@ -66,7 +66,7 @@ public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedF
super.setCtx(ctx, actorCtx); super.setCtx(ctx, actorCtx);
var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration();
metrics = configuration.getMetrics(); metrics = configuration.getMetrics();
deduplicationInterval = configuration.getDeduplicationIntervalMillis(); deduplicationInterval = configuration.getDeduplicationIntervalInSec();
} }
@Override @Override

64
application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java

@ -79,12 +79,16 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
private AssetProfile assetProfile; private AssetProfile assetProfile;
private Asset asset; private Asset asset;
private long deduplicationInterval = 10000; private long deduplicationInterval = 10;
@Before @Before
public void beforeTest() throws Exception { public void beforeTest() throws Exception {
loginSysAdmin(); loginSysAdmin();
updateDefaultTenantProfileConfig(tenantProfileConfig -> {
tenantProfileConfig.setMinAllowedDeduplicationIntervalInSecForCF(1);
});
Tenant tenant = new Tenant(); Tenant tenant = new Tenant();
tenant.setTitle("My tenant"); tenant.setTitle("My tenant");
savedTenant = saveTenant(tenant); savedTenant = saveTenant(tenant);
@ -131,7 +135,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createOccupancyCF(assetProfile.getId()); createOccupancyCF(assetProfile.getId());
await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -149,7 +153,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device3.getId(), "{\"occupied\":true}"); postTelemetry(device3.getId(), "{\"occupied\":true}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -171,7 +175,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
Asset asset2 = createAsset("Asset 2", assetProfile.getId()); Asset asset2 = createAsset("Asset 2", assetProfile.getId());
await().alias("add entity to profile with no related entities and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("add entity to profile with no related entities and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
ObjectNode occupancy = getLatestTelemetry(asset2.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces"); ObjectNode occupancy = getLatestTelemetry(asset2.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces");
@ -184,7 +188,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createEntityRelation(asset2.getId(), device3.getId(), "Contains"); createEntityRelation(asset2.getId(), device3.getId(), "Contains");
createEntityRelation(asset2.getId(), device4.getId(), "Contains"); createEntityRelation(asset2.getId(), device4.getId(), "Contains");
await().alias("create relations and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create relations and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -196,7 +200,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device3.getId(), "{\"occupied\":false}"); postTelemetry(device3.getId(), "{\"occupied\":false}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.MILLISECONDS) await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -218,7 +222,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createOccupancyCF(assetProfile.getId()); createOccupancyCF(assetProfile.getId());
await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -240,7 +244,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device3.getId(), "{\"occupied\":true}"); postTelemetry(device3.getId(), "{\"occupied\":true}");
await().alias("change profile and no aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("change profile and no aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -262,7 +266,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createOccupancyCF(asset2.getId()); createOccupancyCF(asset2.getId());
await().alias("create CF and perform aggregation with default values").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform aggregation with default values").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -280,7 +284,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device1.getId(), "{\"occupied\":false}"); postTelemetry(device1.getId(), "{\"occupied\":false}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -301,7 +305,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device1.getId(), "{\"occupied\":false}"); postTelemetry(device1.getId(), "{\"occupied\":false}");
await().alias("delete cf and update telemetry and no aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("delete cf and update telemetry and no aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -319,13 +323,13 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device1.getId(), "{\"occupied\":false}"); postTelemetry(device1.getId(), "{\"occupied\":false}");
await().alias("update telemetry -> no changes").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) await().alias("update telemetry -> no changes").atMost(deduplicationInterval / 2, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(this::checkInitialCalculationValues); .untilAsserted(this::checkInitialCalculationValues);
postTelemetry(device2.getId(), "{\"occupied\":false}"); postTelemetry(device2.getId(), "{\"occupied\":false}");
await().alias("create CF and perform initial calculation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform initial calculation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -355,7 +359,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createOccupancyCF(asset2.getId()); createOccupancyCF(asset2.getId());
await().alias("create CF and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -368,7 +372,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
doDelete("/api/plugins/telemetry/DEVICE/" + device3.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + thirdTs + "&endTs=" + thirdTs + 1, String.class); doDelete("/api/plugins/telemetry/DEVICE/" + device3.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + thirdTs + "&endTs=" + thirdTs + 1, String.class);
doDelete("/api/plugins/telemetry/DEVICE/" + device4.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + secondTs + "&endTs=" + secondTs + 1, String.class); doDelete("/api/plugins/telemetry/DEVICE/" + device4.getId() + "/timeseries/delete?keys=occupied&deleteAllDataForKeys=false&rewriteLatestIfDeleted=true&deleteLatest=true&startTs=" + secondTs + "&endTs=" + secondTs + 1, String.class);
await().alias("delete latest telemetry and perform aggregation with previous or default values").atMost(deduplicationInterval * 2, TimeUnit.MILLISECONDS) await().alias("delete latest telemetry and perform aggregation with previous or default values").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset2.getId(), Map.of( verifyTelemetry(asset2.getId(), Map.of(
@ -390,7 +394,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
createEntityRelation(asset.getId(), device3.getId(), "Contains"); createEntityRelation(asset.getId(), device3.getId(), "Contains");
await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -408,7 +412,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
deleteEntityRelation(new EntityRelation(asset.getId(), device1.getId(), "Contains", RelationTypeGroup.COMMON)); deleteEntityRelation(new EntityRelation(asset.getId(), device1.getId(), "Contains", RelationTypeGroup.COMMON));
await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -432,7 +436,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
configuration.setRelation(new RelationPathLevel(EntitySearchDirection.FROM, "Has")); configuration.setRelation(new RelationPathLevel(EntitySearchDirection.FROM, "Has"));
saveCalculatedField(cf); saveCalculatedField(cf);
await().alias("update relation path and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update relation path and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -458,7 +462,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
configuration.setArguments(Map.of("oc", argument)); configuration.setArguments(Map.of("oc", argument));
saveCalculatedField(cf); saveCalculatedField(cf);
await().alias("update arguments and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update arguments and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of( verifyTelemetry(asset.getId(), Map.of(
@ -475,7 +479,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device2.getId(), "{\"temperature\":19.6}"); postTelemetry(device2.getId(), "{\"temperature\":19.6}");
CalculatedField cf = createAvgTemperatureCF(asset.getId()); CalculatedField cf = createAvgTemperatureCF(asset.getId());
await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
@ -489,7 +493,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
configuration.setMetrics(Map.of("maxTemperature", aggMetric)); configuration.setMetrics(Map.of("maxTemperature", aggMetric));
saveCalculatedField(cf); saveCalculatedField(cf);
await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24")); verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24"));
@ -498,7 +502,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device1.getId(), "{\"temperature\":101.3}"); postTelemetry(device1.getId(), "{\"temperature\":101.3}");
postTelemetry(device2.getId(), "{\"temperature\":25.8}"); postTelemetry(device2.getId(), "{\"temperature\":25.8}");
await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26")); verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26"));
@ -511,7 +515,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device2.getId(), "{\"temperature\":19.6}"); postTelemetry(device2.getId(), "{\"temperature\":19.6}");
CalculatedField cf = createAvgTemperatureCF(asset.getId()); CalculatedField cf = createAvgTemperatureCF(asset.getId());
await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
@ -524,7 +528,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
configuration.setOutput(output); configuration.setOutput(output);
saveCalculatedField(cf); saveCalculatedField(cf);
await().alias("update output and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update output and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
ArrayNode avgTemperature = getServerAttributes(asset.getId(), "avgTemperature"); ArrayNode avgTemperature = getServerAttributes(asset.getId(), "avgTemperature");
@ -540,17 +544,17 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device2.getId(), "{\"temperature\":19.6}"); postTelemetry(device2.getId(), "{\"temperature\":19.6}");
CalculatedField cf = createAvgTemperatureCF(asset.getId()); CalculatedField cf = createAvgTemperatureCF(asset.getId());
await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create avg temp cf and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
}); });
var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration(); var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) cf.getConfiguration();
configuration.setDeduplicationIntervalMillis(2 * deduplicationInterval); configuration.setDeduplicationIntervalInSec(2 * deduplicationInterval);
saveCalculatedField(cf); saveCalculatedField(cf);
await().alias("update deduplication interval and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS) await().alias("update deduplication interval and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24")); verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
@ -558,7 +562,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
postTelemetry(device2.getId(), "{\"temperature\":32.1}"); postTelemetry(device2.getId(), "{\"temperature\":32.1}");
await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(() -> { .untilAsserted(() -> {
verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28")); verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28"));
@ -566,7 +570,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
} }
private void checkInitialCalculation() { private void checkInitialCalculation() {
await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS) await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS)
.pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
.untilAsserted(this::checkInitialCalculationValues); .untilAsserted(this::checkInitialCalculationValues);
} }
@ -656,7 +660,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
LatestValuesAggregationCalculatedFieldConfiguration configuration = new LatestValuesAggregationCalculatedFieldConfiguration(); LatestValuesAggregationCalculatedFieldConfiguration configuration = new LatestValuesAggregationCalculatedFieldConfiguration();
configuration.setRelation(relation); configuration.setRelation(relation);
configuration.setArguments(inputs); configuration.setArguments(inputs);
configuration.setDeduplicationIntervalMillis(deduplicationInterval); configuration.setDeduplicationIntervalInSec(deduplicationInterval);
configuration.setMetrics(metrics); configuration.setMetrics(metrics);
configuration.setOutput(output); configuration.setOutput(output);

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

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

2
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java

@ -29,7 +29,7 @@ public class LatestValuesAggregationCalculatedFieldConfiguration implements Argu
private RelationPathLevel relation; private RelationPathLevel relation;
private Map<String, Argument> arguments; private Map<String, Argument> arguments;
private long deduplicationIntervalMillis; private long deduplicationIntervalInSec;
private Map<String, AggMetric> metrics; private Map<String, AggMetric> metrics;
private Output output; private Output output;
private boolean useLatestTs; private boolean useLatestTs;

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

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

15
dao/src/main/java/org/thingsboard/server/dao/service/validator/CalculatedFieldDataValidator.java

@ -21,6 +21,7 @@ import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration; import org.thingsboard.server.common.data.cf.configuration.ArgumentsBasedCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration; 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.ScheduledUpdateSupportedCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration;
import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
import org.thingsboard.server.dao.cf.CalculatedFieldDao; import org.thingsboard.server.dao.cf.CalculatedFieldDao;
@ -46,6 +47,7 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
validateCalculatedFieldConfiguration(calculatedField); validateCalculatedFieldConfiguration(calculatedField);
validateSchedulingConfiguration(tenantId, calculatedField); validateSchedulingConfiguration(tenantId, calculatedField);
validateRelationQuerySourceArguments(tenantId, calculatedField); validateRelationQuerySourceArguments(tenantId, calculatedField);
validateAggregationConfiguration(tenantId, calculatedField);
} }
@Override @Override
@ -87,7 +89,7 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
private void validateSchedulingConfiguration(TenantId tenantId, CalculatedField calculatedField) { private void validateSchedulingConfiguration(TenantId tenantId, CalculatedField calculatedField) {
if (!(calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledUpdateCfg) if (!(calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledUpdateCfg)
|| !scheduledUpdateCfg.isScheduledUpdateEnabled()) { || !scheduledUpdateCfg.isScheduledUpdateEnabled()) {
return; return;
} }
long minAllowedScheduledUpdateInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedScheduledUpdateIntervalInSecForCF); long minAllowedScheduledUpdateInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedScheduledUpdateIntervalInSecForCF);
@ -110,6 +112,17 @@ public class CalculatedFieldDataValidator extends DataValidator<CalculatedField>
wrapAsDataValidation(() -> relationQueryDynamicSourceConfiguration.validateMaxRelationLevel(argumentName, maxRelationLevel))); wrapAsDataValidation(() -> relationQueryDynamicSourceConfiguration.validateMaxRelationLevel(argumentName, maxRelationLevel)));
} }
private void validateAggregationConfiguration(TenantId tenantId, CalculatedField calculatedField) {
if (!(calculatedField.getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfiguration)) {
return;
}
long minAllowedDeduplicationInterval = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMinAllowedDeduplicationIntervalInSecForCF);
if (aggConfiguration.getDeduplicationIntervalInSec() < minAllowedDeduplicationInterval) {
throw new IllegalArgumentException("Deduplication interval is less than configured " +
"minimum allowed interval in tenant profile: " + minAllowedDeduplicationInterval);
}
}
private static void wrapAsDataValidation(Runnable validation) { private static void wrapAsDataValidation(Runnable validation) {
try { try {
validation.run(); validation.run();

Loading…
Cancel
Save