ViewVC Help

View File | Revision Log | Show Annotations | Root Listing

root/cvsroot/COMP/CRAB/python/PhEDExDatasvcInfo.py

Comparing COMP/CRAB/python/PhEDExDatasvcInfo.py (file contents):
Revision 1.2 by afanfani, Fri Jul 18 10:00:54 2008 UTC vs.
Revision 1.34 by spiga, Mon Nov 30 16:54:53 2009 UTC

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

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines