Features/ts.mrange groupby

.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);
  });
 });