Merge pull request #246 from Helene/backport_important_fixes_to_v7

Backport important fixes to v7.2

Author: Helene

source/metadata.py

@@ -26,7 +26,7 @@
 from queryHandler.Topo import Topo
 from queryHandler import SensorConfig
 from utils import execution_time
-from messages import MSG
+from messages import ERR, MSG
 from metaclasses import Singleton
 from time import sleep
 
@@ -129,14 +129,15 @@ def __initializeTables(self):
             raise ValueError(MSG['NoSensorConfigData'])
         MAX_ATTEMPTS_COUNT = 3
         for attempt in range(1, MAX_ATTEMPTS_COUNT + 1):
-            self.__metaData = Topo(self.qh.getTopology())
-            if not (self.metaData and self.metaData.topo):
+            topoStr = self.qh.getTopology()
+            if not topoStr:
                 if attempt > MAX_ATTEMPTS_COUNT:
                     break
                 # if no data returned because of the REST HTTP server is still starting, sleep and retry (max 3 times)
                 self.logger.warning(MSG['NoDataStartNextAttempt'].format(attempt, MAX_ATTEMPTS_COUNT))
                 sleep(self.sleepTime)
             else:
+                self.__metaData = Topo(topoStr)
                 foundItems = len(self.metaData.allParents) - 1
                 sensors = self.metaData.sensorsSpec.keys()
                 self.logger.info(MSG['MetaSuccess'])
@@ -155,10 +156,11 @@ def update(self, refresh_all=False):
         if refresh_all:
             self.__sensorsConf = SensorConfig.readSensorsConfigFromMMSDRFS(self.logger)
 
-        self.__metaData = Topo(self.qh.getTopology())
-        if not (self.metaData and self.metaData.topo):
+        topoStr = self.qh.getTopology()
+        if not topoStr:
             self.logger.error(MSG['NoData'])  # Please check the pmcollector is properly configured and running.
-            raise cherrypy.HTTPError(404, MSG[404])
+            raise cherrypy.HTTPError(404, ERR[404])
+        self.__metaData = Topo(topoStr)
         self.logger.details(MSG['MetaSuccess'])
         self.logger.debug(MSG['ReceivAttrValues'].format('parents', ", ".join(self.metaData.allParents)))
         return ({'msg': MSG['MetaSuccess']})

source/opentsdb.py

@@ -352,10 +352,8 @@ def parse_result_tags(self, identifiersMap):
         if identifiersMap:
             for identifiers in identifiersMap:
                 d = defaultdict(dict)
-                for key in identifiers.keys():
-                    d['tags'][key] = identifiers[key]
-                    if d not in self.results:
-                        self.results.append(d)
+                d['tags'] = identifiers
+                self.results.append(d)
 
 
 class SingleTimeSeriesResponse():

source/queryHandler/PerfmonRESTclient.py

