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

Merge pull request #5569 from dmytro-landiak/save-ts-no-latest 

[3.3.3] Save timeseries without latest persistence
pull/5744/head
Igor Kulikov 5 years ago
committed by GitHub
parent
3c75bdf51f 11f0916b34
commit
ab2a7e4357
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 148 additions and 58 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. 58
      application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
  2. 2
      common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java
  3. 29
      dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
  4. 98
      dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java
  5. 2
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
  6. 15
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
  7. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java

58
application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java
View File

@ -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<TsKvEntry> ts, long ttl, FutureCallback<Void> callback) {
doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, true);
}
@Override
public void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback) {
doSaveAndNotify(tenantId, customerId, entityId, ts, ttl, callback, false);
}
private void doSaveAndNotify(TenantId tenantId, CustomerId customerId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> 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<Integer>() {
@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<Integer> getCallback(TenantId tenantId, CustomerId customerId, boolean sysTenant, FutureCallback<Void> 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<TsKvEntry> ts, FutureCallback<Integer> 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<TsKvEntry> ts, long ttl, FutureCallback<Integer> callback) {
ListenableFuture<Integer> saveFuture = tsService.save(tenantId, entityId, ts, ttl);
addCallbacks(tenantId, entityId, ts, callback, saveFuture);
}
private void saveWithoutLatestAndNotifyInternal(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Integer> callback) {
ListenableFuture<Integer> saveFuture = tsService.saveWithoutLatest(tenantId, entityId, ts, ttl);
addCallbacks(tenantId, entityId, ts, callback, saveFuture);
}
private void addCallbacks(TenantId tenantId, EntityId entityId, List<TsKvEntry> ts, FutureCallback<Integer> callback, ListenableFuture<Integer> saveFuture) {
addMainCallback(saveFuture, callback);
addWsCallback(saveFuture, success -> onTimeSeriesUpdate(tenantId, entityId, ts));
if (EntityType.DEVICE.equals(entityId.getEntityType()) || EntityType.ASSET.equals(entityId.getEntityType())) {

2
common/dao-api/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesService.java
View File

@ -41,6 +41,8 @@ public interface TimeseriesService {
ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntry, long ttl);
ListenableFuture<Integer> saveWithoutLatest(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntry, long ttl);
ListenableFuture<List<Void>> saveLatest(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntry);
ListenableFuture<List<Void>> remove(TenantId tenantId, EntityId entityId, List<DeleteTsKvQuery> queries);

29
dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
View File

@ -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<List<Integer>, Integer> SUM_ALL_INTEGERS = new Function<List<Integer>, Integer>() {
@Override
@ -144,12 +145,26 @@ public class BaseTimeseriesService implements TimeseriesService {
@Override
public ListenableFuture<Integer> save(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntries, long ttl) {
List<ListenableFuture<Integer>> futures = Lists.newArrayListWithExpectedSize(tsKvEntries.size() * INSERTS_PER_ENTRY);
return doSave(tenantId, entityId, tsKvEntries, ttl, true);
}
@Override
public ListenableFuture<Integer> saveWithoutLatest(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntries, long ttl) {
return doSave(tenantId, entityId, tsKvEntries, ttl, false);
}
private ListenableFuture<Integer> doSave(TenantId tenantId, EntityId entityId, List<TsKvEntry> tsKvEntries, long ttl, boolean saveLatest) {
int inserts = saveLatest ? INSERTS_PER_ENTRY : INSERTS_PER_ENTRY_WITHOUT_LATEST;
List<ListenableFuture<Integer>> 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<ListenableFuture<Integer>> 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<ListenableFuture<Integer>> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
doSaveAndRegisterFuturesFor(tenantId, futures, entityId, tsKvEntry, ttl);
}
private void doSaveAndRegisterFuturesFor(TenantId tenantId, List<ListenableFuture<Integer>> 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));
}

98
dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java
View File

@ -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<TsKvEntry> tsList = tsService.findAllLatest(tenantId, entityId).get();
private void testLatestTsAndVerify(EntityId entityId) throws ExecutionException, InterruptedException, TimeoutException {
List<TsKvEntry> 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<TsKvEntry> entries = tsService.findLatest(tenantId, deviceId, Collections.singleton(STRING_KEY)).get();
List<TsKvEntry> 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<TsKvEntry> 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<ReadTsKvQuery> queries = new ArrayList<>();
queries.add(new BaseReadTsKvQuery(STRING_KEY, TS - 3, TS, 0, 1000, Aggregation.NONE, "ASC"));
List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
List<TsKvEntry> 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<ReadTsKvQuery> queries = new ArrayList<>();
queries.add(new BaseReadTsKvQuery(STRING_KEY, TS - 3, TS, 0, 1000, Aggregation.NONE, "DESC"));
List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
List<TsKvEntry> 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<TsKvEntry> 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<TsKvEntry> latest = tsService.findLatest(tenantId, deviceId, Collections.singletonList(STRING_KEY)).get();
List<TsKvEntry> 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<TsKvEntry> 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<TsKvEntry> 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> 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) {

2
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineTelemetryService.java
View File

@ -34,6 +34,8 @@ public interface RuleEngineTelemetryService {
void saveAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback);
void saveWithoutLatestAndNotify(TenantId tenantId, CustomerId id, EntityId entityId, List<TsKvEntry> ts, long ttl, FutureCallback<Void> callback);
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, FutureCallback<Void> callback);

15
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
View File

@ -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();

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java
View File

@ -22,11 +22,13 @@ import org.thingsboard.rule.engine.api.NodeConfiguration;
public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration<TbMsgTimeseriesNodeConfiguration> {
private long defaultTTL;
private boolean skipLatestPersistence;
@Override
public TbMsgTimeseriesNodeConfiguration defaultConfiguration() {
TbMsgTimeseriesNodeConfiguration configuration = new TbMsgTimeseriesNodeConfiguration();
configuration.setDefaultTTL(0L);
configuration.setSkipLatestPersistence(false);
return configuration;
}
}

Write Preview
Loading…
Cancel
Save