Features/ts.mrange groupby (#304)

.prettierignore

@@ -1,9 +1,11 @@
 .DS_Store
+.dockerignore
 .gitkeep
 .gitattributes
 .gitignore
 .editorconfig
 .prettierignore
+Dockerfile
 LICENSE
 yarn.lock
 go.mod
@@ -12,6 +14,7 @@ go.sum
 .github
 
 # Folders
+docker-compose
 node_modules
 coverage
 data

docker-compose/cluster.yml

@@ -1,4 +1,4 @@
-version: "3.7"
+version: '3.7'
 
 networks:
   cluster-network:
@@ -46,7 +46,7 @@ services:
     container_name: redis3
     build:
       context: cluster
-    entrypoint: ["/usr/local/bin/startup_cluster.sh"]
+    entrypoint: ['/usr/local/bin/startup_cluster.sh']
     ports:
       - '6381:6379'
       - '16379'

pkg/redis-time-series.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"time"
@@ -97,13 +98,25 @@ func queryTsMRange(from int64, to int64, qm queryModel, client redisClient) back
 		return response
 	}
 
+	var args []interface{}
+	//args := []interface{}{strconv.FormatInt(from, 10), to}
+
 	// Execute command
 	if qm.Aggregation != "" {
-		err = client.RunFlatCmd(&result, qm.Command, strconv.FormatInt(from, 10), to, "AGGREGATION", qm.Aggregation, qm.Bucket, "WITHLABELS", "FILTER", filter)
+		args = []interface{}{to, "AGGREGATION", qm.Aggregation, qm.Bucket, "WITHLABELS", "FILTER", filter}
 	} else {
-		err = client.RunFlatCmd(&result, qm.Command, strconv.FormatInt(from, 10), to, "WITHLABELS", "FILTER", filter)
+		args = []interface{}{to, qm.Bucket, "WITHLABELS", "FILTER", filter}
 	}
 
