Repository: dmwm/wmcore
Issue: ReqMon DataCacheUpdate thread timing out

**Impact of the bug**
ReqMon (aka WMStats)

**Describe the bug**
Likely related to the very high load in the system, as explained here: https://github.com/dmwm/WMCore/issues/9668

Over the last days, WMStats became very sluggish and poking around the logs, I see many timeouts in the `DataCacheUpdate` reqmon cherrypy thread. It seems to have become much worse since yesterday.

From the logs, there are cases that it has to transfer ~200MB worth of data. Given that this is a general purpose, one can multiply that by 5 (1GB of data every 10 to 15min...). It also explains why wmstats sometimes loads and others it remains spinning forever.

**How to reproduce it**
Have 40k workflows in the system with many job failures.

**Expected behavior**
The obvious ones. Cache should always be up-to-date, thus avoiding these requests time out.
These caches were supposed to be "small" as well (<50MB), having such large caches continuously updated in many backends isn't health for the cluster.
We could also calculate the amount of data in the cache and log it (pushing such information to MonIT would be a plus as well).
Anything else that can still be discussed to make this service more robust and resilient to such a high load.
 
**Additional context and error message**
Logs from vocms0766 for today:
```
2020-05-07 02:54:41,440:INFO:DataCacheUpdate:DataCache is updated: 42096
2020-05-07 03:12:55,214:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270001 milliseconds with 115605922 bytes received')
2020-05-07 03:29:13,663:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270001 milliseconds with 53544228 bytes received')
2020-05-07 03:45:05,224:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 68165469 bytes received')
2020-05-07 04:01:29,324:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 225706992 bytes received')
2020-05-07 04:18:01,356:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270008 milliseconds with 121591632 bytes received')
2020-05-07 04:34:01,814:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 23328350 bytes received')
2020-05-07 04:51:40,050:ERROR:DataCacheUpdate:(28, 'Operation timed out after 271232 milliseconds with 22150520 bytes received')
2020-05-07 05:08:50,854:ERROR:DataCacheUpdate:(28, 'Operation timed out after 280071 milliseconds with 20950855 bytes received')
2020-05-07 05:25:37,535:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 54860085 bytes received')
2020-05-07 05:42:15,440:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 62670518 bytes received')
2020-05-07 05:58:33,363:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 122496096 bytes received')
2020-05-07 06:15:27,999:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 63965610 bytes received')
2020-05-07 06:31:27,804:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 213083083 bytes received')
2020-05-07 06:48:05,471:ERROR:DataCacheUpdate:(28, 'Operation timed out after 276537 milliseconds with 215823928 bytes received')
2020-05-07 07:04:44,183:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 25620037 bytes received')
2020-05-07 07:20:36,991:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270000 milliseconds with 99517239 bytes received')
2020-05-07 07:37:25,040:ERROR:DataCacheUpdate:(28, 'Operation timed out after 270001 milliseconds with 113712109 bytes received')
2020-05-07 07:53:46,004:INFO:DataCacheUpdate:DataCache is updated: 42098
```


Changed files:
src/python/WMCore/Services/WMStats/WMStatsReader.py, src/python/WMCore/ReqMgr/DataStructs/RequestStatus.py, src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py

diff --git a/src/python/WMCore/ReqMgr/DataStructs/RequestStatus.py b/src/python/WMCore/ReqMgr/DataStructs/RequestStatus.py
index 02b61a632..f33811999 100644
--- a/src/python/WMCore/ReqMgr/DataStructs/RequestStatus.py
+++ b/src/python/WMCore/ReqMgr/DataStructs/RequestStatus.py
@@ -83,6 +83,25 @@ ACTIVE_STATUS = ["new",
                  "aborted-completed",
                  "rejected"]
 
