Repository: dmwm/wmcore
Issue: MSTransferor interrupted due to bad datastructure reference

**Impact of the bug**
reqmgr2ms

**Describe the bug**
We started seeing those errors [6] in the MSTransferor component. The reason is a broken reference to a value in the following data structure [1]. This happens in this line here [2]. For how it should be done see the next section. 

The second part of this bug is that the exception is not caught and the cycle is interrupted whenever it hits. In the current situation this happens in the very first workflow from every batch of 100 workflows because of the bad reference to one of the fields in the data structure, but even after this is fixed, if a future exception of the kind is not handled properly here, a malformed value returned by the external service (in this case Phedex) can still lead to interruption of the work of the internal component (in this case MSTransferor).

The third part of the bug is that it also prevents the transition of the workflow from 'Assigned' to 'Staging', because this is supposed to be happening in the same component too. The status transition is called here [3], but the chain breaks and the MSTransferor is interrupted just few lines above - here[4]. Luckily we still keep the code of the old function that was used to do that transition [5] and we can revert to it for the moment (the switch is configurable). 

[1]
https://github.com/dmwm/WMCore/blob/f7e830a573d50fb1d7240797f18d809f994b934d/src/python/WMCore/MicroService/DataStructs/Workflow.py#L221-L222

[2]
https://github.com/dmwm/WMCore/blob/f7e830a573d50fb1d7240797f18d809f994b934d/src/python/WMCore/MicroService/DataStructs/Workflow.py#L333

[3]
https://github.com/dmwm/WMCore/blob/f7e830a573d50fb1d7240797f18d809f994b934d/src/python/WMCore/MicroService/Unified/MSTransferor.py#L189-L190
https://github.com/dmwm/WMCore/blob/f7e830a573d50fb1d7240797f18d809f994b934d/src/python/WMCore/MicroService/Unified/MSCore.py#L63-L74

[4]
https://github.com/dmwm/WMCore/blob/f7e830a573d50fb1d7240797f18d809f994b934d/src/python/WMCore/MicroService/Unified/MSTransferor.py#L183

[5]
https://github.com/dmwm/WMCore/pull/9508
https://github.com/dmwm/WMCore/pull/9508/commits/372dafdf11d5f8b26cc4112ae9b241934058a1a1

[6]
```
2020-03-11 20:18:20,918:INFO:MSTransferor: Handling data subscriptions for request: pdmvserv_task_HIG-RunIISummer16NanoAODv6-02991__v1_T_200305_220949_2007
2020-03-11 20:18:20,919:INFO:MSTransferor:   final list of PSNs to be use: [u'T1_DE_KIT', u'T1_ES_PIC', u'T1_FR_CCIN2P3', u'T1_RU_JINR', u'T1_UK_RAL', u'T1_US_FNAL', u'T2_BE_IIHE', u'T2_CH_CERN', u'T2_DE_DESY', u'T2_DE_RWTH', u'T2_ES_CIEMAT', u'T2_FR_GRI
F_IRFU', u'T2_FR_GRIF_LLR', u'T2_FR_IPHC', u'T2_IT_Bari', u'T2_IT_Legnaro', u'T2_UK_London_Brunel', u'T2_UK_SGrid_RALPP', u'T2_US_Caltech', u'T2_US_MIT', u'T2_US_Nebraska', u'T2_US_Purdue', u'T2_US_UCSD', u'T2_US_Vanderbilt', u'T2_US_Wisconsin']
2020-03-11 20:18:20,919:INFO:MSTransferor: List of out-of-space RSEs dropped for 'pdmvserv_task_HIG-RunIISummer16NanoAODv6-02991__v1_T_200305_220949_2007' is: set([u'T1_ES_PIC_Disk', u'T2_US_UCSD', u'T2_CH_CERN', u'T1_FR_CCIN2P3_Disk', u'T2_US_Wisconsin'
, u'T2_DE_DESY', u'T2_DE_RWTH', u'T2_FR_GRIF_IRFU', u'T2_US_Caltech', u'T1_DE_KIT_Disk', u'T2_UK_London_Brunel', u'T2_US_Purdue', u'T1_RU_JINR_Disk', u'T2_UK_SGrid_RALPP', u'T2_US_Vanderbilt', u'T2_US_Nebraska', u'T1_UK_RAL_Disk', u'T1_US_FNAL_Disk'])
2020-03-11 20:18:20,965:ERROR:MSManager: MS daemon error: unsupported operand type(s) for +: 'int' and 'dict'
Traceback (most recent call last):
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/Unified/MSManager.py", line 42, in daemon
    func(reqStatus)
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/Unified/MSManager.py", line 119, in transferor
    res = self.msTransferor.execute(reqStatus)
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/Unified/MSTransferor.py", line 183, in execute
    success, transfers = self.makeTransferRequest(wflow)
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/Unified/MSTransferor.py", line 319, in makeTransferRequest
    for blocks, dataSize, idx in self._decideDataDestination(wflow, dataIn, len(nodes)):
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/Unified/MSTransferor.py", line 465, in _decideDataDestination
    listBlockSets, listSetsSize = wflow.getChunkBlocks(numNodes)
  File "/data/srv/beHG2003k/sw/slc7_amd64_gcc630/cms/reqmgr2ms/0.2.1.pre3/lib/python2.7/site-packages/WMCore/MicroService/DataStructs/Workflow.py", line 333, in getChunkBlocks
    chunkSize = sum(item[1] for item in sortedPrimary) // numChunks
TypeError: unsupported operand type(s) for +: 'int' and 'dict'

```

