Merge pull request #5358 from smatvienko-tb/TbGetTelemetryNode_Aggregation_feature

Tb get telemetry node aggregation feature

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.rule.engine.metadata;
 
-import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.json.JsonWriteFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -35,6 +34,7 @@ import org.thingsboard.rule.engine.api.TbNode;
 import org.thingsboard.rule.engine.api.TbNodeConfiguration;
 import org.thingsboard.rule.engine.api.TbNodeException;
 import org.thingsboard.rule.engine.api.util.TbNodeUtils;
+import org.thingsboard.server.common.data.kv.Aggregation;
 import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
@@ -50,7 +50,6 @@ import java.util.stream.Collectors;
 import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.FETCH_MODE_ALL;
 import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.FETCH_MODE_FIRST;
 import static org.thingsboard.rule.engine.metadata.TbGetTelemetryNodeConfiguration.MAX_FETCH_SIZE;
-import static org.thingsboard.server.common.data.kv.Aggregation.NONE;
 
 /**
  * Created by mshvayka on 04.09.18.
@@ -64,6 +63,7 @@ import static org.thingsboard.server.common.data.kv.Aggregation.NONE;
                 "If selected fetch mode <b>ALL</b> Telemetry will be added like array into Message Metadata where <b>key</b> is Timestamp and <b>value</b> is value of Telemetry.</br>" +
                 "If selected fetch mode <b>FIRST</b> or <b>LAST</b> Telemetry will be added like string without Timestamp.</br>" +
                 "Also, the rule node allows you to select telemetry sampling order: <b>ASC</b> or <b>DESC</b>. </br>" +
+                "Aggregation feature allows you to fetch aggregated telemetry as a single value by <b>AVG, COUNT, SUM, MIN, MAX, NONE</b>. </br>" +
                 "<b>Note</b>: The maximum size of the fetched array is 1000 records.\n ",
         uiResources = {"static/rulenode/rulenode-core-config.js"},
         configDirective = "tbEnrichmentNodeGetTelemetryFromDatabase")
@@ -78,6 +78,7 @@ public class TbGetTelemetryNode implements TbNode {
     private ObjectMapper mapper;
     private String fetchMode;
     private String orderByFetchAll;
+    private Aggregation aggregation;
 
     @Override
     public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
@@ -89,11 +90,20 @@ public class TbGetTelemetryNode implements TbNode {
         if (StringUtils.isEmpty(orderByFetchAll)) {
             orderByFetchAll = ASC_ORDER;
         }
+        aggregation = parseAggregationConfig(config.getAggregation());
+
         mapper = new ObjectMapper();
         mapper.configure(JsonWriteFeature.QUOTE_FIELD_NAMES.mappedFeature(), false);
         mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_FIELD_NAMES, true);
     }
 
+    Aggregation parseAggregationConfig(String aggName) {
+        if (StringUtils.isEmpty(aggName)) {
+            return Aggregation.NONE;
+        }
+        return Aggregation.valueOf(aggName);
+    }
+
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) throws ExecutionException, InterruptedException, TbNodeException {
         if (tsKeyNames.isEmpty()) {
@@ -120,8 +130,14 @@ public class TbGetTelemetryNode implements TbNode {
     }
 
     private List<ReadTsKvQuery> buildQueries(TbMsg msg, List<String> keys) {
+        final Interval interval = getInterval(msg);
+        final long aggIntervalStep = Aggregation.NONE.equals(aggregation) ? 1 :
+                // exact how it validates on BaseTimeseriesService.validate()
+                // see CassandraBaseTimeseriesDao.findAllAsync()
+                interval.getEndTs() - interval.getStartTs();
+
         return keys.stream()
-                .map(key -> new BaseReadTsKvQuery(key, getInterval(msg).getStartTs(), getInterval(msg).getEndTs(), 1, limit, NONE, getOrderBy()))
+                .map(key -> new BaseReadTsKvQuery(key, interval.getStartTs(), interval.getEndTs(), aggIntervalStep, limit, aggregation, getOrderBy()))
                 .collect(Collectors.toList());
     }
 

@@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.metadata;
 
 import lombok.Data;
 import org.thingsboard.rule.engine.api.NodeConfiguration;
+import org.thingsboard.server.common.data.kv.Aggregation;
 
 import java.util.Collections;
 import java.util.List;
@@ -46,6 +47,7 @@ public class TbGetTelemetryNodeConfiguration implements NodeConfiguration<TbGetT
     private String endIntervalTimeUnit;
     private String fetchMode; //FIRST, LAST, ALL
     private String orderBy; //ASC, DESC
+    private String aggregation; //MIN, MAX, AVG, SUM, COUNT, NONE;
     private int limit;
 
     private List<String> latestTsKeyNames;
@@ -63,6 +65,7 @@ public class TbGetTelemetryNodeConfiguration implements NodeConfiguration<TbGetT
         configuration.setStartIntervalPattern("");
         configuration.setEndIntervalPattern("");
         configuration.setOrderBy("ASC");
+        configuration.setAggregation(Aggregation.NONE.name());
         configuration.setLimit(MAX_FETCH_SIZE);
         return configuration;
     }

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.rule.engine.metadata;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.server.common.data.kv.Aggregation;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.willCallRealMethod;
+import static org.mockito.Mockito.mock;
+
+public class TbGetTelemetryNodeTest {
+
+    TbGetTelemetryNode node;
+
+    @Before
+    public void setUp() throws Exception {
+        node = mock(TbGetTelemetryNode.class);
+        willCallRealMethod().given(node).parseAggregationConfig(any());
+    }
+
+    @Test
+    public void givenAggregationAsString_whenParseAggregation_thenReturnEnum() {
+        //compatibility with old configs without "aggregation" parameter
+        assertThat(node.parseAggregationConfig(null), is(Aggregation.NONE));
+        assertThat(node.parseAggregationConfig(""), is(Aggregation.NONE));
+
+        //common values
+        assertThat(node.parseAggregationConfig("MIN"), is(Aggregation.MIN));
+        assertThat(node.parseAggregationConfig("MAX"), is(Aggregation.MAX));
+        assertThat(node.parseAggregationConfig("AVG"), is(Aggregation.AVG));
+        assertThat(node.parseAggregationConfig("SUM"), is(Aggregation.SUM));
+        assertThat(node.parseAggregationConfig("COUNT"), is(Aggregation.COUNT));
+        assertThat(node.parseAggregationConfig("NONE"), is(Aggregation.NONE));
+
+        //all possible values in future
+        for (Aggregation aggEnum : Aggregation.values()) {
+            assertThat(node.parseAggregationConfig(aggEnum.name()), is(aggEnum));
+        }
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void givenAggregationWhiteSpace_whenParseAggregation_thenException() {
+        node.parseAggregationConfig(" ");
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void givenAggregationIncorrect_whenParseAggregation_thenException() {
+        node.parseAggregationConfig("TOP");
+    }
+
+}