Merge pull request #281 from Helene/opentsdb_arrays_true

Support OpenTSDB 2.4 /query/api JSON serializer

Author: Helene

source/collector.py

@@ -205,7 +205,7 @@ def _get_sensor_labels(self):
 
 @classattributes(dict(metricsaggr=None, filters=None, grouptags=None,
                       start='', end='', nsamples=0, duration=0,
-                      dsBucketSize=0, dsOp='', rawData=False),
+                      dsBucketSize=0, dsOp='', rawData=False, dpsArrays=False),
                  ['sensor', 'period'])
 class QueryPolicy(object):
 
@@ -372,10 +372,16 @@ def _collect(self):
                 list(self.request.metricsaggr.keys())[0]))
             rows = res.downsampleResults(self.dsInterval, self.request.dsOp)
 
-        columnValues = defaultdict(dict)
-        for row in rows:
-            for value, columnInfo in zip(row.values, res.columnInfos):
-                columnValues[columnInfo][row.tstamp] = value
+        if self.request.dpsArrays:
+            columnValues = defaultdict(list)
+            for row in rows:
+                for value, columnInfo in zip(row.values, res.columnInfos):
+                    columnValues[columnInfo].append([row.tstamp, value])
+        else:
+            columnValues = defaultdict(dict)
+            for row in rows:
+                for value, columnInfo in zip(row.values, res.columnInfos):
+                    columnValues[columnInfo][row.tstamp] = value
 
         for columnInfo, dps in columnValues.items():
             ts = TimeSeries(columnInfo, dps, self.filtersMap, self.labels)

source/opentsdb.py

@@ -27,7 +27,9 @@
 from collections import defaultdict
 from collector import SensorCollector, QueryPolicy
 from utils import getTimeMultiplier, execution_time, cond_execution_time
-from typing import List
+from typing import List, TypeVar
+
+T = TypeVar('T', dict, list)
 
 
 class OpenTsdbApi(object):
@@ -158,6 +160,9 @@ def build_collector(self, jreq: dict) -> SensorCollector:
                 args['filters'] = filters
                 args['grouptags'] = grouptags
 
+        if 'arrays' in jreq:
+            args['dpsArrays'] = jreq['arrays']
+
         args['rawData'] = q.get('explicitTags', False)
 
         args['sensor'] = sensor
@@ -379,7 +384,7 @@ def GET(self, **params):
     @cherrypy.config(**{'tools.json_in.force': False})
     @cherrypy.tools.json_in()  # @UndefinedVariable
     @cherrypy.tools.json_out()  # @UndefinedVariable
-    def POST(self):
+    def POST(self, **params):
         ''' Process POST. tools.json_in.force is set to False for
         compatability between versions of grafana < 3 and version 3.'''
 
@@ -399,6 +404,9 @@ def POST(self):
                 self.logger.error(MSG['QueryError'].format('empty'))
                 raise cherrypy.HTTPError(400, ERR[400])
 
+            if params and params.get('arrays') == 'true':
+                jreq['arrays'] = True
+
             return self.query(jreq)
 
     def OPTIONS(self):
@@ -446,13 +454,15 @@ def __init__(self, inputQuery, showQuery=False,
         self.tags = tags or defaultdict(list)
         self.aggregatedTags = aggrTags or []
 
-    def to_dict(self, dps: dict = None):
+    def to_dict(self, dps: T = None):
         ''' Converts the SingleTimeSeriesResponse object to dict. '''
         res = self.__dict__
         # Since a single Timeseries might have a huge number of datapoints (dps),
         # first convert object to dict and then fetch the dict of dps to it
         if dps:
             res['dps'] = dps
+        elif isinstance(dps, list):
+            res['dps'] = []
         return res
 
 

tests/test_opentsdb.py

@@ -76,6 +76,34 @@ def query_last_setup():
                                             'index': 0}}
 
 
+def query_arrays_setup():
+    global key1, col1, labels, filtersMap, dps1, ts1, ts2, metricTS, metricTS1, data, data1, jreq
+
+    key1 = Key._from_string('scale-11|CPU|cpu_system', '')
+    col1 = ColumnInfo(name='cpu_user', semType=1, keys=(key1,), column=0)
+    filtersMap = [{'node': 'scale-11'}, {'node': 'scale-12'}, {'node': 'scale-13'}, {'node': 'scale-14'}, {'node': 'scale-15'}, {'node': 'scale-16'}]
+    labels = ['node']
+    dps1 = [[1739214990, 3], [1739215050, 2], [1739215110, 3], [1739215170, 4], [1739215230, 3]]
+    dps2 = []
+    ts1 = TimeSeries(col1, dps1, filtersMap, labels)
+    ts2 = TimeSeries(col1, dps2, filtersMap, labels)
+    metricTS = MetricTimeSeries('cpu_system', '')
+    metricTS.timeseries = [ts1]
+    data = {'cpu_user': metricTS}
+    metricTS1 = MetricTimeSeries('cpu_system', '')
+    metricTS1.timeseries = [ts2]
+    data1 = {'cpu_user': metricTS1}
+    jreq = {'start': 1739214930519, 'end': 1739215230519, 'arrays': True,
+            'inputQuery': {'aggregator': 'noop', 'downsample': '1m-avg',
+                           'filters': [
+                               {'filter': 'scale-11', 'groupBy': False,
+                                'tagk': 'node', 'type': 'pm_filter'
+                                }],
+                           'metric': 'cpu_system', 'index': 0
+                           }
+            }
+
+
 @with_setup(my_setup)
 def test_case01():
     ts = TimeSeries(col3, dps2, filtersMap, labels)
@@ -110,6 +138,7 @@ def test_case03():
         assert 'gpfs_fs_name' in resp[0].get('tags')
         assert 'node' in resp[0].get('tags')
         assert 'gpfs_cluster_name' in resp[0].get('tags')
+        assert isinstance(resp[0].get('dps'), dict)
 
 
 @with_setup(my_setup)
@@ -139,3 +168,30 @@ def test_case05():
         assert resp[0].get('metric') == "cpu_user"
         assert 'gpfs_fs_name' not in resp[0].get('tags')
         assert 'node' in resp[0].get('tags')
+
+
+@with_setup(query_arrays_setup)
+def test_case06():
+    with mock.patch('source.metadata.MetadataHandler') as md:
+        md_instance = md.return_value
+        logger = logging.getLogger(__name__)
+        opentsdb = OpenTsdbApi(logger, md_instance, '9999')
+        resp = opentsdb.format_response(data, jreq)
+        assert set(resp[0].keys()) == set(['metric', 'dps', 'tags', 'aggregatedTags'])
+        assert resp[0].get('metric') == "cpu_system"
+        assert 'node' in resp[0].get('tags')
+        assert isinstance(resp[0].get('dps'), list)
+
+
+@with_setup(query_arrays_setup)
+def test_case07():
+    with mock.patch('source.metadata.MetadataHandler') as md:
+        md_instance = md.return_value
+        logger = logging.getLogger(__name__)
+        opentsdb = OpenTsdbApi(logger, md_instance, '9999')
+        resp = opentsdb.format_response(data1, jreq)
+        assert set(resp[0].keys()) == set(['metric', 'dps', 'tags', 'aggregatedTags'])
+        assert resp[0].get('metric') == "cpu_system"
+        assert 'node' in resp[0].get('tags')
+        assert isinstance(resp[0].get('dps'), list)
+        assert len(resp[0].get('dps')) == 0