**How to reproduce it**
Steps to reproduce the behavior:

**Expected behavior**

Instead of trying to fetch and sum the value like that:
```
chunkSize = sum(item[1] for item in sortedPrimary)
``` 
it should be referred like that:
```
chunkSize = sum(item[1]['blockSize'] for item in sortedPrimary)
```

Also the exception needs to be caught and handled properly so the rest of the cycle is not interrupted.

**Additional context and error message**



Changed files:
src/python/WMCore/MicroService/Unified/MSTransferor.py, test/python/WMCore_t/MicroService_t/DataStructs_t/Workflow_t.py, src/python/WMCore/MicroService/Unified/MSCore.py, src/python/WMCore/MicroService/DataStructs/Workflow.py

diff --git a/src/python/WMCore/MicroService/DataStructs/Workflow.py b/src/python/WMCore/MicroService/DataStructs/Workflow.py
index 813519a6d..64bef4468 100644
--- a/src/python/WMCore/MicroService/DataStructs/Workflow.py
+++ b/src/python/WMCore/MicroService/DataStructs/Workflow.py
@@ -291,9 +291,9 @@ class Workflow(object):
             for parent in list(parents):
                 if parent not in self.getParentBlocks():
                     # then drop this block
-                    parents.remove()
+                    parents.remove(parent)
             self.childToParentBlocks[child] = blocksDict[child]
-            parentBlocks = parentBlocks | blocksDict[child]
+            parentBlocks = parentBlocks | set(blocksDict[child])
 
         # Now remove any parent block that don't need to be transferred
         for block in list(self.getParentBlocks()):
@@ -330,7 +330,12 @@ class Workflow(object):
 
         # create a descendant list of blocks according to their sizes
         sortedPrimary = sorted(self.getPrimaryBlocks().items(), key=operator.itemgetter(1), reverse=True)
-        chunkSize = sum(item[1] for item in sortedPrimary) // numChunks
+        if len(sortedPrimary) < numChunks:
+            msg = "There are less blocks than chunks to create. "
+            msg += "Reducing numChunks from %d to %d" % (numChunks, len(sortedPrimary))
+            self.logger.info(msg)
+            numChunks = len(sortedPrimary)
+        chunkSize = sum(item[1]['blockSize'] for item in sortedPrimary) // numChunks
 
         self.logger.info("Found %d blocks and the avg chunkSize is: %s GB",
                          len(sortedPrimary), gigaBytes(chunkSize))
@@ -347,20 +352,26 @@ class Workflow(object):
                 if not sortedPrimary or idx >= len(sortedPrimary):
                     # then all blocks have been distributed
                     break
-                elif thisChunkSize + sortedPrimary[idx][1] <= chunkSize:
+                elif not thisChunkSize:
+                    # then this site/chunk is empty, assign a block to it
                     thisChunk.add(sortedPrimary[idx][0])