@@ -20,6 +20,7 @@
 @author: HWASSMAN
 '''
 
+import urllib.request
 # catch import failure on AIX since we will not be shipping our third-party libraries on AIX
 try:
     import requests
@@ -32,6 +33,22 @@
 DEFAULT_HEADERS = {"Accept": "application/json",
                    "Content-type": "application/json"}
 
+session = None
+
+
+def get_session():
+    global session
+    if not session:
+        session = requests.Session()
+    return session
+
+
+def close_session():
+    global session
+    if session:
+        session.close()
+        session = None
+
 
 def getAuthHandler(keyName, keyValue):
     if not isinstance(keyName, bytes):
@@ -73,27 +90,39 @@ class perfHTTPrequestHelper(object):
     2. waits until  the server response, finally forwards a result to the QueryHandler
     """
 
-    def __init__(self, logger, reqdata=None, session=None):
-        self.session = session or requests.Session()
+    def __init__(self, logger, reqdata=None, caCert=False):
+        self.session = get_session()
+        self.caCert = caCert
         self.requestData = reqdata
         self.logger = logger
 
     def doRequest(self):
         if self.requestData and isinstance(self.requestData, requests.Request):
+            self.session.verify = self.caCert
             _prepRequest = self.session.prepare_request(self.requestData)
             try:
                 res = self.session.send(_prepRequest)
                 return res
+            except requests.exceptions.ProxyError:
+                close_session()
+                self.logger.debug(f"doRequest __ ProxyError. Found configured proxies: {urllib.request.getproxies()}")
+                res = requests.Response()
+                res.status_code = 503
+                res.reason = "Unable to connect to proxy"
+                return res
             except requests.exceptions.ConnectionError:
+                close_session()
                 res = requests.Response()
                 res.status_code = 503
                 res.reason = "Connection refused from server"
                 return res
             except requests.exceptions.RequestException as e:
+                close_session()
                 self.logger.debug('doRequest __ RequestException. Request data: {}, Response data: {}'.format(e.request, e.response))
                 res = requests.Response()
                 res.status_code = 404
                 res.reason = "The request could not be processed from server"
                 return res
         else:
+            close_session()
             raise TypeError('doRequest __ Error: request data wrong format')

source/queryHandler/Query.py

@@ -47,7 +47,8 @@ class Query(object):
                   "gpfs_disk_usage_name", "gpfs_fset_name", "gpfs_fs_name",
                   "mountPoint", "netdev_name", 'diskdev_name', "node", "db_name",
                   "operation", "protocol", "waiters_time_threshold", "export",
-                  "nodegroup", "account", "filesystem", "tct_csap", "tct_operation", "cloud_nodeclass"])
+                  "nodegroup", "account", "filesystem", "tct_csap", "tct_operation", "cloud_nodeclass",
+                  "gpfs_health_component", "gpfs_health_entity"])
 
     DISK_CAP_METRICS = set(["gpfs_disk_disksize", "gpfs_disk_free_fullkb", "gpfs_disk_free_fragkb",
                             "gpfs_pool_disksize", "gpfs_pool_free_fragkb", "gpfs_pool_free_fullkb",
@@ -91,6 +92,7 @@ def __init__(self, metrics=None, bucketsize=1, filters=None, groupby=None, inclu
         self.timeRep = ' now'         # string time representation
         self.measurements = {}
         self.normalize_rates = True
+        self.rawData = False
         self.key = None
         self.sensor = None
 
@@ -188,26 +190,34 @@ def addMeasurement(self, meassure):
     def __str__(self):
         # dd = '-a' if self.includeDiskData else ''
         # Workaround for RTC Defect 280368: Zimon capacity query does not return all results (seen on CNSA)
-        if (self.metrics
-            and any(str(metric) in self.DISK_CAP_METRICS for metric in self.metrics)
-            ) or (self.sensor
-                  and self.sensor in ("GPFSDiskCap", "GPFSPoolCap", "GPFSInodeCap")
-                  ):
-            dd = '-ar'
-        elif self.includeDiskData:
+        dd = ''
+        if self.includeDiskData:
             dd = '-a'
+        if self.metrics:
+            for metric in self.metrics:
+                if any(map(metric.__contains__, self.DISK_CAP_METRICS)):
+                    dd = '-ar'
+                    break
+        elif (self.sensor and self.sensor in ("GPFSDiskCap",
+                                              "GPFSPoolCap",
+                                              "GPFSInodeCap")
+              ):
+            dd = '-ar'
+
+        if self.rawData:
+            raw = '-z'
         else:
-            dd = ''
+            raw = ''
 
         if self.sensor is not None:
-            queryString = 'get -j {0} group {1} bucket_size {2} {3}'.format(
-                dd, self.sensor, self.bucket_size, self.timeRep)
+            queryString = 'get -j {0} {1} group {2} bucket_size {3} {4}'.format(
+                dd, raw, self.sensor, self.bucket_size, self.timeRep)
         elif self.key is not None:
-            queryString = 'get -j {0} {1} bucket_size {2} {3}'.format(
-                dd, self.key, self.bucket_size, self.timeRep)
+            queryString = 'get -j {0} {1} {2} bucket_size {3} {4}'.format(
+                dd, raw, self.key, self.bucket_size, self.timeRep)
         else:
-            queryString = 'get -j {0} metrics {1} bucket_size {2} {3}'.format(
-                dd, ','.join(self.metrics), self.bucket_size, self.timeRep)
+            queryString = 'get -j {0} {1} metrics {2} bucket_size {3} {4}'.format(
+                dd, raw, ','.join(self.metrics), self.bucket_size, self.timeRep)
 
         if self.filters:
             queryString += ' from ' + ",".join(self.filters)

source/queryHandler/QueryHandler.py

@@ -164,9 +164,10 @@ def __init__(self, query, res_json):
         self.rows = self.__parseRows()
 
         self.index_cache = {}  # (metric, id) -> row value index
-        self.ids = self._findIdentifiers()
+        self.ids = None
 
         if self.query and len(self.query.measurements) > 0:
+            self.ids = self._findIdentifiers()
             self._populate_index_cache()
             self._add_calculated_colunm_headers()
 
@@ -513,9 +514,12 @@ def __do_RESTCall(self, endpoint, requestType='GET', params=None):
 
         try:
             _auth = getAuthHandler(*self.apiKeyData)
-            _reqData = createRequestDataObj(self.logger, requestType, endpoint, self.server, self.port, auth=_auth, params=params)
-            _request = perfHTTPrequestHelper(self.logger, reqdata=_reqData)
-            _request.session.verify = self.caCert
+            _reqData = createRequestDataObj(self.logger, requestType, endpoint,
+                                            self.server, self.port, auth=_auth,
+                                            params=params)
+            _request = perfHTTPrequestHelper(self.logger,
+                                             reqdata=_reqData,
+                                             caCert=self.caCert)
             _response = _request.doRequest()
 
             if _response.status_code == 200:

source/queryHandler/Topo.py

@@ -20,8 +20,8 @@
 @author: HWASSMAN
 '''
 
