diff --git a/src/python/WMCore/MicroService/DataStructs/DefaultStructs.py b/src/python/WMCore/MicroService/DataStructs/DefaultStructs.py index 2b64118731..6773ba0ae4 100644 --- a/src/python/WMCore/MicroService/DataStructs/DefaultStructs.py +++ b/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, diff --git a/src/python/WMCore/MicroService/MSTransferor/MSTransferor.py b/src/python/WMCore/MicroService/MSTransferor/MSTransferor.py index c0c838093e..07d1de86dd 100644 --- a/src/python/WMCore/MicroService/MSTransferor/MSTransferor.py +++ b/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): """ diff --git a/src/python/WMCore/MicroService/MSTransferor/RSEQuotas.py b/src/python/WMCore/MicroService/MSTransferor/RSEQuotas.py index 71c62408d7..bb7c2c0ccc 100644 --- a/src/python/WMCore/MicroService/MSTransferor/RSEQuotas.py +++ b/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" diff --git a/src/python/WMCore/MicroService/MSTransferor/RequestInfo.py b/src/python/WMCore/MicroService/MSTransferor/RequestInfo.py index ad59d8455c..09b134450d 100644 --- a/src/python/WMCore/MicroService/MSTransferor/RequestInfo.py +++ b/src/python/WMCore/MicroService/MSTransferor/RequestInfo.py @@ -10,18 +10,18 @@ # system modules import datetime -import time # WMCore modules from pprint import pformat from copy import deepcopy from Utils.IteratorTools import grouper +from Utils.Timers import timeFunction from WMCore.DataStructs.LumiList import LumiList from WMCore.MicroService.MSTransferor.Workflow import Workflow -from WMCore.MicroService.Tools.PycurlRucio import (getRucioToken, getPileupContainerSizesRucio, +from WMCore.MicroService.Tools.PycurlRucio import (getRucioToken, getPileupContainerSizes, listReplicationRules, getBlocksAndSizeRucio) -from WMCore.MicroService.Tools.Common import (elapsedTime, findBlockParents, - findParent, getBlocksByDsetAndRun, - getFileLumisInBlock, getRunsInBlock) +from WMCore.MicroService.Tools.Common import (findBlockParents, findParent, + getBlocksByDsetAndRun, getRunsInBlock, + getFileLumisInBlock) from WMCore.MicroService.MSCore import MSCore @@ -67,8 +67,9 @@ def __call__(self, reqRecords): # setup the Rucio token self.setupRucio() - # get complete requests information (based on Unified Transferor logic) - self.unified(workflows) + # Now run the heavy stuff (based on the old Unified Transferor logic) + tSpent, resp, funcName = self.unified(workflows) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) return workflows @@ -90,6 +91,7 @@ def setupRucio(self): self.rucioToken, self.tokenValidity = getRucioToken(self.msConfig['rucioAuthUrl'], self.msConfig['rucioAccount']) + @timeFunction def unified(self, workflows): """ Unified Transferor black box @@ -99,37 +101,34 @@ def unified(self, workflows): # make subscriptions based on site white/black lists self.logger.info("Unified method processing %d requests", len(workflows)) - orig = time.time() # start by finding what are the parent datasets for requests requiring it - time0 = time.time() - parentMap = self.getParentDatasets(workflows) + tSpent, parentMap, funcName = self.getParentDatasets(workflows) self.setParentDatasets(workflows, parentMap) - self.logger.debug(elapsedTime(time0, "### getParentDatasets")) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) - # then check the secondary dataset sizes and locations - time0 = time.time() - sizeByDset, locationByDset = self.getSecondaryDatasets(workflows) - locationByDset = self.resolveSecondaryRSEs(locationByDset) - self.setSecondaryDatasets(workflows, sizeByDset, locationByDset) - self.logger.debug(elapsedTime(time0, "### getSecondaryDatasets")) + # then check the secondary dataset sizes + tSpent, sizeByDset, funcName = self.getSecondarySizes(workflows) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) + + # now check the secondary dataset locations for production workflow, then RelVals + for mode in (False, True): + tSpent, locationByDset, funcName = self.getSecondaryLocations(workflows, relvalMode=mode) + locationByDset = self.resolveSecondaryRSEs(locationByDset) + self.setSecondaryDatasets(workflows, sizeByDset, locationByDset, relvalMode=mode) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) # get final primary and parent list of valid blocks, # considering run, block and lumi lists - time0 = time.time() - blocksByDset = self.getInputDataBlocks(workflows) + tSpent, blocksByDset, funcName = self.getInputDataBlocks(workflows) self.setInputDataBlocks(workflows, blocksByDset) - self.logger.debug(elapsedTime(time0, "### getInputDataBlocks")) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) # get a final list of parent blocks - time0 = time.time() - parentageMap = self.getParentChildBlocks(workflows) + tSpent, parentageMap, funcName = self.getParentChildBlocks(workflows) self.setParentChildBlocks(workflows, parentageMap) - self.logger.debug(elapsedTime(time0, "### getParentChildBlocks")) - self.logger.info(elapsedTime(orig, '### total time for unified method')) + self.logger.debug("%s took %.3f secs to execute", funcName, tSpent) self.logger.info("Unified method successfully processed %d requests", len(workflows)) - return workflows - def _workflowRemoval(self, listOfWorkflows, workflowsToRetry): """ Receives the initial list of workflows and another list of workflows @@ -144,37 +143,31 @@ def _workflowRemoval(self, listOfWorkflows, workflowsToRetry): self.logger.warning("Removing workflow that failed processing in MSTransferor: %s", wflow.getName()) listOfWorkflows.remove(wflow) + @timeFunction def getParentDatasets(self, workflows): """ Given a list of requests, find which requests need to process a parent dataset, and discover what the parent dataset name is. :return: dictionary with the child and the parent dataset """ - retryWorkflows = [] - retryDatasets = [] + mapWflowByData = {} datasetByDbs = {} parentByDset = {} for wflow in workflows: if wflow.hasParents(): + mapWflowByData.setdefault(wflow.getInputDataset(), []) + mapWflowByData[wflow.getInputDataset()].append(wflow.getName()) datasetByDbs.setdefault(wflow.getDbsUrl(), set()) datasetByDbs[wflow.getDbsUrl()].add(wflow.getInputDataset()) for dbsUrl, datasets in viewitems(datasetByDbs): self.logger.info("Resolving %d dataset parentage against DBS: %s", len(datasets), dbsUrl) # first find out what's the parent dataset name - parentByDset.update(findParent(datasets, dbsUrl)) - - # now check if any of our calls failed; if so, workflow needs to be skipped from this cycle - # FIXME: isn't there a better way to do this?!? - for dset, value in viewitems(parentByDset): - if value is None: - retryDatasets.append(dset) - if retryDatasets: - for wflow in workflows: - if wflow.hasParents() and wflow.getInputDataset() in retryDatasets: - retryWorkflows.append(wflow) - # remove workflows that failed one or more of the bulk queries to the data-service - self._workflowRemoval(workflows, retryWorkflows) + parents, erroredDsets = findParent(datasets, dbsUrl) + parentByDset.update(parents) + # now remove any workflows that failed the parent lookup resolution + for inputDset in erroredDsets: + self._workflowRemoval(workflows, mapWflowByData[inputDset]) return parentByDset @@ -186,49 +179,68 @@ def setParentDatasets(self, workflows, parentageMap): if wflow.hasParents() and wflow.getInputDataset() in parentageMap: wflow.setParentDataset(parentageMap[wflow.getInputDataset()]) - def getSecondaryDatasets(self, workflows): + @timeFunction + def getSecondarySizes(self, workflows): """ Given a list of requests, list all the pileup datasets and, find their - total dataset sizes and which locations host completed and subscribed datasets. + total dataset sizes. NOTE it only uses valid blocks (i.e., blocks with at least one replica!) :param workflows: a list of Workflow objects - :return: two dictionaries keyed by the dataset. - First contains dataset size as value. - Second contains a list of locations as value. + :return: flat dictionary with the dataset name and its size as value """ - retryWorkflows = [] - retryDatasets = [] + mapWflowByData = {} datasets = set() for wflow in workflows: - datasets = datasets | wflow.getPileupDatasets() + for pileupDset in wflow.getPileupDatasets(): + mapWflowByData.setdefault(pileupDset, []) + mapWflowByData[pileupDset].append(wflow.getName()) + datasets.add(pileupDset) # retrieve pileup container size and locations from Rucio self.logger.info("Fetching pileup dataset sizes for %d datasets against Rucio: %s", len(datasets), self.msConfig['rucioUrl']) - sizesByDset = getPileupContainerSizesRucio(datasets, self.msConfig['rucioUrl'], self.rucioToken) + sizesByDset, erroredDsets = getPileupContainerSizes(datasets, self.msConfig['rucioUrl'], + self.rucioToken) + # now remove any workflows that failed the parent lookup resolution + for inputDset in erroredDsets: + self._workflowRemoval(workflows, mapWflowByData[inputDset]) + return sizesByDset + + @timeFunction + def getSecondaryLocations(self, workflows, relvalMode): + """ + Given a list of requests, list all the pileup datasets and find RSEs + hosting a complete copy of it (under a given rucio account). + :param workflows: a list of Workflow objects + :param relvalMode: flag to deal only with production or relval workflows + :return: flat dictionary with the dataset name and a list of RSE expressions + """ + mapWflowByData = {} + datasets = set() + for wflow in workflows: + if not relvalMode and wflow.isRelVal(): + # we only want non-relval workflows + continue + if relvalMode and not wflow.isRelVal(): + # we only want relval workflows + continue + for pileupDset in wflow.getPileupDatasets(): + mapWflowByData.setdefault(pileupDset, []) + mapWflowByData[pileupDset].append(wflow.getName()) + datasets.add(pileupDset) # then fetch data location for locked data, under our own rucio account - self.logger.info("Fetching pileup container location for %d containers against Rucio: %s", - len(datasets), self.msConfig['rucioUrl']) - locationsByDset = listReplicationRules(datasets, self.msConfig['rucioAccount'], - grouping="A", rucioUrl=self.msConfig['rucioUrl'], - rucioToken=self.rucioToken) - # now check if any of our calls failed; if so, workflow needs to be skipped from this cycle - # FIXME: isn't there a better way to do this?!? - for dset, value in viewitems(sizesByDset): - if value is None: - retryDatasets.append(dset) - for dset, value in viewitems(locationsByDset): - if value is None: - retryDatasets.append(dset) - if retryDatasets: - for wflow in workflows: - for pileup in wflow.getPileupDatasets(): - if pileup in retryDatasets: - retryWorkflows.append(wflow) - # remove workflows that failed one or more of the bulk queries to the data-service - self._workflowRemoval(workflows, retryWorkflows) - return sizesByDset, locationsByDset + rucioAcct = self.msConfig['rucioAccountRelVal'] if relvalMode else self.msConfig['rucioAccount'] + msg = "Resolving pileup container location for {} containers, ".format(len(datasets)) + msg += "under Rucio account: {}, and Rucio url: {}".format(rucioAcct, self.msConfig['rucioUrl']) + self.logger.info(msg) + locationsByDset, erroredDsets = listReplicationRules(datasets, rucioAcct, grouping="A", + rucioUrl=self.msConfig['rucioUrl'], + rucioToken=self.rucioToken) + # now remove any workflows that failed the parent lookup resolution + for inputDset in erroredDsets: + self._workflowRemoval(workflows, mapWflowByData[inputDset]) + return locationsByDset def resolveSecondaryRSEs(self, rsesByContainer): """ @@ -247,15 +259,27 @@ def resolveSecondaryRSEs(self, rsesByContainer): rsesByContainer[contName] = list(set(rseNames)) return rsesByContainer - def setSecondaryDatasets(self, workflows, sizesByDset, locationsByDset): + def setSecondaryDatasets(self, workflows, sizesByDset, locationsByDset, relvalMode): """ Given dictionaries with the pileup dataset size and locations, set the workflow object accordingly. + :param workflows: list of workflow objects + :param sizesByDset: flat dictionary with containers and their sizes + :param locationsByDset: flat dictionary with containers and their location + :param relvalMode: flag to deal only with production or relval workflows + :return: none """ for wflow in workflows: + if not relvalMode and wflow.isRelVal(): + # we only want non-relval workflows + continue + if relvalMode and not wflow.isRelVal(): + # we only want relval workflows + continue for dsetName in wflow.getPileupDatasets(): wflow.setSecondarySummary(dsetName, sizesByDset[dsetName], locationsByDset[dsetName]) + @timeFunction def getInputDataBlocks(self, workflows): """ Given a list of requests, list all the primary and parent datasets and, find @@ -264,30 +288,24 @@ def getInputDataBlocks(self, workflows): :param workflows: a list of Workflow objects :return: dictionary with dataset and a few block information """ - retryWorkflows = [] - retryDatasets = [] + mapWflowByData = {} datasets = set() for wflow in workflows: for dataIn in wflow.getDataCampaignMap(): if dataIn['type'] in ["primary", "parent"]: + mapWflowByData.setdefault(dataIn['name'], []) + mapWflowByData[dataIn['name']].append(wflow.getName()) datasets.add(dataIn['name']) # fetch all block names and their sizes from Rucio self.logger.info("Fetching parent/primary block sizes for %d containers against Rucio: %s", len(datasets), self.msConfig['rucioUrl']) - blocksByDset = getBlocksAndSizeRucio(datasets, self.msConfig['rucioUrl'], self.rucioToken) - - # now check if any of our calls failed; if so, workflow needs to be skipped from this cycle - # FIXME: isn't there a better way to do this?!? - for dsetName in blocksByDset: - if blocksByDset[dsetName] is None: - retryDatasets.append(dsetName) - if retryDatasets: - for wflow in workflows: - if wflow.getInputDataset() in retryDatasets or wflow.getParentDataset() in retryDatasets: - retryWorkflows.append(wflow) - # remove workflows that failed one or more of the bulk queries to the data-service - self._workflowRemoval(workflows, retryWorkflows) + blocksByDset, erroredDsets = getBlocksAndSizeRucio(datasets, self.msConfig['rucioUrl'], + self.rucioToken) + + # now remove any workflows that failed the parent lookup resolution + for inputDset in erroredDsets: + self._workflowRemoval(workflows, mapWflowByData[inputDset]) return blocksByDset def setInputDataBlocks(self, workflows, blocksByDset): @@ -433,6 +451,7 @@ def _removeZeroSizeBlocks(self, blocksDict): blockName, blocksDict[blockName]['blockSize']) return finalBlocks + @timeFunction def getParentChildBlocks(self, workflows): """ Given a list of requests, get their children block, discover their parent blocks @@ -440,13 +459,14 @@ def getParentChildBlocks(self, workflows): :param workflows: list of workflow objects :return: nothing, updates the workflow attributes in place """ - retryWorkflows = [] - retryDatasets = [] + mapWflowByData = {} blocksByDbs = {} parentageMap = {} for wflow in workflows: blocksByDbs.setdefault(wflow.getDbsUrl(), set()) if wflow.getParentDataset(): + mapWflowByData.setdefault(wflow.getParentDataset(), []) + mapWflowByData[wflow.getParentDataset()].append(wflow.getName()) blocksByDbs[wflow.getDbsUrl()] = blocksByDbs[wflow.getDbsUrl()] | set(wflow.getPrimaryBlocks().keys()) for dbsUrl, blocks in viewitems(blocksByDbs): @@ -454,19 +474,12 @@ def getParentChildBlocks(self, workflows): continue self.logger.debug("Fetching DBS parent blocks for %d children blocks...", len(blocks)) # first find out what's the parent dataset name - parentageMap.update(findBlockParents(blocks, dbsUrl)) - - # now check if any of our calls failed; if so, workflow needs to be skipped from this cycle - # FIXME: isn't there a better way to do this?!? - for dset, value in viewitems(parentageMap): - if value is None: - retryDatasets.append(dset) - if retryDatasets: - for wflow in workflows: - if wflow.getParentDataset() in retryDatasets: - retryWorkflows.append(wflow) - # remove workflows that failed one or more of the bulk queries to the data-service - self._workflowRemoval(workflows, retryWorkflows) + parentBlocks, erroredDsets = findBlockParents(blocks, dbsUrl) + parentageMap.update(parentBlocks) + # now remove any workflows that failed the parent lookup resolution + for inputDset in erroredDsets: + self._workflowRemoval(workflows, mapWflowByData[inputDset]) + return parentageMap def setParentChildBlocks(self, workflows, parentageMap): diff --git a/src/python/WMCore/MicroService/Tools/Common.py b/src/python/WMCore/MicroService/Tools/Common.py index 953e764238..60dd2c3a9f 100644 --- a/src/python/WMCore/MicroService/Tools/Common.py +++ b/src/python/WMCore/MicroService/Tools/Common.py @@ -80,141 +80,6 @@ def dbsInfo(datasets, dbsUrl): return datasetBlocks, datasetSizes, datasetTransfers -def getPileupDatasetSizes(datasets, phedexUrl): - """ - Given a list of datasets, find all their blocks with replicas - available, i.e., blocks that have valid files to be processed, - and calculate the total dataset size - :param datasets: list of dataset names - :param phedexUrl: a string with the PhEDEx URL - :return: a dictionary of datasets and their respective sizes - NOTE: Value `None` is returned in case the data-service failed to serve a given request. - """ - sizeByDset = {} - if not datasets: - return sizeByDset - - urls = ['%s/blockreplicas?dataset=%s' % (phedexUrl, dset) for dset in datasets] - logging.info("Executing %d requests against PhEDEx 'blockreplicas' API", len(urls)) - data = multi_getdata(urls, ckey(), cert()) - - for row in data: - dataset = row['url'].split('=')[-1] - if row['data'] is None: - print("Failure in getPileupDatasetSizes for dataset %s. Error: %s %s" % (dataset, - row.get('code'), - row.get('error'))) - sizeByDset.setdefault(dataset, None) - continue - rows = json.loads(row['data']) - sizeByDset.setdefault(dataset, 0) - try: - for item in rows['phedex']['block']: - sizeByDset[dataset] += item['bytes'] - except Exception as exc: - print("Failure in getPileupDatasetSizes for dataset %s. Error: %s" % (dataset, str(exc))) - sizeByDset[dataset] = None - return sizeByDset - - -def getBlockReplicasAndSize(datasets, phedexUrl, group=None): - """ - Given a list of datasets, find all their blocks with replicas - available (thus blocks with at least 1 valid file), completed - and subscribed. - If PhEDEx group is provided, make sure it's subscribed under that - same group. - :param datasets: list of dataset names - :param phedexUrl: a string with the PhEDEx URL - :param group: optional PhEDEx group name - :return: a dictionary in the form of: - {"dataset": - {"block": - {"blockSize": 111, "locations": ["x", "y"]} - } - } - NOTE: Value `None` is returned in case the data-service failed to serve a given request. - """ - dsetBlockSize = {} - if not datasets: - return dsetBlockSize - - urls = ['%s/blockreplicas?dataset=%s' % (phedexUrl, dset) for dset in datasets] - logging.info("Executing %d requests against PhEDEx 'blockreplicas' API", len(urls)) - data = multi_getdata(urls, ckey(), cert()) - - for row in data: - dataset = row['url'].split('=')[-1] - if row['data'] is None: - print("Failure in getBlockReplicasAndSize for dataset %s. Error: %s %s" % (dataset, - row.get('code'), - row.get('error'))) - dsetBlockSize.setdefault(dataset, None) - continue - rows = json.loads(row['data']) - dsetBlockSize.setdefault(dataset, {}) - try: - for item in rows['phedex']['block']: - block = {item['name']: {'blockSize': item['bytes'], 'locations': []}} - for repli in item['replica']: - if repli['complete'] == 'y' and repli['subscribed'] == 'y': - if not group: - block[item['name']]['locations'].append(repli['node']) - elif repli['group'] == group: - block[item['name']]['locations'].append(repli['node']) - dsetBlockSize[dataset].update(block) - except Exception as exc: - print("Failure in getBlockReplicasAndSize for dataset %s. Error: %s" % (dataset, str(exc))) - dsetBlockSize[dataset] = None - return dsetBlockSize - - -def getPileupSubscriptions(datasets, phedexUrl, group=None, percentMin=99): - """ - Provided a list of datasets, find dataset level subscriptions where it's - as complete as `percent_min`. - :param datasets: list of dataset names - :param phedexUrl: a string with the PhEDEx URL - :param group: optional string with the PhEDEx group - :param percent_min: only return subscriptions that are this complete - :return: a dictionary of datasets and a list of their location. - NOTE: Value `None` is returned in case the data-service failed to serve a given request. - """ - locationByDset = {} - if not datasets: - return locationByDset - - if group: - url = "%s/subscriptions?group=%s" % (phedexUrl, group) - url += "&percent_min=%s&dataset=%s" - else: - url = "%s/subscriptions?" % phedexUrl - url += "percent_min=%s&dataset=%s" - urls = [url % (percentMin, dset) for dset in datasets] - - logging.info("Executing %d requests against PhEDEx 'subscriptions' API", len(urls)) - data = multi_getdata(urls, ckey(), cert()) - - for row in data: - dataset = row['url'].rsplit('=')[-1] - if row['data'] is None: - print("Failure in getPileupSubscriptions for dataset %s. Error: %s %s" % (dataset, - row.get('code'), - row.get('error'))) - locationByDset.setdefault(dataset, None) - continue - rows = json.loads(row['data']) - locationByDset.setdefault(dataset, []) - try: - for item in rows['phedex']['dataset']: - for subs in item['subscription']: - locationByDset[dataset].append(subs['node']) - except Exception as exc: - print("Failure in getPileupSubscriptions for dataset %s. Error: %s" % (dataset, str(exc))) - locationByDset[dataset] = None - return locationByDset - - def getBlocksByDsetAndRun(datasetName, runList, dbsUrl): """ Given a dataset name and a list of runs, find all the blocks @@ -282,6 +147,7 @@ def findBlockParents(blocks, dbsUrl): NOTE: Value `None` is returned in case the data-service failed to serve a given request. """ parentsByBlock = {} + erroredCont = [] urls = ['%s/blockparents?block_name=%s' % (dbsUrl, quote(b)) for b in blocks] logging.info("Executing %d requests against DBS 'blockparents' API", len(urls)) data = multi_getdata(urls, ckey(), cert()) @@ -292,7 +158,7 @@ def findBlockParents(blocks, dbsUrl): print("Failure in findBlockParents for block %s. Error: %s %s" % (blockName, row.get('code'), row.get('error'))) - parentsByBlock.setdefault(dataset, None) + erroredCont.append(dataset) continue rows = json.loads(row['data']) try: @@ -305,8 +171,8 @@ def findBlockParents(blocks, dbsUrl): parentsByBlock[dataset][item['this_block_name']].add(item['parent_block_name']) except Exception as exc: print("Failure in findBlockParents for block %s. Error: %s" % (blockName, str(exc))) - parentsByBlock[dataset] = None - return parentsByBlock + erroredCont.append(dataset) + return parentsByBlock, erroredCont def getRunsInBlock(blocks, dbsUrl): @@ -571,13 +437,19 @@ def ioForTask(request, dbsUrl): def findParent(datasets, dbsUrl): """ - Helper function to find the parent dataset. + Provided a list of dataset names, find their parent dataset. It returns a dictionary key'ed by the child dataset NOTE: Value `None` is returned in case the data-service failed to serve a given request. + :param datasets: list of dataset names + :param dbsUrl: string with the dbsurl to contact + :return: a tuple of: + dictionary key'ed by the child dataset name, with the parent as value; + and a list of dataset names that failed to be resolved """ parentByDset = {} + erroredDsets = [] if not datasets: - return parentByDset + return parentByDset, erroredDsets urls = ['%s/datasetparents?dataset=%s' % (dbsUrl, d) for d in datasets] logging.info("Executing %d requests against DBS 'datasetparents' API", len(urls)) @@ -586,10 +458,8 @@ def findParent(datasets, dbsUrl): for row in data: dataset = row['url'].split('=')[-1] if row['data'] is None: - print("Failure in findParent for dataset %s. Error: %s %s" % (dataset, - row.get('code'), - row.get('error'))) - parentByDset.setdefault(dataset, None) + print("Failure in findParent for dataset %s. Error: %s" % (dataset, row)) + erroredDsets.append(dataset) continue rows = json.loads(row['data']) try: @@ -597,5 +467,5 @@ def findParent(datasets, dbsUrl): parentByDset[item['this_dataset']] = item['parent_dataset'] except Exception as exc: print("Failure in findParent for dataset %s. Error: %s" % (dataset, str(exc))) - parentByDset[dataset] = None - return parentByDset + erroredDsets.append(dataset) + return parentByDset, erroredDsets diff --git a/src/python/WMCore/MicroService/Tools/PycurlRucio.py b/src/python/WMCore/MicroService/Tools/PycurlRucio.py index 7756ad4a1c..2a9b41e086 100644 --- a/src/python/WMCore/MicroService/Tools/PycurlRucio.py +++ b/src/python/WMCore/MicroService/Tools/PycurlRucio.py @@ -108,20 +108,21 @@ def renewRucioToken(rucioAuthUrl, userToken): return newExpiration -def getPileupContainerSizesRucio(containers, rucioUrl, rucioToken, scope="cms"): +def getPileupContainerSizes(containers, rucioUrl, rucioToken, scope="cms"): """ - Given a list of containers, find their total size in Rucio + Given a list of container names, find their total size in Rucio :param containers: list of container names :param rucioUrl: a string with the Rucio URL :param rucioToken: a string with the user rucio token :param scope: a string with the Rucio scope of our data - :return: a flat dictionary of container and their respective sizes - NOTE: Value `None` is returned in case the data-service failed to serve a given request. - NOTE: Rucio version of getPileupDatasetSizes() + :return: a tuple of: + flat dictionary of container and their respective sizes + and a list of dataset names that failed to be resolved """ - sizeByDset = {} + sizeByCont = {} + erroredCont = [] if not containers: - return sizeByDset + return sizeByCont, erroredCont headers = {"X-Rucio-Auth-Token": rucioToken} @@ -133,19 +134,18 @@ def getPileupContainerSizesRucio(containers, rucioUrl, rucioToken, scope="cms"): container = row['url'].split('/dids/{}/'.format(scope))[1] container = container.replace("?dynamic=anything", "") if row['data'] is None: - msg = "Failure in getPileupContainerSizesRucio for container {}. Response: {}".format(container, row) + msg = "getPileupContainerSizes failed for container {}. Response: {}".format(container, row) logging.error(msg) - sizeByDset.setdefault(container, None) + erroredCont.append(container) continue response = json.loads(row['data']) try: - sizeByDset.setdefault(container, response['bytes']) + sizeByCont.setdefault(container, response['bytes']) except KeyError: - msg = "getPileupContainerSizesRucio function did not return a valid response for container: %s. Error: %s" - logging.error(msg, container, response) - sizeByDset.setdefault(container, None) - continue - return sizeByDset + msg = "Container {} does not have size in bytes in Rucio. Error: {}".format(container, response) + logging.error(msg) + erroredCont.append(container) + return sizeByCont, erroredCont def listReplicationRules(containers, rucioAccount, grouping, @@ -161,13 +161,14 @@ def listReplicationRules(containers, rucioAccount, grouping, :param rucioUrl: string with the Rucio url :param rucioToken: string with the Rucio token :param scope: string with the data scope - :return: a flat dictionary key'ed by the container name, with a list of RSE - expressions that still need to be resolved - NOTE: Value `None` is returned in case the data-service failed to serve a given request. + :return: a tuple of: + flat dictionary key'ed by the container name, with a list of RSE expressions + and a list of container names that failed to be resolved """ locationByContainer = {} + erroredCont = [] if not containers: - return locationByContainer + return locationByContainer, erroredCont if grouping not in ["A", "D"]: raise RuntimeError("Replication rule grouping value provided ({}) is not allowed!".format(grouping)) @@ -184,7 +185,7 @@ def listReplicationRules(containers, rucioAccount, grouping, if "200 OK" not in row['headers']: msg = "Failure in listReplicationRules for container {}. Response: {}".format(container, row) logging.error(msg) - locationByContainer.setdefault(container, None) + erroredCont.append(container) continue try: locationByContainer.setdefault(container, []) @@ -209,9 +210,8 @@ def listReplicationRules(containers, rucioAccount, grouping, daysStuck = (utcTimeNow - stuckAt) // (24 * 60 * 60) if daysStuck > STUCK_LIMIT: - msg = "Container {} has a STUCK rule for {} days (limit set to: {}).".format(container, - daysStuck, - STUCK_LIMIT) + msg = "Container {} has a STUCK rule for {} days ".format(container, daysStuck) + msg += "(limit set to: {}).".format(STUCK_LIMIT) msg += " Not going to use it! Rule info: {}".format(item) logging.warning(msg) continue @@ -229,9 +229,9 @@ def listReplicationRules(containers, rucioAccount, grouping, msg = "listReplicationRules function did not return a valid response for container: %s." msg += "Server responded with: %s\nError: %s" logging.exception(msg, container, str(exc), row['data']) - locationByContainer.setdefault(container, None) + erroredCont.append(container) continue - return locationByContainer + return locationByContainer, erroredCont def getPileupSubscriptionsRucio(datasets, rucioUrl, rucioToken, scope="cms"): @@ -300,26 +300,21 @@ def getBlocksAndSizeRucio(containers, rucioUrl, rucioToken, scope="cms"): :param rucioUrl: a string with the Rucio URL :param rucioToken: a string with the user rucio token :param scope: a string with the Rucio scope of our data - :return: a dictionary in the form of: - {"dataset": - {"block": - {"blockSize": 111, "locations": ["x", "y"]} - } - } - NOTE: Value `None` is returned in case the data-service failed to serve a given request. - NOTE2: meant to return an output similar to Common.getBlockReplicasAndSize + :return: a tuple of: + a dictionary in the form of: + {"dataset": + {"block": + {"blockSize": 111, "locations": ["x", "y"]}}} + and a list of container names that failed to be resolved """ contBlockSize = {} + erroredCont = [] if not containers: - return contBlockSize + return contBlockSize, erroredCont headers = {"X-Rucio-Auth-Token": rucioToken} urls = [] for cont in containers: - ### FIXME: the long attribute value type has recently changed integer to boolean - ### see PR: https://github.com/rucio/rucio/pull/3949 , which went in in 1.23.5 series - ### we need to make sure CMS production Rucio will be running that version once MicroServices - ### get deployed to CMSWEB urls.append('{}/dids/{}/dids/search?type=dataset&long=True&name={}'.format(rucioUrl, scope, quote(cont + "#*"))) logging.info("Executing %d requests against Rucio DIDs search API for containers", len(urls)) data = multi_getdata(urls, ckey(), cert(), headers=headers) @@ -330,13 +325,13 @@ def getBlocksAndSizeRucio(containers, rucioUrl, rucioToken, scope="cms"): if row['data'] in [None, ""]: msg = "Failure in getBlocksAndSizeRucio function for container {}. Response: {}".format(container, row) logging.error(msg) - contBlockSize[container] = None + erroredCont.append(container) continue for item in parseNewLineJson(row['data']): # NOTE: we do not care about primary block location in Rucio contBlockSize[container][item['name']] = {"blockSize": item['bytes'], "locations": []} - return contBlockSize + return contBlockSize, erroredCont ### NOTE: likely not going to be used for a while diff --git a/test/python/WMCore_t/MicroService_t/Tools_t/Common_t.py b/test/python/WMCore_t/MicroService_t/Tools_t/Common_t.py index d152bbc412..d4f8b4391e 100644 --- a/test/python/WMCore_t/MicroService_t/Tools_t/Common_t.py +++ b/test/python/WMCore_t/MicroService_t/Tools_t/Common_t.py @@ -44,7 +44,7 @@ def testGetEventsLumis(self): def test_findParent(self): "Test function for findParent()" - parents = findParent(self.child, self.dbsUrl) + parents, _ = findParent(self.child, self.dbsUrl) self.assertEqual(parents[self.child[0]], '/SingleElectron/Run2016B-v2/RAW') diff --git a/test/python/WMCore_t/MicroService_t/Tools_t/PycurlRucio_t.py b/test/python/WMCore_t/MicroService_t/Tools_t/PycurlRucio_t.py index 06e1a66ea6..b82996c0be 100644 --- a/test/python/WMCore_t/MicroService_t/Tools_t/PycurlRucio_t.py +++ b/test/python/WMCore_t/MicroService_t/Tools_t/PycurlRucio_t.py @@ -10,7 +10,7 @@ from Utils.PythonVersion import PY3 from WMCore.MicroService.Tools.PycurlRucio import (getRucioToken, parseNewLineJson, - getPileupContainerSizesRucio, listReplicationRules, + getPileupContainerSizes, listReplicationRules, getBlocksAndSizeRucio, stringDateToEpoch) CONT1 = "/NoBPTX/Run2018D-12Nov2019_UL2018-v1/MINIAOD" @@ -46,24 +46,24 @@ def testParseNewLineJson(self): for num, item in enumerate(parseNewLineJson(dataStream)): self.assertItemsEqual(item, validate[str(num + 1)]) - def testGetPileupContainerSizesRucio(self): + def testGetPileupContainerSizes(self): """ Test the getPileupContainerSizesRucio function, which fetches the container total bytes. """ self.rucioToken, self.tokenValidity = getRucioToken(self.rucioAuthUrl, self.rucioAccount) # Test 1: no DIDs provided as input - resp = getPileupContainerSizesRucio([], self.rucioUrl, - self.rucioToken, scope=self.rucioScope) + resp, _ = getPileupContainerSizes([], self.rucioUrl, self.rucioToken, + scope=self.rucioScope) self.assertEqual(resp, {}) # Test 2: multiple valid/invalid DIDs provided as input containers = [CONT1, PU_CONT, self.badDID] - resp = getPileupContainerSizesRucio(containers, self.rucioUrl, - self.rucioToken, scope=self.rucioScope) - self.assertTrue(len(resp) == 3) + resp, failedCont = getPileupContainerSizes(containers, self.rucioUrl, + self.rucioToken, scope=self.rucioScope) + self.assertTrue(len(resp) == 2) self.assertTrue(resp[PU_CONT] > 0) - self.assertIsNone(resp[self.badDID]) + self.assertTrue(self.badDID in failedCont) def testListReplicationRules(self): """ @@ -71,8 +71,8 @@ def testListReplicationRules(self): rules a return a list of RSEs key'ed by the container name. """ self.rucioToken, self.tokenValidity = getRucioToken(self.rucioAuthUrl, self.rucioAccount) - resp = listReplicationRules([], self.rucioAccount, grouping="A", - rucioUrl=self.rucioUrl, rucioToken=self.rucioToken, scope=self.rucioScope) + resp, _ = listReplicationRules([], self.rucioAccount, grouping="A", rucioUrl=self.rucioUrl, + rucioToken=self.rucioToken, scope=self.rucioScope) self.assertEqual(resp, {}) with self.assertRaises(RuntimeError): @@ -80,8 +80,9 @@ def testListReplicationRules(self): rucioUrl=self.rucioUrl, rucioToken=self.rucioToken, scope=self.rucioScope) containers = [PU_CONT, CONT2, self.badDID] - resp = listReplicationRules(containers, self.rucioAccount, grouping="A", - rucioUrl=self.rucioUrl, rucioToken=self.rucioToken, scope=self.rucioScope) + resp, failedContainers = listReplicationRules(containers, self.rucioAccount, grouping="A", + rucioUrl=self.rucioUrl, rucioToken=self.rucioToken, + scope=self.rucioScope) self.assertTrue(len(resp) == 3) self.assertTrue(PU_CONT in resp) self.assertTrue(isinstance(resp[CONT2], list)) @@ -94,16 +95,16 @@ def testGetBlocksAndSizeRucio(self): """ self.rucioToken, self.tokenValidity = getRucioToken(self.rucioAuthUrl, self.rucioAccount) # Test 1: no DIDs provided as input - resp = getBlocksAndSizeRucio([], self.rucioUrl, self.rucioToken, self.rucioScope) + resp, _ = getBlocksAndSizeRucio([], self.rucioUrl, self.rucioToken, self.rucioScope) self.assertEqual(resp, {}) # Test 2: multiple valid/invalid DIDs provided as input containers = [PU_CONT, CONT3, self.badDID] - resp = getBlocksAndSizeRucio(containers, self.rucioUrl, self.rucioToken, self.rucioScope) + resp, failedContainers = getBlocksAndSizeRucio(containers, self.rucioUrl, self.rucioToken, self.rucioScope) self.assertTrue(len(resp) == 3) self.assertTrue(resp[PU_CONT][PU_CONT_BLK]['blockSize'] > 0) self.assertTrue(isinstance(resp[PU_CONT][PU_CONT_BLK]['locations'], list)) self.assertTrue(len(resp[CONT3]) > 0) - self.assertIsNone(resp[self.badDID]) + self.assertTrue(self.badDID in failedContainers) def testStringDateToEpoch(self): """