MSTransferor: use different rucio account for RelVal input data transfers

src/python/WMCore/MicroService/DataStructs/DefaultStructs.py

@@ -28,6 +28,7 @@
                          total_num_requests=0,
                          total_num_campaigns=0,
                          nodes_out_of_space=None,
+                         relval_nodes_out_of_space=None,
                          success_request_transition=0,
                          failed_request_transition=0,
                          problematic_requests=0,

src/python/WMCore/MicroService/MSTransferor/MSTransferor.py

@@ -84,10 +84,16 @@ def __init__(self, msConfig, logger=None):
         self.msConfig.setdefault("rucioRuleWeight", 'ddm_quota')
 
         quotaAccount = self.msConfig["rucioAccount"]
-
         self.rseQuotas = RSEQuotas(quotaAccount, self.msConfig["quotaUsage"],
                                    minimumThreshold=self.msConfig["minimumThreshold"],
                                    verbose=self.msConfig['verbose'], logger=logger)
+
+        # FIXME TODO: remove the setdefault once deployment is properly updated
+        quotaAccountRelVal = self.msConfig.setdefault("rucioAccountRelVal", "wmcore_transferor_relval")
+        self.rseQuotasRelVal = RSEQuotas(quotaAccountRelVal, self.msConfig["quotaUsage"],
+                                         minimumThreshold=self.msConfig["minimumThreshold"],
+                                         verbose=self.msConfig['verbose'], logger=logger)
+
         self.reqInfo = RequestInfo(self.msConfig, self.rucio, self.logger)
 
         self.cric = CRIC(logger=self.logger)
@@ -109,19 +115,12 @@ def __init__(self, msConfig, logger=None):
     def updateCaches(self):
         """
         Fetch some data required for the transferor logic, e.g.:
-         * account limits from Rucio
-         * account usage from Rucio
          * unified configuration
          * all campaign configuration
          * PSN to PNN map from CRIC
+         * account limits from Rucio
+         * account usage from Rucio
         """
-        self.logger.info("Updating RSE/PNN quota and usage")
-        self.rseQuotas.fetchStorageQuota(self.rucio)
-        self.rseQuotas.fetchStorageUsage(self.rucio)
-        self.rseQuotas.evaluateQuotaExceeded()
-        if not self.rseQuotas.getNodeUsage():
-            raise RuntimeWarning("Failed to fetch storage usage stats")
-
         self.logger.info("Updating all local caches...")
         self.dsetCounter = 0
         self.blockCounter = 0
@@ -140,7 +139,15 @@ def updateCaches(self):
             self.campaigns = {}
             for camp in campaigns:
                 self.campaigns[camp['CampaignName']] = camp
