Repository: dmwm/wmcore
Issue: Inconsistency between campaign data location and SiteWhitelist

**Impact of the bug**
MSTransferor

**Describe the bug**
As discussed yesterday in the P&R slack #wmcore-support channel. Some places use the RSE (PNN) information while others contain a list of PSN (cpu resources), which not always have a 1-1 match.

As one can see in the error message reported by MSTransferor at the very bottom, the campaign defines that pileup dataset to be available/placed under `T1_IT_CNAF_Disk`, while the workflow SiteWhitelist contains the processing locations (PSN), thus `T1_IT_CNAF`. Since it does not have any match, MSTransferor refuses to proceed with this pileup data placement.

**How to reproduce it**
Have a Disk endpoint defined in the campaign.

**Expected behavior**
There are different ways to fix it. However, I suspect the best alternative would be to map the campaign RSEs (PNNs) back to their PSN sites. Only then try to match input data expected location against the workflow SiteWhitelist.

**Additional context and error message**
MSTransferor log:
```
./ms-transferor-20210118-ms-transferor-7744c99cd8-5p6m8.log:410:2021-01-18 01:13:24,995:ERROR:MSTransferor: Workflow has been incorrectly assigned: cmsunified_task_TAU-RunIISummer20UL17wmLHEGEN-00007__v1_T_210115_193310_9016. The secondary dataset: /MinBias_TuneCP5_13TeV-pythia8/RunIISummer20UL17SIM-106X_mc2017_realistic_v6-v2/GEN-SIM,belongs to the campaign: RunIISummer20UL17DIGI, with Secondaries location set to: [u'T1_IT_CNAF_Disk']. While the workflow has been assigned to: [u'T1_DE_KIT', u'T1_ES_PIC', u'T1_FR_CCIN2P3', u'T1_IT_CNAF', u'T1_RU_JINR', u'T1_UK_RAL', u'T1_US_FNAL', u'T2_BE_IIHE', u'T2_BE_UCL', u'T2_BR_SPRACE', u'T2_BR_UERJ', u'T2_CH_CERN', u'T2_CH_CERN_HLT', u'T2_CH_CSCS', u'T2_DE_DESY', u'T2_DE_RWTH', u'T2_EE_Estonia', u'T2_ES_CIEMAT', u'T2_FR_CCIN2P3', u'T2_FR_GRIF_IRFU', u'T2_FR_GRIF_LLR', u'T2_FR_IPHC', u'T2_IT_Bari', u'T2_IT_Legnaro', u'T2_IT_Rome', u'T2_KR_KISTI', u'T2_RU_JINR', u'T2_UA_KIPT', u'T2_UK_London_Brunel', u'T2_UK_London_IC', u'T2_UK_SGrid_RALPP', u'T2_US_Caltech', u'T2_US_Florida', u'T2_US_MIT', u'T2_US_Nebraska', u'T2_US_Purdue', u'T2_US_UCSD', u'T2_US_Wisconsin']
```

Changed files:
src/python/WMCore/MicroService/Unified/MSTransferor.py, src/python/WMCore/Services/CRIC/CRIC.py, test/python/WMCore_t/Services_t/CRIC_t/CRIC_t.py, test/python/WMCore_t/MicroService_t/Unified_t/MSTransferor_t.py

diff --git a/src/python/WMCore/MicroService/Unified/MSTransferor.py b/src/python/WMCore/MicroService/Unified/MSTransferor.py
index 54b05c9d9..d10949fb4 100644
--- a/src/python/WMCore/MicroService/Unified/MSTransferor.py
+++ b/src/python/WMCore/MicroService/Unified/MSTransferor.py
@@ -98,6 +98,7 @@ class MSTransferor(MSCore):
         self.uConfig = {}
         self.campaigns = {}
         self.psn2pnnMap = {}
+        self.pnn2psnMap = {}
         self.dsetCounter = 0
         self.blockCounter = 0
         self.emailAlert = EmailAlert(self.msConfig)
@@ -125,6 +126,7 @@ class MSTransferor(MSCore):
         self.uConfig = self.unifiedConfig()
         campaigns = self.reqmgrAux.getCampaignConfig("ALL_DOCS")
         self.psn2pnnMap = self.cric.PSNtoPNNMap()
