Merge pull request #14623 from irynamatveieva/fix/cf-ts-rolling-arg

Fixed processing of telemetry batch in calculated fields

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@@ -36,6 +36,7 @@ import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
 import org.thingsboard.server.common.msg.queue.ServiceType;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.util.ProtoUtils;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
@@ -48,6 +49,7 @@ import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState;
 import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
+import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -62,6 +64,8 @@ import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry.getValueForTsRecord;
+
 
 /**
  * @author Andrew Shvayka
@@ -356,13 +360,39 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
             ReferencedEntityKey key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_LATEST, null);
             Set<String> argNames = args.get(key);
             if (argNames != null) {
-                argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item)));
+                SingleValueArgumentEntry incoming = new SingleValueArgumentEntry(item);
+                argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
+                    if (existing == null) {
+                        return incoming;
+                    }
+                    existing.updateEntry(incoming);
+                    return existing;
+                }));
             }
 
             key = new ReferencedEntityKey(item.getKv().getKey(), ArgumentType.TS_ROLLING, null);
             argNames = args.get(key);
             if (argNames != null) {
-                argNames.forEach(argName -> arguments.put(argName, new SingleValueArgumentEntry(item)));
+                Double recordValue = getValueForTsRecord(ProtoUtils.fromProto(item.getKv()));
+                argNames.forEach(argName -> arguments.compute(argName, (name, existing) -> {
+                    if (existing instanceof TsRollingArgumentEntry rolling) {
+                        if (recordValue != null) {
+                            rolling.getTsRecords().put(item.getTs(), recordValue);
+                        }
+                        return rolling;
+                    }
+                    TsRollingArgumentEntry rolling = new TsRollingArgumentEntry();
+                    if (recordValue != null) {
+                        rolling.getTsRecords().put(item.getTs(), recordValue);
+                    }
+                    if (existing instanceof SingleValueArgumentEntry single) {
+                        Double existingValue = getValueForTsRecord(single.getKvEntryValue());
+                        if (existingValue != null) {
+                            rolling.getTsRecords().put(single.getTs(), existingValue);
+                        }
+                    }
+                    return rolling;
+                }));
             }
         }
         return arguments;

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

@@ -122,20 +122,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
     }
 
     private void addTsRecord(Long ts, KvEntry value) {
-        try {
+        Double recordValue = getValueForTsRecord(value);
-            switch (value.getDataType()) {
+        if (recordValue != null) {
-                case LONG -> value.getLongValue().ifPresent(aLong -> tsRecords.put(ts, aLong.doubleValue()));
+            tsRecords.put(ts, recordValue);
-                case DOUBLE -> value.getDoubleValue().ifPresent(aDouble -> tsRecords.put(ts, aDouble));
-                case BOOLEAN -> value.getBooleanValue().ifPresent(aBoolean -> tsRecords.put(ts, aBoolean ? 1.0 : 0.0));
-                case STRING -> value.getStrValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
-                case JSON -> value.getJsonValue().ifPresent(aString -> tsRecords.put(ts, Double.parseDouble(aString)));
-            }
-        } catch (Exception e) {
-            tsRecords.put(ts, Double.NaN);
-            log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
-        } finally {
-            cleanupExpiredRecords();
         }
+        cleanupExpiredRecords();
     }
 
     private void addTsRecord(Long ts, double value) {
@@ -150,4 +141,19 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
         tsRecords.entrySet().removeIf(tsRecord -> tsRecord.getKey() < System.currentTimeMillis() - timeWindow);
     }
 
+    public static Double getValueForTsRecord(KvEntry value) {
+        try {
+            return switch (value.getDataType()) {
+                case LONG -> value.getLongValue().map(Long::doubleValue).orElse(null);
+                case DOUBLE -> value.getDoubleValue().orElse(null);
+                case BOOLEAN -> value.getBooleanValue().map(b -> b ? 1.0 : 0.0).orElse(null);
+                case STRING -> value.getStrValue().map(Double::parseDouble).orElse(null);
+                case JSON -> value.getJsonValue().map(Double::parseDouble).orElse(null);
+            };
+        } catch (Exception e) {
+            log.debug("Invalid value '{}' for time series rolling arguments. Only numeric values are supported.", value.getValue());
+            return Double.NaN;
+        }
+    }
+
 }

application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java

@@ -796,6 +796,86 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
                 });
     }
 
+    @Test
+    public void testCalculatedFieldWhenBatchOfTelemetrySent() throws Exception {
+        Device testDevice = createDevice("Test device", "1234567890");
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":5}"));
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":10}"));
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":20}"));
+
+        CalculatedField calculatedField = new CalculatedField();
+        calculatedField.setEntityId(testDevice.getId());
+        calculatedField.setType(CalculatedFieldType.SCRIPT);
+        calculatedField.setName("Script CF");
+        calculatedField.setDebugSettings(DebugSettings.all());
+
+        SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration();
+
+        ReferencedEntityKey refEntityKeyA = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null);
+        Argument argumentA = new Argument();
+        argumentA.setRefEntityKey(refEntityKeyA);
+        Argument argumentB = new Argument();
+        ReferencedEntityKey refEntityKeyB = new ReferencedEntityKey("b", ArgumentType.TS_ROLLING, null);
+        argumentB.setTimeWindow(TimeUnit.MINUTES.toMillis(10));
+        argumentB.setLimit(1000);
+        argumentB.setRefEntityKey(refEntityKeyB);
+        config.setArguments(Map.of("a", argumentA, "b", argumentB));
+        config.setExpression("""
+                return {
+                    "latestA": a,
+                    "avgB": b.avg
+                };
+                """);
+
+        Output output = new Output();
+        output.setType(OutputType.TIME_SERIES);
+        config.setOutput(output);
+
+        calculatedField.setConfiguration(config);
+
+        doPost("/api/calculatedField", calculatedField, CalculatedField.class);
+
+        await().alias("create CF -> perform initial calculation").atMost(TIMEOUT, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
+                    assertThat(result).isNotNull();
+                    assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("5");
+                    assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("15.0");
+                });
+
+        long now = System.currentTimeMillis();
+        doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("""
+                [{
+                    "ts": %s,
+                    "values": {
+                        "a": 6,
+                        "b": 100
+                    }
+                }, {
+                    "ts": %s,
+                    "values": {
+                        "a": 7,
+                        "b": 200
+                    }
+                }, {
+                    "ts": %s,
+                    "values": {
+                        "a": 8,
+                        "b": 300
+                    }
+                }]""", now - TimeUnit.MINUTES.toMillis(2), now, now - TimeUnit.MINUTES.toMillis(5))));
+
+        await().alias("update telemetry -> recalculate state").atMost(TIMEOUT, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode result = getLatestTelemetry(testDevice.getId(), "latestA", "avgB");
+                    assertThat(result).isNotNull();
+                    assertThat(result.get("latestA").get(0).get("value").asText()).isEqualTo("7");
+                    assertThat(result.get("avgB").get(0).get("value").asText()).isEqualTo("126.0");
+                });
+    }
+
     private ObjectNode getLatestTelemetry(EntityId entityId, String... keys) throws Exception {
         return doGetAsync("/api/plugins/telemetry/" + entityId.getEntityType() + "/" + entityId.getId() + "/values/timeseries?keys=" + String.join(",", keys), ObjectNode.class);
     }