+	if qm.TsGroupByLabel != "" {
+		if qm.TsReducer == "" {
+			return errorHandler(response, errors.New("reducer not provided for groups, please provide a reducer (e.g. avg, sum) and try again"))
+		}
+		args = append(args, "GROUPBY", qm.TsGroupByLabel, "REDUCE", qm.TsReducer)
+	}
+
+	err = client.RunFlatCmd(&result, qm.Command, strconv.FormatInt(from, 10), args...)
+
 	// Check error
 	if err != nil {
 		return errorHandler(response, err)

pkg/redis-time-series_test.go

@@ -346,6 +346,60 @@ func TestQueryTsMRange(t *testing.T) {
 			"",
 			nil,
 		},
+		{
+			"test groupby/reduction",
+			queryModel{Command: models.TimeSeriesMRange, TsReducer: "SUM", TsGroupByLabel: "reduceLabel"},
+			[]interface{}{
+				[]interface{}{
+					[]byte("foo=bar"),
+					[]interface{}{
+						[]interface{}{
+							[]byte("foo"),
+							[]byte("bar"),
+						},
+						[]interface{}{
+							[]byte("__reducer__"),
+							[]byte("sum"),
+						},
+						[]interface{}{
+							[]byte("__source__"),
+							[]byte("ts:1,ts:2,ts:3"),
+						},
+					},
+					[]interface{}{
+						[]interface{}{int64(1686835010300), []byte("2102")},
+						[]interface{}{int64(1686835011312), []byte("1882")},
+						[]interface{}{int64(1686835013348), []byte("2378")},
+						[]interface{}{int64(1686835014362), []byte("3007")},
+					},
+				},
+			},
+			1686835010300,
+			1686835014362,
+			2,
+			4,
+			[]valueToCheckInResponse{
+				{frameIndex: 0, fieldIndex: 0, rowIndex: 0, value: time.Unix(0, 1686835010300*int64(time.Millisecond))},
+				{frameIndex: 0, fieldIndex: 1, rowIndex: 0, value: float64(2102)},
+			},
+			"foo=bar",
+			"",
+			"",
+			nil,
+		},
+		{"should return error because we missed an actual reducer",
+			queryModel{Command: models.TimeSeriesMRange, Key: "test1", Filter: "filter", TsGroupByLabel: "foo"},
+			interface{}("someString"),
+			0,
+			0,
+			0,
+			0,
+			nil,
+			"",
+			"",
+			"reducer not provided for groups, please provide a reducer (e.g. avg, sum) and try again",
+			nil,
+		},
 		{
 			"should return error if result is string",
 			queryModel{Command: models.TimeSeriesMRange, Key: "test1", Filter: "filter"},

pkg/types.go

@@ -69,6 +69,8 @@ type queryModel struct {
 	Max               string   `json:"max"`
 	ZRangeQuery       string   `json:"zrangeQuery"`
 	Path              string   `json:"path"`
+	TsReducer         string   `json:"tsReducer"`
+	TsGroupByLabel    string   `json:"tsGroupByLabel"`
 	SearchQuery       string   `json:"searchQuery"`
 	SortBy            string   `json:"sortBy"`
 	SortDirection     string   `json:"sortDirection"`

src/components/QueryEditor/QueryEditor.test.tsx

@@ -501,6 +501,16 @@ describe('QueryEditor', () => {
         queryWhenShown: { refId: '', type: QueryTypeValue.TIMESERIES, command: RedisTimeSeries.RANGE },
         queryWhenHidden: { refId: '', type: QueryTypeValue.REDIS, command: Redis.INFO },
       },
+      {
+        name: 'tsGroupByLabel',
+        getComponent: (wrapper: ShallowComponent) =>
+          wrapper.findWhere((node) => {
+            return node.prop('onChange') === wrapper.instance().onTsGroupByLabelChange;
+          }),
+        type: 'string',
+        queryWhenShown: { refId: '', type: QueryTypeValue.TIMESERIES, command: RedisTimeSeries.MRANGE },
+        queryWhenHidden: { refId: '', type: QueryTypeValue.REDIS, command: Redis.INFO },
+      },
       {
         name: 'zrangeQuery',
         getComponent: (wrapper: ShallowComponent) =>

src/components/QueryEditor/QueryEditor.tsx

@@ -31,6 +31,8 @@ import {
   RedisJson,
   RedisQuery,
   RedisTimeSeries,
+  Reducers,
+  ReducerValue,
   ZRangeQuery,
   ZRangeQueryValue,
 } from '../../redis';
@@ -189,6 +191,16 @@ export class QueryEditor extends PureComponent<Props> {
     });
   };
 
+  /**
+   * Ts Reducer Change
+   */
+  onTsReducerChange = this.createSelectFieldHandler<ReducerValue>('tsReducer');
+
+  /**
+   * Group By Change
+   */
+  onTsGroupByLabelChange = this.createTextFieldHandler('tsGroupByLabel');
+
   /**
    * Aggregation change
    */
@@ -332,6 +344,8 @@ export class QueryEditor extends PureComponent<Props> {
       streamingInterval,
       streamingCapacity,
       streamingDataType,
+      tsGroupByLabel,
+      tsReducer,
     } = this.props.query;
     const { onRunQuery, datasource } = this.props;
 
@@ -717,6 +731,30 @@ export class QueryEditor extends PureComponent<Props> {
             </div>
           )}
 
+        {type === QueryTypeValue.TIMESERIES &&
+          command &&
+          CommandParameters.tsGroupBy.includes(command as RedisTimeSeries) && (
+            <div className="gf-form">
+              <FormField
+                labelWidth={8}
+                inputWidth={10}
+                value={tsGroupByLabel}
+                onChange={this.onTsGroupByLabelChange}
+                label="Group By"
+                tooltip="The label to group your time-series by for your reduction"
+              />
+              {tsGroupByLabel && (
+                <Select
+                  options={Reducers}
+                  width={30}
+                  onChange={this.onTsReducerChange}
+                  value={tsReducer}
+                  menuPlacement="bottom"
+                />
+              )}
+            </div>
+          )}
+
         <div className="gf-form">
           <Switch
             label="Streaming"

src/redis/command.ts

@@ -76,6 +76,8 @@ export const CommandParameters = {
   zrangeQuery: [Redis.ZRANGE],
   path: [RedisJson.TYPE, RedisJson.OBJKEYS, RedisJson.GET, RedisJson.OBJLEN, RedisJson.ARRLEN],
   pyFunction: [RedisGears.PYEXECUTE],
+  tsGroupBy: [RedisTimeSeries.MRANGE],
+  tsReducer: [RedisTimeSeries.MRANGE],
   searchQuery: [RediSearch.SEARCH],
   offset: [RediSearch.SEARCH],
   returnFields: [RediSearch.SEARCH],

src/redis/time-series.ts

@@ -57,6 +57,19 @@ export enum AggregationValue {
   SUM = 'sum',
 }
 
+export enum ReducerValue {
+  AVG = 'avg',
+  SUM = 'sum',
+  MIN = 'min',
+  MAX = 'max',
+  RANGE = 'range',
+  COUNT = 'count',
+  STDP = 'std.p',
+  STDS = 'std.s',
+  VARP = 'var.p',
+  VARS = 'var.s',
+}
+
 /**
  * Aggregations
  */
@@ -69,3 +82,27 @@ export const Aggregations: Array<SelectableValue<AggregationValue>> = [
   { label: 'Range', description: 'Diff between maximum and minimum in the bucket', value: AggregationValue.RANGE },
   { label: 'Sum', description: 'Summation', value: AggregationValue.SUM },
 ];
+
+/**
+ * Reducers
+ */
+export const Reducers: Array<SelectableValue<ReducerValue>> = [
+  { label: 'Avg', description: 'Arithmetic mean of all non-NaN values', value: ReducerValue.AVG },
+  { label: 'Sum', description: 'Sum of all non-NaN values', value: ReducerValue.SUM },
+  { label: 'Min', description: 'Minimum non-NaN value', value: ReducerValue.MIN },
+  { label: 'Max', description: 'Maximum non-NaN value', value: ReducerValue.MAX },
+  {
+    label: 'Range',
+    description: 'Difference between maximum non-Nan value and minimum non-NaN value',
+    value: ReducerValue.RANGE,
+  },
+  { label: 'Count', description: 'Number of non-NaN values', value: ReducerValue.COUNT },
+  {
+    label: 'Std Population',
+    description: 'Population standard deviation of all non-NaN values',
+    value: ReducerValue.STDP,
+  },
+  { label: 'Std Sample', description: 'Sample standard deviation of all non-NaN values', value: ReducerValue.STDS },
+  { label: 'Var Population', description: 'Population variance of all non-NaN values', value: ReducerValue.VARP },
+  { label: 'Var Sample', description: 'Sample variance of all non-NaN values', value: ReducerValue.VARS },
+];

src/redis/types.ts

@@ -1,4 +1,4 @@
-import { ZRangeQueryValue } from 'redis';
+import { ReducerValue, ZRangeQueryValue } from 'redis';
 import { DataQuery } from '@grafana/data';
 import { StreamingDataType } from '../constants';
 import { InfoSectionValue } from './info';
@@ -97,6 +97,18 @@ export interface RedisQuery extends DataQuery {
    */
   aggregation?: AggregationValue;
 
+  /**
+   * The reduction to run to sum-up a group
+   *
+   * @type {ReducerValue}
+   */
+  tsReducer?: ReducerValue;
+
+  /**
+   * The label to group time-series by in an TS.MRANGE.
+   */
+  tsGroupByLabel?: string;
+
   /**
    * ZRANGE Query
    *

src/time-series/time-series.test.ts

@@ -72,6 +72,7 @@ describe('TimeSeriesStreaming', () => {
     ]);
     expect(data.fields[0].name === 'value');
     expect(data.fields[0].type === FieldType.number);
-    expect(data.fields[0].values.toArray() === [123]);
+    const fieldsArr = data.fields[0].values.toArray();
+    expect(fieldsArr.length === 1 && fieldsArr[0] === 123);
   });
 });