+### WMSTATS_JOB_INFO + WMSTATS_NO_JOB_INFO is meant to be equal to ACTIVE_STATUS
+WMSTATS_JOB_INFO = ["running-open",
+                    "running-closed",
+                    "force-complete",
+                    "completed",
+                    "closed-out"]
+
+WMSTATS_NO_JOB_INFO = ["new",
+                       "assignment-approved",
+                       "assigned",
+                       "staging",
+                       "staged",
+                       "acquired",
+                       "failed",
+                       "announced",
+                       "aborted",
+                       "aborted-completed",
+                       "rejected"]
+
 ### Used for monitoring in T0-WMStats. See: Services/WMStats/WMStatsReader
 T0_ACTIVE_STATUS = ["new",
                     "Closed",
diff --git a/src/python/WMCore/Services/WMStats/WMStatsReader.py b/src/python/WMCore/Services/WMStats/WMStatsReader.py
index 2f5821017..301173022 100644
--- a/src/python/WMCore/Services/WMStats/WMStatsReader.py
+++ b/src/python/WMCore/Services/WMStats/WMStatsReader.py
@@ -1,11 +1,12 @@
 from __future__ import division, print_function
 
-from Utils.IteratorTools import nestedDictUpdate
+import logging
+from Utils.IteratorTools import nestedDictUpdate, grouper
 from WMCore.Database.CMSCouch import CouchServer
 from WMCore.Lexicon import splitCouchServiceURL, sanitizeURL
 from WMCore.Services.RequestDB.RequestDBReader import RequestDBReader
 from WMCore.Services.WMStats.DataStruct.RequestInfoCollection import RequestInfo
-from WMCore.ReqMgr.DataStructs.RequestStatus import T0_ACTIVE_STATUS, ACTIVE_STATUS
+from WMCore.ReqMgr.DataStructs.RequestStatus import T0_ACTIVE_STATUS, WMSTATS_JOB_INFO, WMSTATS_NO_JOB_INFO
 
 REQUEST_PROPERTY_MAP = {
     "_id": "_id",
@@ -56,7 +57,8 @@ def convertToLegacyFormat(requestDoc):
 
 class WMStatsReader(object):
 
-    def __init__(self, couchURL, appName="WMStats", reqdbURL=None, reqdbCouchApp="ReqMgr"):
+    def __init__(self, couchURL, appName="WMStats", reqdbURL=None,
+                 reqdbCouchApp="ReqMgr", logger=None):
         self._sanitizeURL(couchURL)
         # set the connection for local couchDB call
         self._commonInit(couchURL, appName)
@@ -64,6 +66,7 @@ class WMStatsReader(object):
             self.reqDB = RequestDBReader(reqdbURL, reqdbCouchApp)
         else:
             self.reqDB = None
+        self.logger = logger if logger else logging.getLogger()
 
     def _sanitizeURL(self, couchURL):
         return sanitizeURL(couchURL)['url']
@@ -96,7 +99,7 @@ class WMStatsReader(object):
         return jobInfoByRequestAndAgent
 
     def _updateRequestInfoWithJobInfo(self, requestInfo):
-        if len(requestInfo.keys()) != 0:
+        if requestInfo:
             jobInfoByRequestAndAgent = self.getLatestJobInfoByRequests(requestInfo.keys())
             self._combineRequestAndJobData(requestInfo, jobInfoByRequestAndAgent)
 
@@ -192,15 +195,33 @@ class WMStatsReader(object):
 
     def _getLatestJobInfo(self, keys):
         """
-        keys is [['request_name', 'agent_url'], ....]
-        returns ids
+        Given a list of lists as keys, in the format of:
+            [['request_name', 'agent_url'], ['request_name2', 'agent_url2'], ....]
+        The result format from the latestRequest view is:
+            {u'offset': 527,
+             u'rows': [{u'doc': {u'_rev': u'32-6027014210',
+             ...
+                        u'id': u'cmsgwms-submit6.fnal.gov-cmsunified_ACDC0_task_BTV-RunIISummer19UL18wmLHEGEN-00004__v1_T_200507_162125_3670',
+                        u'key': [u'cmsunified_ACDC0_task_BTV-RunIISummer19UL18wmLHEGEN-00004__v1_T_200507_162125_3670',
+                                 u'cmsgwms-submit6.fnal.gov'],
+                        u'value': None}],
+             u'total_rows': 49606}
         """
-        if len(keys) == 0:
+        if not keys:
             return []
-        options = {"include_docs": True}
+        options = {}
+        options["include_docs"] = True
         options["reduce"] = False
-        result = self._getCouchView("latestRequest", options, keys)
-        return result
+        finalResults = {}
+        # magic number: 5000 keys (need to check which number is optimal)
+        for sliceKeys in grouper(keys, 5000):
+            self.logger.info("Querying latestRequest with %d keys", len(sliceKeys))
+            result = self._getCouchView("latestRequest", options, sliceKeys)
+            if not finalResults and result:
+                finalResults = result
+            elif result.get('rows'):
+                finalResults['rows'].extend(result['rows'])
+        return finalResults
 
     def _getAllDocsByIDs(self, ids, include_docs=True):
         """
@@ -273,9 +294,8 @@ class WMStatsReader(object):
             self._updateRequestInfoWithJobInfo(requestInfo)
         return requestInfo
 
-    def getActiveData(self, jobInfoFlag=False):
-
-        return self.getRequestByStatus(ACTIVE_STATUS, jobInfoFlag)
+    def getActiveData(self, listStatuses, jobInfoFlag=False):
+        return self.getRequestByStatus(listStatuses, jobInfoFlag)
 
     def getT0ActiveData(self, jobInfoFlag=False):
 
@@ -290,18 +310,24 @@ class WMStatsReader(object):
         If legacyFormat is True convert data to old wmstats format from current reqmgr format.
         Shouldn't be set to True unless existing code breaks
         """
-
-        requestInfo = self.reqDB.getRequestByStatus(statusList, True, limit, skip)
-
-        if legacyFormat:
-            # convert the format to wmstas old format
-            for requestName, doc in requestInfo.items():
-                requestInfo[requestName] = convertToLegacyFormat(doc)
-
-        if jobInfoFlag:
-            # get request and agent info
-            self._updateRequestInfoWithJobInfo(requestInfo)
-        return requestInfo
+        results = dict()
+        for status in statusList:
+            self.logger.info("Fetching workflows by status from ReqMgr2, status: %s", status)
+            requestInfo = self.reqDB.getRequestByStatus(status, True, limit, skip)
+            self.logger.info("Found %d workflows in status: %s", len(requestInfo), status)
+
+            if legacyFormat:
+                # convert the format to wmstats old format
+                for requestName, doc in requestInfo.items():
+                    requestInfo[requestName] = convertToLegacyFormat(doc)
+            results.update(requestInfo)
+
+        # now update these requests with agent information too
+        if results and jobInfoFlag:
+            self.logger.info("Now updating these requests with job info...")
+            self._updateRequestInfoWithJobInfo(results)
+
+        return results
 
     def getRequestSummaryWithJobInfo(self, requestName):
         """
diff --git a/src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py b/src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py
index d1f2aa273..1cdff6d47 100644
--- a/src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py
+++ b/src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py
@@ -1,15 +1,15 @@
-'''
-
-'''
 from __future__ import (division, print_function)
 
+import time
 from WMCore.REST.CherryPyPeriodicTask import CherryPyPeriodicTask
 from WMCore.WMStats.DataStructs.DataCache import DataCache
 from WMCore.Services.WMStats.WMStatsReader import WMStatsReader
+from WMCore.ReqMgr.DataStructs.RequestStatus import WMSTATS_JOB_INFO, WMSTATS_NO_JOB_INFO
 
 class DataCacheUpdate(CherryPyPeriodicTask):
 
     def __init__(self, rest, config):
+        self.getJobInfo = getattr(config, "getJobInfo", False)
 
         super(DataCacheUpdate, self).__init__(config)
 
@@ -23,13 +23,21 @@ class DataCacheUpdate(CherryPyPeriodicTask):
         """
         gather active data statistics
         """
+        self.logger.info("Starting gatherActiveDataStats with jobInfo set to: %s", self.getJobInfo)
         try:
+            tStart = time.time()
             if DataCache.islatestJobDataExpired():
                 wmstatsDB = WMStatsReader(config.wmstats_url, reqdbURL=config.reqmgrdb_url,
-                                          reqdbCouchApp="ReqMgr")
-                jobData = wmstatsDB.getActiveData(jobInfoFlag = True)
+                                          reqdbCouchApp="ReqMgr", logger=self.logger)
+                self.logger.info("Getting active data with job info for statuses: %s", WMSTATS_JOB_INFO)
+                jobData = wmstatsDB.getActiveData(WMSTATS_JOB_INFO, jobInfoFlag=self.getJobInfo)
+                self.logger.info("Getting active data with NO job info for statuses: %s", WMSTATS_NO_JOB_INFO)
+                tempData = wmstatsDB.getActiveData(WMSTATS_NO_JOB_INFO, jobInfoFlag=False)
+                jobData.update(tempData)
+                self.logger.info("Running setlatestJobData...")
                 DataCache.setlatestJobData(jobData)
-                self.logger.info("DataCache is updated: %s", len(jobData))
+                self.logger.info("DataCache is up-to-date with %d requests data", len(jobData))
         except Exception as ex:
-            self.logger.error(str(ex))
+            self.logger.exception("Exception updating DataCache. Error: %s", str(ex))
+        self.logger.info("Total time loading data from ReqMgr2 and WMStats: %s", time.time() - tStart)
         return