1 from neco.execution.attribute import Attribute, Flags, Types
2 from neco.execution.trace import Trace, TraceAttr
3 from neco.execution.resource import ResourceManager, clsinit, ResourceState
4 from neco.resources.linux.node import LinuxNode
5 from neco.util import sshfuncs
10 reschedule_delay = "0.5s"
12 # TODO: Resolve wildcards in commands!!
15 class LinuxApplication(ResourceManager):
16 _rtype = "LinuxApplication"
19 def _register_attributes(cls):
20 command = Attribute("command", "Command to execute",
21 flags = Flags.ExecReadOnly)
22 forward_x11 = Attribute("forwardX11", " Enables X11 forwarding for SSH connections",
23 flags = Flags.ExecReadOnly)
24 env = Attribute("env", "Environment variables string for command execution",
25 flags = Flags.ExecReadOnly)
26 sudo = Attribute("sudo", "Run with root privileges",
27 flags = Flags.ExecReadOnly)
28 depends = Attribute("depends",
29 "Space-separated list of packages required to run the application",
30 flags = Flags.ExecReadOnly)
31 sources = Attribute("sources",
32 "Space-separated list of regular files to be deployed in the working "
33 "path prior to building. Archives won't be expanded automatically.",
34 flags = Flags.ExecReadOnly)
35 code = Attribute("code",
36 "Plain text source code to be uploaded to the server. It will be stored "
37 "under ${SOURCES}/code",
38 flags = Flags.ExecReadOnly)
39 build = Attribute("build",
40 "Build commands to execute after deploying the sources. "
41 "Sources will be in the ${SOURCES} folder. "
42 "Example: tar xzf ${SOURCES}/my-app.tgz && cd my-app && ./configure && make && make clean.\n"
43 "Try to make the commands return with a nonzero exit code on error.\n"
44 "Also, do not install any programs here, use the 'install' attribute. This will "
45 "help keep the built files constrained to the build folder (which may "
46 "not be the home folder), and will result in faster deployment. Also, "
47 "make sure to clean up temporary files, to reduce bandwidth usage between "
48 "nodes when transferring built packages.",
49 flags = Flags.ReadOnly)
50 install = Attribute("install",
51 "Commands to transfer built files to their final destinations. "
52 "Sources will be in the initial working folder, and a special "
53 "tag ${SOURCES} can be used to reference the experiment's "
54 "home folder (where the application commands will run).\n"
55 "ALL sources and targets needed for execution must be copied there, "
56 "if building has been enabled.\n"
57 "That is, 'slave' nodes will not automatically get any source files. "
58 "'slave' nodes don't get build dependencies either, so if you need "
59 "make and other tools to install, be sure to provide them as "
60 "actual dependencies instead.",
61 flags = Flags.ReadOnly)
62 stdin = Attribute("stdin", "Standard input", flags = Flags.ExecReadOnly)
63 stdout = Attribute("stdout", "Standard output", flags = Flags.ExecReadOnly)
64 stderr = Attribute("stderr", "Standard error", flags = Flags.ExecReadOnly)
65 tear_down = Attribute("tearDown", "Bash script to be executed before "
66 "releasing the resource",
67 flags = Flags.ReadOnly)
69 cls._register_attribute(command)
70 cls._register_attribute(forward_x11)
71 cls._register_attribute(env)
72 cls._register_attribute(sudo)
73 cls._register_attribute(depends)
74 cls._register_attribute(sources)
75 cls._register_attribute(code)
76 cls._register_attribute(build)
77 cls._register_attribute(install)
78 cls._register_attribute(stdin)
79 cls._register_attribute(stdout)
80 cls._register_attribute(stderr)
81 cls._register_attribute(tear_down)
84 def _register_traces(cls):
85 stdout = Trace("stdout", "Standard output stream")
86 stderr = Trace("stderr", "Standard error stream")
87 buildlog = Trace("buildlog", "Output of the build process")
89 cls._register_trace(stdout)
90 cls._register_trace(stderr)
91 cls._register_trace(buildlog)
93 def __init__(self, ec, guid):
94 super(LinuxApplication, self).__init__(ec, guid)
97 self._home = "app-%s" % self.guid
99 self._logger = logging.getLogger("LinuxApplication")
101 def log_message(self, msg):
102 return " guid %d - host %s - %s " % (self.guid,
103 self.node.get("hostname"), msg)
107 node = self.get_connected(LinuxNode.rtype())
108 if node: return node[0]
113 return os.path.join(self.node.exp_home, self._home)
117 return os.path.join(self.app_home, 'src')
121 return os.path.join(self.app_home, 'build')
131 def trace(self, name, attr = TraceAttr.ALL, block = 512, offset = 0):
132 self.info("Retrieving '%s' trace %s " % (name, attr))
134 path = os.path.join(self.app_home, name)
136 command = "(test -f %s && echo 'success') || echo 'error'" % path
137 (out, err), proc = self.node.execute(command)
139 if (err and proc.poll()) or out.find("error") != -1:
140 msg = " Couldn't find trace %s " % name
141 self.error(msg, out, err)
144 if attr == TraceAttr.PATH:
147 if attr == TraceAttr.ALL:
148 (out, err), proc = self.node.check_output(self.app_home, name)
150 if err and proc.poll():
151 msg = " Couldn't read trace %s " % name
152 self.error(msg, out, err)
157 if attr == TraceAttr.STREAM:
158 cmd = "dd if=%s bs=%d count=1 skip=%d" % (path, block, offset)
159 elif attr == TraceAttr.SIZE:
160 cmd = "stat -c%%s %s " % path
162 (out, err), proc = self.node.execute(cmd)
164 if err and proc.poll():
165 msg = " Couldn't find trace %s " % name
166 self.error(msg, out, err)
169 if attr == TraceAttr.SIZE:
170 out = int(out.strip())
174 def provision(self, filters = None):
175 # create home dir for application
176 self.node.mkdir(self.app_home)
179 self.upload_sources()
187 # install dependencies
188 self.install_dependencies()
196 command = self.get("command")
197 x11 = self.get("forwardX11")
198 if not x11 and command:
199 self.info("Uploading command '%s'" % command)
203 env = self.get("env") or ""
204 for var in env.split(" "):
205 environ += 'export %s\n' % var
207 command = environ + command
209 # If the command runs asynchronous, pre upload the command
210 # to the app.sh file in the remote host
211 dst = os.path.join(self.app_home, "app.sh")
212 command = self.replace_paths(command)
213 self.node.upload(command, dst, text = True)
215 super(LinuxApplication, self).provision()
217 def upload_sources(self):
218 # TODO: check if sources need to be uploaded and upload them
219 sources = self.get("sources")
221 self.info(" Uploading sources ")
223 # create dir for sources
224 self.node.mkdir(self.src_dir)
226 sources = sources.split(' ')
228 http_sources = list()
229 for source in list(sources):
230 if source.startswith("http") or source.startswith("https"):
231 http_sources.append(source)
232 sources.remove(source)
234 # Download http sources
236 cmd = " wget -c --directory-prefix=${SOURCES} "
239 for source in http_sources:
240 cmd += " %s " % (source)
241 verif += " ls ${SOURCES}/%s ;" % os.path.basename(source)
243 # Wget output goes to stderr :S
244 cmd += " 2> /dev/null ; "
247 cmd += " %s " % verif
249 # Upload the command to a file, and execute asynchronously
250 self.upload_and_run(cmd,
251 "http_sources.sh", "http_sources_pid",
252 "http_sources_out", "http_sources_err")
254 self.node.upload(sources, self.src_dir)
256 def upload_code(self):
257 code = self.get("code")
259 # create dir for sources
260 self.node.mkdir(self.src_dir)
262 self.info(" Uploading code ")
264 dst = os.path.join(self.src_dir, "code")
265 self.node.upload(sources, dst, text = True)
267 def upload_stdin(self):
268 stdin = self.get("stdin")
270 # create dir for sources
271 self.info(" Uploading stdin ")
273 dst = os.path.join(self.app_home, "stdin")
274 self.node.upload(stdin, dst, text = True)
276 def install_dependencies(self):
277 depends = self.get("depends")
279 self.info(" Installing dependencies %s" % depends)
280 self.node.install_packages(depends, home = self.app_home)
283 build = self.get("build")
285 self.info(" Building sources ")
287 # create dir for build
288 self.node.mkdir(self.build_dir)
290 # Upload the command to a file, and execute asynchronously
291 self.upload_and_run(build,
292 "build.sh", "build_pid",
293 "build_out", "build_err")
296 install = self.get("install")
298 self.info(" Installing sources ")
300 # Upload the command to a file, and execute asynchronously
301 self.upload_and_run(install,
302 "install.sh", "install_pid",
303 "install_out", "install_err")
306 # Wait until node is associated and deployed
308 if not node or node.state < ResourceState.READY:
309 self.debug("---- RESCHEDULING DEPLOY ---- node state %s " % self.node.state )
310 self.ec.schedule(reschedule_delay, self.deploy)
313 command = self.get("command") or ""
314 self.info(" Deploying command '%s' " % command)
318 self._state = ResourceState.FAILED
321 super(LinuxApplication, self).deploy()
324 command = self.get('command')
325 env = self.get('env')
326 stdin = 'stdin' if self.get('stdin') else None
327 stdout = 'stdout' if self.get('stdout') else 'stdout'
328 stderr = 'stderr' if self.get('stderr') else 'stderr'
329 sudo = self.get('sudo') or False
330 x11 = self.get('forwardX11') or False
333 super(LinuxApplication, self).start()
336 self.info("No command to start ")
337 self._state = ResourceState.FINISHED
340 self.info("Starting command '%s'" % command)
346 for var in env.split(" "):
347 environ += ' %s ' % var
349 command = "(" + environ + " ; " + command + ")"
350 command = self.replace_paths(command)
352 # If the command requires X11 forwarding, we
353 # can't run it asynchronously
354 (out, err), proc = self.node.execute(command,
359 self._state = ResourceState.FINISHED
361 if proc.poll() and err:
364 # Command was previously uploaded, now run the remote
365 # bash file asynchronously
366 cmd = "bash ./app.sh"
367 (out, err), proc = self.node.run(cmd, self.app_home,
373 if proc.poll() and err:
377 pid, ppid = self.node.wait_pid(home = self.app_home)
378 if pid: self._pid = int(pid)
379 if ppid: self._ppid = int(ppid)
381 if not self.pid or not self.ppid:
384 (out, chkerr), proc = self.node.check_output(self.app_home, 'stderr')
386 if failed or out or chkerr:
387 # check if execution errors occurred
388 msg = " Failed to start command '%s' " % command
395 self.error(msg, out, err)
397 msg2 = " Setting state to Failed"
399 self._state = ResourceState.FAILED
401 raise RuntimeError, msg
405 if state == ResourceState.STARTED:
406 self.info("Stopping command %s" % command)
408 (out, err), proc = self.node.kill(self.pid, self.ppid)
411 # check if execution errors occurred
412 msg = " Failed to STOP command '%s' " % self.get("command")
413 self.error(msg, out, err)
414 self._state = ResourceState.FAILED
417 super(LinuxApplication, self).stop()
420 self.info("Releasing resource")
422 tear_down = self.get("tearDown")
424 self.node.execute(tear_down)
427 if self.state == ResourceState.STOPPED:
428 super(LinuxApplication, self).release()
432 if self._state == ResourceState.STARTED:
433 (out, err), proc = self.node.check_output(self.app_home, 'stderr')
436 if err.find("No such file or directory") >= 0 :
437 # The resource is marked as started, but the
438 # command was not yet executed
439 return ResourceState.READY
441 # check if execution errors occurred
442 msg = " Failed to execute command '%s'" % self.get("command")
443 self.error(msg, out, err)
444 self._state = ResourceState.FAILED
446 elif self.pid and self.ppid:
447 status = self.node.status(self.pid, self.ppid)
449 if status == sshfuncs.FINISHED:
450 self._state = ResourceState.FINISHED
454 def upload_and_run(self, cmd, fname, pidfile, outfile, errfile):
455 dst = os.path.join(self.app_home, fname)
456 cmd = self.replace_paths(cmd)
457 self.node.upload(cmd, dst, text = True)
459 cmd = "bash ./%s" % fname
460 (out, err), proc = self.node.run_and_wait(cmd, self.app_home,
464 raise_on_error = True)
466 def replace_paths(self, command):
468 Replace all special path tags with shell-escaped actual paths.
471 return d if d.startswith("/") else os.path.join("${HOME}", d)
474 .replace("${SOURCES}", absolute_dir(self.src_dir))
475 .replace("${BUILD}", absolute_dir(self.build_dir))
476 .replace("${APP_HOME}", absolute_dir(self.app_home))
477 .replace("${NODE_HOME}", absolute_dir(self.node.node_home))
478 .replace("${EXP_HOME}", absolute_dir(self.node.exp_home) )
481 def valid_connection(self, guid):
484 # XXX: What if it is connected to more than one node?
485 resources = self.find_resources(exact_tags = [tags.NODE])
486 self._node = resources[0] if len(resources) == 1 else None