+        self.pnn2psnMap = self.cric.PNNtoPSNMap()
         if not self.uConfig:
             raise RuntimeWarning("Failed to fetch the unified configuration")
         elif not campaigns:
@@ -288,7 +290,7 @@ class MSTransferor(MSCore):
         if not wflow.getInputDataset():
             return
 
-        wflowPnns = self._getPNNs(wflow.getSitelist())
+        wflowPnns = self._getPNNsFromPSNs(wflow.getSitelist())
         primaryAAA = wflow.getReqParam("TrustSitelists")
         msg = "Checking data location for request: %s, TrustSitelists: %s, request white/black list PNNs: %s"
         self.logger.info(msg, wflow.getName(), primaryAAA, wflowPnns)
@@ -406,7 +408,7 @@ class MSTransferor(MSCore):
             # nothing to be done here
             return
 
-        wflowPnns = self._getPNNs(wflow.getSitelist())
+        wflowPnns = self._getPNNsFromPSNs(wflow.getSitelist())
         secondaryAAA = wflow.getReqParam("TrustPUSitelists")
         msg = "Checking secondary data location for request: {}, ".format(wflow.getName())
         msg += "TrustPUSitelists: {}, request white/black list PNNs: {}".format(secondaryAAA, wflowPnns)
@@ -520,23 +522,25 @@ class MSTransferor(MSCore):
                 commonPsns = set()
                 # if the dataset has a location list, use solely that one
                 if campConfig['Secondaries'].get(dsetName, []):
-                    commonPsns = set(psns) & set(campConfig['Secondaries'][dsetName])
+                    campSecPSNs = self._getPSNsFromPNNs(campConfig['Secondaries'][dsetName])
+                    commonPsns = set(psns) & campSecPSNs
                     if not commonPsns:
                         msg = "Workflow has been incorrectly assigned: %s. The secondary dataset: %s,"
                         msg += "belongs to the campaign: %s, with Secondaries location set to: %s. "
                         msg += "While the workflow has been assigned to: %s"
                         self.logger.error(msg, wflow.getName(), dsetName, dataIn['campaign'],
-                                          campConfig['Secondaries'][dsetName], psns)
+                                          campSecPSNs, psns)
                 else:
                     if dsetName.startswith("/Neutrino"):
                         # different PU type use different campaign attributes...
-                        commonPsns = set(psns) & set(campConfig['SecondaryLocation'])
+                        campSecPSNs = self._getPSNsFromPNNs(campConfig['SecondaryLocation'])
+                        commonPsns = set(psns) & campSecPSNs
                         if not commonPsns:
                             msg = "Workflow has been incorrectly assigned: %s. The secondary dataset: %s,"
                             msg += "belongs to the campaign: %s, with SecondaryLocation set to: %s. "
                             msg += "While the workflow has been assigned to: %s"
                             self.logger.error(msg, wflow.getName(), dsetName, dataIn['campaign'],
-                                              campConfig['SecondaryLocation'], psns)
+                                              campSecPSNs, psns)
                     else:
                         if campConfig['SiteWhiteList']:
                             commonPsns = set(psns) & set(campConfig['SiteWhiteList'])
@@ -552,7 +556,7 @@ class MSTransferor(MSCore):
                     # returns an empty set, which will make this workflow to be skipped for the moment
                     return commonPsns
 
-        pnns = self._getPNNs(commonPsns)
+        pnns = self._getPNNsFromPSNs(commonPsns)
         self.logger.info("  found a PSN list: %s, which maps to a list of PNNs: %s", commonPsns, pnns)
         return pnns
 
@@ -768,7 +772,7 @@ class MSTransferor(MSCore):
                 psns = set(psns) - set(campConfig['SiteBlackList'])
 
         self.logger.info("  final list of PSNs to be use: %s", psns)
-        pnns = self._getPNNs(psns)
+        pnns = self._getPNNsFromPSNs(psns)
 
         self.logger.info("List of out-of-space RSEs dropped for '%s' is: %s",
                          wflow.getName(), pnns & self.rseQuotas.getOutOfSpaceRSEs())
