Repository: dmwm/wmcore
Issue: Flaw in the GQ -> LQ data acquisition pagination logic

**Impact of the bug**
WMAgent

**Describe the bug**
We received a report of lack of pressure in the condor pool - via MM this morning - and the reason was that the agents were no longer acquiring work from the global workqueue.

**How to reproduce it**
Very specific scenario, but apparently caused by ~5k ACDCs injected into the system yesterday (likely a correlation of target sites and very high priority).

**Expected behavior**
When the agent is running the GQ to LQ data acquisition (where "data" here is a workqueue element), the expected behavior and logic is:
* it has to iterate over the GQE in slices/pages
* whenever the target number of elements is found (passed in the query string as `num_elem`), stop iterating in the CouchDB list function and return data to the client
* keep iterating through the elements until the whole data is exhausted, or until num_elem are found.

So, for this issue, we need to correct how pagination is done to ensure that the list `workRestrictions` will receive all the required pages from the view function `availableByPriority`, instead of passing over only the first page and returning results to the end user.

**Additional context and error message**
The agent uses `availableWork()` method to pull work from GQ to the LQ. A few options are defined:
https://github.com/dmwm/WMCore/blob/master/src/python/WMCore/WorkQueue/WorkQueueBackend.py#L446-L471

including the use of `skip` and `limit`, to act as a pagination over all the global workqueue elements and evaluate them into pages (slices).

This call gets mapped to something similar to (from the couchdb log):
```
GET /workqueue/_design/WorkQueue/_list/workRestrictions/availableByPriority?
include_docs=true&
descending=true&
resources=%7B%22T2_CH_CERN%22%3A+20975.0%2C+%22T2_CH_CERN_P5%22%3A+187....5.0%7D&
limit=1000&
num_elem=1000&
team=%22production%22&
skip=0
``` 


Changed files:
src/python/WMCore/WorkQueue/WorkQueueBackend.py, etc/WMAgentConfig.py, src/python/WMCore/WorkQueue/WorkQueue.py

diff --git a/etc/WMAgentConfig.py b/etc/WMAgentConfig.py
index 38d3e1a1c..2c7f0464e 100644
--- a/etc/WMAgentConfig.py
+++ b/etc/WMAgentConfig.py
@@ -141,6 +141,10 @@ config.WorkQueueManager.queueParams["ParentQueueCouchUrl"] = "https://cmsweb.cer
 config.WorkQueueManager.queueParams["QueueURL"] = "http://%s:5984" % (config.Agent.hostName)
 config.WorkQueueManager.queueParams["WorkPerCycle"] = 200  # don't pull more than this number of elements per cycle
 config.WorkQueueManager.queueParams["QueueDepth"] = 0.5  # pull work from GQ for only half of the resources
+# number of available elements to be retrieved within a single CouchDB http request
+config.WorkQueueManager.queueParams["RowsPerSlice"] = 2500
+# maximum number of available elements rows to be evaluated when acquiring GQ to LQ work
+config.WorkQueueManager.queueParams["MaxRowsPerCycle"] = 50000
 config.WorkQueueManager.queueParams["rucioAccount"] = "wmcore_transferor"  # account for data locks
 
 
diff --git a/src/python/WMCore/WorkQueue/WorkQueue.py b/src/python/WMCore/WorkQueue/WorkQueue.py
index e4ddfe845..4600f9af1 100644
--- a/src/python/WMCore/WorkQueue/WorkQueue.py
+++ b/src/python/WMCore/WorkQueue/WorkQueue.py
@@ -119,6 +119,8 @@ class WorkQueue(WorkQueueBase):
 
         self.params.setdefault('QueueDepth', 1)  # when less than this locally
         self.params.setdefault('WorkPerCycle', 100)
+        self.params.setdefault('RowsPerSlice', 2500)
+        self.params.setdefault('MaxRowsPerCycle', 50000)
         self.params.setdefault('LocationRefreshInterval', 600)
         self.params.setdefault('FullLocationRefreshInterval', 7200)
         self.params.setdefault('TrackLocationOrSubscription', 'location')
