TbGetTelemetryNode Aggregation feature allows you to fetch aggregated telemetry as a single value by <b>AVG, COUNT, SUM, MIN, MAX, NONE</b>.

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNode.java

@@ -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()) {
@@ -121,7 +131,7 @@ public class TbGetTelemetryNode implements TbNode {
 
     private List<ReadTsKvQuery> buildQueries(TbMsg msg, List<String> keys) {
         return keys.stream()
-                .map(key -> new BaseReadTsKvQuery(key, getInterval(msg).getStartTs(), getInterval(msg).getEndTs(), 1, limit, NONE, getOrderBy()))
+                .map(key -> new BaseReadTsKvQuery(key, getInterval(msg).getStartTs(), getInterval(msg).getEndTs(), 1, limit, aggregation, getOrderBy()))
                 .collect(Collectors.toList());
     }
 

rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNodeConfiguration.java

@@ -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;
     }

rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/metadata/TbGetTelemetryNodeTest.java

@@ -0,0 +1,83 @@
+/**
+ * ThingsBoard, Inc. ("COMPANY") CONFIDENTIAL
+ *
+ * Copyright © 2016-2021 ThingsBoard, Inc. All Rights Reserved.
+ *
+ * NOTICE: All information contained herein is, and remains
+ * the property of ThingsBoard, Inc. and its suppliers,
+ * if any.  The intellectual and technical concepts contained
+ * herein are proprietary to ThingsBoard, Inc.
+ * and its suppliers and may be covered by U.S. and Foreign Patents,
+ * patents in process, and are protected by trade secret or copyright law.
+ *
+ * Dissemination of this information or reproduction of this material is strictly forbidden
+ * unless prior written permission is obtained from COMPANY.
+ *
+ * Access to the source code contained herein is hereby forbidden to anyone except current COMPANY employees,
+ * managers or contractors who have executed Confidentiality and Non-disclosure agreements
+ * explicitly covering such access.
+ *
+ * The copyright notice above does not evidence any actual or intended publication
+ * or disclosure  of  this source code, which includes
+ * information that is confidential and/or proprietary, and is a trade secret, of  COMPANY.
+ * ANY REPRODUCTION, MODIFICATION, DISTRIBUTION, PUBLIC  PERFORMANCE,
+ * OR PUBLIC DISPLAY OF OR THROUGH USE  OF THIS  SOURCE CODE  WITHOUT
+ * THE EXPRESS WRITTEN CONSENT OF COMPANY IS STRICTLY PROHIBITED,
+ * AND IN VIOLATION OF APPLICABLE LAWS AND INTERNATIONAL TREATIES.
+ * THE RECEIPT OR POSSESSION OF THIS SOURCE CODE AND/OR RELATED INFORMATION
+ * DOES NOT CONVEY OR IMPLY ANY RIGHTS TO REPRODUCE, DISCLOSE OR DISTRIBUTE ITS CONTENTS,
+ * OR TO MANUFACTURE, USE, OR SELL ANYTHING THAT IT  MAY DESCRIBE, IN WHOLE OR IN PART.
+ */
+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");
+    }
+
+}