forked from PanDAWMS/pilot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy patharia2cSiteMover.py
790 lines (697 loc) · 37.4 KB
/
aria2cSiteMover.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
import os, re, sys
import commands
from time import time
from subprocess import call
import urllib2
from subprocess import Popen, PIPE
import SiteMover
from futil import *
from PilotErrors import PilotErrors
from pUtil import tolog, readpar, getDirectAccessDic, extractPattern, getSiteInformation
from timed_command import timed_command
from FileStateClient import updateFileState
try:
from rucio.client import Client
except:
tolog("WARNING: Rucio python modules not available")
class replica:
""" Replica """
sfn = None
setname = None
fs = None
filesize = None
csumvalue = None
filename = None
scope = None
# placing the import lfc here breaks compilation on non-lfc sites
# import lfc
# Always called for a single file. If metalink file is found, then all files
# therein are downloaded ont he first single file call. On subsequent calls
# the relevant file should be there and is just checked.
# If no metalink file is found then a single http turl is built by regexp
# from surl
class aria2cSiteMover(SiteMover.SiteMover):
""" SiteMover for aria2c """
copyCommand = "aria2c"
checksum_command = "adler32"
has_mkdir = False
has_df = False
has_getsize = False
has_md5sum = True
has_chmod = False
timeout = 3600
""" get proxy """
try:
sslCert = os.environ['X509_USER_PROXY']
except:
sslCert = ""
try:
sslKey = os.environ['X509_USER_PROXY']
except:
sslKey = ""
try:
sslCertDir = os.environ['X509_CERT_DIR']
except:
os.environ['X509_CERT_DIR'] = "/etc/grid-security/certificates"
sslCertDir = os.environ['X509_CERT_DIR']
try:
rucio_account=os.environ['RUCIO_ACCOUNT']
except:
rucio_account=""
try:
site_name = os.environ['SITE_NAME']
except:
site_name = ""
def __init__(self, setup_path, *args, **kwrds):
self._setup = setup_path
self.copyCommand = 'aria2c'
self.commandInPATH()
rucio_account=self.rucio_account
tolog("Rucio account: %s" %(rucio_account))
if rucio_account == "":
tolog("!!FAILED!!2999!! Rucio account not set!")
raise Exception("!!FAILED!!2999!! Rucio account not set!")
cmd="curl -1 -i -H \"X-Rucio-Account: $RUCIO_ACCOUNT\" --cacert %s --cert %s --key %s --capath %s -X GET https://rucio-auth-prod.cern.ch/auth/x509_proxy| grep 'X-Rucio-Auth-Token:'"%(self.sslKey,self.sslKey,self.sslKey,self.sslCertDir)
tolog("Command to be launched: %s" %(cmd))
token_rucio_cmd=Popen(cmd,stdout=PIPE,stderr=PIPE, shell=True)
token_rucio, stderr= token_rucio_cmd.communicate()
if token_rucio:
if '\r' in token_rucio:
pos2print=token_rucio.find('\r')
token_rucio=token_rucio[:pos2print]
elif '\n' in token_rucio:
pos2print=token_rucio.find('\n')
pos2print=token_rucio.find("CN")
token_rucio2print=token_rucio[:pos2print]+'(Hidden token)'
tolog("Token on file: %s" %(token_rucio2print))
if os.path.exists('token_file'):
os.remove('token_file')
try:
token_file=open('token_file', 'w')
except IOError, e:
tolog ("!!WARNING!! Failed to create file: %s"%(e))
raise Exception("!!FAILED!!1099!! Cannot create file for registering token!")
else:
token_file.write(token_rucio)
else:
tolog("In __init__: Std error from curl: %s" %(stderr))
tolog("!!FAILED!!2999!! Cannot get Rucio token!")
raise Exception("!!FAILED!!2999!! Cannot get Rucio token!")
def commandInPATH(self):
_cmd_str = 'which %s'%self.copyCommand
tolog("Executing command: %s" % (_cmd_str))
s, o = commands.getstatusoutput(_cmd_str)
if s != 0:
tolog("aria2c not found in PATH")
cvmfs_aria2c = '/cvmfs/atlas.cern.ch/repo/sw/local/x86_64-slc6-gcc44-opt/bin/aria2c'
if os.path.exists(cvmfs_aria2c):
tolog("Using %s"%cvmfs_aria2c)
self.copyCommand = cvmfs_aria2c
else:
tolog("aria2c not in PATH or %s"%cvmfs_aria2c)
def get_timeout(self):
return self.timeout
def check_space(self, ub):
""" For when space availability is not verifiable """
return 999999
def getSurl2httpsMap(self):
"""This will come from AGIS but for now from a url, wih fallback to hard-coded value"""
s2hurl = 'http://walkerr.web.cern.ch/walkerr/surl2https.py'
try:
u = urllib2.urlopen(s2hurl)
exec u.read()
except:
tolog('Problem getting surl2https map from: %s'%s2hurl)
surl2https_map = {}
if len(surl2https_map) > 0:
tolog('Using surl2https map from %s of length %d'%(s2hurl,len(surl2https_map)))
self.surl2https_map = surl2https_map
else:
tolog('surl2https_map not set or zero length. Using default.')
self.surl2https_map = {
'atlassrm-fzk.gridka.de':('srm://atlassrm-fzk.gridka.de(:8443/srm/managerv2\?SFN=)*',
'https://f01-060-114-e.gridka.de:2880'),
'srm.ndgf.org':('srm://srm.ndgf.org(:8443/srm/managerv2\?SFN=)*','https://fozzie.ndgf.org:2881'),
'srm.grid.sara.nl':('srm://srm.grid.sara.nl(:8443/srm/managerv2\?SFN=)*','https://bee34.grid.sara.nl'),
'lcg-lrz-se.lrz-muenchen.de':('srm://lcg-lrz-se.lrz.de(:8443/srm/managerv2\?SFN=)*',
'http://lcg-lrz-dc66.grid.lrz-muenchen.de'),
'dcache-se-atlas.desy.de':('srm://dcache-se-atlas.desy.de(:8443/srm/managerv2\?SFN=)*/pnfs/desy.de/atlas',
'https://dcache-door-atlas19.desy.de:2880'),
'golias100.farm.particle.cz':('srm://golias100.farm.particle.cz','https://golias100.farm.particle.cz'),
'grid-se.physik.uni-wuppertal.de':('srm://grid-se.physik.uni-wuppertal.de(:8443/srm/managerv2\?SFN=)*/pnfs/physik.uni-wuppertal.de/data','https://grid-se.physik.uni-wuppertal.de:2881')
}
def surls2metalink(self,replicas,metalinkFile):
""" Convert list of replicas (of multiple files) to metalink
Input argument, replicas, is dict with guid as key, and a list of surls
Mappings from surl to https turl will come from ddm eventually
to cover surls from remote SEs.
For now just add the mapping for the local SE from copysetup.
"""
site_name=self.site_name
local_se_token=site_name+"_DATADISK"
tolog("local SE token: %s"%(local_se_token))
# self.surl2https_map has key is srm hostname, then tuple of (from,to) regexp replace
dirAcc = getDirectAccessDic(readpar('copysetupin'))
if not dirAcc:
dirAcc = getDirectAccessDic(readpar('copysetup'))
# extract srm host for key
srmhost=None
if dirAcc:
srmhost = self.hostFromSurl(dirAcc['oldPrefix'])
for guid in replicas.keys():
reps = replicas[guid]
tolog("Got replicas=%s for guid=%s" % (str(reps), guid))
try:
token_file=open('token_file', 'r')
except IOError, e:
tolog ("!!WARNING!! Failed to open file: %s"%(e))
raise Exception("!!FAILED!!1099!! Cannot open file with token!")
else:
token_rucio=token_file.readline()
pos2print=token_rucio.find("CN")
token_rucio2print=token_rucio[:pos2print]+'(Hidden token)'
tolog("Token I am using: %s" %(token_rucio2print))
httpredirector = readpar('httpredirector')
if not httpredirector:
cmd = "curl -1 -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://rucio-lb-prod.cern.ch/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,reps[0].scope,reps[0].filename)
cmd2print = "curl -1 -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://rucio-lb-prod.cern.ch/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio2print,reps[0].scope,reps[0].filename)
else:
if "http" in httpredirector:
tolog("HTTP redirector I am using: %s" %(httpredirector))
cmd = "curl -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem %s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,httpredirector,reps[0].scope,reps[0].filename)
cmd2print = "curl -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem %s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio2print,httpredirector,reps[0].scope,reps[0].filename)
else:
tolog("HTTP redirector I am using: %s" %(httpredirector))
cmd = "curl -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://%s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,httpredirector,reps[0].scope,reps[0].filename)
cmd2print = "curl -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://%s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio2print,httpredirector,reps[0].scope,reps[0].filename)
tolog("curl command to be executed: %s" %(cmd2print))
metalink_cmd=Popen(cmd, stdout=PIPE,stderr=PIPE, shell=True)
metalink, stderr=metalink_cmd.communicate()
tolog("Metalink given by rucio %s" %(metalink))
if not "location" in metalink:
tolog("In surls2metalink: command std error: %s" %(stderr))
tolog("!!WARNING!!1099!! No metalink to download file, or error in metalink!")
raise Exception("!!FAILED!!1099!! No metalink to download file, or error in metalink!")
else:
mlfile = open(metalinkFile,'w')
mlfile.write(metalink)
mlfile.close()
def hostFromSurl(self,surl):
re_srmhost = re.compile('^srm://([^/|:|\(]*)')
srmhost=re_srmhost.search(surl)
if srmhost:
return srmhost.group(1)
else:
return None
def get_data(self, gpfn, lfn, path, fsize=0, fchecksum=0, guid=0, fscope=None, **pdict):
""" copy input file from SE to local dir """
# determine which timeout option to use
timeout_option = "--connect-timeout 300 --timeout %d" % (self.timeout)
sslCert = self.sslCert
sslKey = self.sslKey
sslCertDir = self.sslCertDir
# used aria2c options:
# --certificate Client certificate file and password (SSL)(proxy)
# --private-key user proxy again
# --ca-certificate: concatenate *.0 in cert dir to make bundle
# --out: <file> Write output to <file> instead of stdout
# --dir: output directory, needed when multiple files(metalink)
# --continue: if file is already there (from previous) then success
# --auto-file-renaming=false : don't rename existing file
error = PilotErrors()
pilotErrorDiag = ""
# Get input parameters from pdict
token = pdict.get('token', None)
jobId = pdict.get('jobId', '')
scope = pdict.get('scope', '')
workDir = pdict.get('workDir', '')
proxycheck = pdict.get('proxycheck', False)
# try to get the direct reading control variable (False for direct reading mode; file should not be copied)
useCT = pdict.get('usect', True)
prodDBlockToken = pdict.get('access', '')
# get the Rucio tracing report
try:
report = pdict['report']
except:
report = {}
else:
# set the proper protocol
report['protocol'] = 'aria2c'
# mark the relative start
report['catStart'] = time()
# the current file
report['filename'] = lfn
# guid
report['guid'] = guid.replace('-','')
# get a proper envsetup
envsetup = self.getEnvsetup(get=True)
#if proxycheck:
# # do we have a valid proxy?
# s, pilotErrorDiag = self.verifyProxy(envsetup=envsetup)
# if s != 0:
# self.prepareReport('PROXYFAIL', report)
# return s, pilotErrorDiag
#else:
# tolog("Proxy verification turned off")
tolog("Proxy verification turned off")
getfile = gpfn
if path == '': path = './'
fullname = os.path.join(path, lfn)
# should the root file be copied or read directly by athena?
directIn, useFileStager = self.getTransferModes()
if directIn:
if useCT:
directIn = False
tolog("Direct access mode is switched off (file will be transferred with the copy tool)")
updateFileState(lfn, workDir, jobId, mode="transfer_mode", state="copy_to_scratch", ftype="input")
else:
# determine if the file is a root file according to its name
rootFile = self.isRootFileName(lfn)
if prodDBlockToken == 'local' or not rootFile:
directIn = False
tolog("Direct access mode has been switched off for this file (will be transferred with the copy tool)")
updateFileState(lfn, workDir, jobId, mode="transfer_mode", state="copy_to_scratch", ftype="input")
elif rootFile:
tolog("Found root file according to file name: %s (will not be transferred in direct reading mode)" % (lfn))
report['relativeStart'] = None
report['transferStart'] = None
self.prepareReport('FOUND_ROOT', report)
if useFileStager:
updateFileState(lfn, workDir, jobId, mode="transfer_mode", state="file_stager", ftype="input")
else:
updateFileState(lfn, workDir, jobId, mode="transfer_mode", state="remote_io", ftype="input")
return 0, pilotErrorDiag
else:
tolog("Normal file transfer")
# Build ca bundle if not already there
cabundleFile='cabundle.pem'
if not os.path.exists(cabundleFile):
_cmd_str = 'cat %s/*.0 > %s'%(sslCertDir,cabundleFile)
tolog("Executing command: %s" % (_cmd_str))
s, o = commands.getstatusoutput(_cmd_str)
# If metalink file not created(including all inputs)
# then make one just for this input
if os.path.exists('AllInput.xml.meta4'):
metalink='AllInput.xml.meta4'
else:
tolog("Getting metalink from Rucio")
rep = replica()
rep.sfn = gpfn
rep.filesize = fsize
rep.filename = lfn
rep.csumvalue = fchecksum
if fscope:
rep.scope = fscope
else:
#scope = extractPattern(gpfn,r'\/rucio\/(.+)\/[a-zA-Z0-9]{2}\/[a-zA-Z0-9]{2}\/')
rep.scope = scope.replace("/",".")
replicas = {guid:[rep]}
self.surls2metalink(replicas,'oneInput.xml.meta4')
metalink='oneInput.xml.meta4'
# build the copy command
#search how many links are available for download
word_occour=0
metaL_file= open(metalink)
for line in metaL_file:
for word in line.strip().split():
if word in ("<url"):
word_occour +=1
tolog("number of links: %s, using only the first" % (str(word_occour)))
#--check-certificate=false makes it easier(sles11)
_cmd_str = '%s -j 1 --ca-certificate=%s --certificate=%s --private-key=%s --auto-file-renaming=false --continue --server-stat-of=aria2cperf.txt %s'%(self.copyCommand, cabundleFile,sslCert,sslCert,metalink)
# invoke the transfer commands
report['relativeStart'] = time()
report['transferStart'] = time()
tolog("Executing command: %s" % (_cmd_str))
s, o = commands.getstatusoutput(_cmd_str)
tolog(o)
if s != 0:
tolog("!!WARNING!!2990!! Command failed: %s" % (_cmd_str))
check_syserr(s, o)
pilotErrorDiag = "aria2c failed: %s" % (o)
tolog("!!WARNING!!2999!! %s" % (pilotErrorDiag))
ec = error.ERR_STAGEINFAILED
return ec, pilotErrorDiag
report['validateStart'] = time()
# get the checksum type (md5sum or adler32)
if fchecksum != 0 and fchecksum != "":
csumtype = self.getChecksumType(fchecksum)
else:
csumtype = "default"
if (fsize != 0 or fchecksum != 0) and self.doFileVerifications():
loc_filename = lfn
dest_file = os.path.join(path, loc_filename)
# get the checksum type (md5sum or adler32)
if fchecksum != 0 and fchecksum != "":
csumtype = self.getChecksumType(fchecksum)
else:
csumtype = "default"
# get remote file size and checksum
ec, pilotErrorDiag, dstfsize, dstfchecksum = self.getLocalFileInfo(dest_file, csumtype=csumtype)
if ec != 0:
self.prepareReport('LOCAL_FILE_INFO_FAIL', report)
return ec, pilotErrorDiag
# compare remote and local file size
if long(fsize) != 0 and long(dstfsize) != long(fsize):
pilotErrorDiag = "Remote and local file sizes do not match for %s (%s != %s)" %\
(os.path.basename(gpfn), str(dstfsize), str(fsize))
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
self.prepareReport('FS_MISMATCH', report)
return error.ERR_GETWRONGSIZE, pilotErrorDiag
# compare remote and local file checksum
if fchecksum and dstfchecksum != fchecksum and not self.isDummyChecksum(fchecksum):
pilotErrorDiag = "Remote and local checksums (of type %s) do not match for %s (%s != %s)" %\
(csumtype, os.path.basename(gpfn), dstfchecksum, fchecksum)
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
# report corrupted file to consistency server
self.reportFileCorruption(gpfn)
if csumtype == "adler32":
self.prepareReport('AD_MISMATCH', report)
return error.ERR_GETADMISMATCH, pilotErrorDiag
else:
self.prepareReport('MD5_MISMATCH', report)
return error.ERR_GETMD5MISMATCH, pilotErrorDiag
updateFileState(lfn, workDir, jobId, mode="file_state", state="transferred", ftype="input")
self.prepareReport('DONE', report)
return 0, pilotErrorDiag
def put_data(self, source, destination, fsize=0, fchecksum=0, **pdict):
""" copy output file from disk to local SE """
# function is based on dCacheSiteMover put function
error = PilotErrors()
pilotErrorDiag = ""
# Get input parameters from pdict
lfn = pdict.get('lfn', '')
guid = pdict.get('guid', '')
token = pdict.get('token', '')
scope = pdict.get('scope', '')
dsname = pdict.get('dsname', '')
testLevel = pdict.get('testLevel', '0')
extradirs = pdict.get('extradirs', '')
proxycheck = pdict.get('proxycheck', False)
experiment = pdict.get('experiment', '')
analysisJob = pdict.get('analJob', False)
prodSourceLabel = pdict.get('prodSourceLabel', '')
# get the site information object
si = getSiteInformation(experiment)
tolog("put_data received prodSourceLabel=%s" % (prodSourceLabel))
if prodSourceLabel == 'ddm' and analysisJob:
tolog("Treating PanDA Mover job as a production job during stage-out")
analysisJob = False
# get the Rucio tracing report
try:
report = pdict['report']
except:
report = {}
else:
# set the proper protocol
report['protocol'] = 'curl'
# mark the relative start
report['catStart'] = time()
# the current file
report['filename'] = lfn
# guid
report['guid'] = guid.replace('-','')
# preparing variables
if fsize == 0 or fchecksum == 0:
ec, pilotErrorDiag, fsize, fchecksum = self.getLocalFileInfo(source, csumtype="adler32")
if ec != 0:
self.prepareReport('LOCAL_FILE_INFO_FAIL', report)
return self.put_data_retfail(ec, pilotErrorDiag)
# now that the file size is known, add it to the tracing report
report['filesize'] = fsize
# get the checksum type
if fchecksum != 0 and fchecksum != "":
csumtype = self.getChecksumType(fchecksum)
else:
csumtype = "default"
# get a proper envsetup
envsetup = self.getEnvsetup()
#if proxycheck:
# s, pilotErrorDiag = self.verifyProxy(envsetup=envsetup, limit=2)
# if s != 0:
# self.prepareReport('NO_PROXY', report)
# return self.put_data_retfail(error.ERR_NOPROXY, pilotErrorDiag)
#else:
# tolog("Proxy verification turned off")
tolog("Proxy verification turned off")
filename = os.path.basename(source)
# get all the proper paths
ec, pilotErrorDiag, tracer_error, dst_gpfn, lfcdir, surl = si.getProperPaths(error, analysisJob, token, prodSourceLabel, dsname, filename, scope=scope, sitemover=self) # quick workaround
if ec != 0:
self.prepareReport(tracer_error, report)
return self.put_data_retfail(ec, pilotErrorDiag)
#here begins the new magic... from Vincenzo Lavorini
sitemover = SiteMover.SiteMover()
v_path = sitemover.getPathFromScope(scope, filename)
rucio_c = Client()
if "ATLAS" in token:
token_ok=token[+5:]
else:
token_ok=token
local_se_token=self.site_name+"_"+token_ok
v_hostname= [j['hostname'] for j in rucio_c.get_protocols(local_se_token)]
v_port= [j['port'] for j in rucio_c.get_protocols(local_se_token)]
v_prefix= [j['prefix'] for j in rucio_c.get_protocols(local_se_token)]
v_address= "https://%s:%s%s"%(v_hostname[0],v_port[0],v_prefix[0])
tolog("prova1 address is %s" % (v_address))
if "rucio/" in v_address and "/rucio" in v_path:
v_address=v_address[:-7]
tolog("prova2 address is %s" % (v_address))
elif "rucio" in v_address and "rucio" in v_path :
v_address=v_address[:-6]
tolog("prova3 address is %s" % (v_address))
full_http_surl=v_address+v_path
tolog("prova3 full_http__surl is %s" % (full_http_surl))
full_surl =surl
if full_surl[:len('token:')] == 'token:':
# remove the space token (e.g. at Taiwan-LCG2) from the SURL info
full_surl = full_surl[full_surl.index('srm://'):]
if testLevel == "1":
source = "thisisjustatest"
# determine which timeout option to use
#commented by Lavorini timeout_option = "--connect-timeout 300 --max-time %d" % (self.timeout)
timeout_option = "--connect-timeout 300"
sslCert = self.sslCert
sslKey = self.sslKey
sslCertDir = self.sslCertDir
# check htcopy if it is existed or env is set properly
#_cmd_str = 'which htcopy'
#try:
# s, o = commands.getstatusoutput(_cmd_str)
#except Exception, e:
# tolog("!!WARNING!!2990!! Exception caught: %s (%d, %s)" % (str(e), s, o))
# o = str(e)
#if s != 0:
# tolog("!!WARNING!!2990!! Command failed: %s" % (_cmd_str))
# o = o.replace('\n', ' ')
# tolog("!!WARNING!!2990!! check PUT command failed. Status=%s Output=%s" % (str(s), str(o)))
#return 999999
# cleanup the SURL if necessary (remove port and srm substring)
#if token:
# used lcg-cp options:
# --srcsetype: specify SRM version
# --verbose: verbosity on
# --vo: specifies the Virtual Organization the user belongs to
# -s: space token description
# -b: BDII disabling
# -t: time-out
# (lcg-cr) -l: specifies the Logical File Name associated with the file. If this option is present, an entry is added to the LFC
# -g: specifies the Grid Unique IDentifier. If this option is not present, a GUID is generated internally
# -d: specifies the destination. It can be the Storage Element fully qualified hostname or an SURL. In the latter case,
# the scheme can be sfn: for a classical SE or srm:. If only the fully qualified hostname is given, a filename is
# generated in the same format as with the Replica Manager
# _cmd_str = '%s lcg-cr --verbose --vo atlas -T srmv2 -s %s -b -t %d -l %s -g %s -d %s file:%s' %\
# (envsetup, token, self.timeout, lfclfn, guid, surl, fppfn)
# usage: lcg-cp [-h,--help] [-i,--insecure] [-c,--config config_file]
# [-n nbstreams] [-s,--sst src_spacetokendesc] [-S,--dst dest_spacetokendesc]
# [-D,--defaultsetype se|srmv1|srmv2] [-T,--srcsetype se|srmv1|srmv2] [-U,--dstsetype se|srmv1|srmv2]
# [-b,--nobdii] [-t timeout] [-v,--verbose] [-V,--vo vo] [--version] src_file dest_file
# surl = putfile[putfile.index('srm://'):]
#_cmd_str = '%s htcopy --ca-path %s --user-cert %s --user-key %s "%s?spacetoken=%s"' % (envsetup, sslCertDir, sslCert, sslKey, full_http_surl, token)
#_cmd_str = '%s lcg-cp --verbose --vo atlas -b %s -U srmv2 -S %s file://%s %s' % (envsetup, timeout_option, token, source, full_surl)
#else:
# surl is the same as putfile
#_cmd_str = '%s htcopy --ca-path %s --user-cert %s --user-key %s "%s"' % (envsetup, sslCertDir, sslCert, sslKey, full_http_surl)
#_cmd_str = '%s lcg-cp --vo atlas --verbose -b %s -U srmv2 file://%s %s' % (envsetup, timeout_option, source, full_surl)
_cmd_str = 'curl -1 --verbose --cert %s --key %s --cacert %s --capath %s -L %s -T %s' % (self.sslKey,self.sslKey,self.sslKey,self.sslCertDir,full_http_surl, source)
tolog("Executing command: %s" % (_cmd_str))
t0 = os.times()
_cmd=Popen(_cmd_str,stdout=PIPE,stderr=PIPE, shell=True )
_cmd_out, _cmd_stderr= _cmd.communicate()
report['relativeStart'] = time()
report['transferStart'] = time()
report['validateStart'] = time()
t1 = os.times()
t = t1[4] - t0[4]
tolog("Curl command output = %s" % (_cmd_out))
tolog("Command finished after %f s" % (t))
if "bytes uploaded" not in _cmd_out:
tolog("!!WARNING!!1137!! Command failed: %s" % (_cmd_str))
'''
# check if file was partially transferred, if so, remove it
_ec = self.removeFile(envsetup, self.timeout, dst_gpfn)
if _ec == -2:
pilotErrorDiag += "(failed to remove file) " # i.e. do not retry stage-out
if "Could not establish context" in o:
pilotErrorDiag += "Could not establish context: Proxy / VO extension of proxy has probably expired"
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
self.prepareReport('CONTEXT_FAIL', report)
return self.put_data_retfail(error.ERR_NOPROXY, pilotErrorDiag)
elif "No such file or directory" in o:
pilotErrorDiag += "No such file or directory: %s" % (o)
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
self.prepareReport('NO_FILE_DIR', report)
return self.put_data_retfail(error.ERR_STAGEOUTFAILED, pilotErrorDiag)
elif "globus_xio: System error" in o:
pilotErrorDiag += "Globus system error: %s" % (o)
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
self.prepareReport('GLOBUS_FAIL', report)
return self.put_data_retfail(error.ERR_PUTGLOBUSSYSERR, pilotErrorDiag)
else:
if len(o) == 0 and t >= self.timeout:
pilotErrorDiag += "Copy command self timed out after %d s" % (t)
tolog("!!WARNING!!2990!! %s" % (pilotErrorDiag))
self.prepareReport('CP_TIMEOUT', report)
return self.put_data_retfail(error.ERR_PUTTIMEOUT, pilotErrorDiag)
else:
if len(o) == 0:
pilotErrorDiag += "Copy command returned error code %d but no output" % (ec)
else:
pilotErrorDiag += o
self.prepareReport('CP_ERROR', report)
return self.put_data_retfail(error.ERR_STAGEOUTFAILED, pilotErrorDiag)
'''
verified = False
#getting the remote checksum from Rucio:
token_file=open('token_fle', 'r')
token_rucio=token_file.readline()
pos2print=token_rucio.find("CN")
token_rucio2print=token_rucio[:pos2print]+'(Hidden token)'
tolog("Token I am using: %s" %(token_rucio2print))
httpredirector = readpar('httpredirector')
trial_n=1
remote_checksum="none"
while (remote_checksum == "none" and trial_n<8):
trial_n+=1
if not httpredirector:
#cmd = "curl -v -1 -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://rucio-lb-prod.cern.ch/replicas/%s/%s?select=geoip |awk \'{FS=\"hash type=\"}; {print $2}\' |awk \'{FS=\">\"}; {print $2}\' |awk \'{FS=\"<\"} {print $1}\'| grep -v \'^$\'"%(token_rucio,scope,filename)
cmd = "curl -v -1 -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://rucio-lb-prod.cern.ch/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,scope,filename)
cmd2print = "curl -v -1 -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://rucio-lb-prod.cern.ch/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio2print,scope,filename)
else:
if "http" in httpredirector:
tolog("HTTP redirector I am using: %s" %(httpredirector))
cmd = "curl -v -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem %s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,httpredirector,scope,filename)
cmd2print = "curl -v -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem %s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucioi2print,httpredirector,scope,filename)
else:
tolog("HTTP redirector I am using: %s" %(httpredirector))
cmd = "curl -v -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://%s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio,httpredirector,reps[0].scope,reps[0].filename)
cmd2print = "curl -v -1 -v -H \"%s\" -H 'Accept: application/metalink4+xml' --cacert cabundle.pem https://%s/replicas/%s/%s?select=geoip&schemes=https,http "%(token_rucio2print,httpredirector,reps[0].scope,reps[0].filename)
tolog("Getting remote checksum: command to be executed: %s" %(cmd2print))
checksum_cmd=Popen(cmd, stdout=PIPE,stderr=PIPE, shell=True)
remote_checksum, stderr=checksum_cmd.communicate()
tolog("Remote checksum as given by rucio %s" %(remote_checksum))
if (remote_checksum == "none"):
tolog("In checking checksum: command std error: %s" %(stderr))
pilotErrorDiag = "Cannot get the checksum of file on SE"
tolog("!!WARNING!!1137!! %s" % (pilotErrorDiag))
tolog("!!WARNING!!1137!! trial numebr %s" % (trial_n))
time.sleep(3)
# try to get the remote checksum with lcg-get-checksum
#remote_checksum = self.lcgGetChecksum(envsetup, self.timeout, full_surl)
#if not remote_checksum:
# # try to grab the remote file info using lcg-ls command
# remote_checksum, remote_fsize = self.getRemoteFileInfo(envsetup, self.timeout, full_surl)
#else:
# tolog("Setting remote file size to None (not needed)")
# remote_fsize = None
# compare the checksums if the remote checksum was extracted
tolog("Remote checksum: %s" % str(remote_checksum))
tolog("Local checksum: %s" % (fchecksum))
if remote_checksum:
if remote_checksum != fchecksum:
pilotErrorDiag = "Remote and local checksums (of type %s) do not match for %s (%s != %s)" %\
(csumtype, os.path.basename(dst_gpfn), remote_checksum, fchecksum)
tolog("!!WARNING!!1800!! %s" % (pilotErrorDiag))
if csumtype == "adler32":
self.prepareReport('AD_MISMATCH', report)
return self.put_data_retfail(error.ERR_PUTADMISMATCH, pilotErrorDiag, surl=full_surl)
else:
self.prepareReport('MD5_MISMATCH', report)
return self.put_data_retfail(error.ERR_PUTMD5MISMATCH, pilotErrorDiag, surl=full_surl)
else:
tolog("Remote and local checksums verified")
verified = True
else:
tolog("Skipped primary checksum verification (remote checksum not known)")
# if lcg-ls could not be used
if "/pnfs/" in surl and not remote_checksum:
# for dCache systems we can test the checksum with the use method
tolog("Detected dCache system: will verify local checksum with the local SE checksum")
# gpfn = srm://head01.aglt2.org:8443/srm/managerv2?SFN=/pnfs/aglt2.org/atlasproddisk/mc08/EVNT/mc08.109270.J0....
path = surl[surl.find('/pnfs/'):]
# path = /pnfs/aglt2.org/atlasproddisk/mc08/EVNT/mc08.109270.J0....#
tolog("File path: %s" % (path))
_filename = os.path.basename(path)
_dir = os.path.dirname(path)
# get the remote checksum
tolog("Local checksum: %s" % (fchecksum))
try:
remote_checksum = self.getdCacheChecksum(_dir, _filename)
except Exception, e:
pilotErrorDiag = "Could not get checksum from dCache: %s (test will be skipped)" % str(e)
tolog('!!WARNING!!2999!! %s' % (pilotErrorDiag))
else:
if remote_checksum == "NOSUCHFILE":
pilotErrorDiag = "The pilot will fail the job since the remote file does not exist"
tolog('!!WARNING!!2999!! %s' % (pilotErrorDiag))
self.prepareReport('NOSUCHFILE', report)
return self.put_data_retfail(error.ERR_NOSUCHFILE, pilotErrorDiag, surl=full_surl)
elif remote_checksum:
tolog("Remote checksum: %s" % (remote_checksum))
else:
tolog("Could not get remote checksum")
if remote_checksum:
if remote_checksum != fchecksum:
pilotErrorDiag = "Remote and local checksums (of type %s) do not match for %s (%s != %s)" %\
(csumtype, _filename, remote_checksum, fchecksum)
if csumtype == "adler32":
self.prepareReport('AD_MISMATCH', report)
return self.put_data_retfail(error.ERR_PUTADMISMATCH, pilotErrorDiag, surl=full_surl)
else:
self.prepareReport('MD5_MISMATCH', report)
return self.put_data_retfail(error.ERR_PUTMD5MISMATCH, pilotErrorDiag, surl=full_surl)
else:
tolog("Remote and local checksums verified")
verified = True
else:
tolog("Skipped secondary checksum test")
# if the checksum could not be verified (as is the case for non-dCache sites) test the file size instead
if not remote_checksum and remote_fsize:
tolog("Local file size: %s" % (fsize))
if remote_fsize and remote_fsize != "" and fsize != "" and fsize:
if str(fsize) != remote_fsize:
pilotErrorDiag = "Remote and local file sizes do not match for %s (%s != %s)" %\
(_filename, remote_fsize, str(fsize))
tolog('!!WARNING!!2999!! %s' % (pilotErrorDiag))
self.prepareReport('FS_MISMATCH', report)
return self.put_data_retfail(error.ERR_PUTWRONGSIZE, pilotErrorDiag, surl=full_surl)
else:
tolog("Remote and local file sizes verified")
verified = True
else:
tolog("Skipped file size test")
# was anything verified?
if not verified:
# fail at this point
pilotErrorDiag = "Neither checksum nor file size could be verified (failing job)"
tolog('!!WARNING!!2999!! %s' % (pilotErrorDiag))
self.prepareReport('NOFILEVERIFICATION', report)
return self.put_data_retfail(error.ERR_NOFILEVERIFICATION, pilotErrorDiag, surl=full_surl)
self.prepareReport('DONE', report)
return 0, pilotErrorDiag, full_surl, fsize, fchecksum, self.arch_type
if __name__ == "__main__":
# surl='https://fozzie.ndgf.org:2881/atlas/disk/atlashotdisk/ddo/DBRelease/v200202/ddo.000001.Atlas.Ideal.DBRelease.v200202/DBRelease-20.2.2.tar.gz'
surl='srm://lcg-lrz-se.lrz-muenchen.de/pnfs/lrz-muenchen.de/data/atlas/dq2/atlashotdisk/ddo/DBRelease/v200201/ddo.000001.Atlas.Ideal.DBRelease.v200201/DBRelease-20.2.1.tar.gz'
mover=aria2cSiteMover("")
#mover.get_data(surl,'somelfn','/tmp',616103906,'checky','guidguid')
mover.get_data(surl,'somelfn','/tmp',616103906,'checky','scope','guidguid')