40 msg.debug(
'Preparing for execution of {0} with inputs {1} and outputs {2}'.format(self.
name, input, output))
43 self.
_exe =
'setsid ' + self.conf.argdict[
'trigExe'].value
46 if (self._inputEventTest
and 'skipEvents' in self.conf.argdict
and
47 self.conf.argdict[
'skipEvents'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor)
is not None):
48 msg.debug(
'Will test for events to process')
49 for dataType
in input:
50 inputEvents = self.conf.dataDictionary[dataType].nentries
51 msg.debug(
'Got {} events for {}'.format(inputEvents, dataType))
52 if not isinstance(inputEvents, int):
53 msg.warning(
'Are input events countable? Got nevents={} so disabling event count check for this input'.format(inputEvents))
54 elif self.conf.argdict[
'skipEvents'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor) >= inputEvents:
56 'No events to process: {0} (skipEvents) >= {1} (inputEvents of {2}'.format(self.conf.argdict[
'skipEvents'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor), inputEvents, dataType))
61 for dataType
in input:
62 inputFiles[dataType] = self.conf.dataDictionary[dataType]
64 for dataType
in output:
65 outputFiles[dataType] = self.conf.dataDictionary[dataType]
68 for dataType, dataArg
in self.conf.dataDictionary.items():
69 if dataArg.io ==
'input' and self.
_name in dataArg.executor:
70 inputFiles[dataArg.subtype] = dataArg
72 msg.info(
'Input Files: {0}; Output Files: {1}'.format(inputFiles, outputFiles))
81 self._extraMetadata[
'inputs'] = list(input)
83 self._extraMetadata[
'outputs'] = list(output)
87 if 'asetup' in self.conf.argdict:
88 asetupString = self.conf.argdict[
'asetup'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor)
90 msg.info(
'Asetup report: {0}'.format(asetupReport()))
94 if 'runInContainer' in self.conf.argdict:
95 OSSetupString = self.conf.argdict[
'runInContainer'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor)
96 msg.info(
'The step {} will be performed in a container running {}, as explicitly requested'.format(self.
_substep, OSSetupString))
99 dbrelease = dbsetup =
None
100 if 'DBRelease' in self.conf.argdict:
101 dbrelease = self.conf.argdict[
'DBRelease'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor)
104 dbdMatch = re.match(
r'DBRelease-([\d\.]+)\.tar\.gz', os.path.basename(dbrelease))
106 msg.debug(
'DBRelease setting {0} matches classic tarball file'.format(dbrelease))
107 if not os.access(dbrelease, os.R_OK):
108 msg.warning(
'Transform was given tarball DBRelease file {0}, but this is not there'.format(dbrelease))
109 msg.warning(
'I will now try to find DBRelease {0} in cvmfs'.format(dbdMatch.group(1)))
110 dbrelease = dbdMatch.group(1)
111 dbsetup = cvmfsDBReleaseCheck(dbrelease)
114 unpacked, dbsetup = unpackDBRelease(tarball=dbrelease, dbversion=dbdMatch.group(1))
117 setupDBRelease(dbsetup)
120 dbsetup = cvmfsDBReleaseCheck(dbrelease)
126 self._prepAthenaCommandLine()
129 if 'athenaHLT' in self.
_exe or 'athenaEF' in self.
_exe:
130 self.
_cmd.remove(
'runargs.BSRDOtoRAW.py')
132 optionList = getTranslated(self.conf.argdict, name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor, output = outputFiles)
133 self.
_cmd.extend(optionList)
135 if self._isCAEnabled():
136 msg.info(
"Running in CA mode")
139 if 'preExec' in self.conf.argdict:
140 self.
_cmd.extend(self.conf.argdict[
'preExec'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor))
141 msg.info(
'Command adjusted for CA to %s', self.
_cmd)
143 msg.error(
"Trig_reco_tf does not support running in legacy mode")
146 if "outputHIST_DEBUGSTREAMMONFile" in self.conf.argdict:
148 dbgAsetupString, dbAlias = dbgStream.dbgPreRun(self.conf.dataDictionary[
'BS_RDO'], self.conf.dataDictionary[
'HIST_DEBUGSTREAMMON'].value, self.conf.argdict)
151 if asetupString
is None and dbgAsetupString
is not None:
152 asetupString = dbgAsetupString
153 msg.info(
'Will use asetup string for debug stream analysis %s', dbgAsetupString)
155 if 'runInContainer' in self.conf.argdict:
156 OSSetupString = self.conf.argdict[
'runInContainer'].returnMyValue(name=self.
_name, substep=self.
_substep, first=self.conf.firstExecutor)
157 msg.info(
'The step {} will be performed in a container running {}, as explicitly requested'.format(self.
_substep, OSSetupString))
160 if 'useDB' in self.conf.argdict
and 'DBserver' not in self.conf.argdict
and dbAlias:
161 msg.warn(
"Database alias will be set to %s", dbAlias)
162 self.
_cmd.append(
"--db-server " + dbAlias)
164 msg.info(
"Flag outputHIST_DEBUGSTREAMMONFile not defined - debug stream analysis will not run.")
168 if 'BS' in self.conf.dataDictionary
or 'DRAW_TRIGCOST' in self.conf.dataDictionary
or 'HIST_DEBUGSTREAMMON' in self.conf.dataDictionary:
172 if len(matchedOutputFileNames) > 0:
173 msg.error(f
'Directoy already contains files with expected output name format ({self.expectedOutputFileName}), please remove/rename these first: {matchedOutputFileNames}')
175 f
'Directory already contains files with expected output name format {self.expectedOutputFileName}, please remove/rename these first: {matchedOutputFileNames}')
178 if OSSetupString
is not None and asetupString
is None:
180 'Athena version must be specified for the substep which requires running inside a container (either via --asetup or from DB)')
183 super(athenaExecutor, self).
preExecute(input, output)
188 msg.info(
'Now writing wrapper for substep executor {0}'.format(self.
_name))
189 self._writeAthenaWrapper(asetup=asetupString, dbsetup=dbsetup, ossetup=OSSetupString)
190 msg.info(
'Athena will be executed in a subshell via {0}'.format(self.
_cmd))
248 def _splitBSfile(self, streamsList, allStreamsFileName, splitFileName):
250 outputStreams =
','.join(str(stream)
for stream
in streamsList)
251 msg.info(
'Splitting stream %s from BS file', outputStreams)
252 splitStreamFailure = 0
254 cmd = f
'trigbs_extractStream.py -s {outputStreams} {allStreamsFileName}'
255 msg.info(
'running command for splitting (in original asetup env): %s', cmd)
256 splitStreamFailure = subprocess.call(cmd, shell=
True)
257 msg.debug(
'trigbs_extractStream.py splitting return code %s', splitStreamFailure)
260 'Exception raised when selecting stream with trigbs_extractStream.py in file {0}: {1}'.format(allStreamsFileName, e))
261 if splitStreamFailure != 0:
262 msg.warning(
'trigbs_extractStream.py returned error (%s) no split BS file created', splitStreamFailure)
266 streamName = outputStreams
if len(streamsList)==1
else 'accepted'
267 expectedStreamFileName = f
'*_{streamName}.*.RAW._*.data'
270 if(len(matchedOutputFileName)):
271 self.
_renamefile(matchedOutputFileName[0], splitFileName)
274 msg.error(
'trigbs_extractStream.py did not created expected file (%s)', expectedStreamFileName)
294 msg.debug(
'Now scanning logfile {0} for HLTMPPU Child Issues'.format(log))
303 myGen = lineByLine(log, substepName=self.
_substep)
305 msg.error(
'Failed to open transform logfile {0}: {1:s}'.format(log, e))
306 for line, lineCounter
in myGen:
308 if 'Child Issue' in line:
310 signal = int((re.search(
'signal ([0-9]*)', line)).group(1))
311 except AttributeError:
314 msg.error(
'Detected issue with HLTChild, setting mother return code to %s', signal)
323 for file
in os.listdir(
'.'):
325 if fnmatch.fnmatch(file,
'athenaHLT:*'):
326 msg.info(
'Merging child log file (%s) into %s', file, self.
_logFileName)
327 with open(file)
as log_file:
329 merged_file.write(
'### Output from {} ###\n'.format(file))
331 for line
in log_file:
332 merged_file.write(line)
334 if 'rejected:' in line
and int(line[14]) != 0:
336 rejected += int(line[14:])
338 if 'accepted:' in line
and int(line[14]) != 0:
340 accepted += int(line[14:])
341 if re.search(
'DFDcmEmuSession.* Communication error', line)
or re.search(
'DFDcmEmuSession.* No new event provided within the timeout limit', line):
342 msg.error(
'Caught DFDcmEmuSession error, aborting job')
345 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
347 dbgStream.getHltDecision(accepted, rejected, self.conf.argdict[
"outputHIST_DEBUGSTREAMMONFile"].value[0])
351 'Exception raised when merging log files into {0}: {1}'.format(self.
_logFileName, e))
353 msg.info(
"Check for expert-monitoring.root file")
358 expectedFileName =
'expert-monitoring.root'
362 msg.info(
'HLT step failed (with status %s) so skip HIST_HLTMON filename check', self.
_rc)
364 elif 'outputHIST_HLTMONFile' in self.conf.argdict:
367 expectedMotherFileName =
'expert-monitoring-mother.root'
368 if(os.path.isfile(expectedFileName)):
369 msg.info(
'Renaming %s to %s', expectedFileName, expectedMotherFileName)
371 os.rename(expectedFileName, expectedMotherFileName)
374 'Exception raised when renaming {0} to {1}: {2}'.format(expectedFileName, expectedMotherFileName, e))
376 msg.error(
'HLTMON argument defined but mother %s not created', expectedFileName)
379 expectedWorkerFileName =
'athenaHLT_workers/athenaHLT-01/' + expectedFileName
380 if(os.path.isfile(expectedWorkerFileName)
and os.path.isfile(expectedMotherFileName)):
381 msg.info(
'Merging worker and mother %s files to %s', expectedFileName, self.conf.argdict[
'outputHIST_HLTMONFile'].value[0])
384 cmd =
'hadd ' + self.conf.argdict[
'outputHIST_HLTMONFile'].value[0] +
' athenaHLT_workers/*/expert-monitoring.root expert-monitoring-mother.root'
385 subprocess.call(cmd, shell=
True)
388 'Exception raised when merging worker and mother {0} files to {1}: {2}'.format(expectedFileName, self.conf.argdict[
'outputHIST_HLTMONFile'].value[0], e))
390 msg.error(
'HLTMON argument defined %s but worker %s not created', self.conf.argdict[
'outputHIST_HLTMONFile'].value[0], expectedFileName)
393 msg.info(
'HLTMON argument not defined so skip %s check', expectedFileName)
395 msg.info(
"Search for created BS files, and rename if single file found")
401 msg.error(
'HLT step failed (with status %s) so skip BS filename check', self.
_rc)
402 elif 'BS' in self.conf.dataDictionary
or 'DRAW_TRIGCOST' in self.conf.dataDictionary
or 'HIST_DEBUGSTREAMMON' in self.conf.dataDictionary:
408 if len(matchedOutputFileNames) == 0:
409 msg.error(
'No BS files created with expected name: %s - please check for earlier failures', self.
expectedOutputFileName)
411 f
'No BS files created with expected name: {self.expectedOutputFileName} - please check for earlier failures')
414 if len(matchedOutputFileNames) == 1:
415 msg.info(
'Single BS file found: will split (if requested) and rename the file')
416 BSFile = matchedOutputFileNames[0]
420 msg.info(
'Multiple BS files found. A single BS file is required by the next transform steps, so they will be merged. Will split the merged file (if requested) and rename the file')
421 mergedBSFile = matchedOutputFileNames[0].
split(
'._0001.data')[0] +
'.mrg._0001.data'
423 mergeFailed = self.
_mergeBSfiles(matchedOutputFileNames, mergedBSFile)
426 'Did not produce a merged BS file with file_merging')
428 BSFile =
'tmp.BS.mrg'
429 msg.info(f
'Renaming temporary merged BS file to {BSFile}')
433 if 'DRAW_TRIGCOST' in self.conf.dataDictionary:
434 splitFailed = self.
_splitBSfile([
'CostMonitoring'], BSFile, self.conf.dataDictionary[
'DRAW_TRIGCOST'].value[0])
437 'Did not produce any BS file when selecting CostMonitoring stream with trigbs_extractStream.py in file')
440 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
444 if 'BS' in self.conf.dataDictionary:
445 argInDict = self.conf.dataDictionary[
'BS']
447 if 'streamSelection' in self.conf.argdict
and self.conf.argdict[
'streamSelection'].value[0] !=
"All":
448 splitEmpty = self.
_splitBSfile(self.conf.argdict[
'streamSelection'].value, BSFile, argInDict.value[0])
450 msg.info(
'Did not produce any BS file when selecting stream with trigbs_extractStream.py in file')
454 cmd_splitFailed =
'trigbs_failedStreamSelection.py ' + BSFile
455 msg.info(
'running command for creating empty file: %s', cmd_splitFailed)
456 subprocess.call(cmd_splitFailed, shell=
True)
459 runnumber = eformat.EventStorage.pickDataReader(BSFile).runNumber()
460 expectedOutputFileName =
'T0debug.00'+str(runnumber)+
'.unknown_debug.unknown.RAW._lb0000._TRF._0001.data'
462 self.
_renamefile(expectedOutputFileName, argInDict.value[0])
464 msg.info(
'Stream "All" requested, so not splitting BS file')
467 msg.info(
'BS output filetype not defined so skip renaming BS')
469 msg.info(
'BS, DRAW_TRIGCOST or HIST_DEBUGSTREAMMON output filetypes not defined so skip BS post processing')
471 msg.info(
'Now run athenaExecutor:postExecute')
474 if "HIST_DEBUGSTREAMMON" in self.conf.dataDictionary:
476 fileNameDbg = self.conf.argdict[
"outputHIST_DEBUGSTREAMMONFile"].value
477 dbgStream.dbgPostRun(argInDict.value[0], fileNameDbg[0], self.conf.argdict, isSplitStream=
True)