Browse Source

Merge branch 'bugfix/lts/cfs-arguments-last-update-ts' into bugfix/cfs-arguments-last-update-ts

pull/14501/head
dshvaika 10 months ago
parent
commit
b1000fb4ec
  1. 14
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java
  2. 18
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java
  3. 11
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java
  4. 18
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java
  5. 4
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java

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

@ -41,6 +41,8 @@ import java.util.stream.Collectors;
@Getter @Getter
public abstract class BaseCalculatedFieldState implements CalculatedFieldState, Closeable { public abstract class BaseCalculatedFieldState implements CalculatedFieldState, Closeable {
protected static final long DEFAULT_LAST_UPDATE_TS = -1L;
protected final EntityId entityId; protected final EntityId entityId;
protected CalculatedFieldCtx ctx; protected CalculatedFieldCtx ctx;
protected TbActorRef actorCtx; protected TbActorRef actorCtx;
@ -48,7 +50,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
protected Map<String, ArgumentEntry> arguments = new HashMap<>(); protected Map<String, ArgumentEntry> arguments = new HashMap<>();
protected boolean sizeExceedsLimit; protected boolean sizeExceedsLimit;
protected long latestTimestamp = -1; protected long latestTimestamp = DEFAULT_LAST_UPDATE_TS;
protected ReadinessStatus readinessStatus; protected ReadinessStatus readinessStatus;
@Setter @Setter
@ -119,7 +121,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
requiredArguments = null; requiredArguments = null;
arguments.clear(); arguments.clear();
sizeExceedsLimit = false; sizeExceedsLimit = false;
latestTimestamp = -1; latestTimestamp = DEFAULT_LAST_UPDATE_TS;
} }
@Override @Override
@ -147,7 +149,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
return valuesNode; return valuesNode;
} }
long latestTs = getLatestTimestamp(); long latestTs = getLatestTimestamp();
if (latestTs == -1) { if (latestTs == DEFAULT_LAST_UPDATE_TS) {
return valuesNode; return valuesNode;
} }
ObjectNode resultNode = JacksonUtil.newObjectNode(); ObjectNode resultNode = JacksonUtil.newObjectNode();
@ -165,12 +167,12 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState,
newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis(); newTs = (lastEntry != null) ? lastEntry.getKey() : System.currentTimeMillis();
} else if (entry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) { } else if (entry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) {
newTs = relatedEntitiesArgumentEntry.getEntityInputs().values().stream() newTs = relatedEntitiesArgumentEntry.getEntityInputs().values().stream()
.mapToLong(e -> (e instanceof SingleValueArgumentEntry s) ? s.getTs() : 0L) .mapToLong(e -> (e instanceof SingleValueArgumentEntry s) ? s.getTs() : DEFAULT_LAST_UPDATE_TS)
.max() .max()
.orElse(0L); .orElse(DEFAULT_LAST_UPDATE_TS);
} else if (entry instanceof GeofencingArgumentEntry geofencingArgumentEntry) { } else if (entry instanceof GeofencingArgumentEntry geofencingArgumentEntry) {
newTs = geofencingArgumentEntry.getZoneStates().values().stream() newTs = geofencingArgumentEntry.getZoneStates().values().stream()
.mapToLong(GeofencingZoneState::getTs).max().orElse(0L); .mapToLong(GeofencingZoneState::getTs).max().orElse(DEFAULT_LAST_UPDATE_TS);
} }
this.latestTimestamp = Math.max(this.latestTimestamp, newTs); this.latestTimestamp = Math.max(this.latestTimestamp, newTs);
} }

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

