-
Notifications
You must be signed in to change notification settings - Fork 108
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #11487 from vkuznet/fix-issue-11426
First implementation of MSPileup data placement logic via MSPileupTasks class
- Loading branch information
Showing
11 changed files
with
1,001 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| #!/usr/bin/env python | ||
| """ | ||
| File : MSPileupReport.py | ||
| Author : Valentin Kuznetsov <vkuznet AT gmail dot com> | ||
| Description: MSPileup report module | ||
| """ | ||
|
|
||
| # WMCore modules | ||
| from Utils.Timers import gmtimeSeconds, encodeTimestamp | ||
|
|
||
|
|
||
| class MSPileupReport(): | ||
| """ | ||
| MSPileupReport class represents MSPileup report object(s) | ||
| """ | ||
| def __init__(self, autoExpire=3600, autoCleanup=False): | ||
| """ | ||
| Constructor for MSPileup object | ||
| """ | ||
| self.docs = [] | ||
| self.autoExpire = autoExpire | ||
| self.autoCleanup = autoCleanup | ||
|
|
||
| def addEntry(self, task, uuid, entry): | ||
| """ | ||
| Add new entry into MSPileup documents | ||
| :param task: task name | ||
| :param uuid: unique id of the entry | ||
| :param entry: entry message or any other object to store | ||
| """ | ||
| if self.autoCleanup: | ||
| self.purgeExpired() | ||
| gmtime = gmtimeSeconds() | ||
| report = {'gmtime': gmtime, 'uuid': uuid, | ||
| 'timestamp': encodeTimestamp(gmtime), | ||
| 'entry': entry, 'task': task} | ||
| self.docs.append(report) | ||
|
|
||
| def purgeExpired(self): | ||
| """ | ||
| Purge expired records from internal docs | ||
| """ | ||
| gmtime = gmtimeSeconds() | ||
| for entry in list(self.docs): | ||
| if gmtime - entry['gmtime'] > self.autoExpire: | ||
| self.docs.remove(entry) | ||
|
|
||
| def getDocuments(self): | ||
| """ | ||
| Return report documents | ||
| """ | ||
| if self.autoCleanup: | ||
| self.purgeExpired() | ||
| return self.docs | ||
|
|
||
| def getReportByUuid(self): | ||
| """ | ||
| Return report documents in dictonary form with uuid's as keys | ||
| """ | ||
| if self.autoCleanup: | ||
| self.purgeExpired() | ||
| rdict = {} | ||
| for doc in self.docs: | ||
| uuid = doc['uuid'] | ||
| timestamp = doc['timestamp'] | ||
| entry = doc['entry'] | ||
| task = doc['task'] | ||
| record = f"{timestamp} {task} task {uuid} {entry}" | ||
| rdict.setdefault(uuid, []).append(record) | ||
| return rdict |
85 changes: 85 additions & 0 deletions
85
src/python/WMCore/MicroService/MSPileup/MSPileupTaskManager.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| #!/usr/bin/env python | ||
| """ | ||
| File : MSPileupTaskManager.py | ||
| Author : Valentin Kuznetsov <vkuznet AT gmail dot com> | ||
| Description: MSPileupTaskManager handles MSPileupTasks | ||
| In particular, it perform the following tasks each polling cycle: | ||
| - fetches pileup sizes for all pileup documents in database back-end | ||
| - update RSE quotas | ||
| - perform monitoring task | ||
| - perform task for active pileups using up-to-date RSE quotas | ||
| - perform task for inactive pileups | ||
| """ | ||
|
|
||
| # system modules | ||
| import os | ||
| from threading import current_thread | ||
|
|
||
| # WMCore modules | ||
| from WMCore.MicroService.MSCore.MSCore import MSCore | ||
| from WMCore.MicroService.DataStructs.DefaultStructs import PILEUP_REPORT | ||
| from WMCore.MicroService.MSPileup.MSPileupData import MSPileupData | ||
| from WMCore.MicroService.MSPileup.MSPileupTasks import MSPileupTasks | ||
| from WMCore.MicroService.MSTransferor.DataStructs.RSEQuotas import RSEQuotas | ||
| from WMCore.MicroService.Tools.PycurlRucio import getPileupContainerSizesRucio, getRucioToken | ||
| from WMCore.Services.Rucio.Rucio import Rucio | ||
|
|
||
|
|
||
| class MSPileupTaskManager(MSCore): | ||
| """ | ||
| MSPileupTaskManager handles MSPileup tasks | ||
| """ | ||
|
|
||
| def __init__(self, msConfig, **kwargs): | ||
| super().__init__(msConfig, **kwargs) | ||
| self.marginSpace = msConfig.get('marginSpace', 1024**4) | ||
| self.rucioAccount = msConfig.get('rucioAccount', 'ms-pileup') | ||
| self.rucioUrl = msConfig.get('rucioHost', 'http://cms-rucio.cern.ch') | ||
| self.rucioAuthUrl = msConfig.get('authHost', 'https://cms-rucio-auth.cern.ch') | ||
| creds = {"client_cert": os.getenv("X509_USER_CERT", "Unknown"), | ||
| "client_key": os.getenv("X509_USER_KEY", "Unknown")} | ||
| configDict = {'rucio_host': self.rucioUrl, 'auth_host': self.rucioAuthUrl, | ||
| 'creds': creds, 'auth_type': 'x509'} | ||
| self.rucioClient = Rucio(self.rucioAccount, configDict=configDict) | ||
| self.dataManager = MSPileupData(msConfig) | ||
| self.mgr = MSPileupTasks(self.dataManager, self.logger, | ||
| self.rucioAccount, self.rucioClient) | ||
| self.rseQuotas = RSEQuotas(self.rucioAccount, msConfig["quotaUsage"], | ||
| minimumThreshold=msConfig["minimumThreshold"], | ||
| verbose=msConfig['verbose'], logger=self.logger) | ||
|
|
||
| def status(self): | ||
| """ | ||
| Provide MSPileupTaskManager status API. | ||
| :return: status dictionary | ||
| """ | ||
| summary = dict(PILEUP_REPORT) | ||
| summary.update({'thread_id': current_thread().name}) | ||
| summary.update({'tasks': self.msg.getReport()) | ||
| return summary | ||
|
|
||
| def executeCycle(self): | ||
| """ | ||
| execute MSPileupTasks polling cycle | ||
| """ | ||
| # get pileup sizes and update them in DB | ||
| spec = {} | ||
| docs = self.dataManager.getPileup(spec) | ||
| rucioToken = getRucioToken(self.rucioAuthUrl, self.rucioAccount) | ||
| containers = [r['pileupName'] for r in docs] | ||
| datasetSizes = getPileupContainerSizesRucio(containers, self.rucioUrl, rucioToken) | ||
| for doc in docs: | ||
| pileupSize = datasetSizes.get(doc['pileupName'], 0) | ||
| doc['pileupSize'] = pileupSize | ||
| self.dataManager.updatePileup(doc) | ||
|
|
||
| # fetch all rse quotas | ||
| self.rseQuotas.fetchStorageUsage(self.rucioClient) | ||
| nodeUsage = self.rseQuotas.getNodeUsage() | ||
|
|
||
| # execute all tasks | ||
| self.mgr.monitoringTask() | ||
| self.mgr.activeTask(nodeUsage=nodeUsage, marginSpace=self.marginSpace) | ||
| self.mgr.inactiveTask() |
Oops, something went wrong.