-
-from collections import defaultdict
+import copy
+from collections import defaultdict, OrderedDict
 from typing import Set, List, Dict, DefaultDict
 
 # dict.iteritems() deprecated in python 3
@@ -34,14 +34,14 @@ class Topo(object):
     '''
 
     def __init__(self, jsonStr=None):
-        self.topo = jsonStr
         self.__metricsDef = defaultdict(dict)   # metrics dictionary, per sensor for all elements in the metadata
+        self.__metricsType = defaultdict(dict)   # metrics types dictionary
         self.__levels = defaultdict(dict)       # component level priority dictionary, per sensor
         self.__ids = {}                                # fieldIds dictionary
         self.__groupKeys = {}
         self.__compTree = {}
-        if self.topo:
-            self._processMetadata(self.topo)
+        if jsonStr:
+            self._processMetadata(jsonStr)
 
     def _processMetadata(self, metadata):
         '''
@@ -55,22 +55,24 @@ def _processMetadata(self, metadata):
             _components = defaultdict(set)
             # filters dictionary, per sensor, per (cluster or cluster_node) component
             _filters = defaultdict(list)
+            # ordered dictionary to store filedLabel:fieldName pairs in order as they occur in metaStr
+            _tags = OrderedDict()
             # name of the  (cluster or cluster_node) component
             label = metaStr.get('fieldLabel')
 
             if label in self.__compTree.keys():
                 _components = self.__compTree[label]['componentsMap']
                 _filters = self.__compTree[label]['filtersMap']
 
-            self._parse_topoJSONStr(self.__metricsDef, self.__levels, self.__ids, self.__groupKeys, _components, _filters, metaStr)
+            self._parse_topoJSONStr(self.__metricsDef, self.__metricsType, self.__levels, self.__ids, self.__groupKeys, _components, _filters, _tags, metaStr)
             tree_entry = {}
             tree_entry['componentsMap'] = _components
             tree_entry['filtersMap'] = _filters
 
             # comp_tree[label] = tree_entry
             self.__compTree[label] = tree_entry
 
