1 |
< |
#!/usr/bin/env python |
2 |
< |
|
1 |
> |
from Actor import * |
2 |
|
import urllib |
3 |
|
from xml.dom.minidom import parse |
4 |
|
from crab_exceptions import * |
5 |
+ |
from WorkSpace import * |
6 |
+ |
from urlparse import urlparse |
7 |
+ |
from LFNBaseName import * |
8 |
+ |
from crab_util import getUserName |
9 |
|
|
10 |
|
class PhEDExDatasvcInfo: |
11 |
< |
""" |
12 |
< |
provides information from PhEDEx Data Service |
13 |
< |
""" |
14 |
< |
## PhEDEx Data Service URL |
15 |
< |
#datasvc_url="https://cmsweb.cern.ch/phedex/test/datasvc/xml/prod" |
16 |
< |
datasvc_url="https://cmsweb.cern.ch/phedex/datasvc/xml/prod" |
11 |
> |
def __init__( self , cfg_params=None, config=None ): |
12 |
> |
|
13 |
> |
## PhEDEx Data Service URL |
14 |
> |
self.datasvc_url="https://cmsweb.cern.ch/phedex/datasvc/xml/prod" |
15 |
> |
|
16 |
> |
self.FacOps_savannah = 'https://savannah.cern.ch/support/?func=additem&group=cmscompinfrasup' |
17 |
> |
self.stage_out_faq='https://twiki.cern.ch/twiki/bin/view/CMS/SWGuideCrabHowTo#Stageout_and_publication' |
18 |
> |
self.dataPub_faq = 'https://twiki.cern.ch/twiki/bin/view/CMS/SWGuideCrabForPublication' |
19 |
> |
|
20 |
> |
self.usePhedex = True |
21 |
> |
self.sched = common.scheduler.name().upper() |
22 |
> |
|
23 |
> |
if config!=None: |
24 |
> |
self.checkConfig(config) |
25 |
> |
else: |
26 |
> |
self.checkCfgConfig(cfg_params) |
27 |
> |
|
28 |
> |
self.protocol = self.srm_version |
29 |
> |
|
30 |
> |
|
31 |
> |
def checkConfig(self,config): |
32 |
> |
""" |
33 |
> |
""" |
34 |
> |
self.srm_version = config.get("srm_version",'srmv2') |
35 |
> |
self.node = config.get('storage_element',None) |
36 |
> |
self.lfn='/store/' |
37 |
> |
|
38 |
> |
def checkCfgConfig(self,cfg_params): |
39 |
> |
""" |
40 |
> |
""" |
41 |
> |
self.datasvc_url = cfg_params.get("USER.datasvc_url",self.datasvc_url) |
42 |
> |
self.srm_version = cfg_params.get("USER.srm_version",'srmv2') |
43 |
> |
self.node = cfg_params.get('USER.storage_element',None) |
44 |
> |
|
45 |
> |
self.publish_data = cfg_params.get("USER.publish_data",0) |
46 |
> |
self.usenamespace = cfg_params.get("USER.usenamespace",0) |
47 |
> |
self.user_remote_dir = cfg_params.get("USER.user_remote_dir",'') |
48 |
> |
if self.user_remote_dir: |
49 |
> |
if ( self.user_remote_dir[-1] != '/' ) : self.user_remote_dir = self.user_remote_dir + '/' |
50 |
> |
|
51 |
> |
self.datasetpath = cfg_params.get("CMSSW.datasetpath") |
52 |
> |
self.publish_data_name = cfg_params.get('USER.publish_data_name','') |
53 |
> |
|
54 |
> |
self.user_port = cfg_params.get("USER.storage_port",'8443') |
55 |
> |
self.user_se_path = cfg_params.get("USER.storage_path",'') |
56 |
> |
if self.user_se_path: |
57 |
> |
if ( self.user_se_path[-1] != '/' ) : self.user_se_path = self.user_se_path + '/' |
58 |
> |
|
59 |
> |
#check if using "private" Storage |
60 |
> |
if not self.node : |
61 |
> |
msg = 'Please specify the storage_element name in your crab.cfg section [USER].\n' |
62 |
> |
msg +='\tFor further information please visit : %s'%self.stage_out_faq |
63 |
> |
raise CrabException(msg) |
64 |
> |
if (self.node.find('T1_') + self.node.find('T2_')+self.node.find('T3_')) == -3: self.usePhedex = False |
65 |
> |
|
66 |
> |
if not self.usePhedex and ( self.user_remote_dir == '' or self.user_se_path == '' ): |
67 |
> |
####### FEDE FOR BUG 73010 ############ |
68 |
> |
msg = 'Error: you are asking to stage out without using CMS Storage Name convention. In this case you \n' |
69 |
> |
msg += ' must specify both user_remote_dir and storage_path in the crab.cfg section [USER].\n' |
70 |
> |
msg += ' For further information please visit : \n\t%s'%self.stage_out_faq |
71 |
> |
task = common._db.getTask() |
72 |
> |
add = '\n\n' |
73 |
> |
import shutil |
74 |
> |
try: |
75 |
> |
add += ' Task not correctly created: removing the working_dir ' + common.work_space._top_dir + ' \n' |
76 |
> |
shutil.rmtree(common.work_space._top_dir) |
77 |
> |
except OSError: |
78 |
> |
add += ' Warning: problems removing the working_dir ' + common.work_space._top_dir + ' \n' |
79 |
> |
add += ' Please remove it by hand' |
80 |
> |
msg += add |
81 |
> |
raise CrabException(msg) |
82 |
> |
######################################### |
83 |
> |
|
84 |
> |
self.forced_path = '/store/user/' |
85 |
> |
if self.sched in ['CAF','LSF','PBS']: |
86 |
> |
self.srm_version = 'direct' |
87 |
> |
self.SE = {'CAF':'caf.cern.ch', 'LSF':'', 'PBS':''} |
88 |
> |
if self.sched == 'CAF': self.forced_path = '/store/caf/user/' |
89 |
> |
|
90 |
> |
if not self.usePhedex: |
91 |
> |
self.forced_path = self.user_remote_dir |
92 |
> |
return |
93 |
> |
|
94 |
> |
def getEndpoint(self): |
95 |
> |
''' |
96 |
> |
Return full SE endpoint and related infos |
97 |
> |
''' |
98 |
> |
self.lfn = self.getLFN() |
99 |
> |
|
100 |
> |
#extract the PFN for the given node,LFN,protocol |
101 |
> |
endpoint = self.getStageoutPFN() |
102 |
> |
if ( endpoint[-1] != '/' ) : endpoint = endpoint + '/' |
103 |
|
|
104 |
< |
def lfn2pfn(self,node,lfn,protocol): |
105 |
< |
""" |
106 |
< |
PhEDEx Data Service lfn2pfn call |
107 |
< |
|
108 |
< |
input: LFN,node name,protocol |
109 |
< |
returns: DOM object with the content of the PhEDEx Data Service call |
110 |
< |
""" |
111 |
< |
params = {'node' : node , 'lfn': lfn , 'protocol': protocol} |
112 |
< |
params = urllib.urlencode(params) |
113 |
< |
#print params |
114 |
< |
datasvc_lfn2pfn="%s/lfn2pfn"%self.datasvc_url |
115 |
< |
urlresults = urllib.urlopen(datasvc_lfn2pfn, params) |
116 |
< |
try: |
117 |
< |
urlresults = parse(urlresults) |
118 |
< |
except: |
119 |
< |
urlresults = None |
120 |
< |
return urlresults |
121 |
< |
|
122 |
< |
def parse_error(self,urlresults): |
123 |
< |
""" |
124 |
< |
look for errors in the DOM object returned by PhEDEx Data Service call |
125 |
< |
""" |
126 |
< |
errormsg = None |
127 |
< |
errors=urlresults.getElementsByTagName('error') |
128 |
< |
for error in errors: |
129 |
< |
errormsg=error.childNodes[0].data |
130 |
< |
if len(error.childNodes)>1: |
131 |
< |
errormsg+=error.childNodes[1].data |
132 |
< |
return errormsg |
133 |
< |
|
134 |
< |
def parse_lfn2pfn(self,urlresults): |
135 |
< |
""" |
136 |
< |
Parse the content of the result of lfn2pfn PhEDEx Data Service call |
137 |
< |
|
138 |
< |
input: DOM object with the content of the lfn2pfn call |
139 |
< |
returns: PFN |
140 |
< |
""" |
141 |
< |
result = urlresults.getElementsByTagName('phedex') |
142 |
< |
if not result: |
143 |
< |
return [] |
144 |
< |
result = result[0] |
145 |
< |
pfn = None |
146 |
< |
mapping = result.getElementsByTagName('mapping') |
147 |
< |
for m in mapping: |
148 |
< |
pfn=m.getAttribute("pfn") |
149 |
< |
if pfn: |
150 |
< |
return pfn |
151 |
< |
|
152 |
< |
def getStageoutPFN(self,node,lfn,protocol): |
153 |
< |
""" |
154 |
< |
input: LFN,node name,protocol |
155 |
< |
returns: PFN |
156 |
< |
""" |
157 |
< |
fullurl="%s/lfn2pfn?node=%s&lfn=%s&protocol=%s"%(self.datasvc_url,node,lfn,protocol) |
158 |
< |
domlfn2pfn = self.lfn2pfn(node,lfn,protocol) |
159 |
< |
if not domlfn2pfn : |
160 |
< |
msg="Unable to get info from %s"%fullurl |
161 |
< |
raise CrabException(msg) |
162 |
< |
|
163 |
< |
errormsg = self.parse_error(domlfn2pfn) |
164 |
< |
if errormsg: |
165 |
< |
msg="Error extracting info from %s due to: %s"%(fullurl,errormsg) |
166 |
< |
raise CrabException(msg) |
167 |
< |
|
168 |
< |
stageoutpfn = self.parse_lfn2pfn(domlfn2pfn) |
169 |
< |
if not stageoutpfn: |
170 |
< |
msg="Unable to get stageout path (PFN) from %s"%fullurl |
171 |
< |
raise CrabException(msg) |
172 |
< |
return stageoutpfn |
173 |
< |
|
174 |
< |
|
175 |
< |
if __name__ == '__main__' : |
176 |
< |
""" |
177 |
< |
""" |
178 |
< |
from crab_logger import Logger |
179 |
< |
from WorkSpace import * |
180 |
< |
continue_dir="/home/fanfani/CRAB" |
181 |
< |
cfg_params={'USER.logdir' : continue_dir } |
182 |
< |
common.work_space = WorkSpace(continue_dir, cfg_params) |
183 |
< |
log = Logger() |
95 |
< |
common.logger = log |
96 |
< |
|
97 |
< |
from LFNBaseName import * |
98 |
< |
# test values |
99 |
< |
lfn = LFNBase("datasetstring") |
100 |
< |
node='T2_IT_Bari' |
101 |
< |
protocol="srmv2" |
102 |
< |
|
103 |
< |
#create an instance of the PhEDExDatasvcInfo object |
104 |
< |
dsvc = PhEDExDatasvcInfo() |
105 |
< |
#extract the PFN for the given node,LFN,protocol |
106 |
< |
print "Stageout to %s"%dsvc.getStageoutPFN(node,lfn,protocol) |
104 |
> |
#extract SE name an SE_PATH (needed for publication) |
105 |
> |
SE, SE_PATH, User = self.splitEndpoint(endpoint) |
106 |
> |
|
107 |
> |
return endpoint, self.lfn , SE, SE_PATH, User |
108 |
> |
|
109 |
> |
def splitEndpoint(self, endpoint): |
110 |
> |
''' |
111 |
> |
Return relevant infos from endpoint |
112 |
> |
''' |
113 |
> |
SE = '' |
114 |
> |
SE_PATH = '' |
115 |
> |
USER = getUserName() |
116 |
> |
if self.usePhedex: |
117 |
> |
if self.protocol == 'direct': |
118 |
> |
query=endpoint |
119 |
> |
SE_PATH = endpoint |
120 |
> |
SE = self.SE[self.sched] |
121 |
> |
else: |
122 |
> |
url = 'http://'+endpoint.split('://')[1] |
123 |
> |
scheme, host, path, params, query, fragment = urlparse(url) |
124 |
> |
SE = self.getAuthoritativeSE() |
125 |
> |
SE_PATH = endpoint.split(host)[1] |
126 |
> |
else: |
127 |
> |
SE = self.node |
128 |
> |
SE_PATH = self.user_se_path + self.user_remote_dir |
129 |
> |
if self.lfn.find('group') != -1: |
130 |
> |
try: |
131 |
> |
USER = (self.lfn.split('group')[1]).split('/')[1] |
132 |
> |
except: |
133 |
> |
pass |
134 |
> |
return SE, SE_PATH, USER |
135 |
> |
|
136 |
> |
def getLFN(self): |
137 |
> |
""" |
138 |
> |
define the LFN composing the needed pieces |
139 |
> |
""" |
140 |
> |
lfn = '' |
141 |
> |
l_User = False |
142 |
> |
if not self.usePhedex and (int(self.publish_data) == 0 and int(self.usenamespace) == 0) : |
143 |
> |
### add here check if user is trying to force a wrong LFN using a T2 TODO |
144 |
> |
## check if storage_name is a T2 (siteDB query) |
145 |
> |
## if yes :match self.user_lfn with LFNBaseName... |
146 |
> |
## if NOT : raise (you are using a T2. It's not allowed stage out into self.user_path+self.user_lfn) |
147 |
> |
lfn = self.user_remote_dir |
148 |
> |
return lfn |
149 |
> |
if self.publish_data_name == '' and int(self.publish_data) == 1: |
150 |
> |
msg = "Error. The [USER] section does not have 'publish_data_name'\n" |
151 |
> |
msg += '\tFor further information please visit : \n\t%s'%self.dataPub_faq |
152 |
> |
raise CrabException(msg) |
153 |
> |
if self.publish_data_name == '' and int(self.usenamespace) == 1: |
154 |
> |
self.publish_data_name = "DefaultDataset" |
155 |
> |
if int(self.publish_data) == 1: |
156 |
> |
if self.sched in ['CAF']: l_User=True |
157 |
> |
primaryDataset = self.computePrimaryDataset() |
158 |
> |
### added the case lfn = LFNBase(self.forced_path, primaryDataset, self.publish_data_name, publish=True) |
159 |
> |
### for the publication in order to be able to check the lfn length |
160 |
> |
lfn = LFNBase(self.forced_path, primaryDataset, self.publish_data_name, publish=True) + '/${PSETHASH}/' |
161 |
> |
elif int(self.usenamespace) == 1: |
162 |
> |
if self.sched in ['CAF']: l_User=True |
163 |
> |
primaryDataset = self.computePrimaryDataset() |
164 |
> |
lfn = LFNBase(self.forced_path, primaryDataset, self.publish_data_name) + '/${PSETHASH}/' |
165 |
> |
else: |
166 |
> |
if self.sched in ['CAF','LSF']: l_User=True |
167 |
> |
lfn = LFNBase(self.forced_path,self.user_remote_dir) |
168 |
> |
if ( lfn[-1] != '/' ) : lfn = lfn + '/' |
169 |
> |
return lfn |
170 |
> |
|
171 |
> |
def computePrimaryDataset(self): |
172 |
> |
""" |
173 |
> |
compute the last part for the LFN in case of publication |
174 |
> |
""" |
175 |
> |
if (self.datasetpath.upper() != 'NONE'): |
176 |
> |
primarydataset = self.datasetpath.split("/")[1] |
177 |
> |
else: |
178 |
> |
primarydataset = self.publish_data_name |
179 |
> |
return primarydataset |
180 |
> |
|
181 |
> |
def domPhedex(self,params,datasvc_baseUrl): |
182 |
> |
""" |
183 |
> |
PhEDEx Data Service lfn2pfn call |
184 |
|
|
185 |
+ |
input: params,datasvc_baseUrl |
186 |
+ |
returns: DOM object with the content of the PhEDEx Data Service call |
187 |
+ |
""" |
188 |
+ |
params = urllib.urlencode(params) |
189 |
+ |
try: |
190 |
+ |
urlresults = urllib.urlopen(datasvc_baseUrl, params) |
191 |
+ |
urlresults = parse(urlresults) |
192 |
+ |
except IOError: |
193 |
+ |
msg="Unable to access PhEDEx Data Service at %s"%datasvc_baseUrl |
194 |
+ |
raise CrabException(msg) |
195 |
+ |
except: |
196 |
+ |
urlresults = None |
197 |
+ |
|
198 |
+ |
return urlresults |
199 |
+ |
|
200 |
+ |
def parse_error(self,urlresults): |
201 |
+ |
""" |
202 |
+ |
look for errors in the DOM object returned by PhEDEx Data Service call |
203 |
+ |
""" |
204 |
+ |
errormsg = None |
205 |
+ |
errors=urlresults.getElementsByTagName('error') |
206 |
+ |
for error in errors: |
207 |
+ |
errormsg=error.childNodes[0].data |
208 |
+ |
if len(error.childNodes)>1: |
209 |
+ |
errormsg+=error.childNodes[1].data |
210 |
+ |
return errormsg |
211 |
+ |
|
212 |
+ |
def parse_lfn2pfn(self,urlresults): |
213 |
+ |
""" |
214 |
+ |
Parse the content of the result of lfn2pfn PhEDEx Data Service call |
215 |
+ |
|
216 |
+ |
input: DOM object with the content of the lfn2pfn call |
217 |
+ |
returns: PFN |
218 |
+ |
""" |
219 |
+ |
result = urlresults.getElementsByTagName('phedex') |
220 |
+ |
|
221 |
+ |
if not result: |
222 |
+ |
return [] |
223 |
+ |
result = result[0] |
224 |
+ |
pfn = None |
225 |
+ |
mapping = result.getElementsByTagName('mapping') |
226 |
+ |
for m in mapping: |
227 |
+ |
pfn=m.getAttribute("pfn") |
228 |
+ |
if pfn: |
229 |
+ |
return pfn |
230 |
+ |
|
231 |
+ |
def getStageoutPFN( self ): |
232 |
+ |
""" |
233 |
+ |
input: LFN,node name,protocol |
234 |
+ |
returns: PFN |
235 |
+ |
""" |
236 |
+ |
if self.usePhedex: |
237 |
+ |
params = {'node' : self.node , 'lfn': self.lfn , 'protocol': self.protocol} |
238 |
+ |
datasvc_lfn2pfn="%s/lfn2pfn"%self.datasvc_url |
239 |
+ |
fullurl="%s/lfn2pfn?node=%s&lfn=%s&protocol=%s"%(self.datasvc_url,self.node,self.lfn,self.protocol) |
240 |
+ |
domlfn2pfn = self.domPhedex(params,datasvc_lfn2pfn) |
241 |
+ |
if not domlfn2pfn : |
242 |
+ |
msg="Unable to get info from %s"%fullurl |
243 |
+ |
raise CrabException(msg) |
244 |
+ |
|
245 |
+ |
errormsg = self.parse_error(domlfn2pfn) |
246 |
+ |
if errormsg: |
247 |
+ |
msg="Error extracting info from %s due to: %s"%(fullurl,errormsg) |
248 |
+ |
raise CrabException(msg) |
249 |
+ |
|
250 |
+ |
stageoutpfn = self.parse_lfn2pfn(domlfn2pfn) |
251 |
+ |
if not stageoutpfn: |
252 |
+ |
msg ='Unable to get stageout path from TFC at Site %s \n'%self.node |
253 |
+ |
msg+=' Please alert the CompInfraSup group through their savannah %s \n'%self.FacOps_savannah |
254 |
+ |
msg+=' reporting: \n' |
255 |
+ |
msg+=' Summary: Unable to get user stageout from TFC at Site %s \n'%self.node |
256 |
+ |
msg+=' OriginalSubmission: stageout path is not retrieved from %s \n'%fullurl |
257 |
+ |
raise CrabException(msg) |
258 |
+ |
else: |
259 |
+ |
if self.sched in ['CAF','LSF','PBS'] : |
260 |
+ |
stageoutpfn = self.user_se_path+self.lfn |
261 |
+ |
else: |
262 |
+ |
stageoutpfn = 'srm://'+self.node+':'+self.user_port+self.user_se_path+self.lfn |
263 |
+ |
|
264 |
+ |
if ( stageoutpfn[-1] != '/' ) : stageoutpfn = stageoutpfn + '/' |
265 |
+ |
return stageoutpfn |
266 |
+ |
|
267 |
+ |
def getAuthoritativeSE(self): |
268 |
+ |
""" |
269 |
+ |
input: node name |
270 |
+ |
returns: AuthoritativeSE |
271 |
+ |
""" |
272 |
+ |
params = {'node' : self.node } |
273 |
+ |
datasvc_nodes="%s/nodes"%self.datasvc_url |
274 |
+ |
fullurl="%s/nodes/?node=%s"%(self.datasvc_url,self.node) |
275 |
+ |
domnodes = self.domPhedex(params,datasvc_nodes) |
276 |
+ |
|
277 |
+ |
if not domnodes : |
278 |
+ |
msg="Unable to get info from %s"%fullurl |
279 |
+ |
raise CrabException(msg) |
280 |
+ |
|
281 |
+ |
errormsg = self.parse_error(domnodes) |
282 |
+ |
if errormsg: |
283 |
+ |
msg="Error extracting info from %s due to: %s"%(fullurl,errormsg) |
284 |
+ |
raise CrabException(msg) |
285 |
+ |
result = domnodes.getElementsByTagName('phedex') |
286 |
+ |
if not result: |
287 |
+ |
return [] |
288 |
+ |
result = result[0] |
289 |
+ |
se = None |
290 |
+ |
node = result.getElementsByTagName('node') |
291 |
+ |
for m in node: |
292 |
+ |
se=m.getAttribute("se") |
293 |
+ |
if se: |
294 |
+ |
return se |
295 |
+ |
|
296 |
+ |
|
297 |
+ |
if __name__ == '__main__': |
298 |
+ |
""" |
299 |
+ |
Sort of unit testing to check Phedex API for whatever site and/or lfn. |
300 |
+ |
Usage: |
301 |
+ |
python PhEDExDatasvcInfo.py --node T2_IT_Bari --lfn /store/maremma |
302 |
+ |
|
303 |
+ |
""" |
304 |
+ |
import getopt,sys |
305 |
+ |
from crab_util import * |
306 |
+ |
import common |
307 |
+ |
klass_name = 'SchedulerGlite' |
308 |
+ |
klass = importName(klass_name, klass_name) |
309 |
+ |
common.scheduler = klass() |
310 |
+ |
|
311 |
+ |
lfn="/store/user/" |
312 |
+ |
node='T2_IT_Bari' |
313 |
+ |
valid = ['node=','lfn='] |
314 |
+ |
try: |
315 |
+ |
opts, args = getopt.getopt(sys.argv[1:], "", valid) |
316 |
+ |
except getopt.GetoptError, ex: |
317 |
+ |
print str(ex) |
318 |
+ |
sys.exit(1) |
319 |
+ |
for o, a in opts: |
320 |
+ |
if o == "--node": |
321 |
+ |
node = a |
322 |
+ |
if o == "--lfn": |
323 |
+ |
lfn = a |
324 |
+ |
|
325 |
+ |
mycfg_params = { 'USER.storage_element': node } |
326 |
+ |
dsvc = PhEDExDatasvcInfo(mycfg_params) |
327 |
+ |
dsvc.lfn = lfn |
328 |
+ |
print dsvc.getStageoutPFN() |
329 |
+ |
|