@@ -310,13 +312,16 @@ class WorkQueue(WorkQueueBase):
         """
         excludeWorkflows = excludeWorkflows or []
         results = []
-        numElems = self.params['WorkPerCycle']
         if not self.backend.isAvailable():
             self.logger.warning('Backend busy or down: skipping fetching of work')
             return results
 
+        # TODO AMR: perhaps numElems limit should be removed for LQ -> WMBS acquisition
         matches, _ = self.backend.availableWork(jobSlots, siteJobCounts,
-                                                excludeWorkflows=excludeWorkflows, numElems=numElems)
+                                                excludeWorkflows=excludeWorkflows,
+                                                numElems=self.params['WorkPerCycle'],
+                                                rowsPerSlice=self.params['RowsPerSlice'],
+                                                maxRows=self.params['MaxRowsPerCycle'])
 
         self.logger.info('Got %i elements matching the constraints', len(matches))
         if not matches:
@@ -825,9 +830,11 @@ class WorkQueue(WorkQueueBase):
         return (resources, jobCounts)
 
     def getAvailableWorkfromParent(self, resources, jobCounts, printFlag=False):
-        numElems = self.params['WorkPerCycle']
         self.logger.info("Going to fetch work from the parent queue: %s", self.parent_queue.queueUrl)
-        work, _ = self.parent_queue.availableWork(resources, jobCounts, self.params['Team'], numElems=numElems)
+        work, _ = self.parent_queue.availableWork(resources, jobCounts, self.params['Team'],
+                                                  numElems=self.params['WorkPerCycle'],
+                                                  rowsPerSlice=self.params['RowsPerSlice'],
+                                                  maxRows=self.params['MaxRowsPerCycle'])
         if not work:
             self._printLog('No available work in parent queue.', printFlag, "warning")
         return work
diff --git a/src/python/WMCore/WorkQueue/WorkQueueBackend.py b/src/python/WMCore/WorkQueue/WorkQueueBackend.py
index a23c00a18..d3d069f06 100644
--- a/src/python/WMCore/WorkQueue/WorkQueueBackend.py
+++ b/src/python/WMCore/WorkQueue/WorkQueueBackend.py
@@ -6,6 +6,7 @@ Interface to WorkQueue persistent storage
 """
 
 from builtins import object
+from math import ceil
 
 from future.utils import viewitems
 
@@ -411,8 +412,8 @@ class WorkQueueBackend(object):
                          len(elements), self.queueUrl)
         return elements, siteJobCounts
 
-    def availableWork(self, thresholds, siteJobCounts, team=None,
-                      excludeWorkflows=None, numElems=9999999):
+    def availableWork(self, thresholds, siteJobCounts, team=None, excludeWorkflows=None,
+                      numElems=1000, rowsPerSlice=1000, maxRows=1000):
         """
         Get work - either from local or global queue - which is available to be run.
 
@@ -422,117 +423,143 @@ class WorkQueueBackend(object):
             is a dictionary with the number of jobs running at a given priority.
         :param team: a string with the team name we want to pull work for
         :param excludeWorkflows: list of (aborted) workflows that should not be accepted
-        :param numElems: integer with the maximum number of elements to be accepted (default
-            to a very large number when pulling work from local queue, read unlimited)
+        :param numElems: integer with the maximum number of elements to be accepted.
+        :param rowsPerSlice: integer defining the amount of rows for each slice (slices
+             of a couchdb view request).
+        :param maxRows: maximum number of available elements (rows) to be considered
+            when pulling work down to the agent.
         :return: a tuple with the elements accepted and an overview of job counts per site
         """
         excludeWorkflows = excludeWorkflows or []
-        elements = []
+        acceptedElems = []
         # If there are no sites, punt early.
         if not thresholds:
             self.logger.error("No thresholds is set: Please check")
-            return elements, siteJobCounts
+            return acceptedElems, siteJobCounts
 
         self.logger.info("Current siteJobCounts:")
         for site, jobsByPrio in viewitems(siteJobCounts):
             self.logger.info("    %s : %s", site, jobsByPrio)
 
+        # Find out how many elements are in Available status
+        numAvail = self.queueLength()
+        self.logger.info("Current amount of WQEs in Available status: %s", numAvail)
+
         self.logger.info("Getting up to %d available work from %s", numElems, self.queueUrl)
         self.logger.info("  for team name: %s", team)
-        self.logger.info("  with excludeWorkflows: %s", excludeWorkflows)
+        self.logger.info("  with excludeWorkflows count: %s", len(excludeWorkflows))
+        self.logger.debug("  with excludeWorkflows: %s", excludeWorkflows)
         self.logger.info("  for thresholds: %s", thresholds)
 