-    def _parse_topoJSONStr(self, metrics, levels, ids, groupKeys, components, filters, metaStr):
+    def _parse_topoJSONStr(self, metrics, metricsType, levels, ids, groupKeys, components, filters, _tags, metaStr):
         '''
         This function parses the 'node' or 'attribute' object found in the given JSON string (metaStr) in
         the componets or metrics dictionary. Also the used metric filters (per sensor) will be stored
@@ -80,15 +82,24 @@ def _parse_topoJSONStr(self, metrics, levels, ids, groupKeys, components, filter
 
         field_value = metaStr['fieldLabel']
         field_name = metaStr['fieldName']
+        field_type = metaStr['fieldSemantics']
 
         # check if entity is a component
         if metaStr['type'] == 'node':
+            if field_name == "sensor":
+                # remove all partialKey tags from _tags dict except parent since we step in the new sensor level metaKey
+                parent = _tags.popitem(last=False)
+                _tags.clear()
+                _tags.update([parent])
+            else:
+                _tags[field_name] = field_value
+
             if field_value not in components[field_name]:
                 components[field_name].add(field_value)
             # check if metaStr includes next level metaStr
             if 'keys' in metaStr and len(metaStr['keys']) > 0:
                 for metaKey in metaStr['keys']:
-                    self._parse_topoJSONStr(metrics, levels, ids, groupKeys, components, filters, metaKey)
+                    self._parse_topoJSONStr(metrics, metricsType, levels, ids, groupKeys, components, filters, _tags, metaKey)
 
         # check if entity is a metric
         elif metaStr['type'] == 'attribute':
@@ -100,21 +111,22 @@ def _parse_topoJSONStr(self, metrics, levels, ids, groupKeys, components, filter
 
             if field_name not in iterval(metrics[sensor]):
                 metrics[sensor][field_id] = field_name
+                metricsType[field_name] = field_type
 
             if groupKey not in groupKeys:
                 # parse sensor relevant data f.e. groupKey, filters, levels
                 groupKeys[groupKey] = len(groupKeys) + 1
-                tags = {}
-                levTree = {}
-                for i, compValue in enumerate(partKey):
-                    for compLabel in components:
-                        if compValue in components[compLabel]:
-                            levTree[i + 1] = compLabel
-                            tags[compLabel] = compValue
-                # if tags not in filters[sensor]:
-                # not needed as groupkeys check will allow this to be reached only once
-                filters[sensor].append(tags)
+                group_tags = copy.deepcopy(_tags)
+
+                # if not all((value in tags.values()) for value in partKey):
+                #     rint("different key values")
+
+                filters[sensor].append(group_tags)
                 if sensor not in levels:
+                    levTree = {}
+                    for i, tag_key in enumerate(group_tags.keys()):
+                        levTree[i + 1] = tag_key
+
                     levels[sensor] = levTree
 
             # parse key id
@@ -177,6 +189,11 @@ def metricsSpec(self):
         ''' Returns all defined metrics as dictionary of (metric_name : metric_id) items '''
         return self.__metricsDef
 
+    @property
+    def metricsType(self):
+        ''' Returns a dictionary of (metric_name : metric_type) items '''
+        return self.__metricsType
+
     @property
     def getAllEnabledMetricsNames(self):
         ''' Returns list of all found metrics names'''
@@ -201,7 +218,7 @@ def getSensorForMetric(self, searchMetric):
         if (searchMetric.find("(") >= 0):
             searchMetric = searchMetric[searchMetric.find("(") + 1:-1]
         for sensor, metrics in self.__metricsDef.items():
-            if searchMetric in metrics.values():
+            if searchMetric in set(metrics.values()):
                 return sensor
         return None
 
@@ -237,8 +254,11 @@ def getAllFilterMapsForSensor(self, searchSensor):
         based on metadata topology returned from zimon "topo".
         '''
         filtersMaps = []
-        if searchSensor in self.allFiltersMaps.keys():
-            filtersMaps.extend(self.allFiltersMaps[searchSensor])
+        if searchSensor in set(self.sensorsSpec.keys()):
+            for entryName in self.__compTree.keys():
+                values = self.__compTree[entryName]['filtersMap'].get(searchSensor, [])
+                if len(values) > 0:
+                    filtersMaps.extend(values)
         return filtersMaps
 
     def getAllFilterMapsForMetric(self, searchMetric):
@@ -274,13 +294,11 @@ def getAllFilterKeysForMetric(self, searchMetric):
         return keys
 
     def getAllFilterKeysForSensor(self, searchSensor):
-        keys = []
+        filter_keys = set()
         filtersMap = self.getAllFilterMapsForSensor(searchSensor)
-        for a in filtersMap:
-            keys.extend(list(a.keys()))
-        if len(keys) > 1:
-            return list(set(keys))
-        return keys
+        for filter in filtersMap:
+            filter_keys.update(filter.keys())
+        return list(filter_keys)
 
     def getAllFilterKeysForMeasurementsMetrics(self, searchMetrics):
         filterKeys = []