9 print 'Not injecting to wmagent in --show mode. Need to run the worklfows.'
11 if opt.wmcontrol==
'init':
14 if opt.wmcontrol==
'test':
17 if opt.wmcontrol==
'submit' and opt.nThreads==0:
18 print 'Not injecting to wmagent in -j 0 mode. Need to run the worklfows.'
20 if opt.wmcontrol==
'force':
21 print "This is an expert setting, you'd better know what you're doing"
25 from modules.wma
import upload_to_couch
26 (filePath,labelInCouch,user,group,where) = arguments
27 cacheId=upload_to_couch(filePath,
40 self.
testMode=((mode!=
'submit')
and (mode!=
'force'))
53 self.
user = os.getenv(
'USER')
61 if not os.getenv(
'WMCORE_ROOT'):
62 print '\n\twmclient is not setup properly. Will not be able to upload or submit requests.\n'
67 print '\n\tFound wmclient\n'
70 "RequestType" :
"TaskChain",
71 "Requestor": self.
user,
73 "CMSSWVersion": os.getenv(
'CMSSW_VERSION'),
74 "Campaign": os.getenv(
'CMSSW_VERSION'),
75 "ScramArch": os.getenv(
'SCRAM_ARCH'),
76 "ProcessingVersion": self.
version,
78 "CouchURL": self.
couch,
79 "ConfigCacheURL": self.
couch,
80 "DbsUrl":
"https://cmsweb.cern.ch/dbs/prod/global/DBSReader",
84 "unmergedLFNBase" :
"/store/unmerged",
85 "mergedLFNBase" :
"/store/relval",
86 "dashboardActivity" :
"relval",
88 "SizePerEvent" : 1234,
93 "EnableHarvesting" :
"True",
94 "DQMUploadUrl" :
"https://cmsweb.cern.ch/dqm/relval",
95 "DQMConfigCacheID" :
None
100 "ConfigCacheID" :
None,
102 "SplittingAlgo" :
"EventBased",
103 "EventsPerJob" :
None,
104 "RequestNumEvents" :
None,
105 "Seeding" :
"AutomaticSeeding",
106 "PrimaryDataset" :
None,
111 "TaskName" :
"DigiHLT",
112 "ConfigCacheID" :
None,
114 "InputDataset" :
None,
115 "SplittingAlgo" :
"LumiBased",
123 "InputFromOutputModule" :
None,
124 "ConfigCacheID" :
None,
126 "SplittingAlgo" :
"LumiBased",
135 def prepare(self,mReader, directories, mode='init'):
139 pprint.pprint(wmsplit)
141 print "Not set up for step splitting"
145 for (n,dir)
in directories.items():
147 print "inspecting",dir
149 for (x,s)
in mReader.workFlowSteps.items():
154 chainDict[
'RequestString']=
'RV'+chainDict[
'CMSSWVersion']+s[1].
split(
'+')[0]
161 if 'INPUT' in step
or (
not isinstance(s[2][index],str)):
162 nextHasDSInput=s[2][index]
170 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).
read())
172 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created"
175 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=
'RelVal'+s[1].
split(
'+')[0]
176 if not '--relval' in s[2][index]:
177 print 'Impossible to create task from scratch without splitting information with --relval'
180 arg=s[2][index].
split()
181 ns=
map(int,arg[arg.index(
'--relval')+1].
split(
','))
182 chainDict[
'nowmTasklist'][-1][
'RequestNumEvents'] = ns[0]
183 chainDict[
'nowmTasklist'][-1][
'EventsPerJob'] = ns[1]
184 if 'FASTSIM' in s[2][index]
or '--fast' in s[2][index]:
185 thisLabel+=
'_FastSim'
190 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).
read())
192 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created"
194 chainDict[
'nowmTasklist'][-1][
'InputDataset']=nextHasDSInput.dataSet
195 splitForThisWf=nextHasDSInput.split
196 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=splitForThisWf
198 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=wmsplit[step]
200 if len(nextHasDSInput.run):
201 chainDict[
'nowmTasklist'][-1][
'RunWhitelist']=nextHasDSInput.run
203 if '--data' in s[2][index]
and nextHasDSInput.label:
204 thisLabel+=
'_RelVal_%s'%nextHasDSInput.label
205 if 'filter' in chainDict[
'nowmTasklist'][-1][
'nowmIO']:
206 print "This has an input DS and a filter sequence: very likely to be the PyQuen sample"
207 processStrPrefix=
'PU_'
208 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=
'RelVal'+s[1].
split(
'+')[0]
214 chainDict[
'nowmTasklist'][-1][
'nowmIO']=json.loads(open(
'%s/%s.io'%(dir,step)).
read())
216 print "Failed to find",
'%s/%s.io'%(dir,step),
".The workflows were probably not run on cfg not created"
219 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=splitForThisWf
221 chainDict[
'nowmTasklist'][-1][
'LumisPerJob']=wmsplit[step]
222 chainDict[
'nowmTasklist'][-1][
'PrimaryDataset']=
'RelVal'+s[1].
split(
'+')[0]
224 chainDict[
'nowmTasklist'][-1][
'TaskName']=step
225 chainDict[
'nowmTasklist'][-1][
'ConfigCacheID']=
'%s/%s.py'%(dir,step)
226 chainDict[
'nowmTasklist'][-1][
'GlobalTag']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'GT']
227 chainDict[
'GlobalTag']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'GT']
228 if 'pileup' in chainDict[
'nowmTasklist'][-1][
'nowmIO']:
229 chainDict[
'nowmTasklist'][-1][
'MCPileup']=chainDict[
'nowmTasklist'][-1][
'nowmIO'][
'pileup']
230 if '--pileup' in s[2][index]:
231 processStrPrefix=
'PU_'
234 chainDict[
'AcquisitionEra'][step]=chainDict[
'CMSSWVersion']
235 chainDict[
'ProcessingString'][step]=processStrPrefix+chainDict[
'nowmTasklist'][-1][
'GlobalTag'].
replace(
'::All',
'')+thisLabel
238 chainDict[
'nowmTasklist'][-1][
'AcquisitionEra']=chainDict[
'CMSSWVersion']
239 chainDict[
'nowmTasklist'][-1][
'ProcessingString']=processStrPrefix+chainDict[
'nowmTasklist'][-1][
'GlobalTag'].
replace(
'::All',
'')+thisLabel
248 for i_second
in reversed(range(len(chainDict[
'nowmTasklist']))):
249 t_second=chainDict[
'nowmTasklist'][i_second]
251 if 'primary' in t_second[
'nowmIO']:
253 primary=t_second[
'nowmIO'][
'primary'][0].
replace(
'file:',
'')
254 for i_input
in reversed(range(0,i_second)):
255 t_input=chainDict[
'nowmTasklist'][i_input]
256 for (om,o)
in t_input[
'nowmIO'].items():
259 t_second[
'InputTask'] = t_input[
'TaskName']
260 t_second[
'InputFromOutputModule'] = om
262 if t_second[
'TaskName'].startswith(
'HARVEST'):
264 chainDict[
'DQMConfigCacheID']=t_second[
'ConfigCacheID']
274 chainDict[
'AcquisitionEra'] = chainDict[
'AcquisitionEra'].
values()[0]
280 if type(i)==int
and i < len(chainDict[
'nowmTasklist']):
281 chainDict[
'nowmTasklist'][i][
'KeepOutput']=
True
282 for (i,t)
in enumerate(chainDict[
'nowmTasklist']):
283 if t[
'TaskName'].startswith(
'HARVEST'):
287 elif t[
'TaskName']
in self.
keep:
291 chainDict[
'Task%d'%(itask)]=t
298 chainDict[
'TaskChain']=itask
300 chainDict.pop(
'nowmTasklist')
307 labelInCouch=self.
label+
'_'+label
308 cacheName=filePath.split(
'/')[-1]
311 print '\tFake upload of',filePath,
'to couch with label',labelInCouch
315 from modules.wma
import upload_to_couch,DATABASE_NAME
317 print '\n\tUnable to find wmcontrol modules. Please include it in your python path\n'
322 print "Not re-uploading",filePath,
"to",where,
"for",label
325 print "Loading",filePath,
"to",where,
"for",label
327 pool = multiprocessing.Pool(1)
328 cacheIds = pool.map( upload_to_couch_oneArg, [(filePath,labelInCouch,self.
user,self.
group,where)] )
329 cacheId = cacheIds[0]
334 for (n,d)
in self.chainDicts.items():
336 if it.startswith(
"Task")
and it!=
'TaskChain':
338 couchID=self.
uploadConf(d[it][
'ConfigCacheID'],
339 str(n)+d[it][
'TaskName'],
342 print d[it][
'ConfigCacheID'],
" uploaded to couchDB for",str(n),
"with ID",couchID
343 d[it][
'ConfigCacheID']=couchID
344 if it ==
'DQMConfigCacheID':
345 couchID=self.
uploadConf(d[
'DQMConfigCacheID'],
349 print d[
'DQMConfigCacheID'],
"uploaded to couchDB for",str(n),
"with ID",couchID
350 d[
'DQMConfigCacheID']=couchID
355 from modules.wma
import makeRequest,approveRequest
356 from wmcontrol
import random_sleep
357 print '\n\tFound wmcontrol\n'
359 print '\n\tUnable to find wmcontrol modules. Please include it in your python path\n'
365 for (n,d)
in self.chainDicts.items():
367 print "Only viewing request",n
368 print pprint.pprint(d)
371 print "For eyes before submitting",n
372 print pprint.pprint(d)
373 print "Submitting",n,
"..........."
374 workFlow=makeRequest(self.
wmagent,d,encodeDict=
True)
375 approveRequest(self.
wmagent,workFlow)
376 print "...........",n,
"submitted"
def performInjectionOptionTest
def upload_to_couch_oneArg