idds/tests/migrating_requests_v1_to_v2.py

0001 #!/usr/bin/env python
0002 #
0003 # Licensed under the Apache License, Version 2.0 (the "License");
0004 # You may not use this file except in compliance with the License.
0005 # You may obtain a copy of the License at
0006 # http://www.apache.org/licenses/LICENSE-2.0OA
0007 #
0008 # Authors:
0009 # - Wen Guan, <wen.guan@cern.ch>, 2020
0010
0011 from idds.common.constants import RequestType
0012 from idds.common.utils import get_rest_host
0013 from idds.core.requests import get_requests
0014 from idds.workflowv2.workflow import Workflow
0015
0016 from idds.atlas.workflowv2.atlasstageinwork import ATLASStageinWork
0017
0018 from idds.client.clientmanager import ClientManager
0019
0020
0021 def convert_req2reqv2(req):
0022     # v1: {'created_at': datetime.datetime(2020, 11, 3, 10, 9, 32), 'substatus': None, 'priority': 0, 'transform_tag': '2', 'requester': 'panda', 'request_metadata': {'workload_id': 23083304, 'rule_id': 'bef3da17f17c49ac97863bb9e96af672'}, 'name': 'valid1.361027.Pythia8EvtGen_A14NNPDF23LO_jetjet_JZ7W.simul.HITS.e5112_s3227_tid12560193_00', 'request_id': 3775, 'accessed_at': datetime.datetime(2020, 11, 3, 10, 9, 32), 'updated_at': datetime.datetime(2020, 11, 3, 10, 9, 32), 'locking': <RequestLocking.Idle: 0>, 'status': <RequestStatus.Cancelled: 9>, 'workload_id': 23083304, 'request_type': <RequestType.StageIn: 2>, 'errors': None, 'processing_metadata': None, 'scope': 'valid1', 'expired_at': datetime.datetime(2020, 12, 3, 10, 9, 32), 'next_poll_at': datetime.datetime(2020, 11, 3, 10, 9, 32)}  # noqa E501
0023
0024     if req['request_type'] == RequestType.StageIn:
0025         request_metadata = req['request_metadata']
0026         work = ATLASStageinWork(executable=None, arguments=None, parameters=None, setup=None,
0027                                 exec_type='local', sandbox=None,
0028                                 primary_input_collection={'scope': req['scope'], 'name': req['name']},
0029                                 other_input_collections=None,
0030                                 output_collections={'scope': req['scope'], 'name': req['name'] + '.idds.stagein'},
0031                                 log_collections=None,
0032                                 logger=None,
0033                                 max_waiting_time=request_metadata.get('max_waiting_time', 3600 * 7 * 24),
0034                                 src_rse=request_metadata.get('src_rse', None),
0035                                 dest_rse=request_metadata.get('dest_rse', None),
0036                                 rule_id=request_metadata.get('rule_id', None))
0037     if req['request_type'] == RequestType.Workflow:
0038         pass
0039         ori_workflow = req['request_metadata']['workflow']
0040         ori_work = ori_workflow.works[ori_workflow.primary_initial_work]
0041         input_coll = ori_work.collections[ori_work.primary_input_collection]
0042         work = ATLASStageinWork(executable=None, arguments=None, parameters=None, setup=None,
0043                                 exec_type='local', sandbox=None,
0044                                 primary_input_collection={'scope': input_coll['scope'], 'name': input_coll['name']},
0045                                 other_input_collections=None,
0046                                 output_collections={'scope': input_coll['scope'], 'name': input_coll['name'] + '.idds.stagein'},
0047                                 log_collections=None,
0048                                 logger=None,
0049                                 max_waiting_time=ori_work.max_waiting_time,
0050                                 src_rse=ori_work.src_rse,
0051                                 dest_rse=ori_work.dest_rse,
0052                                 rule_id=ori_work.rule_id)
0053
0054     workload_id = req['workload_id']
0055     if not workload_id and 'workload_id' in request_metadata:
0056         workload_id = request_metadata['workload_id']
0057
0058     wf = Workflow()
0059     wf.set_workload_id(workload_id)
0060     wf.add_work(work)
0061
0062     host = get_rest_host()
0063     wm = ClientManager(host=host)
0064     request_id = wm.submit(wf)
0065     # print(request_id)
0066     return request_id
0067
0068
0069 reqs = get_requests()
0070 print(len(reqs))
0071 for req in reqs:
0072     # if req['request_id'] in [3743, 3755, 3769, 3775]:
0073     # if req['request_id'] in [3787, 3789, 3791]:
0074     if req['request_id'] in [11673, 17615, 19289, 19397]:
0075         # print(req)
0076         # print(req['request_metadata']['workflow'].to_dict())
0077         new_req_id = convert_req2reqv2(req)
0078         print("convert old request %s to new request %s" % (req['request_id'], new_req_id))
0079     # print(req['request_metadata']['workflow'].to_dict())
0080     pass