Browse Source

fixed error when no telemetry in db

pull/12620/head
IrynaMatveieva 1 year ago
parent
commit
7fcd948071
  1. 3
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
  2. 5
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
  3. 2
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  4. 19
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java
  5. 9
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  6. 4
      application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java
  7. 3
      common/data/src/main/java/org/thingsboard/server/common/data/event/EventFilter.java

3
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java

@ -32,8 +32,8 @@ import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.queue.util.AfterStartUp;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.queue.util.AfterStartUp;
import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
import java.util.Collections;
@ -117,6 +117,7 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
CalculatedField calculatedField = getCalculatedField(calculatedFieldId);
if (calculatedField != null) {
ctx = new CalculatedFieldCtx(calculatedField, tbelInvokeService, apiLimitService);
ctx.init();
calculatedFieldsCtx.put(calculatedFieldId, ctx);
log.debug("[{}] Put calculated field ctx into cache: {}", calculatedFieldId, ctx);
}

5
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java

@ -39,6 +39,7 @@ import org.thingsboard.server.actors.calculatedField.MultipleTbCallback;
import org.thingsboard.server.cluster.TbClusterService;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
@ -68,7 +69,6 @@ import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.dao.attributes.AttributesService;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
@ -447,6 +447,9 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
private KvEntry createDefaultKvEntry(Argument argument) {
String key = argument.getRefEntityKey().getKey();
String defaultValue = argument.getDefaultValue();
if (StringUtils.isBlank(defaultValue)) {
return new StringDataEntry(key, null);
}
if (NumberUtils.isParsable(defaultValue)) {
return new DoubleDataEntry(key, Double.parseDouble(defaultValue));
}

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

@ -52,7 +52,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState {
ArgumentEntry newEntry = entry.getValue();
ArgumentEntry existingEntry = arguments.get(key);
if (existingEntry == null) {
if (existingEntry == null || existingEntry == SingleValueArgumentEntry.EMPTY || existingEntry == TsRollingArgumentEntry.EMPTY) {
validateNewEntry(newEntry);
arguments.put(key, newEntry);
stateUpdated = true;

19
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/RocksDBStateService.java

@ -63,7 +63,7 @@ public class RocksDBStateService implements CalculatedFieldStateService {
CalculatedFieldStateProto stateProto = toProto(stateId, state);
long maxStateSizeInKBytes = ctx.getMaxStateSizeInKBytes();
if (maxStateSizeInKBytes <= 0 || stateProto.getSerializedSize() <= ctx.getMaxStateSizeInKBytes()) {
rocksDBService.put(toProto(stateId), toProto(stateId, state));
rocksDBService.put(toProto(stateId), stateProto);
}
callback.onSuccess();
}
@ -111,8 +111,11 @@ public class RocksDBStateService implements CalculatedFieldStateService {
private SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) {
SingleValueArgumentProto.Builder builder = SingleValueArgumentProto.newBuilder()
.setArgName(argName)
.setValue(KvProtoUtil.toTsValueProto(entry.getTs(), entry.getKvEntryValue()));
.setArgName(argName);
if (entry != SingleValueArgumentEntry.EMPTY) {
builder.setValue(KvProtoUtil.toTsValueProto(entry.getTs(), entry.getKvEntryValue()));
}
Optional.ofNullable(entry.getVersion()).ifPresent(builder::setVersion);
@ -122,7 +125,9 @@ public class RocksDBStateService implements CalculatedFieldStateService {
private TsValueListProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) {
TsValueListProto.Builder builder = TsValueListProto.newBuilder().setKey(argName);
entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(KvProtoUtil.toTsValueProto(ts, value)));
if (entry != TsRollingArgumentEntry.EMPTY) {
entry.getTsRecords().forEach((ts, value) -> builder.addTsValue(KvProtoUtil.toTsValueProto(ts, value)));
}
return builder.build();
}
@ -151,6 +156,9 @@ public class RocksDBStateService implements CalculatedFieldStateService {
}
private SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) {
if (!proto.hasValue()) {
return (SingleValueArgumentEntry) SingleValueArgumentEntry.EMPTY;
}
TsValueProto tsValueProto = proto.getValue();
long ts = tsValueProto.getTs();
BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getArgName(), tsValueProto);
@ -158,6 +166,9 @@ public class RocksDBStateService implements CalculatedFieldStateService {
}
private TsRollingArgumentEntry fromRollingArgumentProto(TsValueListProto proto) {
if (proto.getTsValueCount() <= 0) {
return (TsRollingArgumentEntry) TsRollingArgumentEntry.EMPTY;
}
TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>();
proto.getTsValueList().forEach(tsValueProto -> {
BasicKvEntry kvEntry = (BasicKvEntry) KvProtoUtil.fromTsValueProto(proto.getKey(), tsValueProto);

9
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java

@ -94,10 +94,17 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
Long newVersion = singleValueEntry.getVersion();
if (newVersion == null || this.version == null || newVersion > this.version) {
this.ts = singleValueEntry.getTs();
this.kvEntryValue = singleValueEntry.getKvEntryValue();
this.version = newVersion;
// TODO: should we persist updated ts and version values?
BasicKvEntry newValue = singleValueEntry.getKvEntryValue();
if (this.kvEntryValue.getValue().equals(newValue.getValue())) {
return false;
}
this.kvEntryValue = singleValueEntry.getKvEntryValue();
return true;
}
} else {
throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType());
}

4
application/src/test/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntryTest.java

@ -69,4 +69,8 @@ public class SingleValueArgumentEntryTest {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 18L), 234L))).isFalse();
}
@Test
void testUpdateEntryWhenValueWasNotChanged() {
assertThat(entry.updateEntry(new SingleValueArgumentEntry(ts + 18, new LongDataEntry("key", 11L), 237L))).isFalse();
}
}

3
common/data/src/main/java/org/thingsboard/server/common/data/event/EventFilter.java

@ -29,7 +29,8 @@ import io.swagger.v3.oas.annotations.media.Schema;
@JsonSubTypes.Type(value = RuleChainDebugEventFilter.class, name = "DEBUG_RULE_CHAIN"),
@JsonSubTypes.Type(value = ErrorEventFilter.class, name = "ERROR"),
@JsonSubTypes.Type(value = LifeCycleEventFilter.class, name = "LC_EVENT"),
@JsonSubTypes.Type(value = StatisticsEventFilter.class, name = "STATS")
@JsonSubTypes.Type(value = StatisticsEventFilter.class, name = "STATS"),
@JsonSubTypes.Type(value = CalculatedFieldDebugEventFilter.class, name = "DEBUG_CALCULATED_FIELD")
})
public interface EventFilter {

Loading…
Cancel
Save