1 |
< |
#!/usr/bin/env python2 |
2 |
< |
import sys, os, string, re |
3 |
< |
from DBSInfo import * |
4 |
< |
|
1 |
> |
#!/usr/bin/env python |
2 |
> |
import exceptions |
3 |
> |
import DBSAPI.dbsApi |
4 |
> |
from DBSAPI.dbsApiException import * |
5 |
> |
import common |
6 |
> |
from crab_util import * |
7 |
> |
|
8 |
> |
|
9 |
> |
# ####################################### |
10 |
> |
class DBSError(exceptions.Exception): |
11 |
> |
def __init__(self, errorName, errorMessage): |
12 |
> |
args='\nERROR DBS %s : %s \n'%(errorName,errorMessage) |
13 |
> |
exceptions.Exception.__init__(self, args) |
14 |
> |
pass |
15 |
> |
|
16 |
> |
def getErrorMessage(self): |
17 |
> |
""" Return error message """ |
18 |
> |
return "%s" % (self.args) |
19 |
> |
|
20 |
> |
# ####################################### |
21 |
> |
class DBSInvalidDataTierError(exceptions.Exception): |
22 |
> |
def __init__(self, errorName, errorMessage): |
23 |
> |
args='\nERROR DBS %s : %s \n'%(errorName,errorMessage) |
24 |
> |
exceptions.Exception.__init__(self, args) |
25 |
> |
pass |
26 |
> |
|
27 |
> |
def getErrorMessage(self): |
28 |
> |
""" Return error message """ |
29 |
> |
return "%s" % (self.args) |
30 |
> |
|
31 |
> |
# ####################################### |
32 |
> |
class DBSInfoError: |
33 |
> |
def __init__(self, url): |
34 |
> |
print '\nERROR accessing DBS url : '+url+'\n' |
35 |
> |
pass |
36 |
|
|
37 |
|
# #################################### |
38 |
|
class DataDiscoveryError(exceptions.Exception): |
39 |
< |
def __init__(self, errorMessage): |
40 |
< |
args=errorMessage |
41 |
< |
exceptions.Exception.__init__(self, args) |
42 |
< |
pass |
43 |
< |
|
44 |
< |
def getErrorMessage(self): |
45 |
< |
""" Return exception error """ |
46 |
< |
return "%s" % (self.args) |
39 |
> |
def __init__(self, errorMessage): |
40 |
> |
self.args=errorMessage |
41 |
> |
exceptions.Exception.__init__(self, self.args) |
42 |
> |
pass |
43 |
> |
|
44 |
> |
def getErrorMessage(self): |
45 |
> |
""" Return exception error """ |
46 |
> |
return "%s" % (self.args) |
47 |
|
|
48 |
|
# #################################### |
49 |
|
class NotExistingDatasetError(exceptions.Exception): |
50 |
< |
def __init__(self, errorMessage): |
51 |
< |
args=errorMessage |
52 |
< |
exceptions.Exception.__init__(self, args) |
53 |
< |
pass |
54 |
< |
|
55 |
< |
def getErrorMessage(self): |
56 |
< |
""" Return exception error """ |
57 |
< |
return "%s" % (self.args) |
50 |
> |
def __init__(self, errorMessage): |
51 |
> |
self.args=errorMessage |
52 |
> |
exceptions.Exception.__init__(self, self.args) |
53 |
> |
pass |
54 |
> |
|
55 |
> |
def getErrorMessage(self): |
56 |
> |
""" Return exception error """ |
57 |
> |
return "%s" % (self.args) |
58 |
|
|
59 |
|
# #################################### |
60 |
|
class NoDataTierinProvenanceError(exceptions.Exception): |
61 |
< |
def __init__(self, errorMessage): |
62 |
< |
args=errorMessage |
63 |
< |
exceptions.Exception.__init__(self, args) |
64 |
< |
pass |
65 |
< |
|
66 |
< |
def getErrorMessage(self): |
67 |
< |
""" Return exception error """ |
68 |
< |
return "%s" % (self.args) |
61 |
> |
def __init__(self, errorMessage): |
62 |
> |
self.args=errorMessage |
63 |
> |
exceptions.Exception.__init__(self, self.args) |
64 |
> |
pass |
65 |
> |
|
66 |
> |
def getErrorMessage(self): |
67 |
> |
""" Return exception error """ |
68 |
> |
return "%s" % (self.args) |
69 |
|
|
70 |
|
# #################################### |
71 |
|
# class to find and extact info from published data |
72 |
|
class DataDiscovery: |
73 |
< |
def __init__(self, owner, dataset, dataTiers, cfg_params): |
73 |
> |
def __init__(self, datasetPath, cfg_params): |
74 |
|
|
75 |
< |
# Attributes |
76 |
< |
self.dbsdataset='/'+dataset+'/datatier/'+owner |
46 |
< |
self.dataTiers = dataTiers |
75 |
> |
# Attributes |
76 |
> |
self.datasetPath = datasetPath |
77 |
|
self.cfg_params = cfg_params |
78 |
|
|
79 |
< |
self.dbspaths= [] # DBS output: list of dbspaths for all data |
80 |
< |
self.allblocks = [] # DBS output: list of map fileblocks-totevts for all dataset-owners |
81 |
< |
self.blocksinfo = {} # DBS output: map fileblocks-totevts for the primary block, used internally to this class |
82 |
< |
#DBS output: max events computed by method getMaxEvents |
79 |
> |
self.eventsPerBlock = {} # DBS output: map fileblocks-events for collection |
80 |
> |
self.eventsPerFile = {} # DBS output: map files-events |
81 |
> |
self.blocksinfo = {} # DBS output: map fileblocks-files |
82 |
> |
self.maxEvents = 0 # DBS output: max events |
83 |
|
|
84 |
|
# #################################### |
85 |
|
def fetchDBSInfo(self): |
86 |
|
""" |
87 |
|
Contact DBS |
88 |
|
""" |
59 |
– |
parents = [] |
60 |
– |
parentsblocksinfo = {} |
89 |
|
|
90 |
< |
## add the PU among the required data tiers if the Digi are requested |
91 |
< |
if (self.dataTiers.count('Digi')>0) & (self.dataTiers.count('PU')<=0) : |
92 |
< |
self.dataTiers.append('PU') |
90 |
> |
## get DBS URL |
91 |
> |
dbs_url="http://cmsdbsprod.cern.ch/cms_dbs_prod_global/servlet/DBSServlet" |
92 |
> |
if (self.cfg_params.has_key('CMSSW.dbs_url')): |
93 |
> |
dbs_url=self.cfg_params['CMSSW.dbs_url'] |
94 |
> |
|
95 |
> |
common.logger.debug(3,"Accessing DBS at: "+dbs_url) |
96 |
> |
|
97 |
> |
## check if runs are selected |
98 |
> |
runselection = [] |
99 |
> |
if (self.cfg_params.has_key('CMSSW.runselection')): |
100 |
> |
runselection = parseRange2(self.cfg_params['CMSSW.runselection']) |
101 |
> |
|
102 |
> |
## service API |
103 |
> |
args = {} |
104 |
> |
args['url'] = dbs_url |
105 |
> |
args['level'] = 'CRITICAL' |
106 |
|
|
107 |
< |
## get info about the requested dataset |
67 |
< |
dbs=DBSInfo(self.dbsdataset,self.dataTiers) |
107 |
> |
api = DBSAPI.dbsApi.DbsApi(args) |
108 |
|
try: |
109 |
< |
self.blocksinfo=dbs.getDatasetContents() |
110 |
< |
except DBSError, ex: |
111 |
< |
raise DataDiscoveryError(ex.getErrorMessage()) |
112 |
< |
|
113 |
< |
if len(self.blocksinfo)<=0: |
114 |
< |
msg="\nERROR Data %s do not exist in DBS! \n Check the dataset/owner variables in crab.cfg !"%self.dbsdataset |
115 |
< |
raise NotExistingDatasetError(msg) |
116 |
< |
|
117 |
< |
currentdatatier=string.split(self.blocksinfo.keys()[0],'/')[2] |
118 |
< |
fakedatatier=string.split(self.dbsdataset,'/')[2] |
119 |
< |
currentdbsdataset=string.replace(self.dbsdataset, fakedatatier, currentdatatier) |
120 |
< |
|
121 |
< |
self.dbspaths.append(currentdbsdataset) # add the requested dbspath |
122 |
< |
|
123 |
< |
## get info about the parents |
124 |
< |
try: |
125 |
< |
parents=dbs.getDatasetProvenance() |
126 |
< |
except DBSInvalidDataTierError, ex: |
127 |
< |
msg=ex.getErrorMessage()+' \n Check the data_tier variable in crab.cfg !\n' |
128 |
< |
raise DataDiscoveryError(msg) |
129 |
< |
except DBSError, ex: |
130 |
< |
raise DataDiscoveryError(ex.getErrorMessage()) |
131 |
< |
|
132 |
< |
## check that the user asks for parent Data Tier really existing in the DBS provenance |
133 |
< |
self.checkParentDataTier(parents, self.dataTiers, currentdbsdataset) |
134 |
< |
|
135 |
< |
## for each parent get the corresponding fileblocks |
136 |
< |
for aparent in parents: |
137 |
< |
## fill a list of dbspaths |
138 |
< |
parentdbsdataset=aparent.getDatasetPath() |
139 |
< |
self.dbspaths.append(parentdbsdataset) |
140 |
< |
pdbs=DBSInfo(parentdbsdataset,[]) |
141 |
< |
try: |
142 |
< |
parentsblocksinfo=pdbs.getDatasetContents() |
143 |
< |
except DBSError, ex: |
144 |
< |
raise DataDiscoveryError(ex.getErrorMessage()) |
145 |
< |
|
146 |
< |
self.allblocks.append(parentsblocksinfo.keys()) # add parent fileblocksinfo |
147 |
< |
|
148 |
< |
## all the required blocks |
149 |
< |
self.allblocks.append(self.blocksinfo.keys()) # add also the current fileblocksinfo |
150 |
< |
|
151 |
< |
|
152 |
< |
# ################################################# |
153 |
< |
def checkParentDataTier(self, parents, user_datatiers, currentdbsdataset ): |
154 |
< |
""" |
155 |
< |
check that the data tiers requested by the user really exists in the provenance of the given dataset |
156 |
< |
""" |
157 |
< |
|
158 |
< |
current_datatier=string.split(currentdbsdataset,'/')[2] |
159 |
< |
|
160 |
< |
parent_datatypes=[] |
121 |
< |
for aparent in parents: |
122 |
< |
parent_datatypes.append(aparent.getDataType()) |
123 |
< |
|
124 |
< |
for datatier in user_datatiers: |
125 |
< |
if parent_datatypes.count(datatier)<=0: |
126 |
< |
# the current datatier is not supposed to be in the provenance |
127 |
< |
if not (datatier == current_datatier): |
128 |
< |
msg="\nERROR Data %s not published in DBS with asked data tiers : the data tier not found is %s !\n Check the data_tier variable in crab.cfg !"%(currentdbsdataset,datatier) |
129 |
< |
raise NoDataTierinProvenanceError(msg) |
109 |
> |
if len(runselection) <= 0 : |
110 |
> |
files = api.listDatasetFiles(self.datasetPath) |
111 |
> |
else : |
112 |
> |
files = api.listFiles(path=self.datasetPath, details=True) |
113 |
> |
except DbsBadRequest, msg: |
114 |
> |
raise DataDiscoveryError(msg) |
115 |
> |
except DBSError, msg: |
116 |
> |
raise DataDiscoveryError(msg) |
117 |
> |
|
118 |
> |
# parse files and fill arrays |
119 |
> |
for file in files : |
120 |
> |
filename = file['LogicalFileName'] |
121 |
> |
if filename.find('.dat') < 0 : |
122 |
> |
fileblock = file['Block']['Name'] |
123 |
> |
events = file['NumberOfEvents'] |
124 |
> |
continue_flag = 0 |
125 |
> |
if len(runselection) > 0 : |
126 |
> |
runslist = file['RunsList'] |
127 |
> |
for run in runslist : |
128 |
> |
runnumber = run['RunNumber'] |
129 |
> |
for selected_run in runselection : |
130 |
> |
if runnumber == selected_run : |
131 |
> |
continue_flag = 1 |
132 |
> |
else : |
133 |
> |
continue_flag = 1 |
134 |
> |
|
135 |
> |
if continue_flag == 1 : |
136 |
> |
# number of events per block |
137 |
> |
if fileblock in self.eventsPerBlock.keys() : |
138 |
> |
self.eventsPerBlock[fileblock] += events |
139 |
> |
else : |
140 |
> |
self.eventsPerBlock[fileblock] = events |
141 |
> |
|
142 |
> |
# number of events per file |
143 |
> |
self.eventsPerFile[filename] = events |
144 |
> |
|
145 |
> |
# number of events per block |
146 |
> |
if fileblock in self.blocksinfo.keys() : |
147 |
> |
self.blocksinfo[fileblock].append(filename) |
148 |
> |
else : |
149 |
> |
self.blocksinfo[fileblock] = [filename] |
150 |
> |
|
151 |
> |
# total number of events |
152 |
> |
self.maxEvents += events |
153 |
> |
|
154 |
> |
for block in self.eventsPerBlock.keys() : |
155 |
> |
common.logger.debug(6,"DBSInfo: total nevts %i in block %s "%(self.eventsPerBlock[block],block)) |
156 |
> |
|
157 |
> |
if len(self.eventsPerBlock) <= 0: |
158 |
> |
raise NotExistingDatasetError(("\nNo data for %s in DBS\nPlease check" |
159 |
> |
+ " dataset path variables in crab.cfg") |
160 |
> |
% self.datasetPath) |
161 |
|
|
162 |
|
|
163 |
|
# ################################################# |
164 |
|
def getMaxEvents(self): |
165 |
|
""" |
166 |
< |
max events of the primary dataset-owner |
166 |
> |
max events |
167 |
|
""" |
168 |
< |
## loop over the fileblocks of the primary dataset-owner |
138 |
< |
nevts=0 |
139 |
< |
for blockevts in self.blocksinfo.values(): |
140 |
< |
nevts=nevts+blockevts |
141 |
< |
|
142 |
< |
return nevts |
168 |
> |
return self.maxEvents |
169 |
|
|
170 |
|
# ################################################# |
171 |
< |
def getDBSPaths(self): |
171 |
> |
def getEventsPerBlock(self): |
172 |
|
""" |
173 |
< |
list the DBSpaths for all required data |
173 |
> |
list the event collections structure by fileblock |
174 |
|
""" |
175 |
< |
return self.dbspaths |
175 |
> |
return self.eventsPerBlock |
176 |
|
|
177 |
|
# ################################################# |
178 |
< |
def getEVC(self): |
178 |
> |
def getEventsPerFile(self): |
179 |
|
""" |
180 |
< |
list the event collections structure by fileblock |
180 |
> |
list the event collections structure by file |
181 |
|
""" |
182 |
< |
print "To be used by a more complex job splitting... TODO later... " |
157 |
< |
print "it requires changes in what's returned by DBSInfo.getDatasetContents and then fetchDBSInfo" |
182 |
> |
return self.eventsPerFile |
183 |
|
|
184 |
|
# ################################################# |
185 |
< |
def getFileBlocks(self): |
185 |
> |
def getFiles(self): |
186 |
|
""" |
187 |
< |
fileblocks for all required dataset-owners |
187 |
> |
return files grouped by fileblock |
188 |
|
""" |
189 |
< |
return self.allblocks |
189 |
> |
return self.blocksinfo |
190 |
|
|
191 |
|
######################################################################## |
167 |
– |
|
168 |
– |
|