-        # FIXME: magic numbers
-        docsSliceSize = 1000
         options = {}
         options['include_docs'] = True
         options['descending'] = True
         options['resources'] = thresholds
-        options['limit'] = docsSliceSize
+        options['limit'] = rowsPerSlice
         # FIXME: num_elem option can likely be deprecated, but it needs synchronization
         # between agents and global workqueue... for now, make sure it can return the slice size
-        options['num_elem'] = docsSliceSize
+        options['num_elem'] = rowsPerSlice
         if team:
             options['team'] = team
 
         # Fetch workqueue elements in slices, using the CouchDB "limit" and "skip"
         # options for couch views. Conditions to stop this loop are:
-        #  a) have a hard stop at 50k+1 (we might have to make this configurable)
-        #  b) stop as soon as an empty slice is returned by Couch (thus all docs have
-        #     already been retrieve)
-        #  c) or, once "numElems" elements have been accepted
-        numSkip = 0
-        breakOut = False
-        while True:
-            if breakOut:
-                # then we have reached the maximum number of elements to be accepted
-                break
-            self.logger.info("  with limit docs: %s, and skip first %s docs", docsSliceSize, numSkip)
-            options['skip'] = numSkip
+        #  a) stop once total_rows is reached (exhausted all available elements)
+        #  b) hit maximum allowed elements/rows to be considered for data acquisition (maxRows)
+        #  c) or, once the targeted number of elements has been accepted (numElems)
+        numSlices = ceil(numAvail / rowsPerSlice)
+        numSlices = min(numSlices, int(maxRows / rowsPerSlice))
+        for sliceNum in range(numSlices):
+            # documents to skip as a function of the slice number
+            options['skip'] = sliceNum * rowsPerSlice
+            self.logger.info("  for slice: %s, with rows range [%s - %s]",
+                             sliceNum, options['skip'], options['skip'] + options['limit'])
 
             result = self.db.loadList('WorkQueue', 'workRestrictions', 'availableByPriority', options)
-            result = json.loads(result)
-            if result:
-                self.logger.info("Retrieved %d elements from workRestrictions list for: %s",
-                                 len(result), self.queueUrl)
+            # now check the remaining restrictions and priority
+            wqeSlots = numElems - len(acceptedElems)
+            elems = self._evalAvailableWork(json.loads(result), thresholds, siteJobCounts,
+                                            excludeWorkflows, wqeSlots)
+            acceptedElems.extend(elems)
+            if len(acceptedElems) >= numElems:
+                msg = f"Reached maximum number of elements to be accepted, "
+                msg += f"configured to: {numElems}, from queue: {self.queueUrl}"
+                self.logger.info(msg)
+                break
+
+        self.logger.info("Total of %d elements passed location and siteJobCounts restrictions for: %s",
+                         len(acceptedElems), self.queueUrl)
+        return acceptedElems, siteJobCounts
+
+    def _evalAvailableWork(self, listElems, thresholds, siteJobCounts,
+                           excludeWorkflows, numElems):
+        """
+        Evaluate work available in workqueue and decide whether it can be
+        accepted or not.
+
+        :param listElems: list of dictionaries that correspond to the workqueue elements.
+        :param thresholds: a dictionary key'ed by the site name, values representing the
+            maximum number of jobs allowed at that site.
+        :param siteJobCounts: a dictionary-of-dictionaries key'ed by the site name; value
+            is a dictionary with the number of jobs running at a given priority.
+            NOTE that it is updated in place.
+        :param excludeWorkflows: list of (aborted) workflows that should not be accepted
+        :param numElems: integer with the maximum number of elements to be accepted (default
+            to a very large number when pulling work from local queue, read unlimited)
+        :return: a tuple with the elements accepted and an overview of job counts per site
+        """
+        elems = []
+        self.logger.info("Retrieved %d elements from workRestrictions list for: %s",
+                         len(listElems), self.queueUrl)
+        # Convert python dictionary into Couch WQE objects, skipping aborted workflows
+        # And sort them by creation time and priority, such that highest priority and
+        # oldest elements come first in the list
+        sortedElements = []
+        for i in listElems:
+            element = CouchWorkQueueElement.fromDocument(self.db, i)
+            # make sure not to acquire work for aborted or force-completed workflows
+            if element['RequestName'] in excludeWorkflows:
+                msg = "Skipping aborted/force-completed workflow: %s, work id: %s"
+                self.logger.info(msg, element['RequestName'], element._id)
             else:
-                self.logger.info("All the workqueue elements have been exhausted for: %s ", self.queueUrl)
+                sortedElements.append(element)
+        sortAvailableElements(sortedElements)
+
+        for element in sortedElements:
+            if numElems <= 0:
+                # it means we accepted the configured number of elements
                 break
-            # update number of documents to skip in the next cycle
-            numSkip += docsSliceSize
-
-            # Convert python dictionary into Couch WQE objects, skipping aborted workflows
-            # And sort them by creation time and priority, such that highest priority and
-            # oldest elements come first in the list
-            sortedElements = []
-            for i in result:
-                element = CouchWorkQueueElement.fromDocument(self.db, i)
-                # make sure not to acquire work for aborted or force-completed workflows
-                if element['RequestName'] in excludeWorkflows:
-                    msg = "Skipping aborted/force-completed workflow: %s, work id: %s"
-                    self.logger.info(msg, element['RequestName'], element._id)
-                else:
-                    sortedElements.append(element)
-            sortAvailableElements(sortedElements)
-
-            for element in sortedElements:
-                if numElems <= 0:
-                    msg = "Reached maximum number of elements to be accepted, "
-                    msg += "configured to: {}, from queue: {}".format(len(elements), self.queueUrl)
-                    self.logger.info(msg)
-                    breakOut = True  # get out of the outer loop as well
-                    break
-                commonSites = possibleSites(element)
-                prio = element['Priority']
-                # shuffle list of common sites all the time to give everyone the same chance
-                random.shuffle(commonSites)
-                possibleSite = None
-                for site in commonSites:
-                    if site in thresholds:
-                        # Count the number of jobs currently running of greater priority, if they
-                        # are less than the site thresholds, then accept this element
-                        curJobCount = sum([x[1] if x[0] >= prio else 0 for x in viewitems(siteJobCounts.get(site, {}))])
-                        self.logger.debug(
-                            "Job Count: %s, site: %s thresholds: %s" % (curJobCount, site, thresholds[site]))
-                        if curJobCount < thresholds[site]:
-                            possibleSite = site
-                            break
-
-                if possibleSite:
-                    self.logger.info("Accepting workflow: %s, with prio: %s, element id: %s, for site: %s",
-                                     element['RequestName'], prio, element.id, possibleSite)
-                    numElems -= 1
-                    elements.append(element)
-                    siteJobCounts.setdefault(possibleSite, {})
-                    siteJobCounts[possibleSite][prio] = siteJobCounts[possibleSite].setdefault(prio, 0) + \\
-                                                        element['Jobs'] * element.get('blowupFactor', 1.0)
-                else:
-                    self.logger.debug("No available resources for %s with doc id %s",
-                                      element['RequestName'], element.id)
+            commonSites = possibleSites(element)
+            prio = element['Priority']
+            # shuffle list of common sites all the time to give everyone the same chance
+            random.shuffle(commonSites)
+            possibleSite = None
+            for site in commonSites:
+                if site in thresholds:
+                    # Count the number of jobs currently running of greater priority, if they
+                    # are less than the site thresholds, then accept this element
+                    curJobCount = sum([x[1] if x[0] >= prio else 0 for x in viewitems(siteJobCounts.get(site, {}))])
+                    self.logger.debug("Job Count: %s, site: %s thresholds: %s",
+                                      curJobCount, site, thresholds[site])
+                    if curJobCount < thresholds[site]:
+                        possibleSite = site
+                        break
+
+            if possibleSite:
+                self.logger.info("Accepting workflow: %s, with prio: %s, element id: %s, for site: %s",
+                                 element['RequestName'], prio, element.id, possibleSite)
+                numElems -= 1
+                elems.append(element)
+                siteJobCounts.setdefault(possibleSite, {})
+                siteJobCounts[possibleSite][prio] = siteJobCounts[possibleSite].setdefault(prio, 0) + \\
+                                                    element['Jobs'] * element.get('blowupFactor', 1.0)
+            else:
+                self.logger.debug("No available resources for %s with doc id %s",
+                                  element['RequestName'], element.id)
 
         self.logger.info("And %d elements passed location and siteJobCounts restrictions for: %s",
-                         len(elements), self.queueUrl)
-        return elements, siteJobCounts
+                         len(elems), self.queueUrl)
+        return elems
 
     def getActiveData(self):
         """Get data items we have work in the queue for"""