@@ -831,9 +835,10 @@ class MSTransferor(MSCore):
         self.logger.error("Failed to create transfer document in CouchDB. Will retry again later.")
         return False
 
-    def _getPNNs(self, psnList):
+    def _getPNNsFromPSNs(self, psnList):
         """
-        Given a list/set of PSNs, return a set of valid PNNs
+        Given a list/set of PSNs, return a set of valid PNNs.
+        Note that T3, Tape and a few other PNNs are never returned.
         """
         pnns = set()
         for psn in psnList:
@@ -846,6 +851,20 @@ class MSTransferor(MSCore):
                     pnns.add(pnn)
         return pnns
 
+    def _getPSNsFromPNNs(self, pnnList):
+        """
+        Given a list/set of PNNs, return a set of valid PSNs.
+        Note that T3 sites are never returned.
+        """
+        psns = set()
+        for pnn in pnnList:
+            for psn in self.pnn2psnMap.get(pnn, []):
+                if psn.startswith("T3_"):
+                    pass
+                else:
+                    psns.add(psn)
+        return psns
+
     def _diskPNNs(self, pnnList):
         """
         Provided a list of PNN locations, return another list of
diff --git a/src/python/WMCore/Services/CRIC/CRIC.py b/src/python/WMCore/Services/CRIC/CRIC.py
index 67087cb4a..63cd0023f 100644
--- a/src/python/WMCore/Services/CRIC/CRIC.py
+++ b/src/python/WMCore/Services/CRIC/CRIC.py
@@ -202,7 +202,7 @@ class CRIC(Service):
         """
         Given a PSN regex pattern, return a map of PSN to PNNs
         :param psnPattern: a pattern string