-        self.rseQuotas.printQuotaSummary()
+        for clsObject in (self.rseQuotas, self.rseQuotasRelVal):
+            self.logger.info("Updating RSE quota/usage for Rucio account: %s", clsObject.dataAcct)
+            getattr(clsObject, "fetchStorageQuota")(self.rucio)
+            getattr(clsObject, "fetchStorageUsage")(self.rucio)
+            getattr(clsObject, "evaluateQuotaExceeded")()
+            getattr(clsObject, "printQuotaSummary")()
+            if not getattr(clsObject, "getNodeUsage")():
+                msg = "Failed to fetch storage usage stats for account: {}".format(clsObject.dataAcct)
+                raise RuntimeWarning(msg)
 
     def execute(self, reqStatus):
         """
@@ -169,6 +176,7 @@ def execute(self, reqStatus):
             self.updateCaches()
             self.updateReportDict(summary, "total_num_campaigns", len(self.campaigns))
             self.updateReportDict(summary, "nodes_out_of_space", list(self.rseQuotas.getOutOfSpaceRSEs()))
+            self.updateReportDict(summary, "relval_nodes_out_of_space", list(self.rseQuotasRelVal.getOutOfSpaceRSEs()))
         except RuntimeWarning as ex:
             msg = "All retries exhausted! Last error was: '%s'" % str(ex)
             msg += "\nRetrying to update caches again in the next cycle."
@@ -217,7 +225,8 @@ def execute(self, reqStatus):
                     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
+                                     wflow.getName(), pformat(transfers))
+                    # then create a document in ReqMgr Aux DB
                     if self.createTransferDoc(wflow.getName(), transfers):
                         self.logger.info("Transfer document successfully created in CouchDB for: %s", wflow.getName())
                         # then move this request to staging status
@@ -247,6 +256,7 @@ def execute(self, reqStatus):
         self.updateReportDict(summary, "num_datasets_subscribed", self.dsetCounter)
         self.updateReportDict(summary, "num_blocks_subscribed", self.blockCounter)
         self.updateReportDict(summary, "nodes_out_of_space", list(self.rseQuotas.getOutOfSpaceRSEs()))
+        self.updateReportDict(summary, "relval_nodes_out_of_space", list(self.rseQuotasRelVal.getOutOfSpaceRSEs()))
         return summary
 
     def getRequestRecords(self, reqStatus):
@@ -591,6 +601,7 @@ def makeTransferRequest(self, wflow):
 
         self.logger.info("Handling data subscriptions for request: %s", wflow.getName())
 
+        rseQuotaObject = self.rseQuotasRelVal if wflow.isRelVal() else self.rseQuotas
         for dataIn in wflow.getDataCampaignMap():
             if dataIn["type"] == "parent":
                 msg = "Skipping 'parent' data subscription (done with the 'primary' data), for: %s" % dataIn
@@ -600,9 +611,9 @@ def makeTransferRequest(self, wflow):
                 # secondary already in place
                 continue
 
-            if wflow.getPURSElist() and not wflow.isRelVal():
+            if wflow.getPURSElist():
                 # then the whole workflow is very much limited to a single site
-                nodes = list(wflow.getPURSElist() & self.rseQuotas.getAvailableRSEs())
+                nodes = list(wflow.getPURSElist() & rseQuotaObject.getAvailableRSEs())
                 if not nodes:
                     msg = "Workflow: %s can only run in RSEs with no available space: %s. "
                     msg += "Skipping this workflow until space gets released"
@@ -641,7 +652,7 @@ def makeTransferRequest(self, wflow):
                         transRec['transferIDs'].update(transferId)
                     else:
                         transRec['transferIDs'].add(transferId)
-                    self.rseQuotas.updateNodeUsage(nodes[idx], dataSize)
+                    rseQuotaObject.updateNodeUsage(nodes[idx], dataSize)
 
                 # and update some instance caches
                 if subLevel == 'container':
@@ -653,7 +664,7 @@ def makeTransferRequest(self, wflow):
             response.append(transRec)
 
         # once the workflow has been completely processed, update the node usage
-        self.rseQuotas.evaluateQuotaExceeded()
+        rseQuotaObject.evaluateQuotaExceeded()
         return success, response
 
     def makeTransferRucio(self, wflow, dataIn, subLevel, blocks, dataSize, nodes, nodeIdx):
@@ -672,11 +683,12 @@ def makeTransferRucio(self, wflow, dataIn, subLevel, blocks, dataSize, nodes, no
         success, transferId = True, set()
         subLevel = "ALL" if subLevel == "container" else "DATASET"
         dids = blocks if blocks else [dataIn['name']]
+        rucioAcct = self.msConfig['rucioAccountRelVal'] if wflow.isRelVal() else self.msConfig['rucioAccount']
 
         ruleAttrs = {'copies': 1,
                      'activity': 'Production Input',
                      'lifetime': self.msConfig['rulesLifetime'],
-                     'account': self.msConfig['rucioAccount'],
+                     'account': rucioAcct,
                      'grouping': subLevel,
                      'weight': self.msConfig['rucioRuleWeight'],
                      'meta': {'workflow_group': wflow.getWorkflowGroup()},
@@ -788,14 +800,10 @@ def _getValidSites(self, wflow, dataIn):
         self.logger.info("  final list of PSNs to be use: %s", psns)
         pnns = self._getPNNsFromPSNs(psns)
 
-        if wflow.isRelVal():
-            self.logger.info("RelVal workflow '%s' ignores sites out of quota", wflow.getName())
-            return list(pnns)
-
+        rseQuotaObject = self.rseQuotasRelVal if wflow.isRelVal() else self.rseQuotas
         self.logger.info("List of out-of-space RSEs dropped for '%s' is: %s",
-                         wflow.getName(), pnns & self.rseQuotas.getOutOfSpaceRSEs())
-        return list(pnns & self.rseQuotas.getAvailableRSEs())
-
+                         wflow.getName(), pnns & rseQuotaObject.getOutOfSpaceRSEs())
+        return list(pnns & rseQuotaObject.getAvailableRSEs())
 
     def _decideDataDestination(self, wflow, dataIn, numNodes):
         """

src/python/WMCore/MicroService/MSTransferor/RSEQuotas.py

@@ -145,7 +145,7 @@ def printQuotaSummary(self):
         """
         Print a summary of the current quotas, space usage and space available
         """
-        self.logger.info("Summary of the current quotas in Terabytes:")
+        self.logger.info("Current quotas in Terabytes (account: %s):", self.dataAcct)
         for node in sorted(self.nodeUsage.keys()):
             msg = "  %s:\t\tbytes_limit: %.2f, bytes_used: %.2f, bytes_remaining: %.2f, "
             msg += "quota: %.2f, quota_avail: %.2f"