diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index a3771988cf..232f8bfd00 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -20,8 +20,10 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; +import org.jetbrains.annotations.NotNull; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; +import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.ApiUsageRecordKey; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.EntityView; @@ -45,7 +47,6 @@ import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.usagestats.TbApiUsageClient; import org.thingsboard.server.service.apiusage.TbApiUsageStateService; -import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.service.subscription.TbSubscriptionUtils; import javax.annotation.Nullable; @@ -118,28 +119,46 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Override public void saveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback) { + doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, true); + } + + @Override + public void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback) { + doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, false); + } + + private void doSaveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List ts, long ttl, FutureCallback callback, boolean saveLatest) { checkInternalEntity(entityId); boolean sysTenant = TenantId.SYS_TENANT_ID.equals(tenantId) || tenantId == null; if (sysTenant || apiUsageStateService.getApiUsageState(tenantId).isDbStorageEnabled()) { - saveAndNotifyInternal(tenantId, entityId, ts, ttl, new FutureCallback() { - @Override - public void onSuccess(Integer result) { - if (!sysTenant && result != null && result > 0) { - apiUsageClient.report(tenantId, customerId, ApiUsageRecordKey.STORAGE_DP_COUNT, result); - } - callback.onSuccess(null); - } - - @Override - public void onFailure(Throwable t) { - callback.onFailure(t); - } - }); + if (saveLatest) { + saveAndNotifyInternal(tenantId, entityId, ts, ttl, getCallback(tenantId, customerId, sysTenant, callback)); + } else { + saveWithoutLatestAndNotifyInternal(tenantId, entityId, ts, ttl, getCallback(tenantId, customerId, sysTenant, callback)); + } } else { callback.onFailure(new RuntimeException("DB storage writes are disabled due to API limits!")); } } + @NotNull + private FutureCallback getCallback(TenantId tenantId, CustomerId customerId, boolean sysTenant, FutureCallback callback) { + return new FutureCallback<>() { + @Override + public void onSuccess(Integer result) { + if (!sysTenant && result != null && result > 0) { + apiUsageClient.report(tenantId, customerId, ApiUsageRecordKey.STORAGE_DP_COUNT, result); + } + callback.onSuccess(null); + } + + @Override + public void onFailure(Throwable t) { + callback.onFailure(t); + } + }; + } + @Override public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback) { saveAndNotifyInternal(tenantId, entityId, ts, 0L, callback); @@ -148,6 +167,15 @@ public class DefaultTelemetrySubscriptionService extends AbstractSubscriptionSer @Override public void saveAndNotifyInternal(TenantId tenantId, EntityId entityId, List ts, long ttl, FutureCallback callback) { ListenableFuture saveFuture = tsService.save(tenantId, entityId, ts, ttl); + addCallbacks(tenantId, entityId, ts, callback, saveFuture); + } + + private void saveWithoutLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List ts, long ttl, FutureCallback callback) { + ListenableFuture saveFuture = tsService.saveWithoutLatest(tenantId, entityId, ts, ttl); + addCallbacks(tenantId, entityId, ts, callback, saveFuture); + } + + private void addCallbacks(TenantId tenantId, EntityId entityId, List ts, FutureCallback callback, ListenableFuture saveFuture) { addMainCallback(saveFuture, callback); addWsCallback(saveFuture, success -> onTimeSeriesUpdate(tenantId, entityId, ts)); if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) { diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java index b1a2541fc7..4418a826f3 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java @@ -41,6 +41,8 @@ public interface TimeseriesService { ListenableFuture save(TenantId tenantId, EntityId entityId, List tsKvEntry, long ttl); + ListenableFuture saveWithoutLatest(TenantId tenantId, EntityId entityId, List tsKvEntry, long ttl); + ListenableFuture> saveLatest(TenantId tenantId, EntityId entityId, List tsKvEntry); ListenableFuture> remove(TenantId tenantId, EntityId entityId, List queries); diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java index 3170f6c256..da3fe05024 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java @@ -56,6 +56,7 @@ import static org.apache.commons.lang3.StringUtils.isBlank; public class BaseTimeseriesService implements TimeseriesService { private static final int INSERTS_PER_ENTRY = 3; + private static final int INSERTS_PER_ENTRY_WITHOUT_LATEST = 2; private static final int DELETES_PER_ENTRY = INSERTS_PER_ENTRY; public static final Function, Integer> SUM_ALL_INTEGERS = new Function, Integer>() { @Override @@ -144,12 +145,26 @@ public class BaseTimeseriesService implements TimeseriesService { @Override public ListenableFuture save(TenantId tenantId, EntityId entityId, List tsKvEntries, long ttl) { - List> futures = Lists.newArrayListWithExpectedSize(tsKvEntries.size() * INSERTS_PER_ENTRY); + return doSave(tenantId, entityId, tsKvEntries, ttl, true); + } + + @Override + public ListenableFuture saveWithoutLatest(TenantId tenantId, EntityId entityId, List tsKvEntries, long ttl) { + return doSave(tenantId, entityId, tsKvEntries, ttl, false); + } + + private ListenableFuture doSave(TenantId tenantId, EntityId entityId, List tsKvEntries, long ttl, boolean saveLatest) { + int inserts = saveLatest ? INSERTS_PER_ENTRY : INSERTS_PER_ENTRY_WITHOUT_LATEST; + List> futures = Lists.newArrayListWithExpectedSize(tsKvEntries.size() * inserts); for (TsKvEntry tsKvEntry : tsKvEntries) { if (tsKvEntry == null) { throw new IncorrectParameterException("Key value entry can't be null"); } - saveAndRegisterFutures(tenantId, futures, entityId, tsKvEntry, ttl); + if (saveLatest) { + saveAndRegisterFutures(tenantId, futures, entityId, tsKvEntry, ttl); + } else { + saveWithoutLatestAndRegisterFutures(tenantId, futures, entityId, tsKvEntry, ttl); + } } return Futures.transform(Futures.allAsList(futures), SUM_ALL_INTEGERS, MoreExecutors.directExecutor()); } @@ -167,11 +182,19 @@ public class BaseTimeseriesService implements TimeseriesService { } private void saveAndRegisterFutures(TenantId tenantId, List> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { + doSaveAndRegisterFuturesFor(tenantId, futures, entityId, tsKvEntry, ttl); + futures.add(Futures.transform(timeseriesLatestDao.saveLatest(tenantId, entityId, tsKvEntry), v -> 0, MoreExecutors.directExecutor())); + } + + private void saveWithoutLatestAndRegisterFutures(TenantId tenantId, List> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { + doSaveAndRegisterFuturesFor(tenantId, futures, entityId, tsKvEntry, ttl); + } + + private void doSaveAndRegisterFuturesFor(TenantId tenantId, List> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW)) { throw new IncorrectParameterException("Telemetry data can't be stored for entity view. Read only"); } futures.add(timeseriesDao.savePartition(tenantId, entityId, tsKvEntry.getTs(), tsKvEntry.getKey())); - futures.add(Futures.transform(timeseriesLatestDao.saveLatest(tenantId, entityId, tsKvEntry), v -> 0, MoreExecutors.directExecutor())); futures.add(timeseriesDao.save(tenantId, entityId, tsKvEntry, ttl)); } diff --git a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java index b7872fc239..bba679d509 100644 --- a/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java +++ b/dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java @@ -17,7 +17,10 @@ package org.thingsboard.server.dao.service.timeseries; import com.datastax.oss.driver.api.core.uuid.Uuids; import lombok.extern.slf4j.Slf4j; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import org.thingsboard.server.common.data.EntityView; import org.thingsboard.server.common.data.Tenant; import org.thingsboard.server.common.data.id.DeviceId; @@ -42,6 +45,8 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; @@ -52,6 +57,7 @@ import static org.junit.Assert.assertNotNull; @Slf4j public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { + static final int MAX_TIMEOUT = 30; private static final String STRING_KEY = "stringKey"; private static final String LONG_KEY = "longKey"; @@ -93,8 +99,8 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { testLatestTsAndVerify(deviceId); } - private void testLatestTsAndVerify(EntityId entityId) throws ExecutionException, InterruptedException { - List tsList = tsService.findAllLatest(tenantId, entityId).get(); + private void testLatestTsAndVerify(EntityId entityId) throws ExecutionException, InterruptedException, TimeoutException { + List tsList = tsService.findAllLatest(tenantId, entityId).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertNotNull(tsList); assertEquals(4, tsList.size()); @@ -134,11 +140,24 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { saveEntries(deviceId, TS - 1); saveEntries(deviceId, TS); - List entries = tsService.findLatest(tenantId, deviceId, Collections.singleton(STRING_KEY)).get(); + List entries = tsService.findLatest(tenantId, deviceId, Collections.singleton(STRING_KEY)).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(1, entries.size()); Assert.assertEquals(toTsEntry(TS, stringKvEntry), entries.get(0)); } + @Test + public void testFindLatestWithoutLatestUpdate() throws Exception { + DeviceId deviceId = new DeviceId(Uuids.timeBased()); + + saveEntries(deviceId, TS - 2); + saveEntries(deviceId, TS - 1); + saveEntriesWithoutLatest(deviceId, TS); + + List entries = tsService.findLatest(tenantId, deviceId, Collections.singleton(STRING_KEY)).get(MAX_TIMEOUT, TimeUnit.SECONDS); + Assert.assertEquals(1, entries.size()); + Assert.assertEquals(toTsEntry(TS - 1, stringKvEntry), entries.get(0)); + } + @Test public void testFindByQueryAscOrder() throws Exception { DeviceId deviceId = new DeviceId(Uuids.timeBased()); @@ -150,7 +169,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { List queries = new ArrayList<>(); queries.add(new BaseReadTsKvQuery(STRING_KEY, TS - 3, TS, 0, 1000, Aggregation.NONE, "ASC")); - List entries = tsService.findAll(tenantId, deviceId, queries).get(); + List entries = tsService.findAll(tenantId, deviceId, queries).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(3, entries.size()); Assert.assertEquals(toTsEntry(TS - 3, stringKvEntry), entries.get(0)); Assert.assertEquals(toTsEntry(TS - 2, stringKvEntry), entries.get(1)); @@ -158,7 +177,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { EntityView entityView = saveAndCreateEntityView(deviceId, Arrays.asList(STRING_KEY)); - entries = tsService.findAll(tenantId, entityView.getId(), queries).get(); + entries = tsService.findAll(tenantId, entityView.getId(), queries).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(3, entries.size()); Assert.assertEquals(toTsEntry(TS - 3, stringKvEntry), entries.get(0)); Assert.assertEquals(toTsEntry(TS - 2, stringKvEntry), entries.get(1)); @@ -176,7 +195,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { List queries = new ArrayList<>(); queries.add(new BaseReadTsKvQuery(STRING_KEY, TS - 3, TS, 0, 1000, Aggregation.NONE, "DESC")); - List entries = tsService.findAll(tenantId, deviceId, queries).get(); + List entries = tsService.findAll(tenantId, deviceId, queries).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(3, entries.size()); Assert.assertEquals(toTsEntry(TS - 1, stringKvEntry), entries.get(0)); Assert.assertEquals(toTsEntry(TS - 2, stringKvEntry), entries.get(1)); @@ -184,7 +203,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { EntityView entityView = saveAndCreateEntityView(deviceId, Arrays.asList(STRING_KEY)); - entries = tsService.findAll(tenantId, entityView.getId(), queries).get(); + entries = tsService.findAll(tenantId, entityView.getId(), queries).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(3, entries.size()); Assert.assertEquals(toTsEntry(TS - 1, stringKvEntry), entries.get(0)); Assert.assertEquals(toTsEntry(TS - 2, stringKvEntry), entries.get(1)); @@ -201,13 +220,13 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { saveEntries(deviceId, 40000); tsService.remove(tenantId, deviceId, Collections.singletonList( - new BaseDeleteTsKvQuery(STRING_KEY, 25000, 45000, true))).get(); + new BaseDeleteTsKvQuery(STRING_KEY, 25000, 45000, true))).get(MAX_TIMEOUT, TimeUnit.SECONDS); List list = tsService.findAll(tenantId, deviceId, Collections.singletonList( - new BaseReadTsKvQuery(STRING_KEY, 5000, 45000, 10000, 10, Aggregation.NONE))).get(); + new BaseReadTsKvQuery(STRING_KEY, 5000, 45000, 10000, 10, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(2, list.size()); - List latest = tsService.findLatest(tenantId, deviceId, Collections.singletonList(STRING_KEY)).get(); + List latest = tsService.findLatest(tenantId, deviceId, Collections.singletonList(STRING_KEY)).get(MAX_TIMEOUT, TimeUnit.SECONDS); Assert.assertEquals(20000, latest.get(0).getTs()); } @@ -226,7 +245,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { entries.add(save(deviceId, 55000, 600)); List list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.NONE))).get(); + 60000, 20000, 3, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(55000, list.get(0).getTs()); assertEquals(java.util.Optional.of(600L), list.get(0).getLongValue()); @@ -238,7 +257,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(400L), list.get(2).getLongValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.AVG))).get(); + 60000, 20000, 3, Aggregation.AVG))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); assertEquals(java.util.Optional.of(150.0), list.get(0).getDoubleValue()); @@ -250,7 +269,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(550.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.SUM))).get(); + 60000, 20000, 3, Aggregation.SUM))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -263,7 +282,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(1100L), list.get(2).getLongValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.MIN))).get(); + 60000, 20000, 3, Aggregation.MIN))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -276,7 +295,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(500L), list.get(2).getLongValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.MAX))).get(); + 60000, 20000, 3, Aggregation.MAX))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -289,7 +308,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(600L), list.get(2).getLongValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.COUNT))).get(); + 60000, 20000, 3, Aggregation.COUNT))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -310,7 +329,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { entries.add(save(deviceId, 115000, "C2")); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 60000, - 120000, 20000, 3, Aggregation.NONE))).get(); + 120000, 20000, 3, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(115000, list.get(0).getTs()); assertEquals(java.util.Optional.of("C2"), list.get(0).getStrValue()); @@ -323,7 +342,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 60000, - 120000, 20000, 3, Aggregation.MIN))).get(); + 120000, 20000, 3, Aggregation.MIN))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(70000, list.get(0).getTs()); @@ -336,7 +355,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of("C1"), list.get(2).getStrValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 60000, - 120000, 20000, 3, Aggregation.MAX))).get(); + 120000, 20000, 3, Aggregation.MAX))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(70000, list.get(0).getTs()); @@ -349,7 +368,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of("C2"), list.get(2).getStrValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 60000, - 120000, 20000, 3, Aggregation.COUNT))).get(); + 120000, 20000, 3, Aggregation.COUNT))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(70000, list.get(0).getTs()); @@ -377,7 +396,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { entries.add(save(deviceId, 55000, 600.0)); List list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.NONE))).get(); + 60000, 20000, 3, Aggregation.NONE))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(55000, list.get(0).getTs()); assertEquals(java.util.Optional.of(600.0), list.get(0).getDoubleValue()); @@ -389,7 +408,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(400.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.AVG))).get(); + 60000, 20000, 3, Aggregation.AVG))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); assertEquals(java.util.Optional.of(150.0), list.get(0).getDoubleValue()); @@ -401,7 +420,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(550.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.SUM))).get(); + 60000, 20000, 3, Aggregation.SUM))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -414,7 +433,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(1100.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.MIN))).get(); + 60000, 20000, 3, Aggregation.MIN))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -427,7 +446,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(500.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.MAX))).get(); + 60000, 20000, 3, Aggregation.MAX))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -440,7 +459,7 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { assertEquals(java.util.Optional.of(600.0), list.get(2).getDoubleValue()); list = tsService.findAll(tenantId, deviceId, Collections.singletonList(new BaseReadTsKvQuery(LONG_KEY, 0, - 60000, 20000, 3, Aggregation.COUNT))).get(); + 60000, 20000, 3, Aggregation.COUNT))).get(MAX_TIMEOUT, TimeUnit.SECONDS); assertEquals(3, list.size()); assertEquals(10000, list.get(0).getTs()); @@ -455,28 +474,37 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest { private TsKvEntry save(DeviceId deviceId, long ts, long value) throws Exception { TsKvEntry entry = new BasicTsKvEntry(ts, new LongDataEntry(LONG_KEY, value)); - tsService.save(tenantId, deviceId, entry).get(); + tsService.save(tenantId, deviceId, entry).get(MAX_TIMEOUT, TimeUnit.SECONDS); return entry; } private TsKvEntry save(DeviceId deviceId, long ts, double value) throws Exception { TsKvEntry entry = new BasicTsKvEntry(ts, new DoubleDataEntry(LONG_KEY, value)); - tsService.save(tenantId, deviceId, entry).get(); + tsService.save(tenantId, deviceId, entry).get(MAX_TIMEOUT, TimeUnit.SECONDS); return entry; } private TsKvEntry save(DeviceId deviceId, long ts, String value) throws Exception { TsKvEntry entry = new BasicTsKvEntry(ts, new StringDataEntry(LONG_KEY, value)); - tsService.save(tenantId, deviceId, entry).get(); + tsService.save(tenantId, deviceId, entry).get(MAX_TIMEOUT, TimeUnit.SECONDS); return entry; } - private void saveEntries(DeviceId deviceId, long ts) throws ExecutionException, InterruptedException { - tsService.save(tenantId, deviceId, toTsEntry(ts, stringKvEntry)).get(); - tsService.save(tenantId, deviceId, toTsEntry(ts, longKvEntry)).get(); - tsService.save(tenantId, deviceId, toTsEntry(ts, doubleKvEntry)).get(); - tsService.save(tenantId, deviceId, toTsEntry(ts, booleanKvEntry)).get(); + private void saveEntries(DeviceId deviceId, long ts) throws ExecutionException, InterruptedException, TimeoutException { + tsService.save(tenantId, deviceId, toTsEntry(ts, stringKvEntry)).get(MAX_TIMEOUT, TimeUnit.SECONDS); + tsService.save(tenantId, deviceId, toTsEntry(ts, longKvEntry)).get(MAX_TIMEOUT, TimeUnit.SECONDS); + tsService.save(tenantId, deviceId, toTsEntry(ts, doubleKvEntry)).get(MAX_TIMEOUT, TimeUnit.SECONDS); + tsService.save(tenantId, deviceId, toTsEntry(ts, booleanKvEntry)).get(MAX_TIMEOUT, TimeUnit.SECONDS); + } + + private void saveEntriesWithoutLatest(DeviceId deviceId, long ts) throws ExecutionException, InterruptedException, TimeoutException { + List tsKvEntry = List.of( + toTsEntry(ts, stringKvEntry), + toTsEntry(ts, longKvEntry), + toTsEntry(ts, doubleKvEntry), + toTsEntry(ts, booleanKvEntry)); + tsService.saveWithoutLatest(tenantId, deviceId, tsKvEntry, 0).get(MAX_TIMEOUT, TimeUnit.SECONDS); } private static TsKvEntry toTsEntry(long ts, KvEntry entry) { diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java index e641c83144..b2fd7d2a97 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java @@ -34,6 +34,8 @@ public interface RuleEngineTelemetryService { void saveAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List ts, long ttl, FutureCallback callback); + void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List ts, long ttl, FutureCallback callback); + void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, FutureCallback callback); void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List attributes, boolean notifyDevice, FutureCallback callback); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java index c99a575cc3..1d52b7cdfa 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java @@ -23,9 +23,8 @@ import org.thingsboard.rule.engine.api.TbContext; import org.thingsboard.rule.engine.api.TbNode; import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeException; -import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.rule.engine.api.util.TbNodeUtils; -import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.kv.BasicTsKvEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; @@ -46,7 +45,9 @@ import java.util.concurrent.TimeUnit; name = "save timeseries", configClazz = TbMsgTimeseriesNodeConfiguration.class, nodeDescription = "Saves timeseries data", - nodeDetails = "Saves timeseries telemetry data based on configurable TTL parameter. Expects messages with 'POST_TELEMETRY_REQUEST' message type", + nodeDetails = "Saves timeseries telemetry data based on configurable TTL parameter. Expects messages with 'POST_TELEMETRY_REQUEST' message type. " + + "Timestamp in milliseconds will be taken from metadata.ts, otherwise 'now' timestamp will be applied. " + + "Allows stopping updating values for incoming keys in the latest ts_kv table if 'skipLatestPersistence' is set to true.", uiResources = {"static/rulenode/rulenode-core-config.js"}, configDirective = "tbActionNodeTimeseriesConfig", icon = "file_upload" @@ -94,7 +95,11 @@ public class TbMsgTimeseriesNode implements TbNode { if (ttl == 0L) { ttl = tenantProfileDefaultStorageTtl; } - ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getCustomerId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); + if (config.isSkipLatestPersistence()) { + ctx.getTelemetryService().saveWithoutLatestAndNotify(ctx.getTenantId(), msg.getCustomerId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); + } else { + ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getCustomerId(), msg.getOriginator(), tsKvEntryList, ttl, new TelemetryNodeCallback(ctx, msg)); + } } public static long getTs(TbMsg msg) { @@ -103,7 +108,7 @@ public class TbMsgTimeseriesNode implements TbNode { if (!StringUtils.isEmpty(tsStr)) { try { ts = Long.parseLong(tsStr); - } catch (NumberFormatException e) { + } catch (NumberFormatException ignored) { } } else { ts = msg.getTs(); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java index fb63fbdbde..bb661df905 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java @@ -22,11 +22,13 @@ import org.thingsboard.rule.engine.api.NodeConfiguration; public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration { private long defaultTTL; + private boolean skipLatestPersistence; @Override public TbMsgTimeseriesNodeConfiguration defaultConfiguration() { TbMsgTimeseriesNodeConfiguration configuration = new TbMsgTimeseriesNodeConfiguration(); configuration.setDefaultTTL(0L); + configuration.setSkipLatestPersistence(false); return configuration; } }