-                    thisChunkSize += sortedPrimary[idx][1]
+                    thisChunkSize += sortedPrimary[idx][1]['blockSize']
+                    sortedPrimary.pop(idx)
+                elif thisChunkSize + sortedPrimary[idx][1]['blockSize'] <= chunkSize:
+                    thisChunk.add(sortedPrimary[idx][0])
+                    thisChunkSize += sortedPrimary[idx][1]['blockSize']
                     sortedPrimary.pop(idx)
                 else:
                     idx += 1
-            blockChunks.append(thisChunk)
-            sizeChunks.append(thisChunkSize)
+            if thisChunk:
+                blockChunks.append(thisChunk)
+                sizeChunks.append(thisChunkSize)
 
         # now take care of the leftovers... in a round-robin style....
         while sortedPrimary:
             for chunkNum in range(numChunks):
                 blockChunks[chunkNum].add(sortedPrimary[0][0])
-                sizeChunks[chunkNum] += sortedPrimary[0][1]
+                sizeChunks[chunkNum] += sortedPrimary[0][1]['blockSize']
                 sortedPrimary.pop(0)
                 if not sortedPrimary:
                     break
@@ -384,7 +395,7 @@ class Workflow(object):
             # of blocks within the chunk and update the chunk size as well
             blockChunks[chunkNum].update(parentSet)
             for parent in parentSet:
-                sizeChunks[chunkNum] += parentsSize[parent]
+                sizeChunks[chunkNum] += parentsSize[parent]['blockSize']
         self.logger.info("Created %d primary+parent data chunks out of %d chunks",
                          len(blockChunks), numChunks)
         self.logger.info("    with chunk size distribution: %s", sizeChunks)
diff --git a/src/python/WMCore/MicroService/Unified/MSCore.py b/src/python/WMCore/MicroService/Unified/MSCore.py
index 7de2dd110..57c816777 100644
--- a/src/python/WMCore/MicroService/Unified/MSCore.py
+++ b/src/python/WMCore/MicroService/Unified/MSCore.py
@@ -69,7 +69,7 @@ class MSCore(object):
                 self.logger.info('%s updating %s status to: %s', prefix, reqName, reqStatus)
                 self.reqmgr2.updateRequestStatus(reqName, reqStatus)
             else:
-                self.logger.info('%s FAKE updating %s status to: %s', prefix, reqName, reqStatus)
+                self.logger.info('DRY-RUN:: %s updating %s status to: %s', prefix, reqName, reqStatus)
         except Exception as err:
             self.logger.exception("Failed to change request status. Error: %s", str(err))
 
diff --git a/src/python/WMCore/MicroService/Unified/MSTransferor.py b/src/python/WMCore/MicroService/Unified/MSTransferor.py
index 9211bb504..e0e607ef4 100644
--- a/src/python/WMCore/MicroService/Unified/MSTransferor.py
+++ b/src/python/WMCore/MicroService/Unified/MSTransferor.py
@@ -164,7 +164,7 @@ class MSTransferor(MSCore):
             self.updateReportDict(summary, "error", msg)
             return summary
         except Exception as ex:
-            msg = "Unknown exception updating caches. Error: %s", str(ex)
+            msg = "Unknown exception updating caches. Error: %s" % str(ex)
             self.logger.exception(msg)
             self.updateReportDict(summary, "error", msg)
             return summary
@@ -180,7 +180,13 @@ class MSTransferor(MSCore):
                 # first check which data is already in place and filter them out
                 self.checkDataLocation(wflow)
 
-                success, transfers = self.makeTransferRequest(wflow)
+                try:
+                    success, transfers = self.makeTransferRequest(wflow)
+                except Exception as ex:
+                    success = False
+                    msg = "Unknown exception while making Transfer Request for %s " % wflow.getName()
+                    msg += "\\tError: %s" % str(ex)
+                    self.logger.exception(msg)
                 if success:
                     self.logger.info("Transfer requests successful for %s. Summary: %s",
                                      wflow.getName(), pformat(transfers))                    # then create a document in ReqMgr Aux DB
@@ -358,7 +364,7 @@ class MSTransferor(MSCore):
                     else:
                         self.blockCounter += len(blocks)
                 else:
-                    self.logger.info("NOT making any data subscriptions, as it's disabled in the configuration")
+                    self.logger.info("DRY-RUN: making subscription: %s", subscription)
 
             transRec['transferIDs'] = list(transRec['transferIDs'])
             response.append(transRec)
