ViewVC Help

View File | Revision Log | Show Annotations | Root Listing

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

Comparing COMP/CRAB/python/cmscp.py (file contents):
Revision 1.1 by spiga, Sun Sep 21 10:34:04 2008 UTC vs.
Revision 1.63 by mcinquil, Thu Oct 8 15:15:17 2009 UTC

#	Line 1 | Line 1
1		#!/usr/bin/env python
2	<
3	<	import sys, getopt, string
4	<	import os, popen2
5	<	from ProdCommon.Storage.SEAPI.SElement import SElement, FullPath
2	>	import sys, os
3	>	from ProdCommon.Storage.SEAPI.SElement import SElement, FullPath
4		from ProdCommon.Storage.SEAPI.SBinterface import *
5	<
5	>	from ProdCommon.Storage.SEAPI.Exceptions import *
6
7
8		class cmscp:
9	<	def __init__(self, argv):
9	>	def __init__(self, args):
10		"""
11		cmscp
12	<
15	<	safe copy of local file in current directory to remote SE via lcg_cp/srmcp,
12	>	safe copy of local file  to/from remote SE via lcg_cp/srmcp,
13		including success checking  version also for CAF using rfcp command to copy the output to SE
14		input:
15		$1 middleware (CAF, LSF, LCG, OSG)
16	<	$2 local file (the physical path of output file respect to current working directory)
17	<	$3 file name (the output file name)
18	<	$4 remote SE_PATH (absolute)
19	<	$5 remote SE
20	<	$6 srm version (only in the case of LCG or OSG)
16	>	$2 local file (the absolute path of output file or just the name if it's in top dir)
17	>	$3 if needed: file name (the output file name)
18	>	$5 remote SE (complete endpoint)
19	>	$6 srm version
20	>	--lfn $LFNBaseName
21		output:
22		return 0 if all ok
23		return 60307 if srmcp failed
24		return 60303 if file already exists in the SE
25		"""
26	+
27		#set default
28	+	self.params = {"source":'', "destination":'','destinationDir':'', "inputFileList":'', "outputFileList":'', \
29	+	"protocol":'', "option":'', "middleware":'', "srm_version":'srmv2', "lfn":'' }
30		self.debug = 0
31	<	self.source = ''
32	<	self.destination = ''
33	<	self.file_to_copy = []
34	<	self.remote_file_name = []
35	<	self.protocol = ''
36	<	self.middleware = ''
37	<	self.srmv = ''
38	<
39	<	try:
40	<	opts, args = getopt.getopt(argv, "", ["source=", "destination=", "inputFileList=", "outputFileList=", \
41	<	"protocol=", "middleware=", "srm_version=", "debug", "help"])
42	<	except getopt.GetoptError:
43	<	print self.usage()
44	<	sys.exit(2)
45	<
46	<	self.setAndCheck(opts)
47	<
31	>	self.local_stage = 0
32	>	self.params.update( args )
33	>
34		return
35
36	<	def setAndCheck( self, opts ):
36	>	def processOptions( self ):
37		"""
38	<	Set and check command line parameter
38	>	check command line parameter
39		"""
40	<	if not opts :
41	<	print self.usage()
42	<	sys.exit()
43	<	for opt, arg in opts :
44	<	if opt  == "--help" :
45	<	print self.usage()
46	<	sys.exit()
47	<	elif opt == "--debug" :
48	<	self.debug = 1
49	<	elif opt == "--source" :
50	<	self.source = arg
51	<	elif opt == "--destination":
52	<	self.destination = arg
53	<	elif opt == "--inputFileList":
68	<	infile = arg
69	<	elif opt == "--outputFileList":
70	<	out_file
71	<	elif opt == "--protocol":
72	<	self.protocol = arg
73	<	elif opt == "--middleware":
74	<	self.middleware = arg
75	<	elif opt == "--srm_version":
76	<	self.srmv = arg
77	<
78	<	# source and dest cannot be undefined at same time
79	<	if self.source == '' and self.destination == '':
80	<	print self.usage()
81	<	sys.exit()
82	<	# if middleware is not defined --> protocol cannot be empty
83	<	if self.middleware == '' and self.protocol == '':
84	<	print self.usage()
85	<	sys.exit()
86	<	# input file must be defined
87	<	if infile == '':
88	<	print self.usage()
89	<	sys.exit()
40	>	if 'help' in self.params.keys(): HelpOptions()
41	>	if 'debug' in self.params.keys(): self.debug = 1
42	>	if 'local_stage' in self.params.keys(): self.local_stage = 1
43	>
44	>	# source and dest cannot be undefined at same time
45	>	if not self.params['source']  and not self.params['destination'] :
46	>	HelpOptions()
47	>	# if middleware is not defined --> protocol cannot be empty
48	>	if not self.params['middleware'] and not self.params['protocol'] :
49	>	HelpOptions()
50	>
51	>	# input file must be defined
52	>	if not self.params['inputFileList'] :
53	>	HelpOptions()
54		else:
55	<	if infile.find(','):
56	<	[self.file_to_copy.append(x.strip()) for x in infile.split(',')]
57	<	else:
58	<	self.file_to_copy.append(infile)
59	<
55	>	file_to_copy=[]
56	>	if self.params['inputFileList'].find(','):
57	>	[file_to_copy.append(x.strip()) for x in self.params['inputFileList'].split(',')]
58	>	else:
59	>	file_to_copy.append(self.params['inputFileList'])
60	>	self.params['inputFileList'] = file_to_copy
61	>
62	>	if not self.params['lfn'] and self.local_stage == 1 : HelpOptions()
63	>
64		## TO DO:
65		#### add check for outFiles
66		#### add map {'inFileNAME':'outFileNAME'} to change out name
67
100	–	return
68
69	<	def run( self ):
69	>	def run( self ):
70		"""
71	<	Check if running on UI (no $middleware) or
72	<	on WN (on the Grid), and take different action
71	>	Check if running on UI (no $middleware) or
72	>	on WN (on the Grid), and take different action
73		"""
74	<	if self.middleware :
75	<	results = self.stager()
74	>	self.processOptions()
75	>	if self.debug: print 'calling run() : \n'
76	>	# stage out from WN
77	>	if self.params['middleware'] :
78	>	results = self.stager(self.params['middleware'],self.params['inputFileList'])
79	>	self.finalReport(results)
80	>	# Local interaction with SE
81		else:
82	<	results = self.copy( self.file_to_copy, self.protocol )
82	>	results = self.copy(self.params['inputFileList'], self.params['protocol'], self.params['option'] )
83	>	return results
84
85	<	self.finalReport(results,self.middleware)
85	>	def checkLcgUtils( self ):
86	>	"""
87	>	_checkLcgUtils_
88	>	check the lcg-utils version and report
89	>	"""
90	>	import commands
91	>	cmd = "lcg-cp --version | grep lcg_util"
92	>	status, output = commands.getstatusoutput( cmd )
93	>	num_ver = -1
94	>	if output.find("not found") == -1 or status == 0:
95	>	temp = output.split("-")
96	>	version = ""
97	>	if len(temp) >= 2:
98	>	version = output.split("-")[1]
99	>	temp = version.split(".")
100	>	if len(temp) >= 1:
101	>	num_ver = int(temp[0])*10
102	>	num_ver += int(temp[1])
103	>	return num_ver
104
105	<	return
115	<
116	<	def setProtocol( self ):
105	>	def setProtocol( self, middleware ):
106		"""
107		define the allowed potocols based on $middlware
108	<	which depend on scheduler
108	>	which depend on scheduler
109		"""
110	<	if self.middleware.lower() in ['osg','lcg']:
111	<	supported_protocol = ['srm-lcg','srmv2']
112	<	elif self.middleware.lower() in ['lsf','caf']:
113	<	supported_protocol = ['rfio']
110	>	# default To be used with "middleware"
111	>	if self.debug:
112	>	print 'setProtocol() :\n'
113	>	print '\tmiddleware =  %s utils \n'%middleware
114	>
115	>	lcgOpt={'srmv1':'-b -D srmv1  -t 2400 --verbose',
116	>	'srmv2':'-b -D srmv2  -t 2400 --verbose'}
117	>	if self.checkLcgUtils() >= 17:
118	>	lcgOpt={'srmv1':'-b -D srmv1 --srm-timeout 2400 --sendreceive-timeout 2400 --connect-timeout 2400 --verbose',
119	>	'srmv2':'-b -D srmv2 --srm-timeout 2400 --sendreceive-timeout 2400 --connect-timeout 2400 --verbose'}
120	>
121	>	srmOpt={'srmv1':' -report ./srmcp.report -retry_timeout 480000 -retry_num 3 -streams_num=1 ',
122	>	'srmv2':' -report=./srmcp.report -retry_timeout=480000 -retry_num=3 '}
123	>	rfioOpt=''
124	>
125	>	supported_protocol = None
126	>	if middleware.lower() in ['osg','lcg','condor','sge']:
127	>	supported_protocol = [('srm-lcg',lcgOpt[self.params['srm_version']]),\
128	>	(self.params['srm_version'],srmOpt[self.params['srm_version']])]
129	>	elif middleware.lower() in ['lsf','caf']:
130	>	supported_protocol = [('rfio',rfioOpt)]
131	>	elif middleware.lower() in ['pbs']:
132	>	supported_protocol = [('rfio',rfioOpt),('local','')]
133	>	elif middleware.lower() in ['arc']:
134	>	supported_protocol = [('srmv2','-debug'),('srmv1','-debug')]
135		else:
136	<	## here we can add support for any kind of protocol,
136	>	## here we can add support for any kind of protocol,
137		## maybe some local schedulers need something dedicated
138		pass
139		return supported_protocol
140
141	<	def stager( self ):
141	>
142	>	def checkCopy (self, copy_results, len_list_files, prot, lfn='', se=''):
143		"""
144	<	Implement the logic for remote stage out
144	>	Checks the status of copy and update result dictionary
145		"""
146	<	protocols = self.setProtocol()
147	<	count=0
148	<	list_files = self.file_to_copy
149	<	results={}
150	<	for prot in protocols:
151	<	if self.debug: print 'Trying stage out with %s utils \n'%prot
152	<	copy_results = self.copy( list_files, prot )
153	<	list_retry = []
154	<	list_existing = []
155	<	list_ok = []
156	<	for file, dict in copy_results.iteritems():
157	<	er_code = dict['erCode']
158	<	if er_code == '60307': list_retry.append( file )
159	<	elif er_code == '60303': list_existing.append( file )
160	<	else:
161	<	list_ok.append(file)
162	<	reason = 'Copy succedeed with %s utils'%prot
163	<	upDict = self.updateReport(file, er_code, reason)
164	<	copy_results.update(upDict)
165	<	results.update(copy_results)
166	<	if len(list_ok) != 0:
167	<	msg = 'Copy of %s succedeed with %s utils\n'%(str(list_ok),prot)
168	<	# print msg
169	<	if len(list_ok) == len(list_files) :
170	<	break
146	>	list_retry = []
147	>	list_retry_localSE = []
148	>	list_not_existing = []
149	>	list_ok = []
150	>
151	>	if self.debug:
152	>	print 'in checkCopy() :\n'
153	>	for file, dict in copy_results.iteritems():
154	>	er_code = dict['erCode']
155	>	if er_code == '0':
156	>	list_ok.append(file)
157	>	reason = 'Copy succedeed with %s utils'%prot
158	>	dict['reason'] = reason
159	>	elif er_code == '60302':
160	>	list_not_existing.append( file )
161	>	elif er_code == '10041':
162	>	list_retry.append( file )
163	>	## WHAT TO DO IN GENERAL FAILURE CONDITION
164	>	else:
165	>	list_retry_localSE.append( file )
166	>
167	>	if self.debug:
168	>	print "\t file %s \n"%file
169	>	print "\t dict['erCode'] %s \n"%dict['erCode']
170	>	print "\t dict['reason'] %s \n"%dict['reason']
171	>
172	>	if (lfn != '') and (se != ''):
173	>	upDict = self.updateReport(file, er_code, dict['reason'], lfn, se)
174		else:
175	<	#       print 'Copy of files %s failed using %s...\n'%(str(list_retry)+str(list_existing),prot)
176	<	if len(list_retry): list_files = list_retry
177	<	else: break
178	<	count =+1
175	>	upDict = self.updateReport(file, er_code, dict['reason'])
176	>
177	>	copy_results.update(upDict)
178	>
179	>	msg = ''
180	>	if len(list_ok) != 0:
181	>	msg += '\tCopy of %s succedeed with %s utils\n'%(str(list_ok),prot)
182	>	if len(list_ok) != len_list_files :
183	>	msg += '\tCopy of %s failed using %s for files \n'%(str(list_retry),prot)
184	>	msg += '\tCopy of %s failed using %s : files not found \n'%(str(list_not_existing),prot)
185	>	if self.debug : print msg
186	>
187	>	return copy_results, list_ok, list_retry, list_retry_localSE
188	>
189	>	def LocalCopy(self, list_retry, results):
190	>	"""
191	>	Tries the stage out to the CloseSE
192	>	"""
193	>	if self.debug:
194	>	print 'in LocalCopy() :\n'
195	>	print '\t list_retry %s utils \n'%list_retry
196	>	print '\t len(list_retry) %s \n'%len(list_retry)
197	>
198	>	list_files = list_retry
199	>	self.params['inputFilesList']=list_files
200	>
201	>	### copy backup
202	>	from ProdCommon.FwkJobRep.SiteLocalConfig import loadSiteLocalConfig
203	>	siteCfg = loadSiteLocalConfig()
204	>	seName = siteCfg.localStageOut.get("se-name", None)
205	>	catalog = siteCfg.localStageOut.get("catalog", None)
206	>	implName = siteCfg.localStageOut.get("command", None)
207	>	if (implName == 'srm'):
208	>	implName='srmv1'
209	>	self.params['srm_version']=implName
210	>	##### to be improved ###############
211	>	if (implName == 'rfcp'):
212	>	self.params['middleware']='lsf'
213	>	####################################
214	>
215	>	self.params['protocol']=implName
216	>	tfc = siteCfg.trivialFileCatalog()
217	>
218	>	if self.debug:
219	>	print '\t siteCFG %s \n'%siteCfg
220	>	print '\t seName %s \n'%seName
221	>	print '\t catalog %s \n'%catalog
222	>	print "\t self.params['protocol'] %s \n"%self.params['protocol']
223	>	print '\t tfc %s '%tfc
224	>	print "\t self.params['inputFilesList'] %s \n"%self.params['inputFilesList']
225	>
226	>	file_backup=[]
227	>	for input in self.params['inputFilesList']:
228	>	file = self.params['lfn'] + os.path.basename(input)
229	>	surl = tfc.matchLFN(tfc.preferredProtocol, file)
230	>	file_backup.append(surl)
231	>	if self.debug:
232	>	print '\t lfn %s \n'%self.params['lfn']
233	>	print '\t file %s \n'%file
234	>	print '\t surl %s \n'%surl
235	>
236	>	destination=os.path.dirname(file_backup[0])
237	>	### FEDE added check for final /
238	>	if ( destination[-1] != '/' ) : destination = destination + '/'
239	>	#####################################
240	>	self.params['destination']=destination
241	>
242	>	if self.debug:
243	>	print "\t self.params['destination']%s \n"%self.params['destination']
244	>	print "\t self.params['protocol'] %s \n"%self.params['protocol']
245	>	print "\t self.params['option']%s \n"%self.params['option']
246	>
247	>	for prot, opt in self.setProtocol( self.params['middleware'] ):
248	>	if self.debug: print '\tIn LocalCopy trying the stage out with %s utils \n'%prot
249	>	localCopy_results = self.copy( self.params['inputFileList'], prot, opt )
250	>	if localCopy_results.keys() == [''] or localCopy_results.keys() == '' :
251	>	results.update(localCopy_results)
252	>	else:
253	>	localCopy_results, list_ok, list_retry, list_retry_localSE = self.checkCopy(localCopy_results, len(list_files), prot, self.params['lfn'], seName)
254	>	results.update(localCopy_results)
255	>	if len(list_ok) == len(list_files) :
256	>	break
257	>	if len(list_retry):
258	>	list_files = list_retry
259	>	else: break
260	>	if self.debug:
261	>	print "\t localCopy_results = %s \n"%localCopy_results
262	>
263	>	return results
264	>
265	>	def stager( self, middleware, list_files ):
266	>	"""
267	>	Implement the logic for remote stage out
268	>	"""
269
270	<	#### TODO Daniele
271	<	#check is something fails and created related dict
272	<	#      backup = self.analyzeResults(results)
273	<
274	<	#      if backup :
275	<	#          msg = 'WARNING: backup logic is under implementation\n'
276	<	#          #backupDict = self.backup()
277	<	#          ### NOTA DEVE RITORNARE UN DICT comprensivo di LFN SE Name
278	<	#          results.update(backupDict)
279	<	#          print msg
270	>	if self.debug:
271	>	print 'stager() :\n'
272	>	print '\tmiddleware %s\n'%middleware
273	>	print '\tlist_files %s\n'%list_files
274	>
275	>	results={}
276	>	for prot, opt in self.setProtocol( middleware ):
277	>	if self.debug: print '\tTrying the stage out with %s utils \n'%prot
278	>	copy_results = self.copy( list_files, prot, opt )
279	>	if copy_results.keys() == [''] or copy_results.keys() == '' :
280	>	results.update(copy_results)
281	>	else:
282	>	copy_results, list_ok, list_retry, list_retry_localSE = self.checkCopy(copy_results, len(list_files), prot)
283	>	results.update(copy_results)
284	>	if len(list_ok) == len(list_files) :
285	>	break
286	>	if len(list_retry):
287	>	list_files = list_retry
288	>	else: break
289	>
290	>	if self.local_stage:
291	>	if len(list_retry_localSE):
292	>	results = self.LocalCopy(list_retry_localSE, results)
293	>
294	>	if self.debug:
295	>	print "\t results %s \n"%results
296		return results
297
298		def initializeApi(self, protocol ):
299		"""
300	<	Instantiate storage interface
300	>	Instantiate storage interface
301		"""
302	<	source_prot = protocol
303	<	dest_prot = protocol
304	<	if self.source == '' : source_prot = 'local'
305	<	Source_SE  = self.storageInterface( self.source, source_prot )
306	<	if self.destination == '' : dest_prot = 'local'
307	<	Destination_SE = self.storageInterface( self.destination, dest_prot )
302	>	if self.debug : print 'initializeApi() :\n'
303	>	self.source_prot = protocol
304	>	self.dest_prot = protocol
305	>	if not self.params['source'] : self.source_prot = 'local'
306	>	Source_SE  = self.storageInterface( self.params['source'], self.source_prot )
307	>	if not self.params['destination'] : self.dest_prot = 'local'
308	>	Destination_SE = self.storageInterface( self.params['destination'], self.dest_prot )
309
310		if self.debug :
311	<	print '(source=%s,  protocol=%s)'%(self.source, source_prot)
312	<	print '(destination=%s,  protocol=%s)'%(self.destination, dest_prot)
311	>	msg  = '\t(source=%s,  protocol=%s)'%(self.params['source'], self.source_prot)
312	>	msg += '\t(destination=%s,  protocol=%s)'%(self.params['destination'], self.dest_prot)
313	>	print msg
314
315		return Source_SE, Destination_SE
316
317	<	def copy( self, list_file, protocol ):
317	>	def copy( self, list_file, protocol, options ):
318		"""
319	<	Make the real file copy using SE API
319	>	Make the real file copy using SE API
320		"""
321	+	msg = ""
322		if self.debug :
323	<	print 'copy(): using %s protocol'%protocol
324	<	Source_SE, Destination_SE = self.initializeApi( protocol )
323	>	msg  = 'copy() :\n'
324	>	msg += '\tusing %s protocol\n'%protocol
325	>	print msg
326	>	try:
327	>	Source_SE, Destination_SE = self.initializeApi( protocol )
328	>	except Exception, ex:
329	>	return self.updateReport('', '-1', str(ex))
330
331	<	# create remote dir
332	<	if protocol in ['gridftp','rfio']:
333	<	self.createDir( Destination_SE, protocol )
331	>	# create remote dir
332	>	if Destination_SE.protocol in ['gridftp','rfio','srmv2']:
333	>	try:
334	>	self.createDir( Destination_SE, Destination_SE.protocol )
335	>	except OperationException, ex:
336	>	return self.updateReport('', '60316', str(ex))
337	>	## when the client commands are not found (wrong env or really missing)
338	>	except MissingCommand, ex:
339	>	msg = "ERROR %s %s" %(str(ex), str(ex.detail))
340	>	return self.updateReport('', '10041', msg)
341
342		## prepare for real copy  ##
343	<	sbi = SBinterface( Source_SE, Destination_SE )
344	<	sbi_dest = SBinterface(Destination_SE)
343	>	try :
344	>	sbi = SBinterface( Source_SE, Destination_SE )
345	>	sbi_dest = SBinterface(Destination_SE)
346	>	sbi_source = SBinterface(Source_SE)
347	>	except ProtocolMismatch, ex:
348	>	msg  = "ERROR : Unable to create SBinterface with %s protocol"%protocol
349	>	msg += str(ex)
350	>	return self.updateReport('', '-1', msg)
351
352		results = {}
353	<	## loop over the complete list of files
354	<	for filetocopy in list_file:
355	<	if self.debug : print 'start real copy for %s'%filetocopy
356	<	ErCode, msg = self.checkFileExist( sbi_dest, os.path.basename(filetocopy) )
357	<	if ErCode == '0':
358	<	ErCode, msg = self.makeCopy( sbi, filetocopy )
359	<	if self.debug : print 'Copy results for %s is %s'%( os.path.basename(filetocopy) ,ErCode)
353	>	## loop over the complete list of files
354	>	for filetocopy in list_file:
355	>	if self.debug : print '\tStart real copy for %s'%filetocopy
356	>	try :
357	>	ErCode, msg = self.checkFileExist( sbi_source, sbi_dest, filetocopy, options )
358	>	except Exception, ex:
359	>	ErCode = '60307'
360	>	msg = str(ex)
361	>	if ErCode == '0':
362	>	ErCode, msg = self.makeCopy( sbi, filetocopy , options, protocol,sbi_dest )
363	>	if self.debug : print '\tCopy results for %s is %s'%( os.path.basename(filetocopy), ErCode)
364		results.update( self.updateReport(filetocopy, ErCode, msg))
365		return results
221	–
222	–	def updateReport(self, file, erCode, reason, lfn='', se='' ):
223	–	"""
224	–	Update the final stage out infos
225	–	"""
226	–	jobStageInfo={}
227	–	jobStageInfo['erCode']=erCode
228	–	jobStageInfo['reason']=reason
229	–	jobStageInfo['lfn']=lfn
230	–	jobStageInfo['se']=se
231	–
232	–	report = { file : jobStageInfo}
233	–	return report
366
235	–	def finalReport( self , results, middleware ):
236	–	"""
237	–	It should return a clear list of LFNs for each SE where data are stored.
238	–	allow "crab -copyLocal" or better "crab -copyOutput". TO_DO.
239	–	"""
240	–	if middleware:
241	–	outFile = open('cmscpReport.sh',"a")
242	–	cmscp_exit_status = 0
243	–	txt = ''
244	–	for file, dict in results.iteritems():
245	–	if dict['lfn']=='':
246	–	lfn = '$LFNBaseName/'+os.path.basename(file)
247	–	se  = '$SE'
248	–	else:
249	–	lfn = dict['lfn']+os.pat.basename(file)
250	–	se = dict['se']
251	–	#dict['lfn'] # to be implemented
252	–	txt +=  'echo "Report for File: '+file+'"\n'
253	–	txt +=  'echo "LFN: '+lfn+'"\n'
254	–	txt +=  'echo "StorageElement: '+se+'"\n'
255	–	txt += 'echo "StageOutExitStatusReason ='+dict['reason']+'" | tee -a $RUNTIME_AREA/$repo\n'
256	–	txt += 'echo "StageOutSE = '+dict['se']+'" >> $RUNTIME_AREA/$repo\n'
257	–	if dict['erCode'] != '0':
258	–	cmscp_exit_status = dict['erCode']
259	–	txt += '\n'
260	–	txt += 'export StageOutExitStatus='+str(cmscp_exit_status)+'\n'
261	–	txt +=  'echo "StageOutExitStatus = '+str(cmscp_exit_status)+'" | tee -a $RUNTIME_AREA/$repo\n'
262	–	outFile.write(str(txt))
263	–	outFile.close()
264	–	else:
265	–	for file, code in results.iteritems():
266	–	print 'error code = %s for file %s'%(code,file)
267	–	return
367
368		def storageInterface( self, endpoint, protocol ):
369		"""
370	<	Create the storage interface.
370	>	Create the storage interface.
371		"""
372	+	if self.debug : print 'storageInterface():\n'
373		try:
374		interface = SElement( FullPath(endpoint), protocol )
375	<	except Exception, ex:
376	<	msg = ''
377	<	if self.debug : msg = str(ex)+'\n'
378	<	msg += "ERROR : Unable to create interface with %s protocol\n"%protocol
279	<	print msg
375	>	except ProtocolUnknown, ex:
376	>	msg  = "ERROR : Unable to create interface with %s protocol"%protocol
377	>	msg += str(ex)
378	>	raise Exception(msg)
379
380		return interface
381
283	–	def checkDir(self, Destination_SE, protocol):
284	–	'''
285	–	ToBeImplemented NEEDED for castor
286	–	'''
287	–	return
288	–
382		def createDir(self, Destination_SE, protocol):
383		"""
384	<	Create remote dir for gsiftp/rfio REALLY TEMPORARY
385	<	this should be transparent at SE API level.
384	>	Create remote dir for gsiftp REALLY TEMPORARY
385	>	this should be transparent at SE API level.
386		"""
387	<	ErCode = '0'
388	<	msg_1 = ''
387	>	if self.debug : print 'createDir():\n'
388	>	msg = ''
389		try:
390		action = SBinterface( Destination_SE )
391		action.createDir()
392	<	if self.debug: print "The directory has been created using protocol %s\n"%protocol
393	<	except Exception, ex:
394	<	msg = ''
395	<	if self.debug : msg = str(ex)+'\n'
396	<	msg_1 = "ERROR: problem with the directory creation using %s protocol \n"%protocol
397	<	msg += msg_1
398	<	ErCode = '60316'
399	<	#print msg
400	<
401	<	return ErCode, msg_1
392	>	if self.debug: print "\tThe directory has been created using protocol %s"%protocol
393	>	except TransferException, ex:
394	>	msg  = "ERROR: problem with the directory creation using %s protocol "%protocol
395	>	msg += str(ex)
396	>	if self.debug :
397	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
398	>	dbgmsg += '\t'+str(ex.output)+'\n'
399	>	print dbgmsg
400	>	raise Exception(msg)
401	>	except OperationException, ex:
402	>	msg  = "ERROR: problem with the directory creation using %s protocol "%protocol
403	>	msg += str(ex)
404	>	if self.debug : print '\t'+msg+'\n\t'+str(ex.detail)+'\n'
405	>	raise Exception(msg)
406	>	except MissingDestination, ex:
407	>	msg  = "ERROR: problem with the directory creation using %s protocol "%protocol
408	>	msg += str(ex)
409	>	if self.debug : print '\t'+msg+'\n\t'+str(ex.detail)+'\n'
410	>	raise Exception(msg)
411	>	except AlreadyExistsException, ex:
412	>	if self.debug: print "\tThe directory already exist"
413	>	pass
414	>	return msg
415
416	<	def checkFileExist(self, sbi, filetocopy):
416	>	def checkFileExist( self, sbi_source, sbi_dest, filetocopy, option ):
417		"""
418	<	Check if file to copy already exist
419	<	"""
420	<	try:
421	<	check = sbi.checkExists(filetocopy)
316	<	except Exception, ex:
317	<	msg = ''
318	<	if self.debug : msg = str(ex)+'\n'
319	<	msg += "ERROR: problem with check File Exist using %s protocol \n"%protocol
320	<	# print msg
418	>	Check both if source file exist AND
419	>	if destination file ALREADY exist.
420	>	"""
421	>	if self.debug : print 'checkFileExist():\n'
422		ErCode = '0'
423		msg = ''
424	<	if check :
425	<	ErCode = '60303'
426	<	msg = "file %s already exist"%filetocopy
427	<	print msg
424	>	f_tocopy=filetocopy
425	>	if self.source_prot != 'local':f_tocopy = os.path.basename(filetocopy)
426	>	try:
427	>	checkSource = sbi_source.checkExists( f_tocopy , opt=option )
428	>	if self.debug : print '\tCheck for local file %s exist succeded \n'%f_tocopy
429	>	except OperationException, ex:
430	>	msg  ='ERROR: problems checkig if source file %s exist'%filetocopy
431	>	msg += str(ex)
432	>	if self.debug :
433	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
434	>	dbgmsg += '\t'+str(ex.output)+'\n'
435	>	print dbgmsg
436	>	raise Exception(msg)
437	>	except WrongOption, ex:
438	>	msg  ='ERROR problems checkig if source file % exist'%filetocopy
439	>	msg += str(ex)
440	>	if self.debug :
441	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
442	>	dbgmsg += '\t'+str(ex.output)+'\n'
443	>	print dbgmsg
444	>	raise Exception(msg)
445	>	except MissingDestination, ex:
446	>	msg  ='ERROR problems checkig if source file % exist'%filetocopy
447	>	msg += str(ex)
448	>	if self.debug : print '\t'+msg+'\n\t'+str(ex.detail)+'\n'
449	>	raise Exception(msg)
450	>	## when the client commands are not found (wrong env or really missing)
451	>	except MissingCommand, ex:
452	>	ErCode = '10041'
453	>	msg = "ERROR %s %s" %(str(ex), str(ex.detail))
454	>	return ErCode, msg
455	>	if not checkSource :
456	>	ErCode = '60302'
457	>	msg = "ERROR file %s do not exist"%os.path.basename(filetocopy)
458	>	return ErCode, msg
459	>	f_tocopy=filetocopy
460	>	if self.dest_prot != 'local':f_tocopy = os.path.basename(filetocopy)
461	>	try:
462	>	check = sbi_dest.checkExists( f_tocopy, opt=option )
463	>	if self.debug : print '\tCheck for remote file %s exist succeded \n'%f_tocopy
464	>	except OperationException, ex:
465	>	msg  = 'ERROR: problems checkig if file %s already exist'%filetocopy
466	>	msg += str(ex)
467	>	if self.debug :
468	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
469	>	dbgmsg += '\t'+str(ex.output)+'\n'
470	>	print dbgmsg
471	>	raise Exception(msg)
472	>	except WrongOption, ex:
473	>	msg  = 'ERROR problems checkig if file % already exist'%filetocopy
474	>	msg += str(ex)
475	>	if self.debug :
476	>	msg += '\t'+msg+'\n\t'+str(ex.detail)+'\n'
477	>	msg += '\t'+str(ex.output)+'\n'
478	>	raise Exception(msg)
479	>	except MissingDestination, ex:
480	>	msg  ='ERROR problems checkig if source file % exist'%filetocopy
481	>	msg += str(ex)
482	>	if self.debug : print '\t'+msg+'\n\t'+str(ex.detail)+'\n'
483	>	raise Exception(msg)
484	>	## when the client commands are not found (wrong env or really missing)
485	>	except MissingCommand, ex:
486	>	ErCode = '10041'
487	>	msg = "ERROR %s %s" %(str(ex), str(ex.detail))
488	>	return ErCode, msg
489	>	if check :
490	>	ErCode = '60303'
491	>	msg = "file %s already exist"%os.path.basename(filetocopy)
492
493	<	return ErCode,msg
493	>	return ErCode, msg
494
495	<	def makeCopy(self, sbi, filetocopy ):
495	>	def makeCopy(self, sbi, filetocopy, option, protocol, sbi_dest ):
496		"""
497	<	call the copy API.
497	>	call the copy API.
498		"""
499	<	path = os.path.dirname(filetocopy)
499	>	if self.debug : print 'makeCopy():\n'
500	>	path = os.path.dirname(filetocopy)
501		file_name =  os.path.basename(filetocopy)
502		source_file = filetocopy
503		dest_file = file_name ## to be improved supporting changing file name  TODO
504	<	if self.source == '' and path == '':
504	>	if self.params['source'] == '' and path == '':
505		source_file = os.path.abspath(filetocopy)
506	<	elif self.destination =='':
507	<	dest_file = os.path.join(os.getcwd(),file_name)
508	<	elif self.source != '' and self.destination != '' :
509	<	source_file = file_name
506	>	elif self.params['destination'] =='':
507	>	destDir = self.params.get('destinationDir',os.getcwd())
508	>	dest_file = os.path.join(destDir,file_name)
509	>	elif self.params['source'] != '' and self.params['destination'] != '' :
510	>	source_file = file_name
511	>
512		ErCode = '0'
513		msg = ''
514	+
515	+	if  self.params['option'].find('space_token')>0:
516	+	space_tocken=self.params['option'].split('=')[1]
517	+	if protocol == 'srmv2': option = '%s -space_tocken=%s'%(option,space_tocken)
518	+	if protocol == 'srm-lcg': option = '%s -S %s'%(option,space_tocken)
519		try:
520	<	pippo = sbi.copy( source_file , dest_file )
521	<	if self.protocol == 'srm' : self.checkSize( sbi, filetocopy )
522	<	except Exception, ex:
523	<	msg = ''
524	<	if self.debug : msg = str(ex)+'\n'
525	<	msg = "Problem copying %s file with %s command"%( filetocopy, protocol )
520	>	sbi.copy( source_file , dest_file , opt = option)
521	>	except TransferException, ex:
522	>	msg  = "Problem copying %s file" % filetocopy
523	>	msg += str(ex)
524	>	if self.debug :
525	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
526	>	dbgmsg += '\t'+str(ex.output)+'\n'
527	>	print dbgmsg
528		ErCode = '60307'
529	<	#print msg
530	<
529	>	except WrongOption, ex:
530	>	msg  = "Problem copying %s file" % filetocopy
531	>	msg += str(ex)
532	>	if self.debug :
533	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
534	>	dbgmsg += '\t'+str(ex.output)+'\n'
535	>	print dbgmsg
536	>	except SizeZeroException, ex:
537	>	msg  = "Problem copying %s file" % filetocopy
538	>	msg += str(ex)
539	>	if self.debug :
540	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
541	>	dbgmsg += '\t'+str(ex.output)+'\n'
542	>	print dbgmsg
543	>	ErCode = '60307'
544	>	## when the client commands are not found (wrong env or really missing)
545	>	except MissingCommand, ex:
546	>	ErCode = '10041'
547	>	msg  = "Problem copying %s file" % filetocopy
548	>	msg += str(ex)
549	>	if self.debug :
550	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
551	>	dbgmsg += '\t'+str(ex.output)+'\n'
552	>	print dbgmsg
553	>	except AuthorizationException, ex:
554	>	ErCode = '60307'
555	>	msg  = "Problem copying %s file" % filetocopy
556	>	msg += str(ex)
557	>	if self.debug :
558	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
559	>	dbgmsg += '\t'+str(ex.output)+'\n'
560	>	print dbgmsg
561	>	if ErCode == '0' and protocol.find('srmv') == 0:
562	>	remote_file_size = -1
563	>	local_file_size = os.path.getsize( source_file )
564	>	try:
565	>	remote_file_size = sbi_dest.getSize( dest_file, opt=option )
566	>	if self.debug : print '\t Check of remote size succeded for file %s\n'%dest_file
567	>	except TransferException, ex:
568	>	msg  = "Problem checking the size of %s file" % filetocopy
569	>	msg += str(ex)
570	>	if self.debug :
571	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
572	>	dbgmsg += '\t'+str(ex.output)+'\n'
573	>	print dbgmsg
574	>	ErCode = '60307'
575	>	except WrongOption, ex:
576	>	msg  = "Problem checking the size of %s file" % filetocopy
577	>	msg += str(ex)
578	>	if self.debug :
579	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
580	>	dbgmsg += '\t'+str(ex.output)+'\n'
581	>	print dbgmsg
582	>	ErCode = '60307'
583	>	if local_file_size != remote_file_size:
584	>	msg = "File size dosn't match: local size = %s ; remote size = %s " % (local_file_size, remote_file_size)
585	>	ErCode = '60307'
586	>
587	>	if ErCode != '0':
588	>	try :
589	>	self.removeFile( sbi_dest, dest_file, option )
590	>	except Exception, ex:
591	>	msg += '\n'+str(ex)
592		return ErCode, msg
593	<
594	<	'''
595	<	def checkSize()
593	>
594	>	def removeFile( self, sbi_dest, filetocopy, option ):
595	>	"""
596	>	"""
597	>	if self.debug : print 'removeFile():\n'
598	>	f_tocopy=filetocopy
599	>	if self.dest_prot != 'local':f_tocopy = os.path.basename(filetocopy)
600	>	try:
601	>	sbi_dest.delete( f_tocopy, opt=option )
602	>	if self.debug : '\t deletion of file %s succeeded\n'%str(filetocopy)
603	>	except OperationException, ex:
604	>	msg  ='ERROR: problems removing partially staged file %s'%filetocopy
605	>	msg += str(ex)
606	>	if self.debug :
607	>	dbgmsg  = '\t'+msg+'\n\t'+str(ex.detail)+'\n'
608	>	dbgmsg += '\t'+str(ex.output)+'\n'
609	>	print dbgmsg
610	>	raise Exception(msg)
611	>
612	>	return
613	>
614	>	def updateReport(self, file, erCode, reason, lfn='', se='' ):
615		"""
616	<	Using srm needed a check of the ouptut file size.
616	>	Update the final stage out infos
617		"""
618	<
619	<	echo "--> remoteSize = $remoteSize"
620	<	## for local file
621	<	localSize=$(stat -c%s "$path_out_file")
622	<	echo "-->  localSize = $localSize"
623	<	if [ $localSize != $remoteSize ]; then
624	<	echo "Local fileSize $localSize does not match remote fileSize $remoteSize"
625	<	echo "Copy failed: removing remote file $destination"
626	<	srmrm $destination
627	<	cmscp_exit_status=60307
628	<
629	<
630	<	echo "Problem copying $path_out_file to $destination with srmcp command"
376	<	StageOutExitStatusReason='remote and local file dimension not match'
377	<	echo "StageOutReport = `cat ./srmcp.report`"
378	<	'''
379	<	def backup(self):
380	<	"""
381	<	Check infos from TFC using existing api obtaining:
382	<	1)destination
383	<	2)protocol
618	>	jobStageInfo={}
619	>	jobStageInfo['erCode']=erCode
620	>	jobStageInfo['reason']=reason
621	>	jobStageInfo['lfn']=lfn
622	>	jobStageInfo['se']=se
623	>
624	>	report = { file : jobStageInfo}
625	>	return report
626	>
627	>	def finalReport( self , results ):
628	>	"""
629	>	It a list of LFNs for each SE where data are stored.
630	>	allow "crab -copyLocal" or better "crab -copyOutput". TO_DO.
631		"""
632	+	outFile = open('cmscpReport.sh',"a")
633	+	cmscp_exit_status = 0
634	+	txt = ''
635	+	for file, dict in results.iteritems():
636	+	reason = str(dict['reason'])
637	+	if str(reason).find("'") > -1:
638	+	reason = " ".join(reason.split("'"))
639	+	reason="'%s'"%reason
640	+	if file:
641	+	if dict['lfn']=='':
642	+	lfn = '$LFNBaseName/'+os.path.basename(file)
643	+	se  = '$SE'
644	+	else:
645	+	lfn = dict['lfn']+os.path.basename(file)
646	+	se = dict['se']
647	+	#dict['lfn'] # to be implemented
648	+	txt += 'echo "Report for File: '+file+'"\n'
649	+	txt += 'echo "LFN: '+lfn+'"\n'
650	+	txt += 'echo "StorageElement: '+se+'"\n'
651	+	txt += 'echo "StageOutExitStatusReason = %s" | tee -a $RUNTIME_AREA/$repo\n'%reason
652	+	txt += 'echo "StageOutSE = '+se+'" >> $RUNTIME_AREA/$repo\n'
653	+	#txt += 'export LFNBaseName='+lfn+'\n'
654	+	txt += 'export SE='+se+'\n'
655	+	### FEDE per CopyData ####
656	+
657	+	txt += 'export endpoint='+self.params['destination']+'\n'
658	+
659	+	if dict['erCode'] != '0':
660	+	cmscp_exit_status = dict['erCode']
661	+	else:
662	+	txt += 'echo "StageOutExitStatusReason = %s" | tee -a $RUNTIME_AREA/$repo\n'%reason
663	+	cmscp_exit_status = dict['erCode']
664	+	cmscp_exit_status = dict['erCode']
665	+	txt += '\n'
666	+	txt += 'export StageOutExitStatus='+str(cmscp_exit_status)+'\n'
667	+	txt +=  'echo "StageOutExitStatus = '+str(cmscp_exit_status)+'" | tee -a $RUNTIME_AREA/$repo\n'
668	+	outFile.write(str(txt))
669	+	outFile.close()
670		return
671
387	–	def usage(self):
672
673	<	msg="""
674	<	required parameters:
675	<	--source :: REMOTE           :
676	<	--dest   :: REMOTE           :
677	<	--debug             :
678	<	--inFile :: absPath : or name NOT RELATIVE PATH
679	<	--outFIle :: onlyNAME : NOT YET SUPPORTED
680	<
681	<	optional parameters
682	<	"""
683	<	return msg
673	>	def usage():
674	>
675	>	msg="""
676	>	cmscp:
677	>	safe copy of local file  to/from remote SE via lcg_cp/srmcp,
678	>	including success checking  version also for CAF using rfcp command to copy the output to SE
679	>
680	>	accepted parameters:
681	>	source           =
682	>	destination      =
683	>	inputFileList    =
684	>	outputFileList   =
685	>	protocol         =
686	>	option           =
687	>	middleware       =
688	>	srm_version      =
689	>	destinationDir   =
690	>	lfn=             =
691	>	local_stage      =  activate stage fall back
692	>	debug            =  activate verbose print out
693	>	help             =  print on line man and exit
694	>
695	>	mandatory:
696	>	* "source" and/or "destination" must always be defined
697	>	* either "middleware" or "protocol" must always be defined
698	>	* "inputFileList" must always be defined
699	>	* if "local_stage" = 1 also  "lfn" must be defined
700	>	"""
701	>	print msg
702	>
703	>	return
704	>
705	>	def HelpOptions(opts=[]):
706	>	"""
707	>	Check otps, print help if needed
708	>	prepare dict = { opt : value }
709	>	"""
710	>	dict_args = {}
711	>	if len(opts):
712	>	for opt, arg in opts:
713	>	dict_args[opt.split('--')[1]] = arg
714	>	if opt in ('-h','-help','--help') :
715	>	usage()
716	>	sys.exit(0)
717	>	return dict_args
718	>	else:
719	>	usage()
720	>	sys.exit(0)
721
722		if __name__ == '__main__' :
723	+
724	+	import getopt
725	+
726	+	allowedOpt = ["source=", "destination=", "inputFileList=", "outputFileList=", \
727	+	"protocol=","option=", "middleware=", "srm_version=", \
728	+	"destinationDir=", "lfn=", "local_stage", "debug", "help"]
729	+	try:
730	+	opts, args = getopt.getopt( sys.argv[1:], "", allowedOpt )
731	+	except getopt.GetoptError, err:
732	+	print err
733	+	HelpOptions()
734	+	sys.exit(2)
735	+
736	+	dictArgs = HelpOptions(opts)
737		try:
738	<	cmscp_ = cmscp(sys.argv[1:])
738	>	cmscp_ = cmscp(dictArgs)
739		cmscp_.run()
740	<	except:
741	<	pass
740	>	except Exception, ex :
741	>	print str(ex)
742

Diff Legend

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