-
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.
Add support for pycurl_manager multi_getdata
- Loading branch information
Showing
2 changed files
with
197 additions
and
93 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,176 @@ | ||
| #!/usr/bin/env python | ||
| """ | ||
| _DBSUtils_ | ||
| set of common utilities for DBS3Reader | ||
| """ | ||
| import os | ||
| import json | ||
| import concurrent.futures | ||
|
|
||
| # third-party libraries | ||
| import requests | ||
|
|
||
| from Utils.CertTools import getKeyCertFromEnv, cert, ckey | ||
| from dbs.apis.dbsClient import aggFileLumis, aggFileParents | ||
| from WMCore.Services.pycurl_manager import getdata as multi_getdata | ||
|
|
||
|
|
||
|
|
||
| ######## NOTE: the code below demonstrates how to use Python concurrent.futures | ||
| # we decided to keep it around since it may be used as an example | ||
| # for other concurrent use-cases | ||
| # For logic please see runConcurrentProcess function | ||
| # The wrapperXXX functions can be passed to runConcurrentProcess | ||
| # | ||
| # global DBS URL to be used by stand-alone functions in concurrent calls | ||
| # see wrapperBlock functions | ||
| # DBS_URL = 'https://cmsweb-prod.cern.ch/dbs/prod/global/DBSReader' | ||
|
|
||
| # def dbsRequest(url): | ||
| # """ | ||
| # Wrapper function to perform call to given URL using client's certificates | ||
|
|
||
| # :param url: url name | ||
| # :return: results | ||
| # """ | ||
| # crt = os.path.join(os.getenv('HOME'), '.globus/usercert.pem') | ||
| # key = os.path.join(os.getenv('HOME'), '.globus/userkey.pem') | ||
| # if os.environ.get('X509_USER_PROXY', ''): | ||
| # crt = os.getenv('X509_USER_PROXY') | ||
| # key = os.getenv('X509_USER_PROXY') | ||
| # if not os.path.exists(crt): | ||
| # raise Exception('Unable to locate user X509 certificate file') from None | ||
| # if not os.path.exists(key): | ||
| # raise Exception('Unable to locate user X509 key certificate file') from None | ||
| # caPath = os.environ.get('X509_CERT_DIR', '/etc/grid-security/certificates') | ||
| # if not os.path.exists(caPath): | ||
| # raise Exception('Unable to locate X509_CERT_DIR') from None | ||
| # res = requests.get(url, | ||
| # cert=(crt, key), | ||
| # headers={'Accept-Encoding': 'gzip'}, | ||
| # verify=caPath) | ||
| # return res.json() | ||
|
|
||
| # def wrapperBlockFileParents(blk): | ||
| # """ | ||
| # Wrapper function to perform DBS listFileParents API call for a given block | ||
|
|
||
| # :param blk: block name | ||
| # :return: result of DBS listFileParents API | ||
| # """ | ||
| # blk = blk.replace('#', '%23') | ||
| # url = '{}/fileparents?block_name={}'.format(DBS_URL, blk) | ||
| # return aggFileParents(dbsRequest(url)) | ||
|
|
||
| # def wrapperBlockFileLumis(blk): | ||
| # """ | ||
| # Wrapper function to perform DBS listFileLumis API call for a given block | ||
|
|
||
| # :param blk: block name | ||
| # :return: result of DBS listFileLumis API | ||
| # """ | ||
| # blk = blk.replace('#', '%23') | ||
| # url = '{}/filelumis?block_name={}'.format(DBS_URL, blk) | ||
| # return aggFileLumis(dbsRequest(url)) | ||
|
|
||
| # def runConcurrentProcess(func, blocks, maxWorkers=10): | ||
| # """ | ||
| # run concurrently set of processes for given function and set of blocks. | ||
| # Internally, the code is based on Python concurrent futures primitives, see | ||
| # # https://docs.python.org/3/library/concurrent.futures.html#module-concurrent.futures | ||
|
|
||
| # :param func: the function to run | ||
| # :param blocks: list of blocks our function should process | ||
| # :param maxWorkers: maximum number of workers to use | ||
| # """ | ||
| # with concurrent.futures.ThreadPoolExecutor(maxWorkers) as executor: | ||
| # futures = {executor.submit(func, blk): blk for blk in blocks} | ||
| # rdict = {} | ||
| # for future in concurrent.futures.as_completed(futures): | ||
| # blk = futures[future] | ||
| # try: | ||
| # data = future.result() | ||
| # except Exception as exc: | ||
| # rdict[blk] = [{'error': str(exc)}] | ||
| # else: | ||
| # rdict[blk] = data | ||
| # return rdict | ||
|
|
||
|
|
||
| def dbsListFileParents(dbsUrl, blocks, logger): | ||
| """ | ||
| Concurrent counter part of DBS listFileParents API | ||
| :param dbsUrl: DBS URL | ||
| :param blocks: list of blocks | ||
| :param logger: logger | ||
| :return: list of file parents | ||
| """ | ||
| params = {} | ||
| return dbsParallelApi(dbsUrl, 'fileparents', 'block_name', blocks, params, aggFileParents, logger) | ||
|
|
||
| def dbsListFileLumis(dbsUrl, blocks, logger=None): | ||
| """ | ||
| Concurrent counter part of DBS listFileLumis API | ||
| :param dbsUrl: DBS URL | ||
| :param blocks: list of blocks | ||
| :param logger: logger | ||
| :return: list of file lumis | ||
| """ | ||
| params = {} | ||
| return dbsParallelApi(dbsUrl, 'filelumis', 'block_name', blocks, params, aggFileLumis, logger) | ||
|
|
||
| def dbsParentFilesDetail(dbsUrl, lfns, logger=None): | ||
| """ | ||
| Concurrent counter part of DBS files API | ||
| :param dbsUrl: DBS URL | ||
| :param params: input parameters | ||
| :param logger: logger | ||
| :return: list of file details for given parent lfns | ||
| """ | ||
| params = {'detail': True} | ||
| func = None | ||
| return dbsParallelApi(dbsUrl, 'files', 'logical_file_name', lfns, params, func, logger) | ||
|
|
||
| def dbsParallelApi(dbsUrl, api, attr, values, params, aggFunc, logger=None): | ||
| """ | ||
| Generic paralle DBS calls for given api, attribute and values | ||
| :param dbsUrl: DBS URL | ||
| :param api: api name | ||
| :param attr: api attribute | ||
| :param values: api attribute values | ||
| :param params: optional params dict | ||
| :param aggFunc: aggregation function | ||
| :return: result of DBS api | ||
| """ | ||
| urls = [] | ||
| for item in values: | ||
| if '#' in item: | ||
| item = item.replace('#', '%23') | ||
| url = '%s/%s?%s=%s' % (dbsUrl, api, attr, item) | ||
| for key, val in params.items(): | ||
| url += '&%s=%s' % (key, val) | ||
| urls.append(url) | ||
| if logger: | ||
| logger.info("Executing %d requests against DBS '%s' API", len(urls), api) | ||
| data = multi_getdata(urls, ckey(), cert()) | ||
|
|
||
| rdict = {} | ||
| for row in data: | ||
| item = row['url'].rsplit('=')[-1] | ||
| code = int(row.get('code', 200)) | ||
| if code != 200: | ||
| msg = "Failure in %s for %s. Error: %s %s" % (api, item, | ||
| row.get('code'), | ||
| row.get('error')) | ||
| raise RuntimeError(msg) | ||
| data = row.get('data', []) | ||
| res = json.loads(data) | ||
| if aggFunc: | ||
| rdict[item] = aggFunc(res) | ||
| return rdict |