diff --git a/test/python/WMCore_t/MicroService_t/DataStructs_t/Workflow_t.py b/test/python/WMCore_t/MicroService_t/DataStructs_t/Workflow_t.py
index e38b52f85..408a11056 100644
--- a/test/python/WMCore_t/MicroService_t/DataStructs_t/Workflow_t.py
+++ b/test/python/WMCore_t/MicroService_t/DataStructs_t/Workflow_t.py
@@ -212,6 +212,139 @@ class WorkflowTest(unittest.TestCase):
         self.assertEqual(len(listWflows), 3)
         self.assertEqual(len(badWflows), 2)
 
+    def testParentageRelationship(self):
+        """
+        Test methods related to the primary and parent datasets and blocks
+        """
+        primDict = {"block_A": {"blockSize": 1, "locations": ["Site_A"]},
+                    "block_B": {"blockSize": 2, "locations": ["Site_B"]}}
+        parentDict = {"parent_A": {"blockSize": 11, "locations": ["Site_A"]},
+                      "parent_B": {"blockSize": 12, "locations": ["Site_B"]},
+                      "parent_C": {"blockSize": 13, "locations": ["Site_A", "Site_B"]}}
+        parentage = {"block_A": ["parent_B", "parent_D"],  # parent_D has no replicas!
+                     "block_B": ["parent_A", "parent_C"]}
+        wflow = Workflow("workflow_1", {"RequestType": "TaskChain",
+                                        "InputDataset": "Dataset_name_XXX",
+                                        "IncludeParents": True})
+
+        self.assertEqual(wflow.getParentDataset(), "")
+        wflow.setParentDataset("Parent_dataset_XXX")
+        self.assertEqual(wflow.getParentDataset(), "Parent_dataset_XXX")
+
+        self.assertEqual(wflow.getPrimaryBlocks(), {})
+        wflow.setPrimaryBlocks(primDict)
+        self.assertItemsEqual(wflow.getPrimaryBlocks().keys(), ["block_A", "block_B"])
+
+        self.assertEqual(wflow.getParentBlocks(), {})
+        wflow.setParentBlocks(parentDict)
+        self.assertItemsEqual(wflow.getParentBlocks().keys(), ["parent_A", "parent_B", "parent_C"])
+
+        self.assertEqual(wflow.getChildToParentBlocks(), {})
+        wflow.setChildToParentBlocks(parentage)
+        self.assertItemsEqual(wflow.getChildToParentBlocks(), parentage)
+
+    def testGetChunkBlocks1(self):
+        """
+        Perform single chunk tests on the `getChunkBlocks` method.
+        """
+        primDict = {"block_A": {"blockSize": 1, "locations": ["Site_A", "Site_B"]}}
+        parentDict = {"parent_A": {"blockSize": 11, "locations": ["Site_A", "Site_B"]},
+                      "parent_B": {"blockSize": 8, "locations": []}}
+        wflow = Workflow("workflow_1", {"RequestType": "TaskChain",
+                                        "InputDataset": "Dataset_name_XXX"})
+        wflow.setPrimaryBlocks(primDict)
+        blockChunks, sizeChunks = wflow.getChunkBlocks(1)
+        self.assertEqual(len(blockChunks), 1)
+        self.assertItemsEqual(blockChunks[0], {"block_A"})
+        self.assertEqual(len(sizeChunks), 1)
+        self.assertEqual(sizeChunks[0], 1)
+
+        # now set a parent
+        wflow.setParentDataset("Parent_dataset_XXX")
+        wflow.setParentBlocks(parentDict)
+        blockChunks, sizeChunks = wflow.getChunkBlocks(1)
+        self.assertEqual(len(blockChunks), 1)
+        self.assertItemsEqual(blockChunks[0], {"block_A", "parent_A", "parent_B"})
+        self.assertEqual(len(sizeChunks), 1)
+        self.assertEqual(sizeChunks[0], 20)
+
+    def testGetChunkBlocks2(self):
+        """
+        Perform block distribution among many chunks, testing the `getChunkBlocks` method.
+        """
+        primDict = {"block_A": {"blockSize": 1, "locations": ["Site_A"]},
+                    "block_B": {"blockSize": 2, "locations": ["Site_B"]}}
+        wflow = Workflow("workflow_1", {"RequestType": "TaskChain",
+                                        "InputDataset": "Dataset_name_XXX"})
+        wflow.setPrimaryBlocks(primDict)
+
+        # same number of chunks and primary blocks
+        blockChunks, sizeChunks = wflow.getChunkBlocks(2)
+        self.assertEqual(len(blockChunks), 2)
+        self.assertItemsEqual(blockChunks[0], {"block_B"})
+        self.assertItemsEqual(blockChunks[1], {"block_A"})
+        self.assertEqual(len(sizeChunks), 2)
+        self.assertEqual(sizeChunks[0], 2)
+        self.assertEqual(sizeChunks[1], 1)
+
+        # more chunks than blocks
+        blockChunks, sizeChunks = wflow.getChunkBlocks(5)
+        self.assertEqual(len(blockChunks), 2)
+        self.assertItemsEqual(blockChunks[0], {"block_B"})
+        self.assertItemsEqual(blockChunks[1], {"block_A"})
+        self.assertEqual(len(sizeChunks), 2)
+        self.assertEqual(sizeChunks[0], 2)
+        self.assertEqual(sizeChunks[1], 1)
+
+        # more blocks than chunks
+        primDict.update({"block_C": {"blockSize": 3, "locations": ["Site_C"]},
+                         "block_D": {"blockSize": 4, "locations": ["Site_D"]},
+                         "block_E": {"blockSize": 5, "locations": ["Site_E"]}})
+        wflow.setPrimaryBlocks(primDict)
+        blockChunks, sizeChunks = wflow.getChunkBlocks(3)
+        self.assertEqual(len(blockChunks), 3)
+        self.assertItemsEqual(blockChunks[0], {"block_E"})
+        self.assertItemsEqual(blockChunks[1], {"block_D", "block_A"})
+        self.assertItemsEqual(blockChunks[2], {"block_C", "block_B"})
+        self.assertEqual(len(sizeChunks), 3)
+        self.assertEqual(sizeChunks[0], 5)
+        self.assertEqual(sizeChunks[1], 5)
+        self.assertEqual(sizeChunks[2], 5)
+
+    def testGetChunkBlocks3(self):
+        """
+        Test the `getChunkBlocks` method and especially the parent/child
+        relationship
+        """
+        primDict = {"block_A": {"blockSize": 1, "locations": ["Site_A"]},
+                    "block_B": {"blockSize": 2, "locations": ["Site_B"]}}
+        parentDict = {"parent_A": {"blockSize": 11, "locations": ["Site_A"]},
+                      "parent_B": {"blockSize": 12, "locations": ["Site_B"]},
+                      "parent_C": {"blockSize": 13, "locations": ["Site_A", "Site_B"]}}
+        parentage = {"block_A": ["parent_B", "parent_D"],  # parent_D has no replicas!
+                     "block_B": ["parent_A", "parent_C"]}
+        wflow = Workflow("workflow_1", {"RequestType": "TaskChain",
+                                        "InputDataset": "Dataset_name_XXX"})
+
+        # now set a parent
+        wflow.setParentDataset("Parent_dataset_XXX")
+        wflow.setPrimaryBlocks(primDict)
+        wflow.setParentBlocks(parentDict)
+        wflow.setChildToParentBlocks(parentage)
+
+        blockChunks, sizeChunks = wflow.getChunkBlocks(1)
+        self.assertEqual(len(blockChunks), 1)
+        self.assertItemsEqual(blockChunks[0], {"block_A", "block_B", "parent_A", "parent_B", "parent_C"})
+        self.assertEqual(len(sizeChunks), 1)
+        self.assertEqual(sizeChunks[0], 39)
+
+        blockChunks, sizeChunks = wflow.getChunkBlocks(2)
+        self.assertEqual(len(blockChunks), 2)
+        self.assertItemsEqual(blockChunks[0], {"block_B", "parent_A", "parent_C"})
+        self.assertItemsEqual(blockChunks[1], {"block_A", "parent_B"})
+        self.assertEqual(len(sizeChunks), 2)
+        self.assertEqual(sizeChunks[0], 26)
+        self.assertEqual(sizeChunks[1], 13)
 
 if __name__ == '__main__':
     unittest.main()