@ -79,10 +79,14 @@ import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
@Data @Data
@Slf4j @Slf4j
public class CalculatedFieldCtx implements Closeable { public class CalculatedFieldCtx implements Closeable {
private static final long SCHEDULED_UPDATE_DISABLED_VALUE = -1L;
private CalculatedField calculatedField; private CalculatedField calculatedField;
private CalculatedFieldId cfId; private CalculatedFieldId cfId;
@ -197,7 +201,7 @@ public class CalculatedFieldCtx implements Closeable {
} }
} }
if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) { if (calculatedField.getConfiguration() instanceof ScheduledUpdateSupportedCalculatedFieldConfiguration scheduledConfig) {
this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : -1L; this.scheduledUpdateIntervalMillis = scheduledConfig.isScheduledUpdateEnabled() ? TimeUnit.SECONDS.toMillis(scheduledConfig.getScheduledUpdateInterval()) : SCHEDULED_UPDATE_DISABLED_VALUE;
} }
if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) { if (calculatedField.getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig) {
this.useLatestTs = aggConfig.isUseLatestTs(); this.useLatestTs = aggConfig.isUseLatestTs();
@ -715,8 +719,8 @@ public class CalculatedFieldCtx implements Closeable {
return false; return false;
} }
private boolean isScheduledUpdateEnabled() { private boolean isScheduledUpdateDisabled() {
return scheduledUpdateIntervalMillis != -1; return scheduledUpdateIntervalMillis == SCHEDULED_UPDATE_DISABLED_VALUE;
} }
public boolean shouldFetchRelationQueryDynamicArgumentsFromDb(CalculatedFieldState state) { public boolean shouldFetchRelationQueryDynamicArgumentsFromDb(CalculatedFieldState state) {
@ -726,11 +730,11 @@ public class CalculatedFieldCtx implements Closeable {
return switch (cfType) { return switch (cfType) {
case PROPAGATION -> true; case PROPAGATION -> true;
case GEOFENCING -> { case GEOFENCING -> {
if (!isScheduledUpdateEnabled()) { if (isScheduledUpdateDisabled()) {
yield false; yield false;
} }
var geofencingState = (GeofencingCalculatedFieldState) state; var geofencingState = (GeofencingCalculatedFieldState) state;
if (geofencingState.getLastDynamicArgumentsRefreshTs() == -1L) { if (geofencingState.getLastDynamicArgumentsRefreshTs() == DEFAULT_LAST_UPDATE_TS) {
yield true; yield true;
} }
yield geofencingState.getLastDynamicArgumentsRefreshTs() < yield geofencingState.getLastDynamicArgumentsRefreshTs() <
@ -744,10 +748,10 @@ public class CalculatedFieldCtx implements Closeable {
if (!(state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState)) { if (!(state instanceof RelatedEntitiesAggregationCalculatedFieldState relatedEntitiesAggState)) {
return false; return false;
} }
if (!isScheduledUpdateEnabled()) { if (isScheduledUpdateDisabled()) {
return false; return false;
} }
if (relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() == -1L) { if (relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() == DEFAULT_LAST_UPDATE_TS) {
return true; return true;
} }
return relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() < System.currentTimeMillis() - scheduledUpdateIntervalMillis; return relatedEntitiesAggState.getLastRelatedEntitiesRefreshTs() < System.currentTimeMillis() - scheduledUpdateIntervalMillis;

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

@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.core.type.TypeReference;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.Data; import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.lang.Nullable; import org.springframework.lang.Nullable;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfArg;
@ -34,11 +33,14 @@ import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto; import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS;
@Data @Data
@NoArgsConstructor
@AllArgsConstructor @AllArgsConstructor
public class SingleValueArgumentEntry implements ArgumentEntry { public class SingleValueArgumentEntry implements ArgumentEntry {
public static final Long DEFAULT_VERSION = -1L;
@Nullable @Nullable
protected EntityId entityId; protected EntityId entityId;
@ -48,7 +50,10 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
protected boolean forceResetPrevious; protected boolean forceResetPrevious;
public static final Long DEFAULT_VERSION = -1L; public SingleValueArgumentEntry() {
this.ts = DEFAULT_LAST_UPDATE_TS;
this.version = DEFAULT_VERSION;
}
public SingleValueArgumentEntry(EntityId entityId, ArgumentEntry entry) { public SingleValueArgumentEntry(EntityId entityId, ArgumentEntry entry) {
this(entry); this(entry);

18
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesAggregationCalculatedFieldState.java

@ -57,12 +57,12 @@ import static java.util.concurrent.TimeUnit.SECONDS;
public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState { public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState {
@Setter @Setter
private long lastArgsRefreshTs = -1; private long lastArgsRefreshTs = DEFAULT_LAST_UPDATE_TS;
@Setter @Setter
private long lastMetricsEvalTs = -1; private long lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
@Setter @Setter
private long lastRelatedEntitiesRefreshTs = -1; private long lastRelatedEntitiesRefreshTs = DEFAULT_LAST_UPDATE_TS;
private long deduplicationIntervalMs = -1; private long deduplicationIntervalMs = DEFAULT_LAST_UPDATE_TS;
private Map<String, AggMetric> metrics; private Map<String, AggMetric> metrics;
private ScheduledFuture<?> reevaluationFuture; private ScheduledFuture<?> reevaluationFuture;
@ -102,9 +102,9 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
@Override @Override
public void reset() { // must reset everything dependent on arguments public void reset() { // must reset everything dependent on arguments
super.reset(); super.reset();
lastArgsRefreshTs = -1; lastArgsRefreshTs = DEFAULT_LAST_UPDATE_TS;
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
lastRelatedEntitiesRefreshTs = -1; lastRelatedEntitiesRefreshTs = DEFAULT_LAST_UPDATE_TS;
metrics = null; metrics = null;
} }
@ -153,7 +153,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
} }
public Map<String, ArgumentEntry> updateEntityData(Map<String, ArgumentEntry> fetchedArgs) { public Map<String, ArgumentEntry> updateEntityData(Map<String, ArgumentEntry> fetchedArgs) {
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
return update(fetchedArgs, ctx); return update(fetchedArgs, ctx);
} }
@ -162,7 +162,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry; RelatedEntitiesArgumentEntry aggEntry = (RelatedEntitiesArgumentEntry) argEntry;
aggEntry.getEntityInputs().remove(relatedEntityId); aggEntry.getEntityInputs().remove(relatedEntityId);
}); });
lastMetricsEvalTs = -1; lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
lastArgsRefreshTs = System.currentTimeMillis(); lastArgsRefreshTs = System.currentTimeMillis();
} }

4
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/geofencing/GeofencingCalculatedFieldState.java

@ -58,7 +58,7 @@ import static org.thingsboard.server.common.data.cf.configuration.geofencing.Geo
@EqualsAndHashCode(callSuper = true) @EqualsAndHashCode(callSuper = true)
public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState { public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState {
private long lastDynamicArgumentsRefreshTs = -1; private long lastDynamicArgumentsRefreshTs = DEFAULT_LAST_UPDATE_TS;
public GeofencingCalculatedFieldState(EntityId entityId) { public GeofencingCalculatedFieldState(EntityId entityId) {
super(entityId); super(entityId);
@ -147,7 +147,7 @@ public class GeofencingCalculatedFieldState extends BaseCalculatedFieldState {
@Override @Override
public void reset() { public void reset() {
super.reset(); super.reset();
lastDynamicArgumentsRefreshTs = -1; lastDynamicArgumentsRefreshTs = DEFAULT_LAST_UPDATE_TS;
} }
public void updateLastDynamicArgumentsRefreshTs() { public void updateLastDynamicArgumentsRefreshTs() {

Loading…
Cancel
Save