Memory trace for DataCacheUpdate CP thread

src/python/WMCore/REST/Format.py

@@ -2,6 +2,7 @@
 
 import gzip
 from builtins import str, bytes, object
+from memory_profiler import profile
 
 from Utils.PythonVersion import PY3
 from Utils.Utilities import encodeUnicodeToBytes, encodeUnicodeToBytesConditional
@@ -552,6 +553,7 @@ def _etag_match(status, etagval, match, nomatch):
             if nomatch and ("*" in nomatch or etagval in nomatch):
                 raise cherrypy.HTTPRedirect([], 304)
 
+@profile
 def _etag_tail(head, tail, etag):
     """Generator which first returns anything in `head`, then `tail`.
     Sets ETag header at the end to value of `etag` if it's defined and
@@ -566,6 +568,7 @@ def _etag_tail(head, tail, etag):
     if etagval:
         cherrypy.response.headers["ETag"] = etagval
 
+@profile
 def stream_maybe_etag(size_limit, etag, reply):
     """Maybe generate ETag header for the response, and handle If-Match
     and If-None-Match request headers. Consumes the reply until at most
@@ -589,6 +592,8 @@ def stream_maybe_etag(size_limit, etag, reply):
     req = cherrypy.request
     res = cherrypy.response
     match = [str(x) for x in (req.headers.elements('If-Match') or [])]
+    # FIXME TODO this apparently increases wmstatsserver memory
+    # footprint by half giga byte
     nomatch = [str(x) for x in (req.headers.elements('If-None-Match') or [])]
 
     # If ETag is already set, match conditions and output without buffering.
@@ -606,6 +611,7 @@ def stream_maybe_etag(size_limit, etag, reply):
     # clients including browsers will ignore them.
     size = 0
     result = []
+    ### FIXME TODO: this block apparently leaks memory
     for chunk in reply:
         result.append(chunk)
         size += len(chunk)

src/python/WMCore/REST/Server.py

@@ -3,6 +3,7 @@
 
 import cherrypy
 import inspect
+from memory_profiler import profile
 import os
 import re
 import signal
@@ -728,6 +729,7 @@ def metrics(self):
         return encodeUnicodeToBytes(metrics)
 
     @expose
+    @profile
     def default(self, *args, **kwargs):
         """The HTTP request handler.
 
@@ -760,6 +762,7 @@ def default(self, *args, **kwargs):
 
     default._cp_config = {'response.stream': True}
 
+    @profile
     def _call(self, param):
         """The real HTTP request handler.
 
@@ -859,6 +862,8 @@ def _call(self, param):
         # Format the response.
         response.headers['X-REST-Status'] = 100
         response.headers['Content-Type'] = format
+        # FIXME TODO this apparently increases wmstatsserver memory
+        # footprint by half giga byte
         etagger = apiobj.get('etagger', None) or SHA1ETag()
         reply = stream_compress(fmthandler(obj, etagger),
                                 apiobj.get('compression', self.compression),

src/python/WMCore/Services/WMStats/WMStatsReader.py

@@ -4,6 +4,8 @@
 from future.utils import viewitems
 
 import logging
+from memory_profiler import profile
+
 from Utils.IteratorTools import nestedDictUpdate, grouper
 from WMCore.Database.CMSCouch import CouchServer
 from WMCore.Lexicon import splitCouchServiceURL, sanitizeURL
@@ -93,6 +95,7 @@ def setDefaultStaleOptions(self, options):
             options.update(self.defaultStale)
         return options
 
+    @profile
     def getLatestJobInfoByRequests(self, requestNames):
         jobInfoByRequestAndAgent = {}
 
@@ -101,6 +104,7 @@ def getLatestJobInfoByRequests(self, requestNames):
             jobInfoByRequestAndAgent = self._getLatestJobInfo(requestAndAgentKey)
         return jobInfoByRequestAndAgent
 
+    @profile
     def _updateRequestInfoWithJobInfo(self, requestInfo):
         if requestInfo:
             jobInfoByRequestAndAgent = self.getLatestJobInfoByRequests(list(requestInfo))
@@ -304,6 +308,7 @@ def getT0ActiveData(self, jobInfoFlag=False):
 
         return self.getRequestByStatus(T0_ACTIVE_STATUS, jobInfoFlag)
 
+    @profile
     def getRequestByStatus(self, statusList, jobInfoFlag=False, limit=None, skip=None,
                            legacyFormat=False):
 

src/python/WMCore/WMStats/CherryPyThreads/DataCacheUpdate.py

@@ -1,6 +1,7 @@
 from __future__ import (division, print_function)
 
 import time
+from memory_profiler import profile
 from WMCore.REST.CherryPyPeriodicTask import CherryPyPeriodicTask
 from WMCore.WMStats.DataStructs.DataCache import DataCache
 from WMCore.Services.WMStats.WMStatsReader import WMStatsReader
@@ -10,7 +11,6 @@ class DataCacheUpdate(CherryPyPeriodicTask):
 
     def __init__(self, rest, config):
         self.getJobInfo = getattr(config, "getJobInfo", False)
-
         super(DataCacheUpdate, self).__init__(config)
 
     def setConcurrentTasks(self, config):
@@ -19,6 +19,7 @@ def setConcurrentTasks(self, config):
         """
         self.concurrentTasks = [{'func': self.gatherActiveDataStats, 'duration': 300}]
 
+    @profile
     def gatherActiveDataStats(self, config):
         """
         gather active data statistics

src/python/WMCore/WMStats/DataStructs/DataCache.py

@@ -2,6 +2,8 @@
 from future.utils import viewitems
 
 import time
+
+from Utils.Utilities import getSize
 from WMCore.ReqMgr.DataStructs.Request import RequestInfo, protectedLFNs
 
 class DataCache(object):
@@ -22,6 +24,7 @@ def setDuration(sec):
     @staticmethod
     def getlatestJobData():
         if (DataCache._lastedActiveDataFromAgent):
+            print(f"Size of DataCache: {getSize(DataCache._lastedActiveDataFromAgent)}")
             return DataCache._lastedActiveDataFromAgent["data"]
         else:
             return {}

src/python/WMCore/WMStats/Service/ActiveRequestJobInfo.py

@@ -3,6 +3,7 @@
 Just wait for the server cache to be updated
 """
 from __future__ import (division, print_function)
+from memory_profiler import profile
 from WMCore.REST.Server import RESTEntity, restcall, rows
 from WMCore.REST.Tools import tools
 from WMCore.REST.Error import DataCacheEmpty
@@ -25,6 +26,7 @@ def validate(self, apiobj, method, api, param, safe):
 
     @restcall(formats=[('text/plain', PrettyJSONFormat()), ('application/json', JSONFormat())])
     @tools.expires(secs=-1)
+    @profile()
     def get(self):
         # This assumes DataCahe is periodically updated.
         # If data is not updated, need to check, dataCacheUpdate log