-        :return: a dictionary of PNNs list keyed by their PSN
+        :return: a dictionary key'ed by PSN names, with sets of PNNs as values
         """
         if not isinstance(psnPattern, basestring):
             raise TypeError('psnPattern argument must be of type basestring')
@@ -215,3 +215,21 @@ class CRIC(Service):
             if psnPattern.match(entry['psn_name']):
                 mapping.setdefault(entry['psn_name'], set()).add(entry['phedex_name'])
         return mapping
+
+    def PNNtoPSNMap(self, pnnPattern=''):
+        """
+        Given a PNN regex pattern, return a map of PNN to PSNs
+        :param pnnPattern: a pattern string
+        :return: a dictionary key'ed by PNN names, with sets of PSNs as values
+        """
+        if not isinstance(pnnPattern, basestring):
+            raise TypeError('pnnPattern argument must be of type basestring')
+
+        results = self._CRICSiteQuery(callname='data-processing')
+        mapping = {}
+
+        pnnPattern = re.compile(pnnPattern)
+        for entry in results:
+            if pnnPattern.match(entry['phedex_name']):
+                mapping.setdefault(entry['phedex_name'], set()).add(entry['psn_name'])
+        return mapping
diff --git a/test/python/WMCore_t/MicroService_t/Unified_t/MSTransferor_t.py b/test/python/WMCore_t/MicroService_t/Unified_t/MSTransferor_t.py
index 7894bcbeb..857caecfe 100644
--- a/test/python/WMCore_t/MicroService_t/Unified_t/MSTransferor_t.py
+++ b/test/python/WMCore_t/MicroService_t/Unified_t/MSTransferor_t.py
@@ -28,13 +28,19 @@ class TransferorTest(EmulatedUnitTestCase):
 
     def setUp(self):
         "init test class"
-        self.msConfig = {'verbose': False,
-                         'group': 'DataOps',
+        self.msConfig = {'services': ['transferor'],
+                         'verbose': False,
                          'interval': 1 * 60,
                          'enableStatusTransition': True,
-                         'reqmgrUrl': 'https://cmsweb-testbed.cern.ch/reqmgr2',
+                         'enableDataTransfer': False,
+                         'reqmgr2Url': 'https://cmsweb-testbed.cern.ch/reqmgr2',
                          'reqmgrCacheUrl': 'https://cmsweb-testbed.cern.ch/couchdb/reqmgr_workload_cache',
-                         'phedexUrl': 'https://cmsweb-testbed.cern.ch/phedex/datasvc/json/prod',
+                         'quotaUsage': 0.9,
+                         'rucioAccount': 'wma_test',  # it should be wmcore_transferor
+                         # 'rucioAuthUrl': 'https://cms-rucio-auth.cern.ch',
+                         # 'rucioUrl': 'http://cms-rucio.cern.ch',
+                         'rucioAuthUrl': 'https://cmsrucio-auth-int.cern.ch',
+                         'rucioUrl': 'http://cmsrucio-int.cern.ch',
                          'dbsUrl': 'https://cmsweb-testbed.cern.ch/dbs/int/global/DBSReader'}
 
         self.msTransferor = MSTransferor(self.msConfig)
@@ -43,6 +49,63 @@ class TransferorTest(EmulatedUnitTestCase):
         self.stepChainTempl = getTestFile('data/ReqMgr/requests/Integration/SC_LumiMask_PhEDEx.json')
         super(TransferorTest, self).setUp()
 
+    def testGetPNNsFromPSNs(self):
+        """Test MSTransferor private method _getPNNsFromPSNs()"""
+        self.assertItemsEqual(self.msTransferor.psn2pnnMap, {})
+
+        # now fill up the cache
+        self.msTransferor.psn2pnnMap = self.msTransferor.cric.PSNtoPNNMap()
+
+        self.assertItemsEqual(self.msTransferor._getPNNsFromPSNs([]), set())
+        pnns = self.msTransferor._getPNNsFromPSNs(["T1_IT_CNAF", "T1_IT_CNAF_Disk"])
+        self.assertItemsEqual(pnns, set(["T1_IT_CNAF_Disk"]))
+
+        # dropping T3s and CERNBOX
+        pnns = self.msTransferor._getPNNsFromPSNs(["T1_US_FNAL", "T2_CH_CERN_HLT"])
+        self.assertItemsEqual(pnns, set(["T1_US_FNAL_Disk", "T2_CH_CERN"]))
+
+        # testing with non-existant PSNs
+        psns = self.msTransferor._getPNNsFromPSNs(["T1_US_FNAL_Disk", "T2_CH_CERNBOX"])
+        self.assertItemsEqual(psns, set())
+
+    def testGetPSNsFromPNNs(self):
+        """Test MSTransferor private method _getPSNsFromPNNs()"""
+        self.assertItemsEqual(self.msTransferor.pnn2psnMap, {})
+
+        # now fill up the cache
+        self.msTransferor.pnn2psnMap = self.msTransferor.cric.PNNtoPSNMap()
+
+        self.assertItemsEqual(self.msTransferor._getPSNsFromPNNs([]), set())
+        psns = self.msTransferor._getPSNsFromPNNs(["T1_IT_CNAF", "T1_IT_CNAF_Disk"])
+        self.assertItemsEqual(psns, set(["T1_IT_CNAF"]))
+
+        # test dropping T3s
+        psns = self.msTransferor._getPSNsFromPNNs(["T2_UK_SGrid_RALPP"])
+        self.assertItemsEqual(psns, set(["T2_UK_SGrid_RALPP"]))
+
+        # testing with non-existant PNNs
+        psns = self.msTransferor._getPSNsFromPNNs(["T1_US_FNAL", "T2_CH_CERN_HLT"])
+        self.assertItemsEqual(psns, set())
+
+    def testDiskPNNs(self):
+        """Test MSTransferor private method _diskPNNs()"""
+        # empty list of pnns
+        self.assertItemsEqual(self.msTransferor._diskPNNs([]), set())
+
+        # only PNNs that will be dropped
+        pnns = self.msTransferor._diskPNNs(["T1_US_FNAL_Tape", "T1_US_FNAL_MSS",
+                                            "T2_CH_CERNBOX", "T0_CH_CERN_Export"])
+        self.assertItemsEqual(pnns, set())
+
+        # valid PNNs that can receive data
+        pnns = self.msTransferor._diskPNNs(["T1_US_FNAL_Disk", "T2_CH_CERN", "T2_DE_DESY"])
+        self.assertItemsEqual(pnns, set(["T1_US_FNAL_Disk", "T2_CH_CERN", "T2_DE_DESY"]))
+
+        # finally, a mix of valid and invalid PNNs
+        pnns = self.msTransferor._diskPNNs(["T1_US_FNAL_Disk", "T1_US_FNAL_MSS", "T1_US_FNAL_Tape",
+                                            "T2_CH_CERN", "T2_DE_DESY"])
+        self.assertItemsEqual(pnns, set(["T1_US_FNAL_Disk", "T2_CH_CERN", "T2_DE_DESY"]))
+
     def notestRequestRecord(self):
         """
         Test the requestRecord method
