improve openTSDB format_response

Signed-off-by: hwassman <[email protected]>

Author: Helene

source/collector.py

@@ -53,12 +53,12 @@ def parse_tags(self, filtersMap):
             ident = [key.parent]
             ident.extend(key.identifier)
             logger.trace(MSG['ReceivAttrValues'].format(
-                'Single ts identifiers', ', '.join(ident)))
+                'Single ts identifiers', '|'.join(ident)))
             found = False
             for filtersDict in filtersMap:
                 if set(filtersDict.values()) == set(ident):
-                    logger.trace(MSG['ReceivAttrValues'].format(
-                        'filtersKeys', ', '.join(filtersDict.keys())))
+                    # logger.trace(MSG['ReceivAttrValues'].format(
+                    #    'filtersKeys', ', '.join(filtersDict.keys())))
                     if len(self.columnInfo.keys) == 1:
                         self.tags = filtersDict
                     else:
@@ -72,12 +72,11 @@ def parse_tags(self, filtersMap):
                 local_cache.union(ident)
                 updated_size = len(local_cache)
                 if updated_size > cache_size:
-                    logger.trace(MSG['NewKeyDetected'].format(ident))
-                    local_cache.append(ident)
+                    logger.trace(MSG['NewKeyDetected'].format('|'.join(ident)))
                     md = MetadataHandler()
                     Thread(name='AdHocMetaDataUpdate', target=md.update).start()
                 else:
-                    logger.trace(MSG['NewKeyAlreadyReported'].format(ident))
+                    logger.trace(MSG['NewKeyAlreadyReported'].format('|'.join(ident)))
 
         for _key, _values in tagsDict.items():
             if len(_values) > 1:
@@ -426,7 +425,6 @@ def validate_query_filters(self):
                     400, MSG['AttrNotValid'].format('filter'))
 
     def validate_group_tags(self):
-
         # check groupBy settings
         if self.request.grouptags:
             filter_keys = set()

source/opentsdb.py

@@ -51,15 +51,16 @@ def TOPO(self):
 
     def format_response(self, data: dict, jreq: dict) -> List[dict]:
         respList = []
-        for name, metric in data.items():
+        metrics = set(data.values())
+        for metric in metrics:
             for st in metric.timeseries:
-                res = SingleTimeSeriesResponse(jreq.get('inputQuery'), st.dps,
+                res = SingleTimeSeriesResponse(jreq.get('inputQuery'),
                                                jreq.get('showQuery'),
                                                jreq.get('globalAnnotations'),
                                                st.tags, st.aggregatedTags)
-                cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
-                self.logger.trace(f'OpenTSDB queryResponse: {str(res.__dict__)}')
-                respList.append(res.__dict__)
+                # self.logger.trace(f'OpenTSDB queryResponse for :
+                #                   {data.keys()[0]} with {len(st.dps)} datapoints')
+                respList.append(res.to_dict(st.dps))
         return respList
 
     def query(self, jreq: dict) -> List[dict]:
@@ -101,9 +102,9 @@ def query(self, jreq: dict) -> List[dict]:
             self.logger.trace('Finished custom thread %r.' % collector.thread.name)
 
             coll_resp = self.format_response(collector.metrics, request_data)
-            cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
+            # cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
             resp.extend(coll_resp)
-
+        cherrypy.response.headers['Access-Control-Allow-Origin'] = '*'
         return resp
 
     def build_collector(self, jreq: dict) -> SensorCollector:
@@ -371,16 +372,26 @@ def parse_result_tags(self, identifiersMap):
                         self.results.append(d)
 
 
-class SingleTimeSeriesResponse():
+class SingleTimeSeriesResponse(object):
 
-    def __init__(self, inputQuery, dps, showQuery=False,
+    def __init__(self, inputQuery, showQuery=False,
                  globalAnnotations=False, tags: dict = None,
                  aggrTags: list = None):
         self.metric = inputQuery.get('metric')
-        self.dps = dps
+        self.dps = defaultdict()
         if showQuery:
             self.query = inputQuery
         if globalAnnotations:
             self.globalAnnotations = []
         self.tags = tags or defaultdict(list)
         self.aggregatedTags = aggrTags or []
+
+    def to_dict(self, dps: dict = 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
+        return res
+