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
|
#! /usr/bin/env python
########################################################################################
# #
# Author: Bertrand Neron, #
# Organization:'Biological Software and Databases' Group, Institut Pasteur, Paris. #
# Distributed under GPLv2 Licence. Please refer to the COPYING.LIB document. #
# #
########################################################################################
"""
Classes executing the command and managing the results
"""
import os
import sys
import glob
import time
import cPickle
import atexit
#"the environment variable MOBYLEHOME must be defined in the environment"
#append Mobyle Home to the search modules path
MOBYLEHOME = None
if os.environ.has_key('MOBYLEHOME'):
MOBYLEHOME = os.environ['MOBYLEHOME']
if not MOBYLEHOME:
sys.exit('MOBYLEHOME must be defined in your environment')
if ( os.path.join( MOBYLEHOME , 'Src' ) ) not in sys.path:
sys.path.append( os.path.join( MOBYLEHOME , 'Src' ) )
from Mobyle.MobyleLogger import MLogger
MLogger( child = True )
from logging import getLogger
rc_log = getLogger( 'Mobyle.RunnerChild' )
from Mobyle.JobState import JobState
from Mobyle.Utils import executionLoader , zipFiles , emailResults
from Mobyle.Net import EmailAddress
from Mobyle.Status import Status
from Mobyle.StatusManager import StatusManager
from Mobyle.Admin import Admin
from Mobyle.MobyleError import MobyleError
from Mobyle.ConfigManager import Config
from Mobyle.Registry import registry
_cfg = Config()
__extra_epydoc_fields__ = [('call', 'Called by','Called by')]
class AsynchronJob:
"""
is instantiated in child process instantiate the object corresponding
to the execution manager defined in Config, and after the completion of the job
manage the results
"""
def __init__(self, commandLine, dirPath, serviceName, resultsMask , userEmail = None, email_notify = 'auto' , jobState = None , xmlEnv = None):
"""
@param commandLine: the command to be executed
@type commandLine: String
@param dirPath: the absolute path to directory where the job will be executed (normaly we are already in)
@type dirPath: String
@param serviceName: the name of the service
@type serviceName: string
@param resultsMask: the unix mask to retrieve the results of this job
@type resultsMask: a dictionary { paramName : [ string prompt , ( string class , string or None superclass ) , string mask ] }
@param userEmail: the user email adress
@type userEmail: string
@param email_notify: if the user must be or not notify of the results at the end of the Job.
the 3 authorized values for this argument are:
- 'true' to notify the results to the user
- 'false' to Not notify the results to the user
- 'auto' to notify the results based on the job elapsed time and the config EMAIL_DELAY
@type email_notify: string
@param jobState: the jobState link to this job
@type jobState: a L{JobState} instance
@param xmlEnv: the environement variable need by the program
@type xmlEnv: dictionnary
@call: by the main of this module which is call by L{AsynchronRunner}
"""
self._command = commandLine
self._dirPath = dirPath
self.serviceName = serviceName
self.father_pid = os.getppid()
self.father_done = False
if jobState is None:
self.jobState = JobState( self._dirPath )
else:
self.jobState = jobState
self.userEmail = userEmail
self.email_notify = email_notify
if self._dirPath[-1] == '/':
self._dirPath = self._dirPath[:-1]
self.jobKey = os.path.split( self._dirPath )[ 1 ]
atexit.register( self.childExit , "------------------- %s : %s -------------------" %( serviceName , self.jobKey ) )
t0 = time.time()
############################################
self._run( serviceName , xmlEnv )
#############################################
t1 = time.time()
self.results = {}
for paramName in resultsMask.keys():
resultsFiles = []
#type is a tuple ( klass , superKlass )
masks = resultsMask[ paramName ]
for mask in masks :
for File in glob.glob( mask ):
size = os.path.getsize( File )
if size != 0:
resultsFiles.append(( str( File ) , size , None )) #we have not information about the output format
if resultsFiles:
self.results[ paramName ] = resultsFiles #a list of tuple (string file name , int size , string format or None )
self.jobState.setOutputDataFile( paramName , resultsFiles )
self.jobState.commit()
try:
zipFileName = self.zipResults()
except Exception :
msg = "an error occured during the zipping results :\n\n"
rc_log.critical( "%s/%s : %s" %(self.serviceName, self.jobKey, msg), exc_info=True)
zipFileName = None
if self.userEmail:
if self.email_notify == 'auto':
# we test email_delay() to see if it is >= to 0,
# as it seems that sometimes it is not >0.
if self.email_notify == 'true' or ((t1 - t0) >= _cfg.email_delay()):
emailResults(_cfg,
self.userEmail,
registry,
self.jobState.getID(),
self._dirPath,
self.serviceName,
self.jobKey,
FileName=zipFileName)
else:
pass
def childExit(self , message ):
print >> sys.stderr , message
#rc_log.log( 12 , "runnerChild %d ending, send a SIGCHLD to %d" %( os.getpid() , self.father_pid ) )
def _run(self , serviceName, xmlEnv ):
dispatcher = _cfg.getDispatcher()
execution_config = dispatcher.getExecutionConfig( self.jobState )
try:
exec_engine = executionLoader( execution_config = execution_config )
except MobyleError ,err :
msg = "unknown execution system : %s" %err
rc_log.critical("%s : %s" %( serviceName ,
msg
), exc_info = True
)
sm = StatusManager()
sm.setStatus( self._dirPath , Status( code = 5 , message = 'Mobyle internal server error' ) )
raise MobyleError, msg
except Exception , err:
rc_log.error( str(err ), exc_info=True)
raise err
exec_engine.run( self._command , self._dirPath , serviceName , self.jobState , xmlEnv )
def zipResults(self ):
files2zip = []
for result in self.results.values():
for file_ in result:
#file_ is tuple (string file name , int size , string format or None )
files2zip.append((file_[0], os.path.basename(file_[0])))
xsl_path = os.path.join( _cfg.portal_path() , "xsl" ,)
jobXslPath = os.path.join( xsl_path , "job.xsl" )
files2zip.append( ( jobXslPath , "job.xsl" ) )
identXslPath = os.path.join( xsl_path , "ident.xsl" )
files2zip.append( ( identXslPath , "ident.xsl" ) )
cssPath = os.path.join( _cfg.portal_path() , "css", "mobyle.css" )
files2zip.append( ( cssPath , "mobyle.css" ) )
paramsfiles = self.jobState.getParamfiles()
if paramsfiles:
for paramsfile in paramsfiles:
files2zip.append( ( os.path.join( self._dirPath , paramsfile[0] ) , paramsfile[0] ) )
inputFiles = self.jobState.getInputFiles() #inputFiles = [ ( parameterName , [ (fileName , format or None ) , ...) , ... ]
if inputFiles is not None:
for files in inputFiles:
for item in files[1]: #item = ( filename , size , fmt )
files2zip.append( ( os.path.join( self._dirPath , item[0] ) , item[0] ) )
files2zip.append( ( os.path.join( self._dirPath , "index.xml") , "index.xml" ) )
zipFileName = "%s_%s.zip" %(self.serviceName , self.jobKey )
zip_file = zipFiles( zipFileName , files2zip )
return zip_file
if __name__ == '__main__':
try:
fh = open(".forChild.dump", "r")
fromFather = cPickle.load( fh )
fh.close()
except Exception, err :
rc_log.critical( str( err ) )
raise MobyleError( err )
try:
os.chdir( fromFather[ 'dirPath' ] )
except OSError, err:
msg = fromFather[ 'serviceName' ] + ":" + str( err )
rc_log.critical( msg )
raise MobyleError ,msg
userEmail = fromFather[ 'email']
if userEmail is not None:
userEmail = EmailAddress( userEmail )
child = AsynchronJob( fromFather['commandLine'], # string the unix command line
fromFather['dirPath'], # absolute path of the working directory
fromFather['serviceName'], # string
fromFather['resultsMask'], #
userEmail=userEmail, # Net.EmailAddress to
xmlEnv=fromFather['xmlEnv'], # a dict
email_notify=fromFather['email_notify'] # 'true' , 'false' or 'auto'
)
|