diff --git a/test/python/WMCore_t/Services_t/CRIC_t/CRIC_t.py b/test/python/WMCore_t/Services_t/CRIC_t/CRIC_t.py
index 27a00a3df..0d88a7785 100644
--- a/test/python/WMCore_t/Services_t/CRIC_t/CRIC_t.py
+++ b/test/python/WMCore_t/Services_t/CRIC_t/CRIC_t.py
@@ -239,6 +239,55 @@ class CRICTest(EmulatedUnitTestCase):
 
         return
 
+    def testPNNtoPSNMap(self):
+        """
+        Test API to get a map of PSNs to PNNs
+        """
+        with self.assertRaises(TypeError):
+            self.myCRIC.PNNtoPSNMap(1)
+        with self.assertRaises(TypeError):
+            self.myCRIC.PNNtoPSNMap({'config': 'blah'})
+        with self.assertRaises(TypeError):
+            self.myCRIC.PNNtoPSNMap(['blah'])
+
+        result = self.myCRIC.PNNtoPSNMap()
+        self.assertTrue([psn for psn in result.keys() if psn.startswith('T1_')])
+        self.assertTrue([psn for psn in result.keys() if psn.startswith('T2_')])
+        self.assertTrue([psn for psn in result.keys() if psn.startswith('T3_')])
+        self.assertTrue(len(result) > 50)
+
+        result = self.myCRIC.PNNtoPSNMap(pnnPattern='T1.*')
+        self.assertFalse([psn for psn in result.keys() if not psn.startswith('T1_')])
+        self.assertTrue(len(result) < 10)
+
+        result = self.myCRIC.PNNtoPSNMap(pnnPattern='T2.*')
+        self.assertFalse([psn for psn in result.keys() if not psn.startswith('T2_')])
+        self.assertTrue(len(result) > 10)
+
+        result = self.myCRIC.PNNtoPSNMap(pnnPattern='T3.*')
+        self.assertFalse([psn for psn in result.keys() if not psn.startswith('T3_')])
+        self.assertTrue(len(result) > 10)
+
+        result = self.myCRIC.PNNtoPSNMap('T2_CH_CERN$')
+        self.assertItemsEqual(result.keys(), ['T2_CH_CERN'])
+        self.assertItemsEqual(result['T2_CH_CERN'], ['T2_CH_CERN_HLT', 'T2_CH_CERN'])
+
+        # test a exact site name, which is treated as a regex and yields a confusing result!!!
+        result = self.myCRIC.PNNtoPSNMap('T2_CH_CERN')
+        self.assertItemsEqual(result.keys(), ['T2_CH_CERN', 'T2_CH_CERNBOX'])
+        self.assertItemsEqual(result['T2_CH_CERN'], ['T2_CH_CERN_HLT', 'T2_CH_CERN'])
+        self.assertItemsEqual(result['T2_CH_CERNBOX'], ['T2_CH_CERN'])
+
+        result = self.myCRIC.PNNtoPSNMap('T2_CH_CERN_HLT')
+        self.assertItemsEqual(result, {})
+
+        result = self.myCRIC.PNNtoPSNMap('T1_US_FNAL')
+        self.assertItemsEqual(result.keys(), ['T1_US_FNAL_Disk'])
+        self.assertItemsEqual(result['T1_US_FNAL_Disk'], ['T1_US_FNAL'])
+
+        return
+
+
 
 if __name__ == '__main__':
     unittest.main()